Fix alias error in distributed calls, simplification for nanoseconds in Iceberg

src/Interpreters/GlobalSubqueriesVisitor.cpp

@@ -54,13 +54,16 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names
 
     if (const auto * ast_table_expr = ast->as<ASTTableExpression>())
     {
-        subquery_or_table_name = ast_table_expr->subquery;
-
+        /// `interpretSubquery` reads a table function. A `GLOBAL JOIN s3(...)` is that shape.
         if (ast_table_expr->database_and_table_name)
         {
             subquery_or_table_name = ast_table_expr->database_and_table_name;
             is_table = true;
         }
+        else if (ast_table_expr->table_function)
+            subquery_or_table_name = ast_table_expr->table_function;
+        else
+            subquery_or_table_name = ast_table_expr->subquery;
     }
     else if (ast->as<ASTTableIdentifier>())
     {
@@ -87,14 +90,14 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names
     }
 
     String alias = subquery_or_table_name->tryGetAlias();
-    String external_table_name;
-    if (alias.empty())
-    {
-        auto hash = subquery_or_table_name->getTreeHash(/*ignore_aliases=*/ true);
-        external_table_name = fmt::format("_data_{}", toString(hash));
-    }
-    else
-        external_table_name = alias;
+    const auto hash = subquery_or_table_name->getTreeHash(/*ignore_aliases=*/ true);
+    const String hashed_name = fmt::format("_data_{}", toString(hash));
+    /// The alias is the SQL name. A session temporary table with that name must stay untouched,
+    /// so the broadcast is stored under `_data_{hash}` and the alias is only `AS alias`.
+    const bool alias_taken = !alias.empty()
+        && (external_tables.contains(alias)
+            || static_cast<bool>(getContext()->tryResolveStorageID(StorageID("", alias), Context::ResolveExternal)));
+    const String external_table_name = (alias.empty() || alias_taken) ? hashed_name : alias;
 
     /** We replace the subquery with the name of the temporary table.
         * It is in this form, the request will go to the remote server.
@@ -115,6 +118,7 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names
     if (auto * ast_table_expr = ast->as<ASTTableExpression>())
     {
         ast_table_expr->subquery.reset();
+        ast_table_expr->table_function.reset();
         ast_table_expr->database_and_table_name = database_and_table_name;
 
         ast_table_expr->children.clear();

src/Planner/PlannerJoinTree.cpp

@@ -1491,16 +1491,14 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres
                         auto table_function_ast = table_function_node->toAST();
                         table_function_ast->setAlias({});
 
-                        WriteBufferFromOwnString out;
-                        IAST::FormatSettings format_settings(
-                            /*one_line=*/true,
-                            IdentifierQuotingRule::WhenNecessary,
-                            IdentifierQuotingStyle::Backticks,
-                            /*show_secrets_=*/false);
-
-                        table_function_ast->format(out, format_settings);
-
-                        auto table_function_serialized_ast = std::move(out.str());
+                        /// This serialized AST only travels inside the cluster, over the native
+                        /// protocol, to build the same table function on the executing shard (see
+                        /// `ReadFromTableFunctionStep` / `resolveStorages`). It is never shown to the
+                        /// user, so keep secrets, same as the plain-SQL distributed query path
+                        /// (`IStorageCluster::readFromRemoteInitiator` / `ClusterProxy::rewriteSelectQuery`
+                        /// use `formatWithSecretsOneLine` for the same reason). Hiding them here would
+                        /// send a masked literal (e.g. a redacted S3 secret key) to the remote shard.
+                        auto table_function_serialized_ast = table_function_ast->formatWithSecretsOneLine();
 
                         auto reading_from_table_function = std::make_unique<ReadFromTableFunctionStep>(
                             sample_block,

src/Storages/IStorageCluster.cpp

@@ -42,6 +42,9 @@
 #include <Analyzer/InDepthQueryTreeVisitor.h>
 #include <Analyzer/Utils.h>
 #include <Interpreters/TreeRewriter.h>
+#include <Interpreters/DatabaseCatalog.h>
+#include <Interpreters/GlobalSubqueriesVisitor.h>
+#include <Interpreters/PreparedSets.h>
 #include <Core/Joins.h>
 #include <Parsers/ASTExpressionList.h>
 #include <Parsers/ASTFunction.h>
@@ -306,33 +309,6 @@ void stripObjectStorageRemoteInitiatorSettings(ASTPtr & query)
         select_query->setExpression(ASTSelectQuery::Expression::SETTINGS, {});
 }
 
