finishing

Author: filimonov

docs/en/engines/table-engines/special/hybrid.md

@@ -32,7 +32,7 @@ SET allow_experimental_hybrid_table = 1;
 
 ### Automatic Type Alignment
 
-Hybrid segments can evolve independently, so the same logical column may use different physical types. With the experimental `hybrid_table_auto_cast_columns = 1` (default, requires `allow_experimental_analyzer = 1`), the engine inserts the necessary `CAST` operations into each rewritten query so every shard receives the schema defined by the Hybrid table. This prevents header mismatches without having to edit each query, and you can opt out by setting the flag to `0` if it causes issues in your setup.
+Hybrid segments can evolve independently, so the same logical column may use different physical types. With the experimental `hybrid_table_auto_cast_columns = 1` **(enabled by default and requires `allow_experimental_analyzer = 1`)**, the engine inserts the necessary `CAST` operations into each rewritten query so every shard receives the schema defined by the Hybrid table. You can opt out by setting the flag to `0` if it causes issues.
 
 Segment schemas are cached when you create or attach a Hybrid table. If you alter a segment later (for example change a column type), refresh the Hybrid table (detach/attach or recreate it) so the cached headers stay in sync with the new schema; otherwise the auto-cast feature may miss the change and queries can still fail with header/type errors.
 

src/Analyzer/Passes/HybridCastsPass.cpp

@@ -16,7 +16,6 @@
 
 #include <Core/Settings.h>
 #include <Core/SettingsEnums.h>
-#include <Core/Settings.h>
 #include <Common/Exception.h>
 
 namespace DB
@@ -42,20 +41,6 @@ struct HybridCastTask
 };
 
 // Visitor replaces all usages of the column with CAST(column, type) in the query tree.
