fix prewhere in materialize mysql

base/mysqlxx/Pool.h

@@ -159,9 +159,9 @@ public:
       */
     Pool(const std::string & db_,
          const std::string & server_,
-         const std::string & user_ = "",
+         const std::string & user_,
-         const std::string & password_ = "",
+         const std::string & password_,
-         unsigned port_ = 0,
+         unsigned port_,
          const std::string & socket_ = "",
          unsigned connect_timeout_ = MYSQLXX_DEFAULT_TIMEOUT,
          unsigned rw_timeout_ = MYSQLXX_DEFAULT_RW_TIMEOUT,

src/Databases/DatabaseFactory.cpp

@@ -158,7 +158,7 @@ DatabasePtr DatabaseFactory::getImpl(const ASTCreateQuery & create, const String
 
             const auto & [remote_host_name, remote_port] = parseAddress(host_port, 3306);
             MySQLClient client(remote_host_name, remote_port, mysql_user_name, mysql_user_password);
-            auto mysql_pool = mysqlxx::Pool(mysql_database_name, remote_host_name, mysql_user_name, mysql_user_password);
+            auto mysql_pool = mysqlxx::Pool(mysql_database_name, remote_host_name, mysql_user_name, mysql_user_password, remote_port);
 
 
             auto materialize_mode_settings = std::make_unique<MaterializeMySQLSettings>();

src/Storages/StorageMaterializeMySQL.cpp

@@ -92,7 +92,7 @@ Pipe StorageMaterializeMySQL::read(
     {
         Block pipe_header = pipe.getHeader();
         auto syntax = TreeRewriter(context).analyze(expressions, pipe_header.getNamesAndTypesList());
-        ExpressionActionsPtr expression_actions = ExpressionAnalyzer(expressions, syntax, context).getActions(true);
+        ExpressionActionsPtr expression_actions = ExpressionAnalyzer(expressions, syntax, context).getActions(true, false);
 
         pipe.addSimpleTransform([&](const Block & header)
         {

tests/integration/test_materialize_mysql_database/materialize_with_ddl.py

@@ -842,3 +842,14 @@ def system_tables_test(clickhouse_node, mysql_node, service_name):
     mysql_node.query("CREATE TABLE system_tables_test.test (id int NOT NULL PRIMARY KEY) ENGINE=InnoDB")
     clickhouse_node.query("CREATE DATABASE system_tables_test ENGINE = MaterializeMySQL('{}:3306', 'system_tables_test', 'root', 'clickhouse')".format(service_name))
     check_query(clickhouse_node, "SELECT partition_key, sorting_key, primary_key FROM system.tables WHERE database = 'system_tables_test' AND name = 'test'", "intDiv(id, 4294967)\tid\tid\n")
+
+def move_to_prewhere_and_column_filtering(clickhouse_node, mysql_node, service_name):
+    clickhouse_node.query("DROP DATABASE IF EXISTS cond_on_key_col")
+    mysql_node.query("DROP DATABASE IF EXISTS cond_on_key_col")
+    mysql_node.query("CREATE DATABASE cond_on_key_col")
+    clickhouse_node.query("CREATE DATABASE cond_on_key_col ENGINE = MaterializeMySQL('{}:3306', 'cond_on_key_col', 'root', 'clickhouse')".format(service_name))
+    mysql_node.query("create table cond_on_key_col.products (id int primary key, product_id int not null, catalog_id int not null, brand_id int not null, name text)")
+    mysql_node.query("insert into cond_on_key_col.products (id, name, catalog_id, brand_id, product_id) values (915, 'ertyui', 5287, 15837, 0), (990, 'wer', 1053, 24390, 1), (781, 'qwerty', 1041, 1176, 2);")
+    check_query(clickhouse_node, "SELECT DISTINCT P.id, P.name, P.catalog_id FROM cond_on_key_col.products P WHERE P.name ILIKE '%e%' and P.catalog_id=5287", '915\tertyui\t5287\n')
+    clickhouse_node.query("DROP DATABASE cond_on_key_col")
+    mysql_node.query("DROP DATABASE cond_on_key_col")

tests/integration/test_materialize_mysql_database/test.py

@@ -152,6 +152,7 @@ def test_materialize_database_dml_with_mysql_5_7(started_cluster, started_mysql_
     materialize_with_ddl.dml_with_materialize_mysql_database(clickhouse_node, started_mysql_5_7, "mysql1")
     materialize_with_ddl.materialize_mysql_database_with_views(clickhouse_node, started_mysql_5_7, "mysql1")
     materialize_with_ddl.materialize_mysql_database_with_datetime_and_decimal(clickhouse_node, started_mysql_5_7, "mysql1")
+    materialize_with_ddl.move_to_prewhere_and_column_filtering(clickhouse_node, started_mysql_5_7, "mysql1")
 
 
 @pytest.mark.parametrize(('clickhouse_node'), [node_db_ordinary, node_db_atomic])
@@ -159,6 +160,7 @@ def test_materialize_database_dml_with_mysql_8_0(started_cluster, started_mysql_
     materialize_with_ddl.dml_with_materialize_mysql_database(clickhouse_node, started_mysql_8_0, "mysql8_0")
     materialize_with_ddl.materialize_mysql_database_with_views(clickhouse_node, started_mysql_8_0, "mysql8_0")
     materialize_with_ddl.materialize_mysql_database_with_datetime_and_decimal(clickhouse_node, started_mysql_8_0, "mysql8_0")
+    materialize_with_ddl.move_to_prewhere_and_column_filtering(clickhouse_node, started_mysql_8_0, "mysql8_0")
 
 
 @pytest.mark.parametrize(('clickhouse_node'), [node_db_ordinary, node_db_atomic])