better

Author: filimonov

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

@@ -30,6 +30,10 @@ The Hybrid engine is experimental. Enable it per session (or in the user profile
 SET allow_experimental_hybrid_table = 1;
 ```
 
+### Automatic Type Alignment
+
+Hybrid segments can evolve independently, so the same logical column may use different physical types. When you set the experimental `hybrid_table_auto_cast_columns = 1` (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.
+
 ## Engine definition
 
 ```sql

src/Interpreters/ClusterProxy/SelectStreamFactory.cpp

@@ -15,14 +15,12 @@
 #include <Interpreters/Cluster.h>
 #include <Interpreters/DatabaseCatalog.h>
 #include <Interpreters/AddDefaultDatabaseVisitor.h>
-#include <Interpreters/addTypeConversionToAST.h>
 #include <Interpreters/RequiredSourceColumnsVisitor.h>
 #include <Interpreters/TranslateQualifiedNamesVisitor.h>
 #include <DataTypes/ObjectUtils.h>
 #include <Client/IConnections.h>
 #include <Parsers/ASTSelectQuery.h>
 #include <Parsers/ASTSetQuery.h>
-#include <Parsers/ASTIdentifier.h>
 #include <Processors/QueryPlan/QueryPlan.h>
 #include <Processors/QueryPlan/ReadFromRemote.h>
 #include <Processors/QueryPlan/ExpressionStep.h>
@@ -62,46 +60,6 @@ namespace FailPoints
 namespace ClusterProxy
 {
 
-namespace
-{
-void applyHybridCastsToAST(
-    ASTPtr & node,
-    const ColumnsDescription * metadata_columns,
-    const NameSet * columns_to_cast)
-{
-    if (!metadata_columns || !columns_to_cast || columns_to_cast->empty() || !node)
-        return;
-
-    if (auto * func = node->as<ASTFunction>(); func && func->name == "_CAST")
-        return;
-
-    if (auto * identifier = node->as<ASTIdentifier>())
-    {
-        String candidate = identifier->name();
-        String short_name = candidate;
-
-        auto dot_pos = candidate.rfind('.');
-        if (dot_pos != String::npos && dot_pos + 1 < candidate.size())
-            short_name = candidate.substr(dot_pos + 1);
-
-        if (columns_to_cast->contains(short_name))
-        {
-            if (auto expected_column_opt = metadata_columns->tryGetPhysical(short_name))
-            {
-                auto cast_ast = addTypeConversionToAST(node->clone(), expected_column_opt->type->getName());
-                const auto & alias = identifier->alias.empty() ? short_name : identifier->alias;
-                cast_ast->setAlias(alias);
-                node = cast_ast;
-                return;
-            }
-        }
-    }
-
-    for (auto & child : node->children)
-        applyHybridCastsToAST(child, metadata_columns, columns_to_cast);
-}
-}
-
 /// select query has database, table and table function names as AST pointers
 /// Creates a copy of query, changes database, table and table function names.
 ASTPtr rewriteSelectQuery(
@@ -110,9 +68,7 @@ ASTPtr rewriteSelectQuery(
     const std::string & remote_database,
     const std::string & remote_table,
     ASTPtr table_function_ptr,
-    ASTPtr additional_filter,
-    const NameSet * columns_to_cast,
-    const ColumnsDescription * metadata_columns)
+    ASTPtr additional_filter)
 {
     auto modified_query_ast = query->clone();
 
@@ -166,9 +122,6 @@ ASTPtr rewriteSelectQuery(
 
             RestoreQualifiedNamesVisitor(data).visit(modified_query_ast);
         }
-
-        if (columns_to_cast && !columns_to_cast->empty() && metadata_columns)
-            applyHybridCastsToAST(modified_query_ast, metadata_columns, columns_to_cast);
     }
 
     /// To make local JOIN works, default database should be added to table names.

src/Interpreters/ClusterProxy/SelectStreamFactory.h

@@ -3,12 +3,10 @@
 #include <Analyzer/IQueryTreeNode.h>
 #include <Client/ConnectionPool.h>
 #include <Core/QueryProcessingStage.h>
-#include <Core/Names.h>
 #include <Interpreters/Cluster.h>
 #include <Interpreters/StorageID.h>
 #include <Parsers/IAST_fwd.h>
 #include <Storages/IStorage_fwd.h>
-#include <Storages/ColumnsDescription.h>
 #include <Storages/StorageSnapshot.h>
 
 namespace DB
@@ -44,9 +42,7 @@ ASTPtr rewriteSelectQuery(
     const std::string & remote_database,
     const std::string & remote_table,
     ASTPtr table_function_ptr = nullptr,
-    ASTPtr additional_filter = nullptr,
-    const NameSet * columns_to_cast = nullptr,
-    const ColumnsDescription * metadata_columns = nullptr);
+    ASTPtr additional_filter = nullptr);
 
 using ColumnsDescriptionByShardNum = std::unordered_map<UInt32, ColumnsDescription>;
 using AdditionalShardFilterGenerator = std::function<ASTPtr(uint64_t)>;

src/Storages/StorageDistributed.cpp

@@ -1229,6 +1229,9 @@ void StorageDistributed::read(
     const auto * columns_to_cast_ptr = need_hybrid_casts ? &columns_to_cast_names : nullptr;
     const auto * base_actual_columns_for_cast = !base_segment_columns.empty() ? &base_segment_columns : &metadata_columns;
 
+    if (need_hybrid_casts && !settings[Setting::allow_experimental_analyzer])
+        throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Setting 'hybrid_table_auto_cast_columns' is supported only with allow_experimental_analyzer=1");
+
     auto describe_segment_target = [&](const HybridSegment & segment) -> String
     {
         if (segment.storage_id)
@@ -1319,9 +1322,7 @@ void StorageDistributed::read(
         modified_query_info.query = ClusterProxy::rewriteSelectQuery(
             local_context, modified_query_info.query,
             remote_database, remote_table, remote_table_function_ptr,
-            base_segment_predicate,
-            columns_to_cast_ptr,
-            &metadata_columns);
+            base_segment_predicate);
         log_rewritten_query(base_target, modified_query_info.query);
 
         if (!segments.empty())
@@ -1336,18 +1337,14 @@ void StorageDistributed::read(
                         local_context, additional_query_info.query,
                         segment.storage_id->database_name, segment.storage_id->table_name,
                         nullptr,
-                        segment.predicate_ast,
-                        columns_to_cast_ptr,
-                        &metadata_columns);
+                        segment.predicate_ast);
                 }
                 else
                 {
                     additional_query_info.query = ClusterProxy::rewriteSelectQuery(
                         local_context, additional_query_info.query,
                         "", "", segment.table_function_ast,
-                        segment.predicate_ast,
-                        columns_to_cast_ptr,
-                        &metadata_columns);
+                        segment.predicate_ast);
                 }
 
                 log_rewritten_query(describe_segment_target(segment), additional_query_info.query);

