Merge branch 'fix-test-01287_max_execution_speed' into stripe-log-use-new-serialization-formats

Author: alexey-milovidov

docs/en/interfaces/schema-inference.md

@@ -1449,6 +1449,10 @@ DESC format(JSONEachRow, $$
 
 #### input_format_try_infer_integers {#input-format-try-infer-integers}
 
+:::note
+This setting does not apply to the `JSON` data type.
+:::
+
 If enabled, ClickHouse will try to infer integers instead of floats in schema inference for text formats.
 If all numbers in the column from sample data are integers, the result type will be `Int64`, if at least one number is float, the result type will be `Float64`.
 If the sample data contains only integers and at least one integer is positive and overflows `Int64`, ClickHouse will infer `UInt64`.

src/Analyzer/ConstantNode.cpp

@@ -199,12 +199,12 @@ QueryTreeNodePtr ConstantNode::cloneImpl() const
 
 ASTPtr ConstantNode::toASTImpl(const ConvertToASTOptions & options) const
 {
+    if (!options.add_cast_for_constants)
+        return std::make_shared<ASTLiteral>(getFieldFromColumnForASTLiteral(constant_value.getColumn(), 0, constant_value.getType()));
+
     const auto & constant_value_type = constant_value.getType();
     auto constant_value_ast = std::make_shared<ASTLiteral>(getValue());
 
-    if (!options.add_cast_for_constants)
-        return constant_value_ast;
-
     // Add cast if constant was created as a result of constant folding.
     // Constant folding may lead to type transformation and literal on shard
     // may have a different type.

src/Processors/QueryPlan/Optimizations/filterPushDown.cpp

@@ -387,8 +387,33 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan::
 
     size_t updated_steps = 0;
 
+    /// For the logical join step, we need to merge pre-join actions to filter dag.
+    /// TODO: this should be refactored and replaced with optimizations which
+    /// 1. push filter/expression into JOIN (as post-filter)
+    /// 2. move filter within JOIN step, potentially changing JoinKind
+    /// 3. push filter/expression out of JOIN (from pre-filter)
+    auto fix_predicate_for_join_logical_step = [&](ActionsDAG filter_dag, const ActionsDAG & side_dag)
+    {
+        filter_dag = ActionsDAG::merge(side_dag.clone(), std::move(filter_dag));
+        auto & outputs = filter_dag.getOutputs();
+        outputs.resize(1);
+        outputs.insert(outputs.end(), filter_dag.getInputs().begin(), filter_dag.getInputs().end());
+        filter_dag.removeUnusedActions();
+        return filter_dag;
+    };
+
     if (join_filter_push_down_actions.left_stream_filter_to_push_down)
     {
+        if (logical_join)
+        {
+
+            join_filter_push_down_actions.left_stream_filter_to_push_down = fix_predicate_for_join_logical_step(
+                std::move(*join_filter_push_down_actions.left_stream_filter_to_push_down),
+                *logical_join->getExpressionActions().left_pre_join_actions
+            );
+            join_filter_push_down_actions.left_stream_filter_removes_filter = true;
+        }
+
         const auto & result_name = join_filter_push_down_actions.left_stream_filter_to_push_down->getOutputs()[0]->result_name;
         updated_steps += addNewFilterStepOrThrow(parent_node,
             nodes,
@@ -404,6 +429,16 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan::
 
     if (join_filter_push_down_actions.right_stream_filter_to_push_down && allow_push_down_to_right)
     {
+        if (logical_join)
+        {
+
+            join_filter_push_down_actions.right_stream_filter_to_push_down = fix_predicate_for_join_logical_step(
+                std::move(*join_filter_push_down_actions.right_stream_filter_to_push_down),
+                *logical_join->getExpressionActions().right_pre_join_actions
+            );
+            join_filter_push_down_actions.right_stream_filter_removes_filter = true;
+        }
+
         const auto & result_name = join_filter_push_down_actions.right_stream_filter_to_push_down->getOutputs()[0]->result_name;
         updated_steps += addNewFilterStepOrThrow(parent_node,
             nodes,

src/Storages/tests/gtest_transform_query_for_external_database.cpp

@@ -301,8 +301,7 @@ TEST(TransformQueryForExternalDatabase, Issue7245)
 
     check(state, 1, {"apply_id", "apply_type", "apply_status", "create_time"},
           "SELECT apply_id FROM test.table WHERE apply_type = 2 AND create_time > addDays(toDateTime('2019-01-01 01:02:03', 'UTC'),-7) AND apply_status IN (3,4)",
-          R"(SELECT "apply_id", "apply_type", "apply_status", "create_time" FROM "test"."table" WHERE ("apply_type" = 2) AND ("create_time" > '2018-12-25 01:02:03') AND ("apply_status" IN (3, 4)))",
-          R"(SELECT "apply_id", "apply_type", "apply_status", "create_time" FROM "test"."table" WHERE ("apply_type" = 2) AND ("create_time" > 1545699723) AND ("apply_status" IN (3, 4)))");
+          R"(SELECT "apply_id", "apply_type", "apply_status", "create_time" FROM "test"."table" WHERE ("apply_type" = 2) AND ("create_time" > '2018-12-25 01:02:03') AND ("apply_status" IN (3, 4)))");
 }
 
 TEST(TransformQueryForExternalDatabase, Aliases)
@@ -396,8 +395,7 @@ TEST(TransformQueryForExternalDatabase, ToDate)
 
     check(state, 1, {"a", "b", "foo"},
         "SELECT foo FROM table WHERE a=10 AND b=toDate('2019-10-05', 'UTC')",
-        R"(SELECT "a", "b", "foo" FROM "test"."table" WHERE ("a" = 10) AND ("b" = '2019-10-05'))",
-        R"(SELECT "a", "b", "foo" FROM "test"."table" WHERE ("a" = 10) AND ("b" = 18174))");
+        R"(SELECT "a", "b", "foo" FROM "test"."table" WHERE ("a" = 10) AND ("b" = '2019-10-05'))");
 }
 
 TEST(TransformQueryForExternalDatabase, Analyzer)
@@ -422,8 +420,7 @@ TEST(TransformQueryForExternalDatabase, Analyzer)
 
     check(state, 1, {"is_value"},
         "SELECT is_value FROM table WHERE is_value = true",
-        R"(SELECT "is_value" FROM "test"."table" WHERE "is_value" = true)",
-        R"(SELECT "is_value" FROM "test"."table" WHERE "is_value" = 1)");
+        R"(SELECT "is_value" FROM "test"."table" WHERE "is_value" = true)");
 
     check(state, 1, {"is_value"},
         "SELECT is_value FROM table WHERE is_value = 1",

tests/config/config.d/storage_conf_03517.xml

@@ -0,0 +1,26 @@
+<clickhouse>
+    <storage_configuration>
+        <disks>
+            <disk_s3_plain_rewritable_03517>
+                <type>object_storage</type>
+                <object_storage_type>s3</object_storage_type>
+                <metadata_type>plain_rewritable</metadata_type>
+                <endpoint>http://localhost:11111/test/common/</endpoint>
+                <access_key_id>clickhouse</access_key_id>
+                <secret_access_key>clickhouse</secret_access_key>
+            </disk_s3_plain_rewritable_03517>
+            <disk_cache_03517>
+                <type>cache</type>
+                <disk>disk_s3_plain_rewritable_03517</disk>
+                <path>disks/cache_03517/</path>
+                <max_size>1Mi</max_size>
+                <cache_on_write_operations>1</cache_on_write_operations>
+            </disk_cache_03517>
+            <disk_encrypted_03517>
+                <type>encrypted</type>
+                <disk>disk_cache_03517</disk>
+                <key>1234567812345678</key>
+            </disk_encrypted_03517>
+        </disks>
+    </storage_configuration>
+</clickhouse>

tests/config/install.sh

@@ -255,6 +255,7 @@ if [[ "$EXPORT_S3_STORAGE_POLICIES" == "1" ]]; then
     fi
     ln -sf $SRC_PATH/config.d/storage_conf_02963.xml $DEST_SERVER_PATH/config.d/
     ln -sf $SRC_PATH/config.d/storage_conf_02961.xml $DEST_SERVER_PATH/config.d/
+    ln -sf $SRC_PATH/config.d/storage_conf_03517.xml $DEST_SERVER_PATH/config.d/
     ln -sf $SRC_PATH/users.d/s3_cache.xml $DEST_SERVER_PATH/users.d/
     ln -sf $SRC_PATH/users.d/s3_cache_new.xml $DEST_SERVER_PATH/users.d/
 fi

tests/integration/test_storage_postgresql/test.py

@@ -869,6 +869,29 @@ def _create_and_fill_table(table):
     cursor.execute(f'DROP TABLE "{table}\'"')
 
 
+def test_postgres_datetime(started_cluster):
+    cursor = started_cluster.postgres_conn.cursor()
+    cursor.execute(f"DROP TABLE IF EXISTS test_datetime")
+    cursor.execute("CREATE TABLE test_datetime AS (SELECT '2025-01-02 03:04:05.678900'::timestamptz AS ts, '2025-01-02'::date as d)")
+
+    node1.query("DROP TABLE IF EXISTS test_datetime")
+    node1.query(
+        f"CREATE TABLE test_datetime (ts DateTime64(6, 'UTC'), d Date) ENGINE = PostgreSQL('postgres1:5432', 'postgres', 'test_datetime', 'postgres', '{pg_pass}')"
+    )
+
+    result = node1.query("SELECT ts FROM test_datetime WHERE ts > '2025-01-01'::DateTime")
+    assert result == "2025-01-02 03:04:05.678900\n"
+
+    result = node1.query("SELECT ts FROM test_datetime WHERE ts > '2025-01-01'::DateTime64")
+    assert result == "2025-01-02 03:04:05.678900\n"
+
+    result = node1.query("SELECT ts FROM test_datetime WHERE ts > '2025-01-01'::Nullable(DateTime)")
+    assert result == "2025-01-02 03:04:05.678900\n"
+
+    result = node1.query("SELECT ts FROM test_datetime WHERE ts > '2025-01-01'::Nullable(DateTime64)")
+    assert result == "2025-01-02 03:04:05.678900\n"
+
+
 if __name__ == "__main__":
     cluster.start()
     input("Cluster created, press any key to destroy...")

tests/queries/0_stateless/01287_max_execution_speed.sql

@@ -13,6 +13,7 @@ SELECT count() FROM system.numbers; -- { serverError TOO_SLOW }
 SET min_execution_speed_bytes = 0;
 SELECT 'Ok (2)';
 
+SET timeout_before_checking_execution_speed = 60;
 SET max_execution_speed = 1000000;
 SET max_block_size = 100;
 

tests/queries/0_stateless/03322_initial_query_start_time_check.sql

@@ -2,8 +2,8 @@
 
 DROP TABLE IF EXISTS tmp;
 
-CREATE OR REPLACE VIEW tmp AS SELECT initialQueryStartTime() as it, now() AS t FROM system.one WHERE NOT ignore(sleep(0.1));
-SELECT now()==max(t), initialQueryStartTime()==max(it), initialQueryStartTime()==min(it), initialQueryStartTime()>=now()-1 FROM (SELECT it, t FROM remote('127.0.0.{1..20}', currentDatabase(), tmp)) SETTINGS max_distributed_connections=1, async_socket_for_remote=0;
+CREATE OR REPLACE VIEW tmp AS SELECT initialQueryStartTime() as it, now() AS t FROM system.one WHERE NOT ignore(sleep(0.5));
+SELECT now()==max(t), initialQueryStartTime()==max(it), initialQueryStartTime()==min(it), initialQueryStartTime()>=now()-1 FROM (SELECT it, t FROM remote('127.0.0.{1..10}', currentDatabase(), tmp)) SETTINGS max_distributed_connections=1, async_socket_for_remote=0;
 
 DROP TABLE tmp;
 

tests/queries/0_stateless/03393_join_bug_77848.reference

@@ -0,0 +1,2 @@
+1
+1