-void rewriteASTInFunctionsToGlobalIn(ASTPtr & node)
-{
-    if (!node)
-        return;
-
-    if (auto * function = node->as<ASTFunction>())
-    {
-        if (isNameOfLocalInFunction(function->name) && function->arguments && function->arguments->children.size() >= 2)
-        {
-            const auto & rhs = function->arguments->children[1];
-            if (rhs
-                && (rhs->as<ASTSubquery>() || rhs->as<ASTSelectQuery>() || rhs->as<ASTSelectWithUnionQuery>()
-                    || rhs->as<ASTIdentifier>() || rhs->as<ASTTableIdentifier>()))
-            {
-                function->name = getGlobalInFunctionNameForLocalInFunctionName(function->name);
-            }
-        }
-    }
-
-    for (auto & child : node->children)
-    {
-        if (astIsNestedSelect(child))
-            continue;
-        rewriteASTInFunctionsToGlobalIn(child);
-    }
-}
-
 bool queryHasRightOrFullJoin(const ASTPtr & query)
 {
     if (!query)
@@ -368,6 +344,59 @@ bool queryHasRightOrFullJoin(const ASTPtr & query)
     return false;
 }
 
+bool astContainsTableIdentifier(const ASTPtr & node)
+{
+    if (!node)
+        return false;
+
+    if (node->as<ASTTableIdentifier>())
+        return true;
+
+    for (const auto & child : node->children)
+        if (astContainsTableIdentifier(child))
+            return true;
+
+    return false;
+}
+
+/// A named table in `IN` is read where the client connected. `IN (SELECT 1)` has no table and stays on the replica.
+bool inRightSideIsInitiatorLocal(const ASTPtr & rhs)
+{
+    if (!rhs)
+        return false;
+
+    if (rhs->as<ASTIdentifier>() || rhs->as<ASTTableIdentifier>())
+        return true;
+
+    if (rhs->as<ASTSubquery>() || rhs->as<ASTSelectQuery>() || rhs->as<ASTSelectWithUnionQuery>())
+        return astContainsTableIdentifier(rhs);
+
+    return false;
+}
+
+void rewriteInitiatorLocalInToGlobalIn(ASTPtr & node)
+{
+    if (!node)
+        return;
+
+    if (auto * function = node->as<ASTFunction>())
+    {
+        if (isNameOfLocalInFunction(function->name) && function->arguments && function->arguments->children.size() >= 2
+            && inRightSideIsInitiatorLocal(function->arguments->children[1]))
+            function->name = getGlobalInFunctionNameForLocalInFunctionName(function->name);
+    }
+
+    for (auto & child : node->children)
+    {
+        if (astIsNestedSelect(child))
+            continue;
+        rewriteInitiatorLocalInToGlobalIn(child);
+    }
+}
+
+/// `IN some_table` is read on the initiator. `IN (SELECT 1)` stays on the replica.
+/// `getSampleBlockImpl` sends this same tree, so the interpreter pass must not
+/// turn every subquery into `globalIn` before the broadcast pass runs.
 void rewriteASTJoinsToGlobal(ASTPtr & query)
 {
     ASTSelectQuery * select_query = query->as<ASTSelectQuery>();
@@ -399,7 +428,7 @@ void rewriteASTJoinsToGlobal(ASTPtr & query)
         }
     }
 
-    rewriteASTInFunctionsToGlobalIn(query);
+    rewriteInitiatorLocalInToGlobalIn(query);
 }
 
 }
@@ -409,6 +438,47 @@ void IStorageCluster::rewriteASTForGlobalJoin(ASTPtr & query)
     rewriteASTJoinsToGlobal(query);
 }
 