…eries/0_stateless/03642_hybrid.reference …eries/0_stateless/03643_hybrid.referencetests/queries/0_stateless/03642_hybrid.reference renamed to tests/queries/0_stateless/03643_hybrid.reference tests/queries/0_stateless/03642_hybrid.reference renamed to tests/queries/0_stateless/03643_hybrid.reference

@@ -1,3 +1,4 @@
+Hybrid creation requires allow_experimental_hybrid_table
 Check Hybrid engine is registered
 Hybrid
 Ensure no leftovers before validation checks

…sts/queries/0_stateless/03642_hybrid.sql …sts/queries/0_stateless/03643_hybrid.sqltests/queries/0_stateless/03642_hybrid.sql renamed to tests/queries/0_stateless/03643_hybrid.sql tests/queries/0_stateless/03642_hybrid.sql renamed to tests/queries/0_stateless/03643_hybrid.sql

@@ -346,7 +346,7 @@ DROP TABLE IF EXISTS test_tiered_watermark SYNC;
 DROP TABLE IF EXISTS test_tiered_watermark_after SYNC;
 DROP TABLE IF EXISTS test_tiered_watermark_before SYNC;
 
--- TODO:
+-- TODO: - addressed by 03644_hybrid_auto_cast.sql
 -- Code: 70. DB::Exception: Received from localhost:9000. DB::Exception: Conversion from AggregateFunction(sum, Decimal(38, 0)) to AggregateFunction(sum, UInt32) is not supported: while converting source column `sum(__table1.value)` to destination column `sum(__table1.value)`. (CANNOT_CONVERT_TYPE)
 -- SELECT sum(value) FROM test_tiered_watermark;
 

tests/queries/0_stateless/03644_hybrid_auto_cast.reference

@@ -0,0 +1,13 @@
+hybrid_table_auto_cast_columns = 0, enable_analyzer = 1 (headers mismatch)
+hybrid_table_auto_cast_columns = 0, enable_analyzer = 0 (headers mismatch)
+1
+hybrid_table_auto_cast_columns = 0, enable_analyzer = 1 manual cast
+600
+1
+hybrid_table_auto_cast_columns = 0, enable_analyzer = 0 manual cast
+600
+1
+hybrid_table_auto_cast_columns = 1, enable_analyzer = 1
+600
+1
+hybrid_table_auto_cast_columns = 1, enable_analyzer = 0 (analizer required)

