restore old mongo integration tests

src/Storages/StorageMongoDBPocoLegacy.cpp

@@ -22,6 +22,10 @@
 #include <base/range.h>
 #include <unordered_set>
 
+#include <Parsers/ASTLiteral.h>
+#include <Processors/Sinks/SinkToStorage.h>
+#include <DataTypes/DataTypeArray.h>
+
 namespace DB
 {
 
@@ -90,6 +94,131 @@ void StorageMongoDBPocoLegacy::connectIfNotConnected()
     }
 }
 
+class StorageMongoDBLegacySink : public SinkToStorage
+{
+public:
+    explicit StorageMongoDBLegacySink(
+        const std::string & collection_name_,
+        const std::string & db_name_,
+        const StorageMetadataPtr & metadata_snapshot_,
+        std::shared_ptr<Poco::MongoDB::Connection> connection_)
+        : SinkToStorage(metadata_snapshot_->getSampleBlock())
+        , collection_name(collection_name_)
+        , db_name(db_name_)
+        , metadata_snapshot{metadata_snapshot_}
+        , connection(connection_)
+        , is_wire_protocol_old(isMongoDBWireProtocolOld(*connection_, db_name))
+    {
+    }
+
+    String getName() const override { return "StorageMongoDBLegacySink"; }
+
+    void consume(Chunk & chunk) override
+    {
+        Poco::MongoDB::Database db(db_name);
+        Poco::MongoDB::Document::Vector documents;
+
+        auto block = getHeader().cloneWithColumns(chunk.getColumns());
+
+        size_t num_rows = block.rows();
+        size_t num_cols = block.columns();
+
+        const auto columns = block.getColumns();
+        const auto data_types = block.getDataTypes();
+        const auto data_names = block.getNames();
+
+        documents.reserve(num_rows);
+
+        for (const auto i : collections::range(0, num_rows))
+        {
+            Poco::MongoDB::Document::Ptr document = new Poco::MongoDB::Document();
+
+            for (const auto j : collections::range(0, num_cols))
+            {
+                insertValueIntoMongoDB(*document, data_names[j], *data_types[j], *columns[j], i);
+            }
+
+            documents.push_back(std::move(document));
+        }
+
+        if (is_wire_protocol_old)
+        {
+            Poco::SharedPtr<Poco::MongoDB::InsertRequest> insert_request = db.createInsertRequest(collection_name);
+            insert_request->documents() = std::move(documents);
+            connection->sendRequest(*insert_request);
+        }
+        else
+        {
+            Poco::SharedPtr<Poco::MongoDB::OpMsgMessage> insert_request = db.createOpMsgMessage(collection_name);
+            insert_request->setCommandName(Poco::MongoDB::OpMsgMessage::CMD_INSERT);
+            insert_request->documents() = std::move(documents);
+            connection->sendRequest(*insert_request);
+        }
+    }
+
+private:
+
+    void insertValueIntoMongoDB(
+        Poco::MongoDB::Document & document,
+        const std::string & name,
+        const IDataType & data_type,
+        const IColumn & column,
+        size_t idx)
+    {
+        WhichDataType which(data_type);
+
+        if (which.isArray())
+        {
+            const ColumnArray & column_array = assert_cast<const ColumnArray &>(column);
+            const ColumnArray::Offsets & offsets = column_array.getOffsets();
+
+            size_t offset = offsets[idx - 1];
+            size_t next_offset = offsets[idx];
+
+            const IColumn & nested_column = column_array.getData();
+
+            const auto * array_type = assert_cast<const DataTypeArray *>(&data_type);
+            const DataTypePtr & nested_type = array_type->getNestedType();
+
+            Poco::MongoDB::Array::Ptr array = new Poco::MongoDB::Array();
+            for (size_t i = 0; i + offset < next_offset; ++i)
+            {
+                insertValueIntoMongoDB(*array, Poco::NumberFormatter::format(i), *nested_type, nested_column, i + offset);
+            }
+
+            document.add(name, array);
+            return;
+        }
+
+        /// MongoDB does not support UInt64 type, so just cast it to Int64
+        if (which.isNativeUInt())
+            document.add(name, static_cast<Poco::Int64>(column.getUInt(idx)));
+        else if (which.isNativeInt())
+            document.add(name, static_cast<Poco::Int64>(column.getInt(idx)));
+        else if (which.isFloat32())
+            document.add(name, static_cast<Float64>(column.getFloat32(idx)));
+        else if (which.isFloat64())
+            document.add(name, column.getFloat64(idx));
+        else if (which.isDate())
+            document.add(name, Poco::Timestamp(DateLUT::instance().fromDayNum(DayNum(column.getUInt(idx))) * 1000000));
+        else if (which.isDateTime())
+            document.add(name, Poco::Timestamp(column.getUInt(idx) * 1000000));
+        else
+        {
+            WriteBufferFromOwnString ostr;
+            data_type.getDefaultSerialization()->serializeText(column, idx, ostr, FormatSettings{});
+            document.add(name, ostr.str());
+        }
+    }
+
+    String collection_name;
+    String db_name;
+    StorageMetadataPtr metadata_snapshot;
+    std::shared_ptr<Poco::MongoDB::Connection> connection;
+
+    const bool is_wire_protocol_old;
+};
+
 Pipe StorageMongoDBPocoLegacy::read(
     const Names & column_names,
     const StorageSnapshotPtr & storage_snapshot,
@@ -113,6 +242,13 @@ Pipe StorageMongoDBPocoLegacy::read(
     return Pipe(std::make_shared<MongoDBPocoLegacySource>(connection, database_name, collection_name, Poco::MongoDB::Document{}, sample_block, max_block_size));
 }
 
+
+SinkToStoragePtr StorageMongoDBPocoLegacy::write(const ASTPtr & /* query */, const StorageMetadataPtr & metadata_snapshot, ContextPtr /* context */, bool /*async_insert*/)
+{
+    connectIfNotConnected();
+    return std::make_shared<StorageMongoDBLegacySink>(collection_name, database_name, metadata_snapshot, connection);
+}
+
 StorageMongoDBPocoLegacy::Configuration StorageMongoDBPocoLegacy::getConfiguration(ASTs engine_args, ContextPtr context)
 {
     Configuration configuration;

src/Storages/StorageMongoDBPocoLegacy.h

@@ -42,6 +42,12 @@ public:
         size_t max_block_size,
         size_t num_streams) override;
 
+    SinkToStoragePtr write(
+        const ASTPtr & query,
+        const StorageMetadataPtr & /*metadata_snapshot*/,
+        ContextPtr context,
+        bool async_insert) override;
+
     struct Configuration
     {
         std::string host;

tests/integration/helpers/cluster.py

@@ -2471,7 +2471,7 @@ class ClickHouseCluster:
         while time.time() - start < timeout:
             try:
                 connection.list_database_names()
-                logging.debug(f"Connected to Mongo dbs: {connection.database_names()}")
+                logging.debug(f"Connected to Mongo dbs: {connection.list_database_names()}")
                 return
             except Exception as ex:
                 logging.debug("Can't connect to Mongo " + str(ex))
@@ -4451,7 +4451,7 @@ class ClickHouseInstance:
                 dirs_exist_ok=True,
             )
 
-        if self.with_mongo:
+        if self.with_mongo and os.path.exists(self.mongo_secure_config_dir):
             shutil.copytree(
                 self.mongo_secure_config_dir,
                 p.abspath(p.join(self.path, "mongo_secure_config")),

tests/integration/test_storage_mongodb/configs/feature_flag.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <use_legacy_mongodb_integration>1</use_legacy_mongodb_integration>
+</clickhouse>

tests/integration/test_storage_mongodb/configs/named_collections.xml

@@ -8,9 +8,5 @@
             <database>test</database>
             <collection>simple_table</collection>
         </mongo1>
-        <mongo1_uri>
-            <uri>mongodb://root:clickhouse@mongo1:27017/test</uri>
-            <collection>simple_table_uri</collection>
-        </mongo1_uri>
     </named_collections>
 </clickhouse>

tests/integration/test_storage_mongodb/mongo_secure_config/config.d/ssl_conf.xml

@@ -0,0 +1,8 @@
+<clickhouse>
+    <openSSL>
+        <client>
+            <!-- For self-signed certificate -->
+            <verificationMode>none</verificationMode>
+        </client>
+    </openSSL>
+</clickhouse>

tests/integration/test_storage_mongodb/mongo_secure_config/mongo_cert.pem

@@ -0,0 +1,52 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCUnkHzBu8XdPpD
+3oWJRLBMNEptQwdOUl1QhMyF9wxQ+lya3fHnlsJc2GgWrZHMI/vMr6rJHtNl/92M
+KcychYswKiSKCqpeXOWIotsR9UvTxT9Mvi6Ca3vnkM4h8jt97XnnnXhjSdOUptmQ
+M43zEO0XYekvoHSDuGNdNokEecq9mk64e7bu15DQo8Ck6jDL80opD9UX7diehAop
+vjI2AgrR/+MlxnhCeXqVOLd8ukNtwdmRIWPrKlHcICvabvKHuJpGdUL8GpHPUINd
+qMEUg4wQWKz74IV1aZCdiZICk6yACM/bWed3g1RzV21jAKLaL4xJkUqBiEtcs8Jk
+PT3cVMGHAgMBAAECggEAAul6qiHchB+uQMCWyC5xTeRqAXR3tAv4Tj4fGJjkXY4Z
+OrAjr9Kp38EvX1amgvUWV3FT3NMevDf5xd9OdzAA0g0uJIF+mAhYFW48i1FnQcHQ
+mOf0zmiZR7l8o7ROb3JvooXHxW+ba/qjGPVwC801gJvruehgbOCRxh9DTRp7sH5K
+BmcddhULhKBEQjWUmYNEM3A2axpdi3g1aYKERRLn8J0DXcItTwbxuxbNcs3erl8W
+3yyv/JKmqnWF5sNyX3wEWuQcDEZZy+W7Hn4KPMxyU+WA5el5nJ8kFlxhpInmajwu
+8Ytn6IEyThyXutVomosVBuP16QORl2Nad0hnQO9toQKBgQDDgiehXr3k2wfVaVOD
+PocW4leXausIU2XcCn6FxTG9vLUDMPANw0MxgenC2nrjaUU9J9UjdRYgMcFGWrl4
+E27wEn5e0nZ/Y7F2cfhuOc9vNmZ+eHm2KQRyfAjIVL5Hpldqk2jXyCnLBNeWGHSw
+kPQMU+FLqmrOFUvXlD2my+OSHwKBgQDCmgS9r+xFh4BCB9dY6eyQJF/jYmAQHs26
+80WJ6gAhbUw1O71uDtS9/3PZVXwwNCOHrcc49BPrpJdxGPHGvd2Q5y+j5LDDbQSZ
+aLTiCZ2B0RM5Bd2dXD8gEHN4WCX7pJ/o4kDi4zONBmp5mg/tFfer5z5IU/1P7Wak
+1Mu0JIHzmQKBgDNaNoqeVgaMuYwGtFbez6DlJtiwzrdLIJAheYYte5k4vdruub8D
+sNyKIRp7RJgDCJq9obBEiuE98GRIZDrz78nDMco6QcHIL87KtNRO/vtZMKa7gkyk
+jXR8u9nS2H/9YyytN3amLsQSq4XTOqM+D7xFNAIp6w/ibB9d4quzFj1FAoGBAKTE
+x/LcO897NWuzO/D6z+QUCGR87R15F3SNenmVedrTskz4ciH3yMW+v5ZrPSWLX/IH
+f8GHWD6TM+780eoW5L1GIh5BCjHN4rEJ6O3iekxqfD4x6zzL2F8Lztk8uZxh/Uuw
+FoSFHybvIcQoYAe8K+KPfzq6cqb0OY6i5n920dkxAoGAJkw6ADqsJfH3NR+bQfgF
+oEA1KqriMxyEJm44Y7E80C+iF4iNALF+Er9TSnr4mDxX5e/dW9d1YeS9o0nOfkpF
+MaBmJfxqo4QQJLPRaxYQ2Jhfn7irir4BroxeNXQgNNhgSuKIvkfRyGYwl7P0AT4v
+8H8rkZGneMD3gLB5MfnRhGk=
+-----END PRIVATE KEY-----
+-----BEGIN CERTIFICATE-----
+MIIEGzCCAwOgAwIBAgIUaoGlyuJAyvs6yowFXymfu7seEiUwDQYJKoZIhvcNAQEL
+BQAwgZwxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDbGlja0hvdXNlMRMwEQYDVQQH
+DApDbGlja0hvdXNlMREwDwYDVQQKDAhQZXJzb25hbDETMBEGA1UECwwKQ2xpY2tI
+b3VzZTEkMCIGCSqGSIb3DQEJARYVY2xpY2tob3VzZUBjbGlja2hvdXNlMRUwEwYD
+VQQDDAxtb25nb19zZWN1cmUwHhcNMjQwNTI2MTYwMDMxWhcNMzQwNTI0MTYwMDMx
+WjCBnDELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNsaWNrSG91c2UxEzARBgNVBAcM
+CkNsaWNrSG91c2UxETAPBgNVBAoMCFBlcnNvbmFsMRMwEQYDVQQLDApDbGlja0hv
+dXNlMSQwIgYJKoZIhvcNAQkBFhVjbGlja2hvdXNlQGNsaWNraG91c2UxFTATBgNV
+BAMMDG1vbmdvX3NlY3VyZTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+AJSeQfMG7xd0+kPehYlEsEw0Sm1DB05SXVCEzIX3DFD6XJrd8eeWwlzYaBatkcwj
++8yvqske02X/3YwpzJyFizAqJIoKql5c5Yii2xH1S9PFP0y+LoJre+eQziHyO33t
+eeedeGNJ05Sm2ZAzjfMQ7Rdh6S+gdIO4Y102iQR5yr2aTrh7tu7XkNCjwKTqMMvz
+SikP1Rft2J6ECim+MjYCCtH/4yXGeEJ5epU4t3y6Q23B2ZEhY+sqUdwgK9pu8oe4
+mkZ1Qvwakc9Qg12owRSDjBBYrPvghXVpkJ2JkgKTrIAIz9tZ53eDVHNXbWMAotov
+jEmRSoGIS1yzwmQ9PdxUwYcCAwEAAaNTMFEwHQYDVR0OBBYEFJyz3Kt5XBDg5cvI
+0v1ioqejqX+CMB8GA1UdIwQYMBaAFJyz3Kt5XBDg5cvI0v1ioqejqX+CMA8GA1Ud
+EwEB/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAHAQFA5VMYvaQFnKtKfHg9TF
+qfJ4uM3YsGdgsgmGWgflD1S4Z290H6Q2QvyZAEceTrlJxqArlWlVp5DAU6EeXjEh
+QMAgdkJHF1Hg2jsZKPtdkb88UtuzwAME357T8NtEJSHzNE5QqYwlVM71JkWpdqvA
+UUdOJbWhhJfowIf4tMmL1DUuIy2qYpoP/tEBXEw9uwpmZqb7KELwT3lRyOMaGFN7
+RHVwbvJWlHiu83QDNaWz6ijQkWl3tCN6TWcFD1qc1x8GpMzjbsAAYbCx7fbHM2LD
+9kGSCiyv5K0MLNK5u67RtUFfPHtyD8RA0TtxIZ4PEN/eFANKS2/5NEi1ZuZ5/Pk=
+-----END CERTIFICATE-----

tests/integration/test_storage_mongodb_new/configs/feature_flag.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <use_legacy_mongodb_integration>0</use_legacy_mongodb_integration>
+</clickhouse>

tests/integration/test_storage_mongodb_new/configs/named_collections.xml

@@ -0,0 +1,17 @@
+<clickhouse>
+    <named_collections>
+        <mongo1>
+            <user>root</user>
+            <password>clickhouse</password>
+            <host>mongo1</host>
+            <port>27017</port>
+            <database>test</database>
+            <collection>simple_table</collection>
+        </mongo1>
+        <mongo1_uri>
+            <uri>mongodb://root:clickhouse@mongo1:27017/test</uri>
+            <collection>simple_table_uri</collection>
+        </mongo1_uri>
+    </named_collections>
+
+</clickhouse>

tests/integration/test_storage_mongodb_new/configs/users.xml

@@ -0,0 +1,9 @@
+<clickhouse>
+    <users>
+        <default>
+            <password></password>
+            <profile>default</profile>
+            <named_collection_control>1</named_collection_control>
+        </default>
+    </users>
+</clickhouse>

tests/integration/test_storage_mongodb_new/mongo_secure_config/cert.crt

@@ -0,0 +1,24 @@
+-----BEGIN CERTIFICATE-----
+MIIEGzCCAwOgAwIBAgIUaoGlyuJAyvs6yowFXymfu7seEiUwDQYJKoZIhvcNAQEL
+BQAwgZwxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDbGlja0hvdXNlMRMwEQYDVQQH
+DApDbGlja0hvdXNlMREwDwYDVQQKDAhQZXJzb25hbDETMBEGA1UECwwKQ2xpY2tI
+b3VzZTEkMCIGCSqGSIb3DQEJARYVY2xpY2tob3VzZUBjbGlja2hvdXNlMRUwEwYD
+VQQDDAxtb25nb19zZWN1cmUwHhcNMjQwNTI2MTYwMDMxWhcNMzQwNTI0MTYwMDMx
+WjCBnDELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNsaWNrSG91c2UxEzARBgNVBAcM
+CkNsaWNrSG91c2UxETAPBgNVBAoMCFBlcnNvbmFsMRMwEQYDVQQLDApDbGlja0hv
+dXNlMSQwIgYJKoZIhvcNAQkBFhVjbGlja2hvdXNlQGNsaWNraG91c2UxFTATBgNV
+BAMMDG1vbmdvX3NlY3VyZTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+AJSeQfMG7xd0+kPehYlEsEw0Sm1DB05SXVCEzIX3DFD6XJrd8eeWwlzYaBatkcwj
++8yvqske02X/3YwpzJyFizAqJIoKql5c5Yii2xH1S9PFP0y+LoJre+eQziHyO33t
+eeedeGNJ05Sm2ZAzjfMQ7Rdh6S+gdIO4Y102iQR5yr2aTrh7tu7XkNCjwKTqMMvz
+SikP1Rft2J6ECim+MjYCCtH/4yXGeEJ5epU4t3y6Q23B2ZEhY+sqUdwgK9pu8oe4
+mkZ1Qvwakc9Qg12owRSDjBBYrPvghXVpkJ2JkgKTrIAIz9tZ53eDVHNXbWMAotov
+jEmRSoGIS1yzwmQ9PdxUwYcCAwEAAaNTMFEwHQYDVR0OBBYEFJyz3Kt5XBDg5cvI
+0v1ioqejqX+CMB8GA1UdIwQYMBaAFJyz3Kt5XBDg5cvI0v1ioqejqX+CMA8GA1Ud
+EwEB/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAHAQFA5VMYvaQFnKtKfHg9TF
+qfJ4uM3YsGdgsgmGWgflD1S4Z290H6Q2QvyZAEceTrlJxqArlWlVp5DAU6EeXjEh
+QMAgdkJHF1Hg2jsZKPtdkb88UtuzwAME357T8NtEJSHzNE5QqYwlVM71JkWpdqvA
+UUdOJbWhhJfowIf4tMmL1DUuIy2qYpoP/tEBXEw9uwpmZqb7KELwT3lRyOMaGFN7
+RHVwbvJWlHiu83QDNaWz6ijQkWl3tCN6TWcFD1qc1x8GpMzjbsAAYbCx7fbHM2LD
+9kGSCiyv5K0MLNK5u67RtUFfPHtyD8RA0TtxIZ4PEN/eFANKS2/5NEi1ZuZ5/Pk=
+-----END CERTIFICATE-----

tests/integration/test_storage_mongodb_new/mongo_secure_config/key.pem

@@ -0,0 +1,52 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCUnkHzBu8XdPpD
+3oWJRLBMNEptQwdOUl1QhMyF9wxQ+lya3fHnlsJc2GgWrZHMI/vMr6rJHtNl/92M
+KcychYswKiSKCqpeXOWIotsR9UvTxT9Mvi6Ca3vnkM4h8jt97XnnnXhjSdOUptmQ
+M43zEO0XYekvoHSDuGNdNokEecq9mk64e7bu15DQo8Ck6jDL80opD9UX7diehAop
+vjI2AgrR/+MlxnhCeXqVOLd8ukNtwdmRIWPrKlHcICvabvKHuJpGdUL8GpHPUINd
+qMEUg4wQWKz74IV1aZCdiZICk6yACM/bWed3g1RzV21jAKLaL4xJkUqBiEtcs8Jk
+PT3cVMGHAgMBAAECggEAAul6qiHchB+uQMCWyC5xTeRqAXR3tAv4Tj4fGJjkXY4Z
+OrAjr9Kp38EvX1amgvUWV3FT3NMevDf5xd9OdzAA0g0uJIF+mAhYFW48i1FnQcHQ
+mOf0zmiZR7l8o7ROb3JvooXHxW+ba/qjGPVwC801gJvruehgbOCRxh9DTRp7sH5K
+BmcddhULhKBEQjWUmYNEM3A2axpdi3g1aYKERRLn8J0DXcItTwbxuxbNcs3erl8W
+3yyv/JKmqnWF5sNyX3wEWuQcDEZZy+W7Hn4KPMxyU+WA5el5nJ8kFlxhpInmajwu
+8Ytn6IEyThyXutVomosVBuP16QORl2Nad0hnQO9toQKBgQDDgiehXr3k2wfVaVOD
+PocW4leXausIU2XcCn6FxTG9vLUDMPANw0MxgenC2nrjaUU9J9UjdRYgMcFGWrl4
+E27wEn5e0nZ/Y7F2cfhuOc9vNmZ+eHm2KQRyfAjIVL5Hpldqk2jXyCnLBNeWGHSw
+kPQMU+FLqmrOFUvXlD2my+OSHwKBgQDCmgS9r+xFh4BCB9dY6eyQJF/jYmAQHs26
+80WJ6gAhbUw1O71uDtS9/3PZVXwwNCOHrcc49BPrpJdxGPHGvd2Q5y+j5LDDbQSZ
+aLTiCZ2B0RM5Bd2dXD8gEHN4WCX7pJ/o4kDi4zONBmp5mg/tFfer5z5IU/1P7Wak
+1Mu0JIHzmQKBgDNaNoqeVgaMuYwGtFbez6DlJtiwzrdLIJAheYYte5k4vdruub8D
+sNyKIRp7RJgDCJq9obBEiuE98GRIZDrz78nDMco6QcHIL87KtNRO/vtZMKa7gkyk
+jXR8u9nS2H/9YyytN3amLsQSq4XTOqM+D7xFNAIp6w/ibB9d4quzFj1FAoGBAKTE
+x/LcO897NWuzO/D6z+QUCGR87R15F3SNenmVedrTskz4ciH3yMW+v5ZrPSWLX/IH
+f8GHWD6TM+780eoW5L1GIh5BCjHN4rEJ6O3iekxqfD4x6zzL2F8Lztk8uZxh/Uuw
+FoSFHybvIcQoYAe8K+KPfzq6cqb0OY6i5n920dkxAoGAJkw6ADqsJfH3NR+bQfgF
+oEA1KqriMxyEJm44Y7E80C+iF4iNALF+Er9TSnr4mDxX5e/dW9d1YeS9o0nOfkpF
+MaBmJfxqo4QQJLPRaxYQ2Jhfn7irir4BroxeNXQgNNhgSuKIvkfRyGYwl7P0AT4v
+8H8rkZGneMD3gLB5MfnRhGk=
+-----END PRIVATE KEY-----
+-----BEGIN CERTIFICATE-----
+MIIEGzCCAwOgAwIBAgIUaoGlyuJAyvs6yowFXymfu7seEiUwDQYJKoZIhvcNAQEL
+BQAwgZwxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDbGlja0hvdXNlMRMwEQYDVQQH
+DApDbGlja0hvdXNlMREwDwYDVQQKDAhQZXJzb25hbDETMBEGA1UECwwKQ2xpY2tI
+b3VzZTEkMCIGCSqGSIb3DQEJARYVY2xpY2tob3VzZUBjbGlja2hvdXNlMRUwEwYD
+VQQDDAxtb25nb19zZWN1cmUwHhcNMjQwNTI2MTYwMDMxWhcNMzQwNTI0MTYwMDMx
+WjCBnDELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNsaWNrSG91c2UxEzARBgNVBAcM
+CkNsaWNrSG91c2UxETAPBgNVBAoMCFBlcnNvbmFsMRMwEQYDVQQLDApDbGlja0hv
+dXNlMSQwIgYJKoZIhvcNAQkBFhVjbGlja2hvdXNlQGNsaWNraG91c2UxFTATBgNV
+BAMMDG1vbmdvX3NlY3VyZTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+AJSeQfMG7xd0+kPehYlEsEw0Sm1DB05SXVCEzIX3DFD6XJrd8eeWwlzYaBatkcwj
++8yvqske02X/3YwpzJyFizAqJIoKql5c5Yii2xH1S9PFP0y+LoJre+eQziHyO33t
+eeedeGNJ05Sm2ZAzjfMQ7Rdh6S+gdIO4Y102iQR5yr2aTrh7tu7XkNCjwKTqMMvz
+SikP1Rft2J6ECim+MjYCCtH/4yXGeEJ5epU4t3y6Q23B2ZEhY+sqUdwgK9pu8oe4
+mkZ1Qvwakc9Qg12owRSDjBBYrPvghXVpkJ2JkgKTrIAIz9tZ53eDVHNXbWMAotov
+jEmRSoGIS1yzwmQ9PdxUwYcCAwEAAaNTMFEwHQYDVR0OBBYEFJyz3Kt5XBDg5cvI
+0v1ioqejqX+CMB8GA1UdIwQYMBaAFJyz3Kt5XBDg5cvI0v1ioqejqX+CMA8GA1Ud
+EwEB/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAHAQFA5VMYvaQFnKtKfHg9TF
+qfJ4uM3YsGdgsgmGWgflD1S4Z290H6Q2QvyZAEceTrlJxqArlWlVp5DAU6EeXjEh
+QMAgdkJHF1Hg2jsZKPtdkb88UtuzwAME357T8NtEJSHzNE5QqYwlVM71JkWpdqvA
+UUdOJbWhhJfowIf4tMmL1DUuIy2qYpoP/tEBXEw9uwpmZqb7KELwT3lRyOMaGFN7
+RHVwbvJWlHiu83QDNaWz6ijQkWl3tCN6TWcFD1qc1x8GpMzjbsAAYbCx7fbHM2LD
+9kGSCiyv5K0MLNK5u67RtUFfPHtyD8RA0TtxIZ4PEN/eFANKS2/5NEi1ZuZ5/Pk=
+-----END CERTIFICATE-----

tests/integration/test_storage_mongodb_new/mongo_secure_config/mongo_secure.conf

@@ -0,0 +1,6 @@
+net:
+  ssl:
+    mode: requireSSL
+    PEMKeyFile: /mongo/key.pem
+    CAFile: /mongo/cert.crt
+    allowConnectionsWithoutCertificates: true

tests/integration/test_table_function_mongodb/configs/feature_flag.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <use_legacy_mongodb_integration>1</use_legacy_mongodb_integration>
+</clickhouse>

tests/integration/test_table_function_mongodb/mongo_secure_config/config.d/ssl_conf.xml

@@ -0,0 +1,8 @@
+<clickhouse>
+    <openSSL>
+        <client>
+            <!-- For self-signed certificate -->
+            <verificationMode>none</verificationMode>
+        </client>
+    </openSSL>
+</clickhouse>

tests/integration/test_table_function_mongodb/test.py

@@ -10,11 +10,12 @@ from helpers.cluster import ClickHouseCluster
 def started_cluster(request):
     try:
         cluster = ClickHouseCluster(__file__)
-        cluster.add_instance(
+        node = cluster.add_instance(
             "node",
             with_mongo=True,
             main_configs=[
-                "configs/named_collections.xml",
+                "mongo_secure_config/config.d/ssl_conf.xml",
+                "configs/feature_flag.xml",
             ],
             user_configs=["configs/users.xml"],
         )
@@ -25,33 +26,34 @@ def started_cluster(request):
 
 
 def get_mongo_connection(started_cluster, secure=False, with_credentials=True):
-    if secure:
+    connection_str = ""
-        return pymongo.MongoClient(
-            "mongodb://root:clickhouse@localhost:{}/?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true".format(
-                started_cluster.mongo_secure_port
-            )
-        )
     if with_credentials:
-        return pymongo.MongoClient(
+        connection_str = "mongodb://root:clickhouse@localhost:{}".format(
-            "mongodb://root:clickhouse@localhost:{}".format(started_cluster.mongo_port)
+            started_cluster.mongo_secure_port if secure else started_cluster.mongo_port
         )
-
+    else:
-    return pymongo.MongoClient(
+        connection_str = "mongodb://localhost:{}".format(
-        "mongodb://localhost:{}".format(started_cluster.mongo_no_cred_port)
+            started_cluster.mongo_no_cred_port
-    )
+        )
+    if secure:
+        connection_str += "/?tls=true&tlsAllowInvalidCertificates=true"
+    return pymongo.MongoClient(connection_str)
 
 
+@pytest.mark.parametrize("started_cluster", [False], indirect=["started_cluster"])
 def test_simple_select(started_cluster):
     mongo_connection = get_mongo_connection(started_cluster)
     db = mongo_connection["test"]
     db.add_user("root", "clickhouse")
     simple_mongo_table = db["simple_table"]
-    data = []
-    for i in range(0, 100):
-        data.append({"key": i, "data": hex(i * i)})
-    simple_mongo_table.insert_many(data)
 
     node = started_cluster.instances["node"]
+    for i in range(0, 100):
+        node.query(
+            "INSERT INTO FUNCTION mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String') (key, data) VALUES ({}, '{}')".format(
+                i, hex(i * i)
+            )
+        )
     assert (
         node.query(
             "SELECT COUNT() FROM mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String')"
@@ -73,52 +75,14 @@ def test_simple_select(started_cluster):
 
     assert (
         node.query(
-            "SELECT data FROM mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String') WHERE key = 42"
+            "SELECT data from mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String') where key = 42"
-        )
-        == hex(42 * 42) + "\n"
-    )
-    simple_mongo_table.drop()
-
-
-def test_simple_select_uri(started_cluster):
-    mongo_connection = get_mongo_connection(started_cluster)
-    db = mongo_connection["test"]
-    db.add_user("root", "clickhouse")
-    simple_mongo_table = db["simple_table"]
-    data = []
-    for i in range(0, 100):
-        data.append({"key": i, "data": hex(i * i)})
-    simple_mongo_table.insert_many(data)
-
-    node = started_cluster.instances["node"]
-    assert (
-        node.query(
-            "SELECT COUNT() FROM mongodb('mongodb://root:clickhouse@mongo1:27017/test', 'simple_table', structure='key UInt64, data String')"
-        )
-        == "100\n"
-    )
-    assert (
-        node.query(
-            "SELECT sum(key) FROM mongodb('mongodb://root:clickhouse@mongo1:27017/test', 'simple_table', structure='key UInt64, data String')"
-        )
-        == str(sum(range(0, 100))) + "\n"
-    )
-    assert (
-        node.query(
-            "SELECT sum(key) FROM mongodb('mongodb://root:clickhouse@mongo1:27017/test', 'simple_table', 'key UInt64, data String')"
-        )
-        == str(sum(range(0, 100))) + "\n"
-    )
-
-    assert (
-        node.query(
-            "SELECT data FROM mongodb('mongodb://root:clickhouse@mongo1:27017/test', 'simple_table', structure='key UInt64, data String') WHERE key = 42"
         )
         == hex(42 * 42) + "\n"
     )
     simple_mongo_table.drop()
 
 
+@pytest.mark.parametrize("started_cluster", [False], indirect=["started_cluster"])
 def test_complex_data_type(started_cluster):
     mongo_connection = get_mongo_connection(started_cluster)
     db = mongo_connection["test"]
@@ -133,49 +97,27 @@ def test_complex_data_type(started_cluster):
 
     assert (
         node.query(
-            """
+            "SELECT COUNT() FROM mongodb('mongo1:27017', 'test', 'complex_table', 'root', 'clickhouse', structure='key UInt64, data String, dict Map(UInt64, String)')"
-            SELECT COUNT()
-            FROM mongodb('mongo1:27017',
-                         'test',
-                         'complex_table',
-                         'root',
-                         'clickhouse',
-                         structure='key UInt64, data String, dict Map(UInt64, String)')"""
         )
         == "100\n"
     )
     assert (
         node.query(
-            """
+            "SELECT sum(key) FROM mongodb('mongo1:27017', 'test', 'complex_table', 'root', 'clickhouse', structure='key UInt64, data String, dict Map(UInt64, String)')"
-            SELECT sum(key)
-            FROM mongodb('mongo1:27017',
-                         'test',
-                         'complex_table',
-                         'root',
-                         'clickhouse',
-                         structure='key UInt64, data String, dict Map(UInt64, String)')"""
         )
         == str(sum(range(0, 100))) + "\n"
     )
 
     assert (
         node.query(
-            """
+            "SELECT data from mongodb('mongo1:27017', 'test', 'complex_table', 'root', 'clickhouse', structure='key UInt64, data String, dict Map(UInt64, String)') where key = 42"
-            SELECT data
-            FROM mongodb('mongo1:27017',
-                         'test',
-                         'complex_table',
-                         'root',
-                         'clickhouse',
-                         structure='key UInt64, data String, dict Map(UInt64, String)')
-            WHERE key = 42
-            """
         )
         == hex(42 * 42) + "\n"
     )
     incomplete_mongo_table.drop()
 
 
+@pytest.mark.parametrize("started_cluster", [False], indirect=["started_cluster"])
 def test_incorrect_data_type(started_cluster):
     mongo_connection = get_mongo_connection(started_cluster)
     db = mongo_connection["test"]
@@ -196,6 +138,7 @@ def test_incorrect_data_type(started_cluster):
     strange_mongo_table.drop()
 
 
+@pytest.mark.parametrize("started_cluster", [True], indirect=["started_cluster"])
 def test_secure_connection(started_cluster):
     mongo_connection = get_mongo_connection(started_cluster, secure=True)
     db = mongo_connection["test"]
@@ -210,63 +153,35 @@ def test_secure_connection(started_cluster):
 
     assert (
         node.query(
-            """SELECT COUNT()
+            "SELECT COUNT() FROM mongodb('mongo_secure:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String', options='ssl=true')"
-               FROM mongodb('mongo_secure:27017',
-                            'test',
-                            'simple_table',
-                            'root',
-                            'clickhouse',
-                            structure='key UInt64, data String',
-                            options='tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true')"""
         )
         == "100\n"
     )
     assert (
         node.query(
-            """SELECT sum(key)
+            "SELECT sum(key) FROM mongodb('mongo_secure:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String', options='ssl=true')"
-               FROM mongodb('mongo_secure:27017',
-                            'test',
-                            'simple_table',
-                            'root',
-                            'clickhouse',
-                            structure='key UInt64, data String',
-                            options='tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true')"""
         )
         == str(sum(range(0, 100))) + "\n"
     )
     assert (
         node.query(
-            """SELECT sum(key)
+            "SELECT sum(key) FROM mongodb('mongo_secure:27017', 'test', 'simple_table', 'root', 'clickhouse', 'key UInt64, data String', 'ssl=true')"
-               FROM mongodb('mongo_secure:27017',
-                            'test',
-                            'simple_table',
-                            'root',
-                            'clickhouse',
-                            'key UInt64, data String',
-                            'tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true')"""
         )
         == str(sum(range(0, 100))) + "\n"
     )
 
     assert (
         node.query(
-            """SELECT data
+            "SELECT data from mongodb('mongo_secure:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String', options='ssl=true') where key = 42"
-               FROM mongodb('mongo_secure:27017',
-                            'test',
-                            'simple_table',
-                            'root',
-                            'clickhouse',
-                            'key UInt64, data String',
-                            'tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true')
-               WHERE key = 42"""
         )
         == hex(42 * 42) + "\n"
     )
     simple_mongo_table.drop()
 
 
-def test_secure_connection_with_validation(started_cluster):
+@pytest.mark.parametrize("started_cluster", [False], indirect=["started_cluster"])
-    mongo_connection = get_mongo_connection(started_cluster, secure=True)
+def test_predefined_connection_configuration(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster)
     db = mongo_connection["test"]
     db.add_user("root", "clickhouse")
     simple_mongo_table = db["simple_table"]
@@ -276,73 +191,16 @@ def test_secure_connection_with_validation(started_cluster):
     simple_mongo_table.insert_many(data)
 
     node = started_cluster.instances["node"]
-    with pytest.raises(QueryRuntimeException):
-        node.query(
-            """SELECT COUNT() FROM mongodb('mongo_secure:27017',
-                   'test',
-                   'simple_table',
-                   'root',
-                   'clickhouse',
-                   structure='key UInt64, data String',
-                   options='tls=true')"""
-        )
-
-    simple_mongo_table.drop()
-
-
-def test_secure_connection_uri(started_cluster):
-    mongo_connection = get_mongo_connection(started_cluster, secure=True)
-    db = mongo_connection["test"]
-    db.add_user("root", "clickhouse")
-    simple_mongo_table = db["simple_table"]
-    data = []
-    for i in range(0, 100):
-        data.append({"key": i, "data": hex(i * i)})
-    simple_mongo_table.insert_many(data)
-
-    node = started_cluster.instances["node"]
-
     assert (
         node.query(
-            """SELECT COUNT()
+            "SELECT count() FROM mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String')"
-               FROM mongodb('mongodb://root:clickhouse@mongo_secure:27017/test?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true',
-                            'simple_table',
-                            'key UInt64, data String')"""
         )
         == "100\n"
     )
-    assert (
-        node.query(
-            """SELECT sum(key)
-               FROM mongodb('mongodb://root:clickhouse@mongo_secure:27017/test?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true',
-                            'simple_table',
-                            'key UInt64, data String')"""
-        )
-        == str(sum(range(0, 100))) + "\n"
-    )
-    assert (
-        node.query(
-            """SELECT sum(key)
-               FROM mongodb('mongodb://root:clickhouse@mongo_secure:27017/test?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true',
-                            'simple_table',
-                            'key UInt64, data String')"""
-        )
-        == str(sum(range(0, 100))) + "\n"
-    )
-
-    assert (
-        node.query(
-            """SELECT data
-               FROM mongodb('mongodb://root:clickhouse@mongo_secure:27017/test?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true',
-                            'simple_table',
-                            'key UInt64, data String')
-               WHERE key = 42"""
-        )
-        == hex(42 * 42) + "\n"
-    )
     simple_mongo_table.drop()
 
 
+@pytest.mark.parametrize("started_cluster", [False], indirect=["started_cluster"])
 def test_no_credentials(started_cluster):
     mongo_connection = get_mongo_connection(started_cluster, with_credentials=False)
     db = mongo_connection["test"]
@@ -362,6 +220,7 @@ def test_no_credentials(started_cluster):
     simple_mongo_table.drop()
 
 
+@pytest.mark.parametrize("started_cluster", [False], indirect=["started_cluster"])
 def test_auth_source(started_cluster):
     mongo_connection = get_mongo_connection(started_cluster, with_credentials=False)
     admin_db = mongo_connection["admin"]
@@ -383,10 +242,10 @@ def test_auth_source(started_cluster):
     simple_mongo_table.insert_many(data)
 
     node = started_cluster.instances["node"]
-    with pytest.raises(QueryRuntimeException):
+
-        node.query(
+    node.query_and_get_error(
-            "SELECT count() FROM mongodb('mongo_no_cred:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String')"
+        "SELECT count() FROM mongodb('mongo_no_cred:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String')"
-        )
+    )
 
     assert (
         node.query(
@@ -394,10 +253,10 @@ def test_auth_source(started_cluster):
         )
         == "100\n"
     )
-
     simple_mongo_table.drop()
 
 
+@pytest.mark.parametrize("started_cluster", [False], indirect=["started_cluster"])
 def test_missing_columns(started_cluster):
     mongo_connection = get_mongo_connection(started_cluster)
     db = mongo_connection["test"]

tests/integration/test_table_function_mongodb_new/configs/feature_flag.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <use_legacy_mongodb_integration>0</use_legacy_mongodb_integration>
+</clickhouse>

tests/integration/test_table_function_mongodb_new/configs/users.xml

@@ -0,0 +1,9 @@
+<clickhouse>
+    <users>
+        <default>
+            <password></password>
+            <profile>default</profile>
+            <named_collection_control>1</named_collection_control>
+        </default>
+    </users>
+</clickhouse>

tests/integration/test_table_function_mongodb_new/mongo_secure_config/cert.crt

@@ -0,0 +1,24 @@
+-----BEGIN CERTIFICATE-----
+MIIEGzCCAwOgAwIBAgIUaoGlyuJAyvs6yowFXymfu7seEiUwDQYJKoZIhvcNAQEL
+BQAwgZwxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDbGlja0hvdXNlMRMwEQYDVQQH
+DApDbGlja0hvdXNlMREwDwYDVQQKDAhQZXJzb25hbDETMBEGA1UECwwKQ2xpY2tI
+b3VzZTEkMCIGCSqGSIb3DQEJARYVY2xpY2tob3VzZUBjbGlja2hvdXNlMRUwEwYD
+VQQDDAxtb25nb19zZWN1cmUwHhcNMjQwNTI2MTYwMDMxWhcNMzQwNTI0MTYwMDMx
+WjCBnDELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNsaWNrSG91c2UxEzARBgNVBAcM
+CkNsaWNrSG91c2UxETAPBgNVBAoMCFBlcnNvbmFsMRMwEQYDVQQLDApDbGlja0hv
+dXNlMSQwIgYJKoZIhvcNAQkBFhVjbGlja2hvdXNlQGNsaWNraG91c2UxFTATBgNV
+BAMMDG1vbmdvX3NlY3VyZTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+AJSeQfMG7xd0+kPehYlEsEw0Sm1DB05SXVCEzIX3DFD6XJrd8eeWwlzYaBatkcwj
++8yvqske02X/3YwpzJyFizAqJIoKql5c5Yii2xH1S9PFP0y+LoJre+eQziHyO33t
+eeedeGNJ05Sm2ZAzjfMQ7Rdh6S+gdIO4Y102iQR5yr2aTrh7tu7XkNCjwKTqMMvz
+SikP1Rft2J6ECim+MjYCCtH/4yXGeEJ5epU4t3y6Q23B2ZEhY+sqUdwgK9pu8oe4
+mkZ1Qvwakc9Qg12owRSDjBBYrPvghXVpkJ2JkgKTrIAIz9tZ53eDVHNXbWMAotov
+jEmRSoGIS1yzwmQ9PdxUwYcCAwEAAaNTMFEwHQYDVR0OBBYEFJyz3Kt5XBDg5cvI
+0v1ioqejqX+CMB8GA1UdIwQYMBaAFJyz3Kt5XBDg5cvI0v1ioqejqX+CMA8GA1Ud
+EwEB/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAHAQFA5VMYvaQFnKtKfHg9TF
+qfJ4uM3YsGdgsgmGWgflD1S4Z290H6Q2QvyZAEceTrlJxqArlWlVp5DAU6EeXjEh
+QMAgdkJHF1Hg2jsZKPtdkb88UtuzwAME357T8NtEJSHzNE5QqYwlVM71JkWpdqvA
+UUdOJbWhhJfowIf4tMmL1DUuIy2qYpoP/tEBXEw9uwpmZqb7KELwT3lRyOMaGFN7
+RHVwbvJWlHiu83QDNaWz6ijQkWl3tCN6TWcFD1qc1x8GpMzjbsAAYbCx7fbHM2LD
+9kGSCiyv5K0MLNK5u67RtUFfPHtyD8RA0TtxIZ4PEN/eFANKS2/5NEi1ZuZ5/Pk=
+-----END CERTIFICATE-----

tests/integration/test_table_function_mongodb_new/mongo_secure_config/key.pem

@@ -0,0 +1,52 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCUnkHzBu8XdPpD
+3oWJRLBMNEptQwdOUl1QhMyF9wxQ+lya3fHnlsJc2GgWrZHMI/vMr6rJHtNl/92M
+KcychYswKiSKCqpeXOWIotsR9UvTxT9Mvi6Ca3vnkM4h8jt97XnnnXhjSdOUptmQ
+M43zEO0XYekvoHSDuGNdNokEecq9mk64e7bu15DQo8Ck6jDL80opD9UX7diehAop
+vjI2AgrR/+MlxnhCeXqVOLd8ukNtwdmRIWPrKlHcICvabvKHuJpGdUL8GpHPUINd
+qMEUg4wQWKz74IV1aZCdiZICk6yACM/bWed3g1RzV21jAKLaL4xJkUqBiEtcs8Jk
+PT3cVMGHAgMBAAECggEAAul6qiHchB+uQMCWyC5xTeRqAXR3tAv4Tj4fGJjkXY4Z
+OrAjr9Kp38EvX1amgvUWV3FT3NMevDf5xd9OdzAA0g0uJIF+mAhYFW48i1FnQcHQ
+mOf0zmiZR7l8o7ROb3JvooXHxW+ba/qjGPVwC801gJvruehgbOCRxh9DTRp7sH5K
+BmcddhULhKBEQjWUmYNEM3A2axpdi3g1aYKERRLn8J0DXcItTwbxuxbNcs3erl8W
+3yyv/JKmqnWF5sNyX3wEWuQcDEZZy+W7Hn4KPMxyU+WA5el5nJ8kFlxhpInmajwu
+8Ytn6IEyThyXutVomosVBuP16QORl2Nad0hnQO9toQKBgQDDgiehXr3k2wfVaVOD
+PocW4leXausIU2XcCn6FxTG9vLUDMPANw0MxgenC2nrjaUU9J9UjdRYgMcFGWrl4
+E27wEn5e0nZ/Y7F2cfhuOc9vNmZ+eHm2KQRyfAjIVL5Hpldqk2jXyCnLBNeWGHSw
+kPQMU+FLqmrOFUvXlD2my+OSHwKBgQDCmgS9r+xFh4BCB9dY6eyQJF/jYmAQHs26
+80WJ6gAhbUw1O71uDtS9/3PZVXwwNCOHrcc49BPrpJdxGPHGvd2Q5y+j5LDDbQSZ
+aLTiCZ2B0RM5Bd2dXD8gEHN4WCX7pJ/o4kDi4zONBmp5mg/tFfer5z5IU/1P7Wak
+1Mu0JIHzmQKBgDNaNoqeVgaMuYwGtFbez6DlJtiwzrdLIJAheYYte5k4vdruub8D
+sNyKIRp7RJgDCJq9obBEiuE98GRIZDrz78nDMco6QcHIL87KtNRO/vtZMKa7gkyk
+jXR8u9nS2H/9YyytN3amLsQSq4XTOqM+D7xFNAIp6w/ibB9d4quzFj1FAoGBAKTE
+x/LcO897NWuzO/D6z+QUCGR87R15F3SNenmVedrTskz4ciH3yMW+v5ZrPSWLX/IH
+f8GHWD6TM+780eoW5L1GIh5BCjHN4rEJ6O3iekxqfD4x6zzL2F8Lztk8uZxh/Uuw
+FoSFHybvIcQoYAe8K+KPfzq6cqb0OY6i5n920dkxAoGAJkw6ADqsJfH3NR+bQfgF
+oEA1KqriMxyEJm44Y7E80C+iF4iNALF+Er9TSnr4mDxX5e/dW9d1YeS9o0nOfkpF
+MaBmJfxqo4QQJLPRaxYQ2Jhfn7irir4BroxeNXQgNNhgSuKIvkfRyGYwl7P0AT4v
+8H8rkZGneMD3gLB5MfnRhGk=
+-----END PRIVATE KEY-----
+-----BEGIN CERTIFICATE-----
+MIIEGzCCAwOgAwIBAgIUaoGlyuJAyvs6yowFXymfu7seEiUwDQYJKoZIhvcNAQEL
+BQAwgZwxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDbGlja0hvdXNlMRMwEQYDVQQH
+DApDbGlja0hvdXNlMREwDwYDVQQKDAhQZXJzb25hbDETMBEGA1UECwwKQ2xpY2tI
+b3VzZTEkMCIGCSqGSIb3DQEJARYVY2xpY2tob3VzZUBjbGlja2hvdXNlMRUwEwYD
+VQQDDAxtb25nb19zZWN1cmUwHhcNMjQwNTI2MTYwMDMxWhcNMzQwNTI0MTYwMDMx
+WjCBnDELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNsaWNrSG91c2UxEzARBgNVBAcM
+CkNsaWNrSG91c2UxETAPBgNVBAoMCFBlcnNvbmFsMRMwEQYDVQQLDApDbGlja0hv
+dXNlMSQwIgYJKoZIhvcNAQkBFhVjbGlja2hvdXNlQGNsaWNraG91c2UxFTATBgNV
+BAMMDG1vbmdvX3NlY3VyZTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+AJSeQfMG7xd0+kPehYlEsEw0Sm1DB05SXVCEzIX3DFD6XJrd8eeWwlzYaBatkcwj
++8yvqske02X/3YwpzJyFizAqJIoKql5c5Yii2xH1S9PFP0y+LoJre+eQziHyO33t
+eeedeGNJ05Sm2ZAzjfMQ7Rdh6S+gdIO4Y102iQR5yr2aTrh7tu7XkNCjwKTqMMvz
+SikP1Rft2J6ECim+MjYCCtH/4yXGeEJ5epU4t3y6Q23B2ZEhY+sqUdwgK9pu8oe4
+mkZ1Qvwakc9Qg12owRSDjBBYrPvghXVpkJ2JkgKTrIAIz9tZ53eDVHNXbWMAotov
+jEmRSoGIS1yzwmQ9PdxUwYcCAwEAAaNTMFEwHQYDVR0OBBYEFJyz3Kt5XBDg5cvI
+0v1ioqejqX+CMB8GA1UdIwQYMBaAFJyz3Kt5XBDg5cvI0v1ioqejqX+CMA8GA1Ud
+EwEB/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAHAQFA5VMYvaQFnKtKfHg9TF
+qfJ4uM3YsGdgsgmGWgflD1S4Z290H6Q2QvyZAEceTrlJxqArlWlVp5DAU6EeXjEh
+QMAgdkJHF1Hg2jsZKPtdkb88UtuzwAME357T8NtEJSHzNE5QqYwlVM71JkWpdqvA
+UUdOJbWhhJfowIf4tMmL1DUuIy2qYpoP/tEBXEw9uwpmZqb7KELwT3lRyOMaGFN7
+RHVwbvJWlHiu83QDNaWz6ijQkWl3tCN6TWcFD1qc1x8GpMzjbsAAYbCx7fbHM2LD
+9kGSCiyv5K0MLNK5u67RtUFfPHtyD8RA0TtxIZ4PEN/eFANKS2/5NEi1ZuZ5/Pk=
+-----END CERTIFICATE-----

tests/integration/test_table_function_mongodb_new/mongo_secure_config/mongo_secure.conf

@@ -0,0 +1,6 @@
+net:
+  ssl:
+    mode: requireSSL
+    PEMKeyFile: /mongo/key.pem
+    CAFile: /mongo/cert.crt
+    allowConnectionsWithoutCertificates: true

tests/integration/test_table_function_mongodb_new/test.py

@@ -0,0 +1,419 @@
+import pymongo
+
+import pytest
+from helpers.client import QueryRuntimeException
+
+from helpers.cluster import ClickHouseCluster
+
+
+@pytest.fixture(scope="module")
+def started_cluster(request):
+    try:
+        cluster = ClickHouseCluster(__file__)
+        cluster.add_instance(
+            "node",
+            with_mongo=True,
+            main_configs=[
+                "configs/named_collections.xml",
+                "configs/feature_flag.xml",
+            ],
+            user_configs=["configs/users.xml"],
+        )
+        cluster.start()
+        yield cluster
+    finally:
+        cluster.shutdown()
+
+
+def get_mongo_connection(started_cluster, secure=False, with_credentials=True):
+    if secure:
+        return pymongo.MongoClient(
+            "mongodb://root:clickhouse@localhost:{}/?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true".format(
+                started_cluster.mongo_secure_port
+            )
+        )
+    if with_credentials:
+        return pymongo.MongoClient(
+            "mongodb://root:clickhouse@localhost:{}".format(started_cluster.mongo_port)
+        )
+
+    return pymongo.MongoClient(
+        "mongodb://localhost:{}".format(started_cluster.mongo_no_cred_port)
+    )
+
+
+def test_simple_select(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster)
+    db = mongo_connection["test"]
+    db.add_user("root", "clickhouse")
+    simple_mongo_table = db["simple_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i)})
+    simple_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+    assert (
+        node.query(
+            "SELECT COUNT() FROM mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String')"
+        )
+        == "100\n"
+    )
+    assert (
+        node.query(
+            "SELECT sum(key) FROM mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String')"
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+    assert (
+        node.query(
+            "SELECT sum(key) FROM mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', 'key UInt64, data String')"
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+
+    assert (
+        node.query(
+            "SELECT data FROM mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String') WHERE key = 42"
+        )
+        == hex(42 * 42) + "\n"
+    )
+    simple_mongo_table.drop()
+
+
+def test_simple_select_uri(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster)
+    db = mongo_connection["test"]
+    db.add_user("root", "clickhouse")
+    simple_mongo_table = db["simple_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i)})
+    simple_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+    assert (
+        node.query(
+            "SELECT COUNT() FROM mongodb('mongodb://root:clickhouse@mongo1:27017/test', 'simple_table', structure='key UInt64, data String')"
+        )
+        == "100\n"
+    )
+    assert (
+        node.query(
+            "SELECT sum(key) FROM mongodb('mongodb://root:clickhouse@mongo1:27017/test', 'simple_table', structure='key UInt64, data String')"
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+    assert (
+        node.query(
+            "SELECT sum(key) FROM mongodb('mongodb://root:clickhouse@mongo1:27017/test', 'simple_table', 'key UInt64, data String')"
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+
+    assert (
+        node.query(
+            "SELECT data FROM mongodb('mongodb://root:clickhouse@mongo1:27017/test', 'simple_table', structure='key UInt64, data String') WHERE key = 42"
+        )
+        == hex(42 * 42) + "\n"
+    )
+    simple_mongo_table.drop()
+
+
+def test_complex_data_type(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster)
+    db = mongo_connection["test"]
+    db.add_user("root", "clickhouse")
+    incomplete_mongo_table = db["complex_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i), "dict": {"a": i, "b": str(i)}})
+    incomplete_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+
+    assert (
+        node.query(
+            """
+            SELECT COUNT()
+            FROM mongodb('mongo1:27017',
+                         'test',
+                         'complex_table',
+                         'root',
+                         'clickhouse',
+                         structure='key UInt64, data String, dict Map(UInt64, String)')"""
+        )
+        == "100\n"
+    )
+    assert (
+        node.query(
+            """
+            SELECT sum(key)
+            FROM mongodb('mongo1:27017',
+                         'test',
+                         'complex_table',
+                         'root',
+                         'clickhouse',
+                         structure='key UInt64, data String, dict Map(UInt64, String)')"""
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+
+    assert (
+        node.query(
+            """
+            SELECT data
+            FROM mongodb('mongo1:27017',
+                         'test',
+                         'complex_table',
+                         'root',
+                         'clickhouse',
+                         structure='key UInt64, data String, dict Map(UInt64, String)')
+            WHERE key = 42
+            """
+        )
+        == hex(42 * 42) + "\n"
+    )
+    incomplete_mongo_table.drop()
+
+
+def test_incorrect_data_type(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster)
+    db = mongo_connection["test"]
+    db.add_user("root", "clickhouse")
+    strange_mongo_table = db["strange_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i), "aaaa": "Hello"})
+    strange_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+
+    with pytest.raises(QueryRuntimeException):
+        node.query(
+            "SELECT aaaa FROM mongodb('mongo1:27017', 'test', 'strange_table', 'root', 'clickhouse', structure='key UInt64, data String')"
+        )
+
+    strange_mongo_table.drop()
+
+
+def test_secure_connection(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster, secure=True)
+    db = mongo_connection["test"]
+    db.add_user("root", "clickhouse")
+    simple_mongo_table = db["simple_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i)})
+    simple_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+
+    assert (
+        node.query(
+            """SELECT COUNT()
+               FROM mongodb('mongo_secure:27017',
+                            'test',
+                            'simple_table',
+                            'root',
+                            'clickhouse',
+                            structure='key UInt64, data String',
+                            options='tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true')"""
+        )
+        == "100\n"
+    )
+    assert (
+        node.query(
+            """SELECT sum(key)
+               FROM mongodb('mongo_secure:27017',
+                            'test',
+                            'simple_table',
+                            'root',
+                            'clickhouse',
+                            structure='key UInt64, data String',
+                            options='tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true')"""
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+    assert (
+        node.query(
+            """SELECT sum(key)
+               FROM mongodb('mongo_secure:27017',
+                            'test',
+                            'simple_table',
+                            'root',
+                            'clickhouse',
+                            'key UInt64, data String',
+                            'tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true')"""
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+
+    assert (
+        node.query(
+            """SELECT data
+               FROM mongodb('mongo_secure:27017',
+                            'test',
+                            'simple_table',
+                            'root',
+                            'clickhouse',
+                            'key UInt64, data String',
+                            'tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true')
+               WHERE key = 42"""
+        )
+        == hex(42 * 42) + "\n"
+    )
+    simple_mongo_table.drop()
+
+
+def test_secure_connection_with_validation(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster, secure=True)
+    db = mongo_connection["test"]
+    db.add_user("root", "clickhouse")
+    simple_mongo_table = db["simple_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i)})
+    simple_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+    with pytest.raises(QueryRuntimeException):
+        node.query(
+            """SELECT COUNT() FROM mongodb('mongo_secure:27017',
+                   'test',
+                   'simple_table',
+                   'root',
+                   'clickhouse',
+                   structure='key UInt64, data String',
+                   options='tls=true')"""
+        )
+
+    simple_mongo_table.drop()
+
+
+def test_secure_connection_uri(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster, secure=True)
+    db = mongo_connection["test"]
+    db.add_user("root", "clickhouse")
+    simple_mongo_table = db["simple_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i)})
+    simple_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+
+    assert (
+        node.query(
+            """SELECT COUNT()
+               FROM mongodb('mongodb://root:clickhouse@mongo_secure:27017/test?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true',
+                            'simple_table',
+                            'key UInt64, data String')"""
+        )
+        == "100\n"
+    )
+    assert (
+        node.query(
+            """SELECT sum(key)
+               FROM mongodb('mongodb://root:clickhouse@mongo_secure:27017/test?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true',
+                            'simple_table',
+                            'key UInt64, data String')"""
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+    assert (
+        node.query(
+            """SELECT sum(key)
+               FROM mongodb('mongodb://root:clickhouse@mongo_secure:27017/test?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true',
+                            'simple_table',
+                            'key UInt64, data String')"""
+        )
+        == str(sum(range(0, 100))) + "\n"
+    )
+
+    assert (
+        node.query(
+            """SELECT data
+               FROM mongodb('mongodb://root:clickhouse@mongo_secure:27017/test?tls=true&tlsAllowInvalidCertificates=true&tlsAllowInvalidHostnames=true',
+                            'simple_table',
+                            'key UInt64, data String')
+               WHERE key = 42"""
+        )
+        == hex(42 * 42) + "\n"
+    )
+    simple_mongo_table.drop()
+
+
+def test_no_credentials(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster, with_credentials=False)
+    db = mongo_connection["test"]
+    simple_mongo_table = db["simple_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i)})
+    simple_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+    assert (
+        node.query(
+            "SELECT count() FROM mongodb('mongo_no_cred:27017', 'test', 'simple_table', '', '', structure='key UInt64, data String')"
+        )
+        == "100\n"
+    )
+    simple_mongo_table.drop()
+
+
+def test_auth_source(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster, with_credentials=False)
+    admin_db = mongo_connection["admin"]
+    admin_db.add_user(
+        "root",
+        "clickhouse",
+        roles=[{"role": "userAdminAnyDatabase", "db": "admin"}, "readWriteAnyDatabase"],
+    )
+    simple_mongo_table = admin_db["simple_table"]
+    data = []
+    for i in range(0, 50):
+        data.append({"key": i, "data": hex(i * i)})
+    simple_mongo_table.insert_many(data)
+    db = mongo_connection["test"]
+    simple_mongo_table = db["simple_table"]
+    data = []
+    for i in range(0, 100):
+        data.append({"key": i, "data": hex(i * i)})
+    simple_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+    with pytest.raises(QueryRuntimeException):
+        node.query(
+            "SELECT count() FROM mongodb('mongo_no_cred:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String')"
+        )
+
+    assert (
+        node.query(
+            "SELECT count() FROM mongodb('mongo_no_cred:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data String', options='authSource=admin')"
+        )
+        == "100\n"
+    )
+
+    simple_mongo_table.drop()
+
+
+def test_missing_columns(started_cluster):
+    mongo_connection = get_mongo_connection(started_cluster)
+    db = mongo_connection["test"]
+    db.add_user("root", "clickhouse")
+    simple_mongo_table = db["simple_table"]
+    data = []
+    for i in range(0, 10):
+        data.append({"key": i, "data": hex(i * i)})
+    for i in range(0, 10):
+        data.append({"key": i})
+    simple_mongo_table.insert_many(data)
+
+    node = started_cluster.instances["node"]
+    result = node.query(
+        "SELECT count() FROM mongodb('mongo1:27017', 'test', 'simple_table', 'root', 'clickhouse', structure='key UInt64, data Nullable(String)') WHERE isNull(data)"
+    )
+    assert result == "10\n"
+    simple_mongo_table.drop()