+/// The initiator interpreter usually does this already: `IStorageCluster::isRemote` is true, so
+/// `GlobalSubqueriesVisitor` replaces a `GLOBAL` right side with an external table. This pass
+/// covers a query that still names a local table when `convertToRemote` runs. A second pass
+/// is a no-op once that external table is in the context. Sets are added to the interpreter's
+/// `prepared_sets`, which `DelayedCreatingSetsStep` builds before the rows are sent.
+namespace
+{
+void broadcastOldAnalyzerGlobalJoin(ASTPtr & query_to_send, const SelectQueryInfo & query_info, const ContextPtr & context)
+{
+    /// Idempotent with `rewriteASTForGlobalJoin`: `IN (SELECT 1)` is still local here.
+    rewriteASTJoinsToGlobal(query_to_send);
+
+    auto prepared_sets = query_info.prepared_sets ? query_info.prepared_sets : std::make_shared<PreparedSets>();
+    TemporaryTablesMapping external_tables;
+    bool has_global_subqueries = false;
+
+    /// `table_join` stays null: the right side is read with all of its columns. Passing the
+    /// analyzed join here would rebuild its actions against an AST the first pass may already
+    /// have rewritten to an external table.
+    GlobalSubqueriesVisitor::Data data(
+        context,
+        /* subquery_depth */ 0,
+        /* is_remote */ true,
+        /* is_explain */ false,
+        external_tables,
+        prepared_sets,
+        has_global_subqueries,
+        /* table_join */ nullptr);
+    GlobalSubqueriesVisitor(data).visit(query_to_send);
+
+    /// The read context is const. External tables live on the query context, and a later
+    /// `Context::createCopy` still sees them through `getExternalTables`.
+    auto mutable_context = context->getQueryContext();
+    for (auto & [name, holder] : external_tables)
+    {
+        if (!mutable_context->tryResolveStorageID(StorageID{"", name}, Context::ResolveExternal))
+            mutable_context->addExternalTable(name, std::move(*holder));
+    }
+}
+}
+
 /*
 Try to make subquery to send on nodes
 Converts
@@ -455,14 +525,18 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded(
     }
     case ObjectStorageClusterJoinMode::GLOBAL:
     {
-        /// Old analyzer: ship the left scan only. `INNER`/`LEFT`/`CROSS` stay in `query_to_send`.
+        /// Old analyzer. `RIGHT`/`FULL`/`PASTE` stay on the initiator. `INNER`/`LEFT`/`CROSS`
+        /// stay in `query_to_send`, with the right side replaced by an external table.
         if (!query_info.query_tree)
         {
             if (queryHasRightOrFullJoin(query_to_send))
             {
                 auto info = getQueryJoinInfoFromAST(query_to_send);
                 rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context);
+                return;
             }
+
+            broadcastOldAnalyzerGlobalJoin(query_to_send, query_info, context);
             return;
         }
 
@@ -853,6 +927,12 @@ IStorageCluster::RemoteCallVariables IStorageCluster::convertToRemote(
     if (!table_expression->table_function)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table function in table expression");
 
+    /// `AS alias` is stored on the inner table function. `JoinedTables` reads only the outer
+    /// table function, so `remote()` must carry the same alias. Leave it on the inner function:
+    /// `addTableFunction` sends that node to the remote initiator and does not copy a non-empty
+    /// outer alias onto the replacement.
+    const String table_alias = table_expression->table_function->tryGetAlias();
+
     boost::intrusive_ptr<ASTFunction> remote_query;
 
     if (shard_addresses[0].user_specified)
@@ -868,6 +948,11 @@ IStorageCluster::RemoteCallVariables IStorageCluster::convertToRemote(
         remote_query = makeASTFunction(remote_function_name, make_intrusive<ASTLiteral>(host_name), table_expression->table_function);
     }
 
+    if (!table_alias.empty())
+    {
+        remote_query->setAlias(table_alias);
+    }
+
     table_expression->table_function = remote_query;
 
     auto remote_function = TableFunctionFactory::instance().get(remote_query, new_context);

src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp

@@ -3,6 +3,8 @@
 #include <Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h>
 #include <Core/TypeId.h>
+#include <DataTypes/DataTypeNullable.h>
+#include <DataTypes/DataTypesDecimal.h>
 #include <IO/ReadHelpers.h>
 #include <IO/WriteHelpers.h>
 #include <Core/Field.h>
@@ -155,10 +157,21 @@ DataFileMetaInfo::DataFileMetaInfo(
                 String right_str;
                 if (i_bounds->second.first.tryGet(left_str) && i_bounds->second.second.tryGet(right_str))
                 {
-                    auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true);
-                    auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false);
-                    if (left && right && !DB::Range::less(*right, *left))
-                        hyperrectangle = DB::Range(*left, true, *right, true);
+                    const auto non_nullable_type = removeNullable(type);
+                    if (WhichDataType(non_nullable_type).isDateTime64() && getDecimalScale(*non_nullable_type) == 6)
+                    {
+                        /// Two ranges are the middle band, so this column is not a constant.
+                        const auto timestamp_ranges = Iceberg::deserializeDateTime64MinMaxRanges(left_str, right_str, type);
+                        if (timestamp_ranges.size() == 1)
+                            hyperrectangle = DB::Range(timestamp_ranges[0].lower, true, timestamp_ranges[0].upper, true);
+                    }
+                    else
+                    {
+                        auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true);
+                        auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false);
+                        if (left && right && !DB::Range::less(*right, *left))
+                            hyperrectangle = DB::Range(*left, true, *right, true);
+                    }
                 }
             }
         }