tests/queries/0_stateless/03644_hybrid_auto_cast.sql

@@ -0,0 +1,85 @@
+SET allow_experimental_hybrid_table = 1,
+    prefer_localhost_replica = 0;
+
+DROP TABLE IF EXISTS test_tiered_watermark_after;
+DROP TABLE IF EXISTS test_tiered_watermark_before;
+DROP TABLE IF EXISTS test_tiered_watermark;
+
+CREATE TABLE test_tiered_watermark_after
+(
+    `id` UInt32,
+    `name` String,
+    `date` Date,
+    `value` UInt64,
+    `categories` Array(UInt32)
+)
+ENGINE = MergeTree()
+ORDER BY id;
+
+CREATE TABLE test_tiered_watermark_before
+(
+    `id` Int32,
+    `name` Nullable(String),
+    `date` Date,
+    `value` Decimal128(0),
+    `categories` Array(Int64)
+)
+ENGINE = MergeTree()
+ORDER BY id;
+
+INSERT INTO test_tiered_watermark_after VALUES
+    (11, 'Alice', '2025-08-15', 100, [100, 10]),
+    (12, 'Bob', '2025-08-20', 200, [200, 20]),
+    (13, 'Charlie', '2025-08-25', 300, [300, 30]),
+    (14, 'David', '2025-09-05', 400, [400, 40]),
+    (15, 'Eve', '2025-09-10', 500, [500, 50]),
+    (16, 'Frank', '2025-09-15', 600, [600, 60]);
+
+INSERT INTO test_tiered_watermark_before VALUES
+    (21, 'Alice', '2025-08-15', 100, [100, 10]),
+    (22, 'Bob', '2025-08-20', 200, [200, 20]),
+    (23, 'Charlie', '2025-08-25', 300, [300, 30]),
+    (24, 'David', '2025-09-05', 400, [400, 40]),
+    (25, 'Eve', '2025-09-10', 500, [500, 50]),
+    (26, 'Frank', '2025-09-15', 600, [600, 60]);
+
+CREATE TABLE test_tiered_watermark
+ENGINE = Hybrid(
+    remote('127.0.0.1:9000', currentDatabase(), 'test_tiered_watermark_after'),
+    date >= '2025-09-01',
+    remote('127.0.0.1:9000', currentDatabase(), 'test_tiered_watermark_before'),
+    date < '2025-09-01'
+);
+
+-- the problem
+SELECT 'hybrid_table_auto_cast_columns = 0, enable_analyzer = 1 (headers mismatch)';
+SET hybrid_table_auto_cast_columns = 0, enable_analyzer = 1;
+SELECT max(value) FROM test_tiered_watermark; -- { serverError CANNOT_CONVERT_TYPE }
+SELECT sum(if(arrayExists(x -> (x IN (10)), categories), 1, 0)) AS x FROM test_tiered_watermark; -- { serverError THERE_IS_NO_COLUMN }
+
+SELECT 'hybrid_table_auto_cast_columns = 0, enable_analyzer = 0 (headers mismatch)';
+SET hybrid_table_auto_cast_columns = 0, enable_analyzer = 0;
+SELECT max(value) FROM test_tiered_watermark; -- { serverError CANNOT_CONVERT_TYPE }
+SELECT sum(if(arrayExists(x -> (x IN (10)), categories), 1, 0)) AS x FROM test_tiered_watermark; -- works w/o analyzer
+
+-- workaround - explicit cast
+SELECT 'hybrid_table_auto_cast_columns = 0, enable_analyzer = 1 manual cast';
+SET hybrid_table_auto_cast_columns = 0, enable_analyzer = 1;
+SELECT max(value::UInt32) FROM test_tiered_watermark;
+SELECT sum(if(arrayExists(x -> (x IN (10)), categories::Array(UInt32)), 1, 0)) AS x FROM test_tiered_watermark;
+
+SELECT 'hybrid_table_auto_cast_columns = 0, enable_analyzer = 0 manual cast';
+SET hybrid_table_auto_cast_columns = 0, enable_analyzer = 0;
+SELECT max(value::UInt32) FROM test_tiered_watermark;
+SELECT sum(if(arrayExists(x -> (x IN (10)), categories::Array(UInt32)), 1, 0)) AS x FROM test_tiered_watermark;
+
+-- feature to add casts automatically
+SELECT 'hybrid_table_auto_cast_columns = 1, enable_analyzer = 1';
+SET hybrid_table_auto_cast_columns = 1, enable_analyzer = 1;
+SELECT max(value) FROM test_tiered_watermark;
+SELECT sum(if(arrayExists(x -> (x IN (10)), categories), 1, 0)) AS x FROM test_tiered_watermark;
+
+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 }