-//
-// It normalizes headers coming from different segments when table structure in some segments
-// differs from the Hybrid table definition. For example column X is UInt32 in the Hybrid table,
-// but Int64 in an additional segment.
-//
-// Without these casts ConvertingActions may fail to reconcile mismatched headers when casts are impossible
-// (e.g. AggregateFunction states carry hashed data tied to their argument type and cannot be recast), for example:
-// "Conversion from AggregateFunction(uniq, Decimal(38, 0)) to AggregateFunction(uniq, UInt64) is not supported"
-// (CANNOT_CONVERT_TYPE).
-//
-// Per-segment casts are not reliable because WithMergeState strips aliases, so merged pipelines
-// from different segments would return different headers (with or without CAST), leading to errors
-// like "Cannot find column `max(value)` in source stream, there are only columns: [max(_CAST(value, 'UInt64'))]"
-// (THERE_IS_NO_COLUMN).
 class HybridCastVisitor : public InDepthQueryTreeVisitor<HybridCastVisitor>
 {
 public:
@@ -125,7 +110,7 @@ void collectHybridTables(const QueryTreeNodePtr & join_tree, std::unordered_map<
         {
             ColumnsDescription to_cast = distributed->getColumnsToCast();
             if (!to_cast.empty())
-                cast_map.emplace(join_tree.get(), std::move(to_cast));
+                cast_map.emplace(join_tree.get(), std::move(to_cast)); // repeated table_expression can overwrite
         }
         return;
     }

src/Analyzer/Passes/HybridCastsPass.h

@@ -7,8 +7,20 @@ namespace DB
 {
 
 /// Adds CASTs for Hybrid segments when physical types differ from the Hybrid schema
-/// and reorders the SELECT list to match the schema order (needed because planner
-/// later aligns remote headers by position).
+///
+/// It normalizes headers coming from different segments when table structure in some segments
+/// differs from the Hybrid table definition. For example column X is UInt32 in the Hybrid table,
+/// but Int64 in an additional segment.
+///
+/// Without these casts ConvertingActions may fail to reconcile mismatched headers when casts are impossible
+/// (e.g. AggregateFunction states carry hashed data tied to their argument type and cannot be recast), for example:
+/// "Conversion from AggregateFunction(uniq, Decimal(38, 0)) to AggregateFunction(uniq, UInt64) is not supported"
+/// (CANNOT_CONVERT_TYPE).
+///
+/// Per-segment casts are not reliable because WithMergeState strips aliases, so merged pipelines
+/// from different segments would return different headers (with or without CAST), leading to errors
+/// like "Cannot find column `max(value)` in source stream, there are only columns: [max(_CAST(value, 'UInt64'))]"
+/// (THERE_IS_NO_COLUMN).
 class HybridCastsPass : public IQueryTreePass
 {
 public:

src/Analyzer/QueryTreePassManager.cpp

@@ -267,8 +267,6 @@ void addQueryTreePasses(QueryTreePassManager & manager, bool only_analyze)
     manager.addPass(std::make_unique<ConvertLogicalExpressionToCNFPass>());
     manager.addPass(std::make_unique<RegexpFunctionRewritePass>());
 
-    manager.addPass(std::make_unique<HybridCastsPass>());
-
     manager.addPass(std::make_unique<RewriteSumFunctionWithSumAndCountPass>());
     manager.addPass(std::make_unique<CountDistinctPass>());
     manager.addPass(std::make_unique<UniqToCountPass>());
@@ -312,6 +310,8 @@ void addQueryTreePasses(QueryTreePassManager & manager, bool only_analyze)
     manager.addPass(std::make_unique<ShardNumColumnToFunctionPass>());
 
     manager.addPass(std::make_unique<OptimizeDateOrDateTimeConverterWithPreimagePass>());
+
+    manager.addPass(std::make_unique<HybridCastsPass>());
 }
 
 }

src/Storages/StorageDistributed.cpp

@@ -950,7 +950,7 @@ QueryTreeNodePtr buildQueryTreeDistributed(SelectQueryInfo & query_info,
     const StorageSnapshotPtr & distributed_storage_snapshot,
     const StorageID & remote_storage_id,
     const ASTPtr & remote_table_function,
-    const ASTPtr & additional_filter)
+    const ASTPtr & additional_filter = nullptr)
 {
     auto & planner_context = query_info.planner_context;
     const auto & query_context = planner_context->getQueryContext();
@@ -1091,6 +1091,17 @@ void StorageDistributed::read(
         return String{"<unknown_segment>"};
     };
 
+    auto describe_base_target = [&]() -> String
+    {
+       if (remote_table_function_ptr)
+          return remote_table_function_ptr->formatForLogging();
+       if (!remote_database.empty())
+          return remote_database + "." + remote_table;
+       return remote_table;
+    };
+
+    String base_target = describe_base_target();
+
     const bool log_hybrid_query_rewrites = (!segments.empty() || base_segment_predicate);
 
     auto log_rewritten_query = [&](const String & target, const ASTPtr & ast)
@@ -1101,14 +1112,6 @@ void StorageDistributed::read(
         LOG_TRACE(log, "rewriteSelectQuery (target: {}) -> {}", target, ast->formatForLogging());
     };
 
-    String base_target;
-    if (remote_table_function_ptr)
-        base_target = remote_table_function_ptr->formatForLogging();
-    else if (!remote_database.empty())
-        base_target = remote_database + "." + remote_table;
-    else
-        base_target = remote_table;
-
     if (settings[Setting::allow_experimental_analyzer])
     {
         StorageID remote_storage_id = StorageID::createEmpty();
@@ -2231,26 +2234,31 @@ void StorageDistributed::setHybridLayout(std::vector<HybridSegment> segments_)
     segments = std::move(segments_);
     log = getLogger("Hybrid (" + getStorageID().table_name + ")");
 
-    cached_columns_to_cast = getColumnsToCast();
-    if (!cached_columns_to_cast.empty() && log->is(Poco::Message::PRIO_DEBUG))
-    {
-        std::vector<String> cols;
-        for (const auto & col : cached_columns_to_cast.getAllPhysical())
-            cols.push_back(col.name + " -> " + col.type->getName());
-        LOG_DEBUG(log, "Hybrid auto-cast will apply to: {}", fmt::join(cols, ", "));
-    }
-
     auto virtuals = createVirtuals();
     // or _segment_index?
     virtuals.addEphemeral("_table_index", std::make_shared<DataTypeUInt32>(), "Index of the table function in Hybrid (0 for main table, 1+ for additional segments)");
     setVirtuals(virtuals);
 }
 
+void StorageDistributed::setCachedColumnsToCast(ColumnsDescription columns)
+{
+    cached_columns_to_cast = std::move(columns);
+    if (!cached_columns_to_cast.empty() && log)
+    {
+        Names columns_with_types;
+        columns_with_types.reserve(cached_columns_to_cast.getAllPhysical().size());
+        for (const auto & col : cached_columns_to_cast.getAllPhysical())
+            columns_with_types.emplace_back(col.name + " " + col.type->getName());
+        LOG_DEBUG(log, "Hybrid auto-cast will apply to: [{}]", fmt::join(columns_with_types, ", "));
+    }
+}
+
 ColumnsDescription StorageDistributed::getColumnsToCast() const
 {
     return cached_columns_to_cast;
 }
 
+
 void registerStorageDistributed(StorageFactory & factory)
 {
     factory.registerStorage("Distributed", [](const StorageFactory::Arguments & args)
@@ -2421,7 +2429,7 @@ void registerStorageHybrid(StorageFactory & factory)
                 {
                     throw Exception(
                         ErrorCodes::BAD_ARGUMENTS,
-                        "Hybrid segment '{}' is missing column '{}' required by Hybrid schema",
+                        "Hybrid segment {} is missing column '{}' required by Hybrid schema",
                         segment_name, column.name);
                 }
 

src/Storages/StorageDistributed.h

@@ -58,7 +58,6 @@ class StorageDistributed final : public IStorage, WithContext
         ASTPtr table_function_ast;
         ASTPtr predicate_ast;
         std::optional<StorageID> storage_id; // For table identifiers instead of table functions
-        ColumnsDescription actual_columns;
 
         HybridSegment(ASTPtr table_function_ast_, ASTPtr predicate_ast_)
             : table_function_ast(std::move(table_function_ast_))

tests/queries/0_stateless/03643_hybrid.reference

@@ -174,28 +174,52 @@ Read predicate-filtered data with analyzer enabled and prefer localhost replica
 21	Alice	2025-08-15	100
 22	Bob	2025-08-20	200
 23	Charlie	2025-08-25	300
-Check if the subqueries were recorded in query_log
+Check if the subqueries were recorded in query_log (hybrid_table_auto_cast_columns = 0)
 Row 1:
 ──────
 type:              QueryFinish
 is_initial_query2: 1
 tbl:               ['_table_function.remote','db.test_tiered_watermark']
-qry:               SELECT * FROM test_tiered_watermark ORDER BY id DESC SETTINGS enable_analyzer = 1, prefer_localhost_replica = 0, log_queries=1, serialize_query_plan=0, log_comment = 'test_tiered_watermark', max_threads=1 FORMAT Null;
-log_comment:       test_tiered_watermark
+qry:               SELECT * FROM test_tiered_watermark ORDER BY id DESC SETTINGS enable_analyzer = 1,  hybrid_table_auto_cast_columns = 0, prefer_localhost_replica = 0, log_queries=1, serialize_query_plan=0, log_comment = 'test_tiered_watermark1', max_threads=1 FORMAT Null;
+log_comment:       test_tiered_watermark1
 
 Row 2:
 ──────
 type:              QueryFinish
 is_initial_query2: 0
 tbl:               ['db.test_tiered_watermark_after']
 qry:               SELECT `__table1`.`id` AS `id`, `__table1`.`name` AS `name`, `__table1`.`date` AS `date`, `__table1`.`value` AS `value` FROM `db`.`test_tiered_watermark_after` AS `__table1` WHERE `__table1`.`date` >= '2025-09-01' ORDER BY `__table1`.`id` DESC
-log_comment:       test_tiered_watermark
+log_comment:       test_tiered_watermark1
 
 Row 3:
 ──────
 type:              QueryFinish
 is_initial_query2: 0
 tbl:               ['db.test_tiered_watermark_before']
 qry:               SELECT `__table1`.`id` AS `id`, `__table1`.`name` AS `name`, `__table1`.`date` AS `date`, `__table1`.`value` AS `value` FROM `db`.`test_tiered_watermark_before` AS `__table1` WHERE `__table1`.`date` < '2025-09-01' ORDER BY `__table1`.`id` DESC
-log_comment:       test_tiered_watermark
+log_comment:       test_tiered_watermark1
+Check if the subqueries were recorded in query_log (hybrid_table_auto_cast_columns = 1)
+Row 1:
+──────
+type:              QueryFinish
+is_initial_query2: 1
+tbl:               ['_table_function.remote','db.test_tiered_watermark']
+qry:               SELECT * FROM test_tiered_watermark ORDER BY id DESC SETTINGS enable_analyzer = 1, hybrid_table_auto_cast_columns = 1, prefer_localhost_replica = 0, log_queries=1, serialize_query_plan=0, log_comment = 'test_tiered_watermark2', max_threads=1 FORMAT Null;
+log_comment:       test_tiered_watermark2
+
+Row 2:
+──────
+type:              QueryFinish
+is_initial_query2: 0
+tbl:               ['db.test_tiered_watermark_after']
+qry:               SELECT _CAST(`__table1`.`id`, 'UInt32') AS `id`, _CAST(`__table1`.`name`, 'String') AS `name`, `__table1`.`date` AS `date`, _CAST(`__table1`.`value`, 'UInt32') AS `value` FROM `db`.`test_tiered_watermark_after` AS `__table1` WHERE `__table1`.`date` >= '2025-09-01' ORDER BY `id` DESC
+log_comment:       test_tiered_watermark2
+
+Row 3:
+──────
+type:              QueryFinish
+is_initial_query2: 0
+tbl:               ['db.test_tiered_watermark_before']
+qry:               SELECT _CAST(`__table1`.`id`, 'UInt32') AS `id`, _CAST(`__table1`.`name`, 'String') AS `name`, `__table1`.`date` AS `date`, _CAST(`__table1`.`value`, 'UInt32') AS `value` FROM `db`.`test_tiered_watermark_before` AS `__table1` WHERE `__table1`.`date` < '2025-09-01' ORDER BY _CAST(`__table1`.`id`, 'UInt32') DESC
+log_comment:       test_tiered_watermark2
 Clean up predicate filtering tables

tests/queries/0_stateless/03643_hybrid.sql

@@ -303,19 +303,19 @@ SELECT count() FROM test_tiered_watermark_before WHERE id = 17;
 
 
 SELECT 'Read predicate-filtered data with analyzer disabled and no localhost preference';
-SELECT * FROM test_tiered_watermark ORDER BY id SETTINGS enable_analyzer = 0, prefer_localhost_replica = 0, hybrid_table_auto_cast_columns = 0;
+SELECT * FROM test_tiered_watermark ORDER BY id SETTINGS enable_analyzer = 0, prefer_localhost_replica = 0;
 SELECT 'Read predicate-filtered data with analyzer enabled and no localhost preference';
-SELECT * FROM test_tiered_watermark ORDER BY id SETTINGS enable_analyzer = 1, prefer_localhost_replica = 0, hybrid_table_auto_cast_columns = 0;
+SELECT * FROM test_tiered_watermark ORDER BY id SETTINGS enable_analyzer = 1, prefer_localhost_replica = 0;
 SELECT 'Read predicate-filtered data with analyzer disabled and prefer localhost replica';
 SELECT * FROM test_tiered_watermark ORDER BY id SETTINGS enable_analyzer = 0, prefer_localhost_replica = 1;
 SELECT 'Read predicate-filtered data with analyzer enabled and prefer localhost replica';
-SELECT * FROM test_tiered_watermark ORDER BY id SETTINGS enable_analyzer = 1, prefer_localhost_replica = 1, hybrid_table_auto_cast_columns = 0;
+SELECT * FROM test_tiered_watermark ORDER BY id SETTINGS enable_analyzer = 1, prefer_localhost_replica = 1;
 
 -- other combinations of settings work, but give a bit different content in the query_log
 -- See the problem around is_initial_query described in https://github.com/Altinity/ClickHouse/issues/1077
-SELECT 'Check if the subqueries were recorded in query_log';
+SELECT 'Check if the subqueries were recorded in query_log (hybrid_table_auto_cast_columns = 0)';
 
-SELECT * FROM test_tiered_watermark ORDER BY id DESC SETTINGS enable_analyzer = 1, prefer_localhost_replica = 0, hybrid_table_auto_cast_columns = 0, log_queries=1, serialize_query_plan=0, log_comment = 'test_tiered_watermark', max_threads=1 FORMAT Null;
+SELECT * FROM test_tiered_watermark ORDER BY id DESC SETTINGS enable_analyzer = 1,  hybrid_table_auto_cast_columns = 0, prefer_localhost_replica = 0, log_queries=1, serialize_query_plan=0, log_comment = 'test_tiered_watermark1', max_threads=1 FORMAT Null;
 SYSTEM FLUSH LOGS;
 SELECT
     type,
@@ -331,7 +331,31 @@ WHERE
     FROM system.query_log
     WHERE
         event_time > now() - 300
-        and log_comment = 'test_tiered_watermark'
+        and log_comment = 'test_tiered_watermark1'
+        and current_database = currentDatabase()
+        and query_id = initial_query_id )
+ORDER BY tbl, event_time_microseconds
+FORMAT Vertical;
+
+SELECT 'Check if the subqueries were recorded in query_log (hybrid_table_auto_cast_columns = 1)';
+
+SELECT * FROM test_tiered_watermark ORDER BY id DESC SETTINGS enable_analyzer = 1, hybrid_table_auto_cast_columns = 1, prefer_localhost_replica = 0, log_queries=1, serialize_query_plan=0, log_comment = 'test_tiered_watermark2', max_threads=1 FORMAT Null;
+SYSTEM FLUSH LOGS;
+SELECT
+    type,
+    query_id = initial_query_id AS is_initial_query2,
+    arraySort(arrayMap(x -> replaceAll(x, currentDatabase(), 'db'), tables)) as tbl,
+    replaceAll(query, currentDatabase(), 'db') as qry,
+    log_comment
+FROM system.query_log
+WHERE
+ event_time > now() - 300 AND type = 'QueryFinish' AND
+ initial_query_id IN (
+    SELECT initial_query_id
+    FROM system.query_log
+    WHERE
+        event_time > now() - 300
+        and log_comment = 'test_tiered_watermark2'
         and current_database = currentDatabase()
         and query_id = initial_query_id )
 ORDER BY tbl, event_time_microseconds

tests/queries/0_stateless/03644_hybrid_auto_cast.sql

@@ -81,5 +81,5 @@ SELECT sum(if(arrayExists(x -> (x IN (10)), categories), 1, 0)) AS x FROM test_t
 
 SELECT 'hybrid_table_auto_cast_columns = 1, enable_analyzer = 0 (analizer required)';
 SET hybrid_table_auto_cast_columns = 1, enable_analyzer = 0;
-SELECT max(value) FROM test_tiered_watermark; -- { serverError NOT_IMPLEMENTED }
-SELECT sum(if(arrayExists(x -> (x IN (10)), categories), 1, 0)) AS x FROM test_tiered_watermark; -- { serverError NOT_IMPLEMENTED }
+SELECT max(value) FROM test_tiered_watermark; -- { serverError CANNOT_CONVERT_TYPE }
+