almost working draft

Author: filimonov

src/Core/Settings.cpp

@@ -6928,6 +6928,9 @@ Allows creation of tables with the [TimeSeries](../../engines/table-engines/inte
 )", EXPERIMENTAL) \
     DECLARE(Bool, allow_experimental_hybrid_table, false, R"(
 Allows creation of tables with the [Hybrid](../../engines/table-engines/special/hybrid.md) table engine.
+)", EXPERIMENTAL) \
+    DECLARE(Bool, hybrid_table_auto_cast_columns, false, R"(
+Automatically cast columns to the schema defined in Hybrid tables when remote segments expose different physical types.
 )", EXPERIMENTAL) \
     DECLARE(Bool, allow_experimental_codecs, false, R"(
 If it is set to true, allow to specify experimental compression codecs (but we don't have those yet and this option does nothing).

src/Core/SettingsChangesHistory.cpp

@@ -54,6 +54,8 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"export_merge_tree_partition_manifest_ttl", 180, 180, "New setting."},
             {"export_merge_tree_part_file_already_exists_policy", "skip", "skip", "New setting."},
             {"iceberg_timezone_for_timestamptz", "UTC", "UTC", "New setting."}
+            {"iceberg_timezone_for_timestamptz", "UTC", "UTC", "New setting."},
+            {"hybrid_table_auto_cast_columns", false, false, "New setting to automatically cast Hybrid table columns when segments disagree on types."}
         });
         addSettingsChanges(settings_changes_history, "25.8",
         {

src/Interpreters/ClusterProxy/SelectStreamFactory.cpp

@@ -15,12 +15,14 @@
 #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>
@@ -60,6 +62,46 @@ 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(
@@ -68,7 +110,9 @@ ASTPtr rewriteSelectQuery(
     const std::string & remote_database,
     const std::string & remote_table,
     ASTPtr table_function_ptr,
-    ASTPtr additional_filter)
+    ASTPtr additional_filter,
+    const NameSet * columns_to_cast,
+    const ColumnsDescription * metadata_columns)
 {
     auto modified_query_ast = query->clone();
 
@@ -122,6 +166,9 @@ 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,10 +3,12 @@
 #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
@@ -42,7 +44,9 @@ ASTPtr rewriteSelectQuery(
     const std::string & remote_database,
     const std::string & remote_table,
     ASTPtr table_function_ptr = nullptr,
-    ASTPtr additional_filter = nullptr);
+    ASTPtr additional_filter = nullptr,
+    const NameSet * columns_to_cast = nullptr,
+    const ColumnsDescription * metadata_columns = nullptr);
 
 using ColumnsDescriptionByShardNum = std::unordered_map<UInt32, ColumnsDescription>;
 using AdditionalShardFilterGenerator = std::function<ASTPtr(uint64_t)>;