diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index d0725772d8aa..ed0260f3074d 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -2138,6 +2138,15 @@ Possible values: - `local` — Replaces the database and table in the subquery with local ones for the destination server (shard), leaving the normal `IN`/`JOIN.` - `global` — Replaces the `IN`/`JOIN` query with `GLOBAL IN`/`GLOBAL JOIN.` Right table executes first and is added to the secondary query as temporay table. - `allow` — Default value. Allows the use of these types of subqueries. +- `distributed` — Experimental. Dispatches a whole `JOIN` query to the cluster of its driving table, so the `JOIN` and any `GROUP BY` run on the cluster's nodes instead of on the initiator, which then only merges the partial aggregate states. The driving table is found by walking down the left side of the query -- through `INNER ALL`/`LEFT JOIN`s and through a subquery or CTE that does not itself aggregate, deduplicate, sort or limit -- and must be a `DataLake` catalog table distributed via `object_storage_cluster`. + + This is attempted only for the outermost `SELECT` of an initial query, and it is all-or-nothing: either the entire query is eligible and is dispatched as one unit, or ordinary planning handles the entire query. No narrower, nested candidate is attempted, and nothing falls back per level. + + Eligibility requires that every other table reachable anywhere in the query also resolves through a `DataLake` catalog, has no row-level security policy, and is readable by the current user. An explicit `*Cluster()` table function, an ordinary local or `Distributed` table, or any table failing one of those checks makes the whole query ineligible. The combination is disabled outright when `additional_table_filters` is set or `object_storage_remote_initiator` is enabled. + + Performance characteristic to be aware of: only the driving table is partitioned across the cluster. Every other table in the query is read and recomputed **in full on every node**, including any `JOIN` or `GROUP BY` over them. That is a win when those relations are small relative to the driver and a loss when they are not; ClickHouse does not estimate this cost when deciding to dispatch. + + ClickHouse also does not verify that the tables, dictionaries or user-defined functions the query references are configured identically on every node of the driver's cluster -- as with `Distributed`, that consistency is the deployment's responsibility. )", 0) \ \ DECLARE(UInt64, max_concurrent_queries_for_all_users, 0, R"( @@ -8348,6 +8357,18 @@ Trigger processor to spill data into external storage adpatively. grace join is DECLARE(String, object_storage_cluster, "", R"( Cluster to make distributed requests to object storages with alternative syntax. )", EXPERIMENTAL) \ + DECLARE(String, object_storage_distributed_driver_database, "", R"( +Internal. Set by the initiator on a query dispatched by `object_storage_cluster_join_mode='distributed'`, naming the +database of the one table whose files are distributed across the cluster. Together with +`object_storage_distributed_driver_table` it tells a worker which table reads from the initiator's file-task queue; +every other table in the same query is read in full, locally. Not meant to be set by hand. + +`IMPORTANT` because a worker that silently ignored it would find no driver, read every file of every table, and +return each row once per node. A server too old to know the setting must refuse the query instead. +)", EXPERIMENTAL | IMPORTANT) \ + DECLARE(String, object_storage_distributed_driver_table, "", R"( +Internal. The table name counterpart of `object_storage_distributed_driver_database`. Not meant to be set by hand. +)", EXPERIMENTAL | IMPORTANT) \ DECLARE(UInt64, object_storage_max_nodes, 0, R"( Limit for hosts used for request in object storage cluster table functions - azureBlobStorageCluster, s3Cluster, hdfsCluster, etc. Possible values: diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index f093c4054cf6..0cf3b8841ae1 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -42,6 +42,8 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya", { {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."}, + {"object_storage_distributed_driver_database", "", "", "New internal setting. Names the database of the driving table of an `object_storage_cluster_join_mode='distributed'` dispatch."}, + {"object_storage_distributed_driver_table", "", "", "New internal setting. Names the driving table of an `object_storage_cluster_join_mode='distributed'` dispatch."}, }); addSettingsChanges(settings_changes_history, "26.6", diff --git a/src/Core/SettingsEnums.cpp b/src/Core/SettingsEnums.cpp index e84926119387..d3c1814afbc3 100644 --- a/src/Core/SettingsEnums.cpp +++ b/src/Core/SettingsEnums.cpp @@ -100,9 +100,10 @@ IMPLEMENT_SETTING_ENUM(DistributedProductMode, ErrorCodes::UNKNOWN_DISTRIBUTED_P {"allow", DistributedProductMode::ALLOW}}) IMPLEMENT_SETTING_ENUM(ObjectStorageClusterJoinMode, ErrorCodes::BAD_ARGUMENTS, - {{"local", ObjectStorageClusterJoinMode::LOCAL}, - {"global", ObjectStorageClusterJoinMode::GLOBAL}, - {"allow", ObjectStorageClusterJoinMode::ALLOW}}) + {{"local", ObjectStorageClusterJoinMode::LOCAL}, + {"global", ObjectStorageClusterJoinMode::GLOBAL}, + {"allow", ObjectStorageClusterJoinMode::ALLOW}, + {"distributed", ObjectStorageClusterJoinMode::DISTRIBUTED}}) IMPLEMENT_SETTING_ENUM(QueryResultCacheNondeterministicFunctionHandling, ErrorCodes::BAD_ARGUMENTS, diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h index a17f433ba6ec..64c1bb5e72f2 100644 --- a/src/Core/SettingsEnums.h +++ b/src/Core/SettingsEnums.h @@ -170,7 +170,8 @@ enum class ObjectStorageClusterJoinMode : uint8_t { LOCAL, /// Convert to local query GLOBAL, /// Convert to global query - ALLOW /// Enable + ALLOW, /// Enable + DISTRIBUTED /// Let a distributed object-storage driver own the whole JOIN and run it on cluster workers }; DECLARE_SETTING_ENUM(ObjectStorageClusterJoinMode) diff --git a/src/Planner/Planner.cpp b/src/Planner/Planner.cpp index faf2e42ea909..9d1fbb373541 100644 --- a/src/Planner/Planner.cpp +++ b/src/Planner/Planner.cpp @@ -82,6 +82,8 @@ #include #include #include +#include +#include #include #include #include @@ -2302,7 +2304,31 @@ void Planner::buildPlanForQueryNode() } JoinTreeQueryPlan join_tree_query_plan; - if (planner_context->getMutableQueryContext()->canUseTaskBasedParallelReplicas() + /// object_storage_cluster_join_mode='distributed': only the outermost, initial-query Planner may dispatch. + /// All three guards below are load-bearing; each was added after a live failure: + /// - only_analyze: buildDistributedObjectStorageQueryPlan calls getSampleBlock, which starts its own + /// analyze-only Planner over the same query. Without this guard that Planner dispatches again, and so on + /// -- `Code: 306. TOO_DEEP_RECURSION`. + /// - is_subquery: a nested query dispatched on its own gets its `__tableN` identifiers renumbered locally, + /// which no longer match what the enclosing scope resolved against -- `Not found column + /// __table7.appinfo_ccl in block. There are only columns: __table2.appinfo_ccl, ...`. + /// - INITIAL_QUERY: a worker must execute what it was sent, not dispatch it onwards. + /// Whole-or-nothing for the outermost query: on rejection, ordinary planning handles everything. + std::optional distributed_object_storage_candidate; + if (!select_query_options.only_analyze && !select_query_options.is_subquery + && query_context->getClientInfo().query_kind == ClientInfo::QueryKind::INITIAL_QUERY) + distributed_object_storage_candidate = findDistributedObjectStorageCandidate(query_tree, query_context); + + /// Dispatch can still decline here: the driver has to be nameable unambiguously in the serialized query. + std::optional dispatched_query_plan; + if (distributed_object_storage_candidate) + dispatched_query_plan = buildDistributedObjectStorageQueryPlan(query_tree, *distributed_object_storage_candidate, select_query_info, planner_context); + + if (dispatched_query_plan) + { + join_tree_query_plan = std::move(*dispatched_query_plan); + } + else if (planner_context->getMutableQueryContext()->canUseTaskBasedParallelReplicas() && planner_context->getGlobalPlannerContext()->parallel_replicas_node == &query_node) { join_tree_query_plan = buildQueryPlanForParallelReplicas(query_node, planner_context, select_query_info.storage_limits); diff --git a/src/Planner/buildDistributedObjectStorageQueryPlan.cpp b/src/Planner/buildDistributedObjectStorageQueryPlan.cpp new file mode 100644 index 000000000000..5b57e393c978 --- /dev/null +++ b/src/Planner/buildDistributedObjectStorageQueryPlan.cpp @@ -0,0 +1,215 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace DB +{ +namespace ErrorCodes +{ + extern const int LOGICAL_ERROR; +} + +namespace +{ + +/// Counts how many times `storage_id` is named as a table in `ast`. The count is taken on the serialized query -- +/// the text a worker actually parses -- not on the query tree, because serialization inlines CTE bodies: a CTE +/// referenced twice is one node in the tree but two table references in the SQL. +size_t countTableReferences(const ASTPtr & ast, const StorageID & storage_id) +{ + if (!ast) + return 0; + + size_t count = 0; + if (const auto * identifier = ast->as()) + { + const auto referenced = identifier->getTableId(); + if (referenced.table_name == storage_id.table_name && referenced.database_name == storage_id.database_name) + ++count; + } + + for (const auto & child : ast->children) + count += countTableReferences(child, storage_id); + + return count; +} + + +/// Splits a condition into its top-level `and` operands. Anything that is not an `and` is one atom. +void collectConjunctionAtoms(const QueryTreeNodePtr & node, QueryTreeNodes & atoms) +{ + if (const auto * function_node = node->as(); function_node && function_node->getFunctionName() == "and") + { + for (const auto & argument : function_node->getArguments().getNodes()) + collectConjunctionAtoms(argument, atoms); + return; + } + + atoms.push_back(node); +} + +/// True when every column this condition reads comes from `driver` and nothing in it has to be executed to be +/// understood. A subquery is rejected outright: this condition is evaluated while listing the driver's files, +/// long before there is a pipeline to run one in. +bool readsOnlyDriverColumns(const QueryTreeNodePtr & node, const TableNode * driver) +{ + if (node->as() || node->as()) + return false; + + if (const auto * column_node = node->as()) + return column_node->getColumnSource().get() == driver; + + for (const auto & child : node->getChildren()) + if (child && !readsOnlyDriverColumns(child, driver)) + return false; + + return true; +} + +/// The predicate over the driver's own columns, as an ActionsDAG the file listing can prune with. +/// +/// Correctness rests on two restrictions `findDistributedObjectStorageCandidate` already enforces, and would +/// break if either were relaxed: the driver sits on the left spine of INNER ALL / LEFT joins only, so a driver +/// row dropped here cannot have produced a result row; and every QueryNode crossed to reach it is +/// partition-preserving (`isSafeIntermediateSubquery` -- no GROUP BY, DISTINCT, LIMIT, window), so dropping a +/// row cannot change what the surviving rows compute. +/// +/// Returns nothing when no atom qualifies, which simply means every file is listed. +std::optional buildDriverOnlyFilter( + const DistributedObjectStorageCandidate & candidate, const PlannerContextPtr & planner_context) +{ + QueryTreeNodes atoms; + for (const auto * query_node : candidate.query_nodes_on_driver_path) + { + if (query_node->hasPrewhere()) + collectConjunctionAtoms(query_node->getPrewhere(), atoms); + if (query_node->hasWhere()) + collectConjunctionAtoms(query_node->getWhere(), atoms); + } + + QueryTreeNodes driver_atoms; + for (const auto & atom : atoms) + if (readsOnlyDriverColumns(atom, candidate.driver)) + driver_atoms.push_back(atom->clone()); + + if (driver_atoms.empty()) + return {}; + + const auto context = planner_context->getQueryContext(); + /// mergeConditionNodes always builds an `and`, which needs at least two arguments. + auto condition = driver_atoms.size() == 1 ? driver_atoms.front() : mergeConditionNodes(driver_atoms, context); + + /// Passed explicitly so buildFilterInfo does not go looking for this table expression in the planner + /// context: the dispatch boundary is planned as one unit and never registers the driver on its own. + const auto driver_columns = candidate.driver->getStorageSnapshot()->metadata->getColumns().getNamesOfPhysical(); + NameSet required_names(driver_columns.begin(), driver_columns.end()); + + auto mutable_planner_context = planner_context; + auto filter_info = buildFilterInfo( + std::move(condition), candidate.driver_table_expression, mutable_planner_context, std::move(required_names)); + + return std::move(filter_info.actions); +} + +} + +/// This mirrors buildQueryPlanForParallelReplicas (Planner/findParallelReplicasQuery.cpp) step for step: +/// header of the query -> serialize to SQL -> remote read -> convert the remote header back by position. +/// Keep the two in sync. +/// +/// Unlike parallel replicas, the query sent is the one the user wrote: no table expression is rewritten. Which +/// table drives the dispatch travels beside the query, in `object_storage_distributed_driver_database`/`_table`, +/// and a worker reads exactly that one table from the initiator's file-task queue. +std::optional buildDistributedObjectStorageQueryPlan( + const QueryTreeNodePtr & dispatch_boundary_node, + const DistributedObjectStorageCandidate & candidate, + const SelectQueryInfo & select_query_info, + const PlannerContextPtr & planner_context) +{ + const auto context = planner_context->getQueryContext(); + constexpr auto processed_stage = QueryProcessingStage::WithMergeableState; + + auto * driver_storage = candidate.driver_storage; + const auto & driver_storage_snapshot = candidate.driver->getStorageSnapshot(); + const auto driver_storage_id = candidate.driver->getStorageID(); + + /// The header the query produces at this stage, on the initiator and on every worker alike -- the same tree + /// serves both, so nothing has to be reconciled by name afterwards. + auto [remote_header, new_planner_context] = InterpreterSelectQueryAnalyzer::getSampleBlockAndPlannerContext( + dispatch_boundary_node->clone(), context, SelectQueryOptions(processed_stage).analyze()); + + /// Strip grouping-function specializations in a clone: the workers re-resolve the generic function + /// themselves, but the tree the header came from must keep them. + auto query_tree_for_ast = dispatch_boundary_node->clone(); + removeGroupingFunctionSpecializations(query_tree_for_ast); + ASTPtr query_to_send = queryNodeToDistributedSelectQuery(query_tree_for_ast); + + if (!query_to_send->as()) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Distributed object-storage dispatch: expected a plain SELECT at the dispatch boundary"); + + /// The driver is named by database and table, so it must be unambiguous in the query a worker receives. A + /// self-join, or a CTE over the driver referenced more than once, would leave a worker unable to tell which + /// occurrence owns the file-task queue -- and reading the queue twice is silently wrong, not an error. + /// Fall back to ordinary planning instead. + if (countTableReferences(query_to_send, driver_storage_id) != 1) + return {}; + + /// Travels to the workers with the query. `ReadFromCluster::updateSettings` copies from this context. + auto dispatch_context = Context::createCopy(context); + dispatch_context->setSetting("object_storage_distributed_driver_database", driver_storage_id.getDatabaseName()); + dispatch_context->setSetting("object_storage_distributed_driver_table", driver_storage_id.getTableName()); + + SelectQueryInfo query_info = select_query_info; + query_info.query = query_to_send; + query_info.query_tree = dispatch_boundary_node; + query_info.planner_context = new_planner_context; + + JoinTreeQueryPlan result; + result.stage = processed_stage; + + /// Prunes the driver's file listing. Without it every file of the driver is handed out and each worker + /// opens the ones its partitions cannot match only to discard them. + std::shared_ptr driver_filter; + if (auto filter_dag = buildDriverOnlyFilter(candidate, planner_context)) + { + VirtualColumnUtils::buildSetsForDAGExcludingGlobalIn(*filter_dag, context); + driver_filter = std::make_shared(std::move(*filter_dag)); + } + + driver_storage->readPreparedClusterQuery( + result.query_plan, + driver_storage_snapshot, + query_info, + dispatch_context, + processed_stage, + query_to_send, + remote_header, + std::move(driver_filter)); + + /// No converting step, unlike buildQueryPlanForParallelReplicas: that one rewrites the table expression + /// before serializing, so its two headers are built from different trees and can diverge. Here the query + /// is sent as written and `readPreparedClusterQuery` is given `remote_header` as the source step's own + /// output header, so the plan's current header is that same block. + + return result; +} + +} diff --git a/src/Planner/buildDistributedObjectStorageQueryPlan.h b/src/Planner/buildDistributedObjectStorageQueryPlan.h new file mode 100644 index 000000000000..0e1f31344ab8 --- /dev/null +++ b/src/Planner/buildDistributedObjectStorageQueryPlan.h @@ -0,0 +1,32 @@ +#pragma once + +#include +#include + +namespace DB +{ + +class PlannerContext; +using PlannerContextPtr = std::shared_ptr; +struct SelectQueryInfo; + +/// Builds the whole-query dispatch plan for `candidate`: serializes the query unchanged, announces which table +/// drives it, and reads the result back through a single ReadFromCluster step at WithMergeableState, so the +/// caller's normal finalization (MergingAggregated and the rest) applies on top. +/// +/// No table expression is rewritten. The driving table is named to the workers by database and table, through +/// `object_storage_distributed_driver_database`/`_table`, and a worker reads that one table from the initiator's +/// file-task queue while reading every other table in full, locally. +/// +/// Returns nullopt when the driver cannot be named unambiguously in the serialized query -- it must appear +/// exactly once. The caller then falls back to ordinary planning. +/// +/// Structurally the same as buildQueryPlanForParallelReplicas in Planner/findParallelReplicasQuery.cpp, +/// including the position-based conversion back to the query's header. +std::optional buildDistributedObjectStorageQueryPlan( + const QueryTreeNodePtr & dispatch_boundary_node, + const DistributedObjectStorageCandidate & candidate, + const SelectQueryInfo & select_query_info, + const PlannerContextPtr & planner_context); + +} diff --git a/src/Planner/findDistributedObjectStorageCandidate.cpp b/src/Planner/findDistributedObjectStorageCandidate.cpp new file mode 100644 index 000000000000..92a3c3606564 --- /dev/null +++ b/src/Planner/findDistributedObjectStorageCandidate.cpp @@ -0,0 +1,266 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace DB +{ +namespace Setting +{ + extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; + extern const SettingsBool object_storage_remote_initiator; + extern const SettingsMap additional_table_filters; +} + +namespace +{ + +/// Dispatch drops the initiator's row policies: every table in the dispatched query, driver included, is +/// re-resolved independently by each worker, and a policy the initiator would have applied does not travel +/// with the query text. Reject. +bool hasEffectiveRowPolicy(const TableNode & table_node, const ContextPtr & context) +{ + const auto & storage_id = table_node.getStorageID(); + auto filter = context->getRowPolicyFilter( + storage_id.getDatabaseName(), storage_id.getTableName(), RowPolicyFilterType::SELECT_FILTER); + return filter && !filter->isAlwaysTrue(); +} + +/// Dispatch replaces the per-table access check that `prepareBuildQueryPlanForTableExpression` would have run on +/// each table, so do it here instead. Table-level and non-throwing: a failure falls back to ordinary planning, +/// which then enforces the real column-level rules with its own error. +bool hasSelectAccess(const TableNode & table_node, const ContextPtr & context) +{ + const auto & storage_id = table_node.getStorageID(); + return context->getAccess()->isGranted( + AccessType::SELECT, storage_id.getDatabaseName(), storage_id.getTableName()); +} + +/// Safe to include in the dispatch, as driver or as partner: resolved through a DataLake catalog, so every worker +/// re-resolves it identically; no row policy; visible to the user. +bool isSafeDataLakeLeaf(const TableNode & table_node, const ContextPtr & context) +{ + const auto & storage_id = table_node.getStorageID(); + if (!storage_id.hasDatabase()) + return false; + + if (!dynamic_cast(table_node.getStorage().get())) + return false; + + if (!DatabaseCatalog::instance().isDatalakeCatalog(storage_id.getDatabaseName())) + return false; + + return !hasEffectiveRowPolicy(table_node, context) && hasSelectAccess(table_node, context); +} + +bool isEligibleDriver(const TableNode & table_node, const ContextPtr & context, IStorageCluster *& out_storage) +{ + if (!isSafeDataLakeLeaf(table_node, context)) + return false; + + auto * storage = dynamic_cast(table_node.getStorage().get()); + if (storage->getClusterName(context).empty()) + return false; + + out_storage = storage; + return true; +} + +/// Aggregate/window function in this node's own subtree, not crossing into a nested query. Catches +/// `SELECT count() FROM driver`, which has no GROUP BY node at all. +bool containsAggregateOrWindowFunction(const QueryTreeNodePtr & node) +{ + if (!node) + return false; + + if (const auto * function_node = node->as()) + if (function_node->isAggregateFunction() || function_node->isWindowFunction()) + return true; + + if (node->as() || node->as()) + return false; + + for (const auto & child : node->getChildren()) + if (containsAggregateOrWindowFunction(child)) + return true; + + return false; +} + +/// A subquery crossed on the way down to the driver runs independently on each worker's own partition of the +/// driver, so anything that finalizes across rows (aggregation, DISTINCT, LIMIT, ...) would turn a partial result +/// into a final one whenever a group spans two workers. Reject rather than prove which ones are partition-safe. +/// Does not apply to a JOIN partner's subquery, which every worker recomputes in full. +bool isSafeIntermediateSubquery(const QueryNode & query_node) +{ + return !query_node.isDistinct() && !query_node.hasGroupBy() && !query_node.hasHaving() && !query_node.hasWindow() + && !query_node.hasQualify() && !query_node.hasOrderBy() && !query_node.hasInterpolate() && !query_node.hasLimitBy() + && !query_node.hasLimit() && !query_node.hasOffset() && !containsAggregateOrWindowFunction(query_node.getProjectionNode()) + && !containsAggregateOrWindowFunction(query_node.getWithNode()); +} + +struct DriverPathResult +{ + bool unusable = false; + const TableNode * driver = nullptr; + QueryTreeNodePtr driver_table_expression; + IStorageCluster * driver_storage = nullptr; + + /// Innermost first; the caller reverses and prepends the dispatch boundary. + std::vector query_nodes_on_driver_path; + + /// No JOIN on the path means there is nothing here for this mode to optimize; stock `IStorageCluster::read` + /// already handles a plain single-table cluster read. + bool has_join = false; +}; + +DriverPathResult unusableDriverPath() +{ + DriverPathResult result; + result.unusable = true; + return result; +} + +/// Walks strictly down the left spine for exactly one driver. The right side of a JOIN is never inspected here -- +/// it is validated separately as worker-local content by `allWorkerLocalTableReferencesAreSafe`, which is why a +/// DataLake table (or a whole nested JOIN/GROUP BY) on the right never competes for the driver role. +DriverPathResult findDriverOnLeftSpine(const QueryTreeNodePtr & node, const ContextPtr & context) +{ + if (const auto * table_node = node->as()) + { + IStorageCluster * storage = nullptr; + if (!isEligibleDriver(*table_node, context, storage)) + return unusableDriverPath(); + + DriverPathResult result; + result.driver = table_node; + result.driver_table_expression = node; + result.driver_storage = storage; + return result; + } + + if (const auto * query_node = node->as()) + { + const auto & join_tree = query_node->getJoinTree(); + if (!join_tree) + return unusableDriverPath(); + + auto result = findDriverOnLeftSpine(join_tree, context); + if (result.unusable) + return result; + + /// Crossed as an intermediate subquery, not as the dispatch boundary (that is the node originally passed + /// to findDistributedObjectStorageCandidate). + if (!isSafeIntermediateSubquery(*query_node)) + return unusableDriverPath(); + + result.query_nodes_on_driver_path.push_back(query_node); + return result; + } + + if (const auto * join_node = node->as()) + { + const auto join_kind = join_node->getKind(); + const auto join_strictness = join_node->getStrictness(); + if ((join_kind != JoinKind::Inner || join_strictness != JoinStrictness::All) && join_kind != JoinKind::Left) + return unusableDriverPath(); + + auto left = findDriverOnLeftSpine(join_node->getLeftTableExpression(), context); + if (left.unusable) + return unusableDriverPath(); + + left.has_join = true; + return left; + } + + return unusableDriverPath(); +} + +/// Everything else reachable from the dispatch boundary is recomputed in full on every worker, so every table it +/// reaches must re-resolve identically there. No structural restrictions apply here (unlike the driver's own +/// path), because none of this content is partitioned. +/// +/// Proves this for table references only. Ordinary FunctionNodes are accepted unexamined, so anything the query +/// calls that is server-local or externally backed -- `hostName`, a dictionary via `dictGet`, a user-defined +/// function -- moves from the initiator to the workers and must be present and consistent across the cluster. +/// Same assumption `Distributed` makes; stated in the setting's own documentation. +bool allWorkerLocalTableReferencesAreSafe(const QueryTreeNodePtr & node, const TableNode * driver, const ContextPtr & context) +{ + if (!node) + return true; + + if (const auto * table_node = node->as()) + return table_node == driver || isSafeDataLakeLeaf(*table_node, context); + + if (node->as()) + return false; + + for (const auto & child : node->getChildren()) + if (!allWorkerLocalTableReferencesAreSafe(child, driver, context)) + return false; + + return true; +} + +} + +std::optional findDistributedObjectStorageCandidate( + const QueryTreeNodePtr & query_node, const ContextPtr & context) +{ + if (context->getSettingsRef()[Setting::object_storage_cluster_join_mode] != ObjectStorageClusterJoinMode::DISTRIBUTED) + return {}; + + /// Dispatch goes straight to the driver's cluster, bypassing the remote-initiator topology that + /// `IStorageCluster::read` would apply via convertToRemote. + if (context->getSettingsRef()[Setting::object_storage_remote_initiator]) + return {}; + + /// additional_table_filters is keyed by the initiator's current_database and the query's own naming, both of + /// which shift once the query is serialized for remote execution. Planner.cpp disables parallel replicas for + /// the same reason. + if (!context->getSettingsRef()[Setting::additional_table_filters].value.empty()) + return {}; + + const auto * query_node_typed = query_node->as(); + if (!query_node_typed) + return {}; + + const auto & join_tree = query_node_typed->getJoinTree(); + if (!join_tree) + return {}; + + auto driver_path = findDriverOnLeftSpine(join_tree, context); + if (driver_path.unusable || !driver_path.driver || !driver_path.has_join) + return {}; + + if (!allWorkerLocalTableReferencesAreSafe(query_node, driver_path.driver, context)) + return {}; + + DistributedObjectStorageCandidate candidate; + candidate.driver = driver_path.driver; + candidate.driver_table_expression = driver_path.driver_table_expression; + candidate.driver_storage = driver_path.driver_storage; + + candidate.query_nodes_on_driver_path.push_back(query_node_typed); + candidate.query_nodes_on_driver_path.insert( + candidate.query_nodes_on_driver_path.end(), + driver_path.query_nodes_on_driver_path.rbegin(), + driver_path.query_nodes_on_driver_path.rend()); + + return candidate; +} + +} diff --git a/src/Planner/findDistributedObjectStorageCandidate.h b/src/Planner/findDistributedObjectStorageCandidate.h new file mode 100644 index 000000000000..ab12d2b0ec1c --- /dev/null +++ b/src/Planner/findDistributedObjectStorageCandidate.h @@ -0,0 +1,56 @@ +#pragma once +#include +#include +#include +#include + +namespace DB +{ + +class TableNode; +class QueryNode; +class IStorageCluster; + +class IQueryTreeNode; +using QueryTreeNodePtr = std::shared_ptr; + +class Context; +using ContextPtr = std::shared_ptr; + +/// A whole-query JOIN-pushdown candidate for `object_storage_cluster_join_mode='distributed'`. The entire query +/// passed to findDistributedObjectStorageCandidate is dispatched to `driver`'s cluster as one unit. +struct DistributedObjectStorageCandidate +{ + /// The driving table, reachable from the dispatch boundary only via the left path of every JOIN/subquery + /// crossing. + const TableNode * driver = nullptr; + + /// The same node, kept as a pointer the planner can hand to APIs that take a table expression. + QueryTreeNodePtr driver_table_expression; + + IStorageCluster * driver_storage = nullptr; + + /// Every QueryNode crossed on the way down to the driver, outermost first, including the dispatch + /// boundary itself. Their `WHERE`/`PREWHERE` are the only places a predicate over the driver's own + /// columns can appear such that a driver row failing it cannot reach the result -- which is what makes + /// it safe to prune the driver's files by it. See buildDistributedObjectStorageQueryPlan. + std::vector query_nodes_on_driver_path; +}; + +/// Decides whether `query_node` as a whole can be executed on a single DataLake-catalog driver's cluster. +/// +/// The driver is found by walking strictly down the left spine: QueryNode -> its join tree; INNER ALL or LEFT +/// JOIN -> left operand only; an intermediate QueryNode -> only if partition-preserving; TableNode -> eligible if +/// it resolves through a DataLake catalog and has a non-empty cluster. Everything else reachable from +/// `query_node` is then checked as worker-local content that each worker recomputes in full. That check covers +/// table references only -- other functions the query calls (`dictGet`, a UDF, `hostName`) simply move to the +/// workers and are assumed to be consistent there, as they are for `Distributed`. +/// +/// All-or-nothing for `query_node` itself -- there is no narrower fallback candidate. Returns nullopt when the +/// mode is not 'distributed', there is no JOIN, no eligible driver is reachable, or any unsafe leaf appears +/// anywhere in the subtree (explicit `*Cluster()` table function, local/`Distributed` table, row policy, missing +/// SELECT access, unsupported shape). The caller then falls back to ordinary planning. +std::optional findDistributedObjectStorageCandidate( + const QueryTreeNodePtr & query_node, const ContextPtr & context); + +} diff --git a/src/Planner/tests/gtest_distributed_object_storage_join_dispatch.cpp b/src/Planner/tests/gtest_distributed_object_storage_join_dispatch.cpp new file mode 100644 index 000000000000..b689efd4d1f2 --- /dev/null +++ b/src/Planner/tests/gtest_distributed_object_storage_join_dispatch.cpp @@ -0,0 +1,559 @@ +#include + +#include + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace DB; + +namespace +{ + +NamesAndTypesList driverColumns() +{ + return {{"id", std::make_shared()}}; +} + +NamesAndTypesList lookupColumns() +{ + return {{"lookup_id", std::make_shared()}}; +} + +struct State; + +/// A DatabaseMemory that reports itself as a DataLake catalog: that is what makes the tables inside it eligible +/// for dispatch (findDistributedObjectStorageCandidate asks DatabaseCatalog::isDatalakeCatalog). +class FakeDataLakeDatabase : public DatabaseWithOwnTablesBase +{ +public: + explicit FakeDataLakeDatabase(const String & name_, ContextPtr context_) + : DatabaseWithOwnTablesBase(name_, "FakeDataLakeDatabase(" + name_ + ")", context_) + { + } + + String getEngineName() const override { return "FakeDataLakeCatalog"; } + bool isDatalakeCatalog() const override { return true; } + +private: + ASTPtr getCreateDatabaseQueryImpl() const override { return nullptr; } +}; + +/// Minimal driver stand-in; getTaskIteratorExtension() is only invoked during real pipeline execution, +/// which these tests never trigger -- they only check the plan. +class FakeDriverStorage : public IStorageCluster +{ +public: + FakeDriverStorage(const StorageID & table_id, String cluster_name_) + : IStorageCluster(cluster_name_, table_id, getLogger("test")) + { + StorageInMemoryMetadata metadata; + metadata.setColumns(ColumnsDescription{driverColumns()}); + setInMemoryMetadata(metadata); + } + + std::string getName() const override { return "FakeDriverStorage"; } + + RemoteQueryExecutor::Extension getTaskIteratorExtension( + const ActionsDAG::Node *, const ActionsDAG *, const ContextPtr &, ClusterPtr, StorageMetadataPtr) const override + { + return {}; + } + + /// Whole-query dispatch must never ask a storage to rewrite itself into a cluster table function -- + /// that was the previous design, and the driver now travels as a plain table name announced in the + /// settings. Trips if anything on the dispatch path calls the rewrite hook again. + void updateQueryToSendIfNeeded(ASTPtr &, const StorageSnapshotPtr &, const ContextPtr &, bool) override + { + rewrite_hook_called = true; + } + + mutable bool rewrite_hook_called = false; +}; + +/// Stand-in for a second table from the same DataLake catalog, e.g. ice.geo_location_lookup. +class FakeSafeLookupStorage : public IStorageCluster +{ +public: + explicit FakeSafeLookupStorage(const StorageID & table_id) + : IStorageCluster(/*cluster_name_*/ "", table_id, getLogger("test")) + { + StorageInMemoryMetadata metadata; + metadata.setColumns(ColumnsDescription{lookupColumns()}); + setInMemoryMetadata(metadata); + } + + std::string getName() const override { return "FakeSafeLookupStorage"; } + + RemoteQueryExecutor::Extension getTaskIteratorExtension( + const ActionsDAG::Node *, const ActionsDAG *, const ContextPtr &, ClusterPtr, StorageMetadataPtr) const override + { + return {}; + } + +protected: + /// Empty cluster name -> plain reads (e.g. 'allow' mode) fall back here instead of ReadFromCluster. + void readFallBackToPure( + QueryPlan & query_plan, + const Names & column_names, + const StorageSnapshotPtr & storage_snapshot, + SelectQueryInfo &, + ContextPtr, + QueryProcessingStage::Enum, + size_t, + size_t) override + { + auto header = std::make_shared(storage_snapshot->getSampleBlockForColumns(column_names)); + Pipe pipe(std::make_shared(header)); + query_plan.addStep(std::make_unique(std::move(pipe))); + } +}; + +/// Port 1 is never listening; irrelevant since the test only builds the QueryPlan. +void registerUnreachableCluster(const ContextMutablePtr & context, const String & cluster_name) +{ + std::ostringstream config_text; + config_text << "<" << cluster_name << ">" + << "127.0.0.11" + << ""; + std::istringstream config_stream(config_text.str()); + Poco::AutoPtr config = new Poco::Util::XMLConfiguration(config_stream); + context->setClustersConfig(config, /*enable_discovery=*/false); +} + +/// Test fixture, modelled on src/Planner/tests/gtest_planner_empty_projection.cpp. +struct State +{ + State(const State &) = delete; + + ContextMutablePtr context; + std::shared_ptr driver; + + static State & instance() + { + static State state; + return state; + } + +private: + explicit State() + : context(Context::createCopy(getContext().context)) + { + tryRegisterFunctions(); + tryRegisterAggregateFunctions(); + + /// Default test context leaves query_kind at NO_QUERY; must look like a real initiator query. + ClientInfo client_info = context->getClientInfo(); + client_info.query_kind = ClientInfo::QueryKind::INITIAL_QUERY; + context->setClientInfo(client_info); + + /// Table resolution during analysis requires a real query context (context->getQueryContext() throws + /// THERE_IS_NO_QUERY otherwise); every real query already has one. + context->makeQueryContext(); + + static constexpr auto database_name = "distributed_object_storage_join_dispatch_test_db"; + static constexpr auto cluster_name = "vig-test"; + + DatabasePtr database = std::make_shared(database_name, context); + + driver = std::make_shared(StorageID(database_name, "driver"), cluster_name); + database->attachTable(context, "driver", driver, {}); + + database->attachTable(context, "safe_lookup", std::make_shared(StorageID(database_name, "safe_lookup")), {}); + database->attachTable(context, "dim2", std::make_shared(StorageID(database_name, "dim2")), {}); + + DatabaseCatalog::instance().attachDatabase(database->getDatabaseName(), database); + context->setCurrentDatabase(database_name); + + registerUnreachableCluster(context, cluster_name); + } +}; + +/// getQueryPlan() returns a reference into the interpreter's own move-only plan, so build+explain in one scope. +String planAndExplain(const String & query, const ContextMutablePtr & context) +{ + /// Scope the tripwire to this plan: the driver storage is shared across tests, and a query that falls back + /// to an ordinary cluster read is *supposed* to call the rewrite hook. + State::instance().driver->rewrite_hook_called = false; + + ParserSelectQuery parser; + ASTPtr ast = parseQuery(parser, query, 1000, 1000, 1000000); + auto query_tree = buildQueryTree(ast, context); + QueryTreePassManager pass_manager(context); + addQueryTreePasses(pass_manager); + pass_manager.run(query_tree); + + SelectQueryOptions options; + InterpreterSelectQueryAnalyzer interpreter(query_tree, context, options); + auto & plan = interpreter.getQueryPlan(); + + WriteBufferFromOwnString buffer; + plan.explainPlan(buffer, {.header = true, .description = true, .actions = true}); + return buffer.str(); +} + +/// Like planAndExplain(), but also runs the query plan optimizer (predicate pushdown included) before +/// explaining -- this is what actually drives ReadFromCluster::applyFilters() during a real EXPLAIN/execution, +/// which planAndExplain() alone never touches. Needed to reproduce a live exception: a WHERE on the +/// driver gets pushed down as a filter onto the whole-query ReadFromCluster step, whose SelectQueryInfo used to +/// carry a mismatched planner_context/table_expression pair for a per-table filter lookup that this step isn't. +String planOptimizeAndExplain(const String & query, const ContextMutablePtr & context) +{ + ParserSelectQuery parser; + ASTPtr ast = parseQuery(parser, query, 1000, 1000, 1000000); + auto query_tree = buildQueryTree(ast, context); + QueryTreePassManager pass_manager(context); + addQueryTreePasses(pass_manager); + pass_manager.run(query_tree); + + SelectQueryOptions options; + InterpreterSelectQueryAnalyzer interpreter(query_tree, context, options); + auto & plan = interpreter.getQueryPlan(); + plan.optimize(QueryPlanOptimizationSettings(context)); + + WriteBufferFromOwnString buffer; + plan.explainPlan(buffer, {.header = true, .description = true, .actions = true}); + return buffer.str(); +} + +/// Finds the single ReadFromCluster step in `node`'s subtree, or nullptr. +ReadFromCluster * findReadFromCluster(QueryPlan::Node * node) +{ + if (!node) + return nullptr; + if (auto * read_from_cluster = dynamic_cast(node->step.get())) + return read_from_cluster; + for (auto * child : node->children) + if (auto * found = findReadFromCluster(child)) + return found; + return nullptr; +} + +ReadFromClusterQuery * findReadFromClusterQuery(QueryPlan::Node * node) +{ + if (!node) + return nullptr; + if (auto * read_from_cluster_query = dynamic_cast(node->step.get())) + return read_from_cluster_query; + for (auto * child : node->children) + if (auto * found = findReadFromClusterQuery(child)) + return found; + return nullptr; +} + +} + +/// Core case: the driver is the JOIN's leftmost table, and the whole query dispatches as one +/// ReadFromCluster step with no local JOIN. +TEST(DistributedObjectStorageJoinDispatch, DriverOwnsWholeJoinWhenModeIsDistributed) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planAndExplain("SELECT driver.id, safe_lookup.lookup_id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.lookup_id", state.context); + + EXPECT_NE(plan_text.find("ReadFromCluster"), String::npos) << plan_text; + EXPECT_NE(plan_text.find("INNER JOIN"), String::npos) << "expected the whole JOIN forwarded in ReadFromCluster's query, got:\n" << plan_text; + EXPECT_EQ(plan_text.find("JoinLogical"), String::npos) << "expected no local JOIN step, got:\n" << plan_text; +} + +/// Nothing in the dispatched query is rewritten: the driver crosses the wire as the catalog table the user +/// wrote, and which table drives the dispatch travels separately, in +/// `object_storage_distributed_driver_database`/`_table`. That is only unambiguous while the driver is named +/// once, which the planner checks on the serialized query -- so pin both halves here: no cluster function +/// appears, and the driver's own name appears exactly once. +TEST(DistributedObjectStorageJoinDispatch, DriverCrossesTheWireUnrewrittenAndNamedOnce) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planAndExplain( + "SELECT driver.id, safe_lookup.lookup_id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.lookup_id", state.context); + + ASSERT_NE(plan_text.find("ReadFromCluster"), String::npos) << plan_text; + + EXPECT_FALSE(State::instance().driver->rewrite_hook_called) + << "dispatch rewrote the driver into a cluster table function; it must be sent as a plain table name"; + + size_t driver_mentions = 0; + for (size_t pos = plan_text.find("driver"); pos != String::npos; pos = plan_text.find("driver", pos + 1)) + ++driver_mentions; + EXPECT_GE(driver_mentions, 1u) << "expected the driver to be named in the dispatched query, got:\n" << plan_text; +} + +/// Default mode ('allow'): unaffected, JOIN still executes locally. +TEST(DistributedObjectStorageJoinDispatch, DriverIsWrappedWhenModeIsNotDistributed) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("allow")); + + auto plan_text = planAndExplain("SELECT driver.id, safe_lookup.lookup_id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.lookup_id", state.context); + + EXPECT_NE(plan_text.find("JoinLogical"), String::npos) << "expected a local JOIN step, got:\n" << plan_text; +} + +/// Driver buried in a subquery: the outer JOIN and GROUP BY still dispatch as one ReadFromCluster, +/// with stock MergingAggregated finalization reused on top. +TEST(DistributedObjectStorageJoinDispatch, BuriedDriverOwnsWholeOuterQueryWithGroupBy) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planAndExplain( + "SELECT x.id, count() FROM " + "(SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.lookup_id) AS x " + "INNER JOIN dim2 ON x.id = dim2.lookup_id " + "GROUP BY x.id", + state.context); + + EXPECT_NE(plan_text.find("ReadFromCluster"), String::npos) << plan_text; + EXPECT_NE(plan_text.find("INNER JOIN"), String::npos) << "expected both JOINs forwarded in ReadFromCluster's query, got:\n" << plan_text; + EXPECT_EQ(plan_text.find("JoinLogical"), String::npos) << "expected no local JOIN step, got:\n" << plan_text; + EXPECT_NE(plan_text.find("MergingAggregated"), String::npos) + << "expected stock final-merge aggregation on top of the dispatched read, got:\n" << plan_text; +} + +/// The ReadFromCluster header must reflect the WithMergeableState stage (unmerged aggregate states), not the +/// query's final projection types -- count() is UInt64 only after MergingAggregated, AggregateFunction(count) +/// beforehand. A header built from the final projection instead would declare the wrong type here, undetected +/// by plan structure alone since ReadFromCluster's header is never validated against nothing at plan time. +TEST(DistributedObjectStorageJoinDispatch, ReadFromClusterHeaderCarriesUnmergedAggregateState) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planAndExplain( + "SELECT x.id, count() FROM " + "(SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.lookup_id) AS x " + "INNER JOIN dim2 ON x.id = dim2.lookup_id " + "GROUP BY x.id", + state.context); + + auto read_from_cluster_pos = plan_text.find("ReadFromCluster"); + ASSERT_NE(read_from_cluster_pos, String::npos) << plan_text; + auto merging_aggregated_pos = plan_text.find("MergingAggregated"); + ASSERT_NE(merging_aggregated_pos, String::npos) << plan_text; + + /// explainPlan() prints children before parents, so ReadFromCluster's own header block sits between the + /// two step names. + auto read_from_cluster_block = plan_text.substr(read_from_cluster_pos, merging_aggregated_pos - read_from_cluster_pos); + EXPECT_NE(read_from_cluster_block.find("AggregateFunction(count"), String::npos) + << "expected ReadFromCluster's header to carry the unmerged aggregate state type, got:\n" << read_from_cluster_block; +} + +/// A realistic projection: a CASE expression reading columns from both sides of the JOIN, plus count(), +/// GROUP BY, ORDER BY and LIMIT all on the dispatch boundary itself. Exercises the header/rename machinery +/// with more than one plain passthrough column, unlike the trivial single-column projections above. +TEST(DistributedObjectStorageJoinDispatch, ComplexProjectionOverBothJoinSidesBuildsSuccessfully) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planAndExplain( + "SELECT " + " CASE WHEN safe_lookup.lookup_id > 0 THEN driver.id ELSE safe_lookup.lookup_id END AS dst_city, " + " count() AS c " + "FROM driver LEFT JOIN safe_lookup ON driver.id = safe_lookup.lookup_id " + "GROUP BY dst_city " + "ORDER BY dst_city " + "LIMIT 10", + state.context); + + EXPECT_NE(plan_text.find("ReadFromCluster"), String::npos) << plan_text; + EXPECT_NE(plan_text.find("LEFT JOIN"), String::npos) << "expected the whole JOIN forwarded in ReadFromCluster's query, got:\n" << plan_text; + EXPECT_EQ(plan_text.find("JoinLogical"), String::npos) << "expected no local JOIN step, got:\n" << plan_text; + EXPECT_NE(plan_text.find("MergingAggregated"), String::npos) + << "expected stock final-merge aggregation on top of the dispatched read, got:\n" << plan_text; +} + +/// Regression for a live exception: a WHERE on the driver survives real plan optimization (not just +/// candidate discovery/plan construction), which pushes it down as a filter onto the whole-query +/// ReadFromCluster step and calls ReadFromCluster::applyFilters() -> SourceStepWithFilter::applyFilters() -> +/// SelectQueryInfo::buildNodeNameToInputNodeColumn(), which looks up a per-table `table_expression` in a +/// `planner_context` that here describes the *whole* dispatched query -- it throws, and used to dereference a +/// null table_expression while formatting that very error. In whole-query mode applyFilters must therefore use +/// SourceStepWithFilterBase::applyFilters, which does not build that per-table mapping at all. +TEST(DistributedObjectStorageJoinDispatch, ComplexProjectionWithWhereSurvivesPlanOptimization) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planOptimizeAndExplain( + "SELECT " + " CASE WHEN safe_lookup.lookup_id > 0 THEN driver.id ELSE safe_lookup.lookup_id END AS dst_city, " + " count() AS c " + "FROM driver LEFT JOIN safe_lookup ON driver.id = safe_lookup.lookup_id " + "WHERE driver.id > 0 " + "GROUP BY dst_city " + "ORDER BY dst_city " + "LIMIT 10", + state.context); + + EXPECT_NE(plan_text.find("ReadFromCluster"), String::npos) << plan_text; + EXPECT_EQ(plan_text.find("JoinLogical"), String::npos) << "expected no local JOIN step, got:\n" << plan_text; +} + +/// The full buried-driver shape: driver behind one intermediate subquery, outer LEFT JOIN against a RHS subquery that +/// itself LEFT JOINs a further GROUP BY subquery and also has its own GROUP BY. None of that RHS content is a +/// competing driver -- it's recomputed whole on every worker -- and the whole thing must still build into a +/// single dispatched ReadFromCluster with stock finalization for the outer GROUP BY on top. +TEST(DistributedObjectStorageJoinDispatch, BuriedDriverWithNestedRightSideBuildsSuccessfully) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planAndExplain( + "SELECT transaction_event.id, count() FROM " + "(SELECT driver.id FROM driver) AS transaction_event " + "LEFT JOIN " + "(SELECT safe_lookup.lookup_id AS id FROM safe_lookup LEFT JOIN " + "(SELECT dim2.lookup_id AS id FROM dim2 GROUP BY dim2.lookup_id) AS policy_matches " + "ON safe_lookup.lookup_id = policy_matches.id GROUP BY safe_lookup.lookup_id) AS alert_events " + "ON transaction_event.id = alert_events.id " + "GROUP BY transaction_event.id", + state.context); + + EXPECT_NE(plan_text.find("ReadFromCluster"), String::npos) << plan_text; + EXPECT_EQ(plan_text.find("JoinLogical"), String::npos) << "expected no local JOIN step, got:\n" << plan_text; + EXPECT_NE(plan_text.find("MergingAggregated"), String::npos) + << "expected stock final-merge aggregation on top of the dispatched read, got:\n" << plan_text; +} + +/// The same shape, asserting that dispatch rewrites nothing: the driver (`driver`, buried inside +/// `transaction_event`) and every other DataLake table reachable from the RHS (`safe_lookup`, `dim2`) all cross +/// the wire as ordinary catalog identifiers. Only the announced driver reads the initiator's file-task queue. +TEST(DistributedObjectStorageJoinDispatch, DispatchesBuriedDriverWithoutRewritingAnyTable) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planAndExplain( + "SELECT transaction_event.id, count() FROM " + "(SELECT driver.id FROM driver) AS transaction_event " + "LEFT JOIN " + "(SELECT safe_lookup.lookup_id AS id FROM safe_lookup LEFT JOIN " + "(SELECT dim2.lookup_id AS id FROM dim2 GROUP BY dim2.lookup_id) AS policy_matches " + "ON safe_lookup.lookup_id = policy_matches.id GROUP BY safe_lookup.lookup_id) AS alert_events " + "ON transaction_event.id = alert_events.id " + "GROUP BY transaction_event.id", + state.context); + + EXPECT_FALSE(State::instance().driver->rewrite_hook_called) + << "dispatch rewrote the driver into a cluster table function; it must be sent as a plain table name"; + + EXPECT_NE(plan_text.find("safe_lookup"), String::npos) << "expected safe_lookup to remain an ordinary catalog identifier, got:\n" << plan_text; + EXPECT_NE(plan_text.find("dim2"), String::npos) << "expected dim2 to remain an ordinary catalog identifier, got:\n" << plan_text; +} + +/// The same shape written with CTEs rather than derived tables. The setting documents CTE support, and the +/// analyzer resolves a CTE into a QueryNode just as it does a derived table -- but a CTE reference serializes to +/// its bare name unless the body is inlined, which would not resolve on a worker. +/// queryNodeToDistributedSelectQuery is what inlines it; this pins that down, including for a CTE +/// (`policy_matches`) referenced from inside another CTE. +TEST(DistributedObjectStorageJoinDispatch, BuriedDriverWithCommonTableExpressions) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto plan_text = planAndExplain( + "WITH transaction_event AS (SELECT driver.id FROM driver), " + "policy_matches AS (SELECT dim2.lookup_id AS id FROM dim2 GROUP BY dim2.lookup_id), " + "alert_events AS (SELECT safe_lookup.lookup_id AS id FROM safe_lookup " + "LEFT JOIN policy_matches ON safe_lookup.lookup_id = policy_matches.id " + "GROUP BY safe_lookup.lookup_id) " + "SELECT transaction_event.id, count() FROM transaction_event " + "LEFT JOIN alert_events ON transaction_event.id = alert_events.id " + "GROUP BY transaction_event.id", + state.context); + + EXPECT_NE(plan_text.find("ReadFromCluster"), String::npos) << plan_text; + EXPECT_EQ(plan_text.find("JoinLogical"), String::npos) << "expected no local JOIN step, got:\n" << plan_text; + EXPECT_NE(plan_text.find("MergingAggregated"), String::npos) + << "expected stock final-merge aggregation on top of the dispatched read, got:\n" << plan_text; + + /// No dangling CTE name: every CTE body must appear inlined in the forwarded query, and nothing is rewritten. + EXPECT_FALSE(State::instance().driver->rewrite_hook_called) + << "dispatch rewrote the driver into a cluster table function; it must be sent as a plain table name"; + + EXPECT_NE(plan_text.find("safe_lookup"), String::npos) << plan_text; + EXPECT_NE(plan_text.find("dim2"), String::npos) << plan_text; +} + +/// Whole-query dispatch gets its own source step rather than reusing the single-table one. That is what makes +/// a whole class of mismatch unrepresentable: ReadFromCluster carries a StorageSnapshot and required columns +/// describing one table, which for a dispatched query would describe the driver while the step's output is the +/// query's result. ReadFromClusterQuery has neither, so there is nothing to disagree. +TEST(DistributedObjectStorageJoinDispatch, DispatchUsesItsOwnSourceStepNotATableRead) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + ParserSelectQuery parser; + String query = + "SELECT " + " CASE WHEN safe_lookup.lookup_id > 0 THEN driver.id ELSE safe_lookup.lookup_id END AS dst_city, " + " count() AS c " + "FROM driver LEFT JOIN safe_lookup ON driver.id = safe_lookup.lookup_id " + "GROUP BY dst_city"; + ASTPtr ast = parseQuery(parser, query, 1000, 1000, 1000000); + auto query_tree = buildQueryTree(ast, state.context); + QueryTreePassManager pass_manager(state.context); + addQueryTreePasses(pass_manager); + pass_manager.run(query_tree); + + SelectQueryOptions options; + InterpreterSelectQueryAnalyzer interpreter(query_tree, state.context, options); + auto & plan = interpreter.getQueryPlan(); + + EXPECT_NE(findReadFromClusterQuery(plan.getRootNode()), nullptr) + << "expected the dispatch to use its own source step"; + EXPECT_EQ(findReadFromCluster(plan.getRootNode()), nullptr) + << "expected no single-table cluster read: the dispatched query is not one table's rows"; +} + +/// The mode can also arrive via a query-level SETTINGS clause rather than context->setSetting(); the header's +/// hook-disabling context copy (Context::createCopy(context)) must still see it and disable the hook for the +/// sample-block analysis, or this recurses instead of the ambient session-level context not mattering here. +TEST(DistributedObjectStorageJoinDispatch, HeaderComputationHandlesQueryLevelModeSetting) +{ + auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("allow")); + + auto plan_text = planAndExplain( + "SELECT driver.id, count() FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.lookup_id " + "GROUP BY driver.id SETTINGS object_storage_cluster_join_mode = 'distributed'", + state.context); + + EXPECT_NE(plan_text.find("ReadFromCluster"), String::npos) << plan_text; + EXPECT_NE(plan_text.find("AggregateFunction(count"), String::npos) + << "expected header computation to still see WithMergeableState types under a query-level SETTINGS override, got:\n" << plan_text; +} diff --git a/src/Planner/tests/gtest_find_distributed_object_storage_candidate.cpp b/src/Planner/tests/gtest_find_distributed_object_storage_candidate.cpp new file mode 100644 index 000000000000..16b903734b4e --- /dev/null +++ b/src/Planner/tests/gtest_find_distributed_object_storage_candidate.cpp @@ -0,0 +1,620 @@ +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace DB; + +namespace +{ + +NamesAndTypesList testColumns() +{ + return {{"id", std::make_shared()}}; +} + +/// A DatabaseMemory that reports itself as a DataLake catalog, which is what makes the tables inside it +/// eligible (findDistributedObjectStorageCandidate asks DatabaseCatalog::isDatalakeCatalog). +class FakeDataLakeDatabase : public DatabaseWithOwnTablesBase +{ +public: + explicit FakeDataLakeDatabase(const String & name_, ContextPtr context_) + : DatabaseWithOwnTablesBase(name_, "FakeDataLakeDatabase(" + name_ + ")", context_) + { + } + + String getEngineName() const override { return "FakeDataLakeCatalog"; } + bool isDatalakeCatalog() const override { return true; } + +private: + ASTPtr getCreateDatabaseQueryImpl() const override { return nullptr; } +}; + +/// Minimal IStorageCluster test double, configurable per instance. +class FakeClusterStorage : public IStorageCluster +{ +public: + FakeClusterStorage(const StorageID & table_id, String cluster_name_) + : IStorageCluster(cluster_name_, table_id, getLogger("test")) + { + StorageInMemoryMetadata metadata; + metadata.setColumns(ColumnsDescription{testColumns()}); + setInMemoryMetadata(metadata); + } + + std::string getName() const override { return "FakeClusterStorage"; } + + RemoteQueryExecutor::Extension getTaskIteratorExtension( + const ActionsDAG::Node *, const ActionsDAG *, const ContextPtr &, ClusterPtr, StorageMetadataPtr) const override + { + return {}; + } +}; + +/// Modelled on src/Planner/tests/gtest_planner_empty_projection.cpp. +struct State +{ + State(const State &) = delete; + + ContextMutablePtr context; + + static const State & instance() + { + static State state; + return state; + } + +private: + explicit State() + : context(Context::createCopy(getContext().context)) + { + tryRegisterFunctions(); + tryRegisterAggregateFunctions(); + + static constexpr auto database_name = "find_distributed_object_storage_candidate_test_db"; + DatabasePtr database = std::make_shared(database_name, context); + + auto attach_cluster_table = [&](const DatabasePtr & db, const String & table_name, String cluster_name) + { + db->attachTable( + context, + table_name, + std::make_shared(StorageID(db->getDatabaseName(), table_name), std::move(cluster_name)), + {}); + }; + + /// A distributed driver, e.g. ice.event_page. + attach_cluster_table(database, "driver", "vig-test"); + /// A second DataLake-catalog table under the same cluster -- never an independent driver on the + /// right of a JOIN, just a plain (if wasteful) safe partner. + attach_cluster_table(database, "second_datalake_table", "vig-test"); + /// A safe co-resolved table with no cluster dispatch of its own (e.g. ice.geo_location_lookup). + attach_cluster_table(database, "safe_lookup", ""); + + /// A Cluster-engine table in an ordinary database -- not resolved through a DataLake catalog, so not + /// safe to re-resolve on a worker. + static constexpr auto plain_database_name = "find_distributed_object_storage_candidate_test_plain_db"; + DatabasePtr plain_database = std::make_shared(plain_database_name, context); + attach_cluster_table(plain_database, "unsafe_cluster_table", "some-cluster"); + DatabaseCatalog::instance().attachDatabase(plain_database->getDatabaseName(), plain_database); + + database->attachTable( + context, + "local_table", + std::make_shared( + StorageID(database_name, "local_table"), ColumnsDescription{testColumns()}, ConstraintsDescription{}, String{}, MemorySettings{}), + {}); + + DatabaseCatalog::instance().attachDatabase(database->getDatabaseName(), database); + context->setCurrentDatabase(database_name); + } +}; + +QueryTreeNodePtr analyze(const String & query, const ContextMutablePtr & context) +{ + ParserSelectQuery parser; + ASTPtr ast = parseQuery(parser, query, 1000, 1000, 1000000); + auto query_tree = buildQueryTree(ast, context); + QueryTreePassManager pass_manager(context); + addQueryTreePasses(pass_manager); + pass_manager.run(query_tree); + return query_tree; +} + +/// AccessControl is shared across every Context::createCopy() of the same test-global context, so entity +/// names must be unique per call, not just per helper, or a second test's insert() throws "already exists". +size_t nextTestAccessEntitySuffix() +{ + static std::atomic counter{0}; + return counter++; +} + +/// A context copy whose user has no grants at all -- Context::getAccess() gives the global context (no +/// setUserID() call) full access unconditionally, so a real, minimally-privileged user is needed to exercise +/// the access-denied path at all. +ContextMutablePtr contextWithNoGrants(const ContextMutablePtr & base_context) +{ + auto context = Context::createCopy(base_context); + context->getAccessControl().addMemoryStorage("find_distributed_object_storage_candidate_test_storage", /*allow_backup_*/ false); + auto user = std::make_shared(); + user->setName(fmt::format("find_distributed_object_storage_candidate_test_no_grants_user_{}", nextTestAccessEntitySuffix())); + auto user_id = context->getAccessControl().insert(user); + context->setUser(user_id); + return context; +} + +/// A context copy whose user has full access, but a nontrivial row policy applies to `table_name`. +ContextMutablePtr contextWithRowPolicy(const ContextMutablePtr & base_context, const String & database_name, const String & table_name) +{ + auto context = Context::createCopy(base_context); + context->getAccessControl().addMemoryStorage("find_distributed_object_storage_candidate_test_storage", /*allow_backup_*/ false); + auto suffix = nextTestAccessEntitySuffix(); + + auto user = std::make_shared(); + user->setName(fmt::format("find_distributed_object_storage_candidate_test_row_policy_user_{}", suffix)); + user->access.grant(AccessType::ALL); + auto user_id = context->getAccessControl().insert(user); + + auto policy = std::make_shared(); + policy->setFullName(fmt::format("find_distributed_object_storage_candidate_test_policy_{}", suffix), database_name, table_name); + policy->filters[static_cast(RowPolicyFilterType::SELECT_FILTER)] = "id != -1"; + policy->to_roles = RolesOrUsersSet(user_id); + context->getAccessControl().insert(policy); + + context->setUser(user_id); + return context; +} + +} + +TEST(FindDistributedObjectStorageCandidate, AcceptsDriverJoinedWithSafeDataLakeTable) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id", state.context); + + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); + ASSERT_NE(candidate->driver, nullptr); + EXPECT_EQ(candidate->driver->getStorageID().table_name, "driver"); +} + +TEST(FindDistributedObjectStorageCandidate, DoesNotInterceptPlainDistributedObjectStorageQuery) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM driver", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsOrdinaryLocalTableAsRhs) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN local_table ON driver.id = local_table.id", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsWhenModeIsNotDistributed) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("allow")); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsWhenRemoteInitiatorIsSet) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + state.context->setSetting("object_storage_remote_initiator", true); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); + + state.context->setSetting("object_storage_remote_initiator", false); +} + +/// additional_table_filters matching depends on the initiator's current_database and the query's own +/// aliasing, both of which shift once forwarded as fully serialized remote SQL (the same reasoning +/// Planner.cpp already uses to disable parallel replicas for this combination). Rejected as a blanket +/// disablement whenever the setting is nonempty at all, regardless of which table it names. +TEST(FindDistributedObjectStorageCandidate, RejectsWhenAdditionalTableFiltersIsSet) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + state.context->setSetting("additional_table_filters", String("{'driver': 'id > 0'}")); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); + + state.context->setSetting("additional_table_filters", String("")); +} + +/// Ordinary per-table planning checks SELECT access on every table it plans (checkAccessRights() in +/// PlannerJoinTree.cpp); this whole-query dispatch replaces that walk entirely, so it must perform the +/// same check itself for the driver and every JOIN partner, or a user without SELECT could gain access to +/// the driver's data simply by enabling 'distributed' mode. +TEST(FindDistributedObjectStorageCandidate, RejectsDriverWithoutSelectAccess) +{ + const auto & state = State::instance(); + auto context = contextWithNoGrants(state.context); + context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id", context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, context).has_value()); +} + +/// Same access check, but the table without SELECT access is buried inside a subquery rather +/// than at the dispatch boundary's own top-level JOIN -- the recursive intermediate-subquery walk must reach +/// it too, not just the tables directly visible at the outermost level. +TEST(FindDistributedObjectStorageCandidate, RejectsBuriedDriverWithoutSelectAccess) +{ + const auto & state = State::instance(); + auto context = contextWithNoGrants(state.context); + context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT x.id FROM (SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id) AS x " + "INNER JOIN second_datalake_table ON x.id = second_datalake_table.id", + context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, context).has_value()); +} + +/// A row policy on the driver itself: a worker re-resolves the driver from its own catalog and the initiator's +/// policy does not travel with the query text, so it could never be reapplied there. +TEST(FindDistributedObjectStorageCandidate, RejectsDriverWithRowPolicy) +{ + const auto & state = State::instance(); + auto context = contextWithRowPolicy(state.context, "find_distributed_object_storage_candidate_test_db", "driver"); + context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id", context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, context).has_value()); +} + +/// A row policy on a non-driver JOIN partner: unlike the driver, this table stays a plain catalog identifier +/// and is independently re-resolved by DatabaseDataLake on each worker -- there is no cheap way to prove that +/// re-resolution applies the same effective policy as the initiator's own user, so it must conservatively +/// block whole-query dispatch too, not just a policy on the driver. +TEST(FindDistributedObjectStorageCandidate, RejectsNonDriverPartnerWithRowPolicy) +{ + const auto & state = State::instance(); + auto context = contextWithRowPolicy(state.context, "find_distributed_object_storage_candidate_test_db", "safe_lookup"); + context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id", context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, AcceptsSecondDataLakeTableAsRhsEvenThoughItIsAlsoDispatchCapable) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree + = analyze("SELECT driver.id FROM driver INNER JOIN second_datalake_table ON driver.id = second_datalake_table.id", state.context); + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); + EXPECT_EQ(candidate->driver->getStorageID().table_name, "driver"); +} + +/// A second driver-capable table sitting in its own subquery on the right of the outer JOIN is never inspected +/// as a competing driver at all -- it's just worker-local content, safe because it's a DataLake-catalog table, +/// same as AcceptsSecondDataLakeTableAsRhsEvenThoughItIsAlsoDispatchCapable above but one level deeper. +TEST(FindDistributedObjectStorageCandidate, AcceptsSecondDriverCapableSubqueryAsRhs) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT x.id FROM (SELECT driver.id FROM driver) AS x " + "INNER JOIN (SELECT second_datalake_table.id FROM second_datalake_table) AS y ON x.id = y.id", + state.context); + + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); + EXPECT_EQ(candidate->driver->getStorageID().table_name, "driver"); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsUnsafeClusterTableAsDriver) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree + = analyze("SELECT t.id FROM find_distributed_object_storage_candidate_test_plain_db.unsafe_cluster_table AS t INNER JOIN safe_lookup ON t.id = safe_lookup.id", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsUnsafeIStorageClusterTableAsRhs) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree + = analyze("SELECT driver.id FROM driver INNER JOIN find_distributed_object_storage_candidate_test_plain_db.unsafe_cluster_table AS t ON driver.id = t.id", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +/// Nested JoinNodes within one QueryNode (chained JOINs) must be fully understood, not just one JOIN. +TEST(FindDistributedObjectStorageCandidate, AcceptsDriverInMultiJoinQueryNode) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT driver.id FROM driver " + "INNER JOIN safe_lookup ON driver.id = safe_lookup.id " + "INNER JOIN second_datalake_table ON driver.id = second_datalake_table.id", + state.context); + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); + EXPECT_EQ(candidate->driver->getStorageID().table_name, "driver"); +} + +/// A GROUP BY inside the intermediate subquery `x` (between the dispatch boundary and the driver, on the +/// driver's own left path) would be executed independently per worker partition if the whole query were +/// dispatched -- unsafe, since a group spanning multiple workers' partitions would never get merged. No +/// fallback: the whole query is simply not a candidate; stock planning handles it (which plans `x`'s own +/// GROUP BY correctly, as the boundary of its own ordinary subquery plan). +TEST(FindDistributedObjectStorageCandidate, RejectsWhenDriverPathIntermediateSubqueryHasGroupBy) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT x.id, x.c FROM " + "(SELECT driver.id, count() AS c FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id GROUP BY driver.id) AS x " + "INNER JOIN second_datalake_table ON x.id = second_datalake_table.id", + state.context); + + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +/// Same hazard as GROUP BY: a LIMIT inside a driver-path intermediate subquery would apply per worker +/// partition, not globally, if the whole query were dispatched. No fallback. +TEST(FindDistributedObjectStorageCandidate, RejectsWhenDriverPathIntermediateSubqueryHasLimit) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT x.id FROM " + "(SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id LIMIT 10) AS x " + "INNER JOIN second_datalake_table ON x.id = second_datalake_table.id", + state.context); + + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +/// A plain WHERE/projection-only intermediate subquery is partition-preserving (each row stays independent), +/// so the outer query is still the accepted dispatch boundary, same as before this safety check existed. +TEST(FindDistributedObjectStorageCandidate, AcceptsOuterCandidateWhenIntermediateSubqueryIsWhereProjectionOnly) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT x.id FROM " + "(SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id WHERE driver.id > 0) AS x " + "INNER JOIN second_datalake_table ON x.id = second_datalake_table.id", + state.context); + + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); +} + +/// The right side of a JOIN is never searched for a driver, no matter what it contains -- so a driver-bearing +/// subquery sitting there simply isn't found; the left side alone decides whether there's a candidate at all +/// (here it doesn't have one, since `safe_lookup` has no cluster of its own). No fallback: rejected outright. +TEST(FindDistributedObjectStorageCandidate, RejectsWhenLeftSpineHasNoDriverEvenThoughRhsSubqueryDoes) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT x.id FROM safe_lookup LEFT JOIN " + "(SELECT driver.id FROM driver) AS x " + "ON safe_lookup.id = x.id", + state.context); + + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsUnsafeSubqueryInWhereClause) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id " + "WHERE driver.id IN (SELECT id FROM local_table)", + state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsCrossJoin) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM driver CROSS JOIN safe_lookup", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsDriverOnTheRightOfRightJoin) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM safe_lookup RIGHT JOIN driver ON safe_lookup.id = driver.id", state.context); + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, RejectsExplicitClusterTableFunctionAsRhs) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze("SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id", state.context); + + /// Build a TableFunctionNode directly -- no real TableFunctionFactory registration needed. + auto & query_node = query_tree->as(); + auto & join_node = query_node.getJoinTree()->as(); + + auto table_function_node = std::make_shared("icebergS3Cluster"); + auto explicit_cluster_storage = std::make_shared( + StorageID("system", "explicit_cluster_table"), "other-cluster"); + table_function_node->resolve(nullptr, explicit_cluster_storage, state.context, {}); + join_node.getRightTableExpression() = table_function_node; + + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +/// Driver buried in a subquery joined against another safe table; the outer query is the candidate. +TEST(FindDistributedObjectStorageCandidate, AcceptsBuriedDriverWithSafeOuterJoin) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT x.id FROM (SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id) AS x " + "INNER JOIN second_datalake_table ON x.id = second_datalake_table.id", + state.context); + + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); + EXPECT_EQ(candidate->driver->getStorageID().table_name, "driver"); +} + +/// Same shape, but the outer JOIN partner is an ordinary local table: whole-query dispatch would have to +/// forward it too (it's part of the same dispatch boundary), which isn't safe -- rejected outright, no +/// fallback to dispatching the inner subquery alone even though it would be safe on its own. +TEST(FindDistributedObjectStorageCandidate, RejectsWhenOuterJoinPartnerIsUnsafe) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT x.id FROM (SELECT driver.id FROM driver INNER JOIN safe_lookup ON driver.id = safe_lookup.id) AS x " + "INNER JOIN local_table ON x.id = local_table.id", + state.context); + + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +/// The full buried-driver shape: the driver sits behind one intermediate subquery (`transaction_event`, standing in for +/// `txnlog`) on the left of the outer LEFT JOIN; the right side (`alert_events`) is itself a LEFT JOIN against +/// a further subquery with its own GROUP BY (`policy_matches`), and `alert_events` itself also has a GROUP BY. +/// None of that RHS structure is inspected for a competing driver or restricted for GROUP BY/JOIN -- it's +/// worker-local content, recomputed whole on every worker. The outer query's own GROUP BY/ORDER BY/LIMIT are +/// the dispatch boundary's own, handled by stock finalization on top of the dispatched read. +TEST(FindDistributedObjectStorageCandidate, AcceptsBuriedDriverWithNestedRightSide) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT transaction_event.id, count() AS c FROM " + "(SELECT driver.id FROM driver WHERE driver.id > 0) AS transaction_event " + "LEFT JOIN " + "(SELECT safe_lookup.id FROM safe_lookup LEFT JOIN " + "(SELECT second_datalake_table.id FROM second_datalake_table GROUP BY second_datalake_table.id) AS policy_matches " + "ON safe_lookup.id = policy_matches.id GROUP BY safe_lookup.id) AS alert_events " + "ON transaction_event.id = alert_events.id " + "GROUP BY transaction_event.id " + "ORDER BY transaction_event.id " + "LIMIT 10", + state.context); + + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); + EXPECT_EQ(candidate->driver->getStorageID().table_name, "driver"); +} + +/// Direct-driver shape: one root LEFT JOIN between the driver and a safe lookup table, with WHERE, GROUP BY, +/// ORDER BY and LIMIT all sitting directly on the dispatch boundary itself (not an intermediate subquery) -- +/// freely allowed there, unlike on a driver-path intermediate subquery. +/// The buried-driver shape expressed with CTEs: the driver sits inside a CTE used as the JOIN's left side. The analyzer +/// resolves a CTE reference into the same QueryNode a derived table would produce, so the left-spine walk must +/// find the driver through it exactly as it does through a subquery. +TEST(FindDistributedObjectStorageCandidate, AcceptsBuriedDriverInsideCommonTableExpression) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "WITH transaction_event AS (SELECT driver.id FROM driver), " + "alert_events AS (SELECT safe_lookup.id FROM safe_lookup) " + "SELECT transaction_event.id, count() FROM transaction_event " + "LEFT JOIN alert_events ON transaction_event.id = alert_events.id " + "GROUP BY transaction_event.id", + state.context); + + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); + EXPECT_EQ(candidate->driver->getStorageID().getTableName(), "driver"); +} + +/// A CTE on the driver's own left path is still an intermediate subquery: if it aggregates, each worker would +/// finalize its own slice of the driver as though it were the whole group. +TEST(FindDistributedObjectStorageCandidate, RejectsAggregatingCommonTableExpressionOnDriverPath) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "WITH transaction_event AS (SELECT driver.id FROM driver GROUP BY driver.id), " + "alert_events AS (SELECT safe_lookup.id FROM safe_lookup) " + "SELECT transaction_event.id, count() FROM transaction_event " + "LEFT JOIN alert_events ON transaction_event.id = alert_events.id " + "GROUP BY transaction_event.id", + state.context); + + EXPECT_FALSE(findDistributedObjectStorageCandidate(query_tree, state.context).has_value()); +} + +TEST(FindDistributedObjectStorageCandidate, AcceptsDirectDriverWithFilterAndAggregation) +{ + const auto & state = State::instance(); + state.context->setSetting("object_storage_cluster_join_mode", String("distributed")); + + auto query_tree = analyze( + "SELECT driver.id, count() FROM driver LEFT JOIN safe_lookup ON driver.id = safe_lookup.id " + "WHERE driver.id > 0 GROUP BY driver.id ORDER BY driver.id LIMIT 10", + state.context); + + auto candidate = findDistributedObjectStorageCandidate(query_tree, state.context); + ASSERT_TRUE(candidate.has_value()); + EXPECT_EQ(candidate->driver->getStorageID().table_name, "driver"); +} diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index fe1a51a861c0..436ef17967b4 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -45,6 +45,10 @@ #include #include #include +#include +#include +#include +#include #include @@ -71,6 +75,9 @@ namespace Setting extern const SettingsBool object_storage_remote_initiator; extern const SettingsString object_storage_remote_initiator_cluster; extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; + extern const SettingsString object_storage_distributed_driver_database; + extern const SettingsString object_storage_distributed_driver_table; + extern const SettingsString object_storage_cluster; } namespace ErrorCodes @@ -132,6 +139,93 @@ ActionsDAG andListingFilterDAGs(ActionsDAG first, ActionsDAG second) } +namespace +{ + +/// A worker applies the query's own SETTINGS clause on top of the settings it received, so anything +/// normalized only in the context is undone by a value written into the query text. Every such normalization +/// therefore comes in a pair, and the two `prepare...ForRemoteExecution` functions below are the query-text +/// half -- one per kind of remote read, so neither can be given a subset of the other's by accident. +ASTSetQuery * getQuerySettings(ASTPtr & query) +{ + auto * select_query = query->as(); + if (!select_query) + return nullptr; + + auto settings_ast = select_query->settings(); + return settings_ast ? &settings_ast->as() : nullptr; +} + +void dropEmptySettings(ASTPtr & query, bool changed) +{ + auto * select_query = query->as(); + if (changed && select_query && select_query->settings() + && select_query->settings()->as().changes.empty()) + select_query->setExpression(ASTSelectQuery::Expression::SETTINGS, {}); +} + +void downgradeJoinModeInQuerySettings(ASTPtr & query) +{ + auto * settings = getQuerySettings(query); + if (!settings) + return; + + bool changed = false; + for (auto & change : settings->changes) + { + if (change.name != "object_storage_cluster_join_mode") + continue; + if (change.value.safeGet() != "distributed") + continue; + change.value = Field(String("allow")); + changed = true; + } + + dropEmptySettings(query, changed); +} + +void dropDriverAnnouncementFromQuerySettings(ASTPtr & query) +{ + auto * settings = getQuerySettings(query); + if (!settings) + return; + + bool changed = settings->changes.removeSetting("object_storage_distributed_driver_database"); + changed |= settings->changes.removeSetting("object_storage_distributed_driver_table"); + dropEmptySettings(query, changed); +} + +void dropClusterFromQuerySettings(ASTPtr & query) +{ + auto * settings = getQuerySettings(query); + if (!settings) + return; + + dropEmptySettings(query, settings->changes.removeSetting("object_storage_cluster")); +} + +/// An ordinary cluster read, including one reached after whole-query dispatch was declined. It must behave +/// exactly like `allow`, and it must not carry a driver name: this read supplies a task iterator, so its +/// workers see `collaborate_with_initiator` and a table matching that name would read this step's file-task +/// queue instead of listing its own files. +void prepareOrdinaryClusterQueryForRemoteExecution(ASTPtr & query) +{ + downgradeJoinModeInQuerySettings(query); + dropDriverAnnouncementFromQuerySettings(query); +} + +/// A whole-query dispatch. `object_storage_cluster` outranks a table's own cluster in getClusterName, so +/// leaving it set would make every table in the dispatched query fan out again from each worker. The driver +/// name goes too: the initiator puts the one it chose in the settings it sends, and a value in the query text +/// would override it on the worker and hand the queue to a table the planner did not pick. +void prepareWholeQueryDispatchForRemoteExecution(ASTPtr & query) +{ + dropClusterFromQuerySettings(query); + dropDriverAnnouncementFromQuerySettings(query); +} + +} + void ReadFromCluster::applyFilters(ActionDAGNodes added_filter_nodes) { SourceStepWithFilter::applyFilters(std::move(added_filter_nodes)); @@ -158,9 +252,8 @@ void ReadFromCluster::createExtension() if (extension) return; - const ActionsDAG * filter = listing_filter_dag - ? listing_filter_dag.get() - : (filter_actions_dag ? filter_actions_dag.get() : query_info.filter_actions_dag.get()); + const ActionsDAG * filter + = listing_filter_dag ? listing_filter_dag.get() : (filter_actions_dag ? filter_actions_dag.get() : query_info.filter_actions_dag.get()); const ActionsDAG::Node * predicate = filter ? filter->getOutputs().at(0) : nullptr; extension = storage->getTaskIteratorExtension( predicate, @@ -415,6 +508,9 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( } case ObjectStorageClusterJoinMode::ALLOW: // Do nothing special return; + case ObjectStorageClusterJoinMode::DISTRIBUTED: + /// A whole-query dispatch never goes through read() -- see readPreparedClusterQuery() below. + return; } } @@ -548,6 +644,47 @@ void IStorageCluster::read( query_plan.addStep(std::move(reading)); } +/// Smaller sibling of read(): the caller already has query_to_send and sample_block, so no join-stripping, +/// no sample-block computation, and no RestoreQualifiedNamesVisitor (which assumes position 0). +void IStorageCluster::readPreparedClusterQuery( + QueryPlan & query_plan, + const StorageSnapshotPtr & storage_snapshot, + SelectQueryInfo & query_info, + ContextPtr context, + QueryProcessingStage::Enum processed_stage, + ASTPtr query_to_send, + SharedHeader sample_block, + std::shared_ptr driver_filter) +{ + auto cluster_name_from_settings = getClusterName(context); + const auto & settings = context->getSettingsRef(); + auto cluster = getClusterImpl(context, cluster_name_from_settings, isObjectStorage() ? settings[Setting::object_storage_max_nodes] : 0); + + AddDefaultDatabaseVisitor visitor(context, context->getCurrentDatabase(), + /* only_replace_current_database_function_= */false, + /* only_replace_in_join_= */true); + visitor.visit(query_to_send); + + std::optional external_tables = std::nullopt; + if (query_info.planner_context && query_info.planner_context->getMutableQueryContext()) + external_tables = query_info.planner_context->getMutableQueryContext()->getExternalTables(); + + auto reading = std::make_unique( + sample_block, + std::static_pointer_cast(shared_from_this()), + storage_snapshot, + context, + std::move(query_to_send), + processed_stage, + cluster, + log, + std::move(external_tables), + std::move(driver_filter)); + + query_plan.addStep(std::move(reading)); +} + + IStorageCluster::RemoteCallVariables IStorageCluster::convertToRemote( ClusterPtr cluster, ContextPtr context, @@ -647,14 +784,26 @@ SinkToStoragePtr IStorageCluster::write( throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method write is not supported by storage {}", getName()); } -void ReadFromCluster::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) +namespace { - const Scalars & scalars = context->hasQueryContext() ? context->getQueryContext()->getScalars() : Scalars{}; - const bool add_agg_info = processed_stage == QueryProcessingStage::WithMergeableState; - Pipes pipes; - auto new_context = updateSettings(context->getSettingsRef()); +/// The cluster-function transport, shared by ReadFromCluster and ReadFromClusterQuery: every replica is taken +/// as a shard, each gets one connection, and all of them pull work from the same task iterator. What differs +/// between the two callers is only what the query is and where the tasks come from. +Pipe buildClusterFunctionRemotePipe( + const ASTPtr & query_to_send, + const SharedHeader & output_header, + const ContextPtr & new_context, + const ClusterPtr & cluster, + QueryProcessingStage::Enum processed_stage, + const RemoteQueryExecutor::Extension & extension, + const std::optional & external_tables, + LoggerPtr log) +{ + const Scalars & scalars = new_context->hasQueryContext() ? new_context->getQueryContext()->getScalars() : Scalars{}; + const bool add_agg_info = processed_stage == QueryProcessingStage::WithMergeableState; const auto & current_settings = new_context->getSettingsRef(); + auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithFailover(current_settings); size_t replica_index = 0; @@ -662,10 +811,9 @@ void ReadFromCluster::initializePipeline(QueryPipelineBuilder & pipeline, const if (current_settings[Setting::max_parallel_replicas] > 1) max_replicas_to_use = std::min(max_replicas_to_use, current_settings[Setting::max_parallel_replicas].value); - createExtension(); - ProfileEvents::increment(ProfileEvents::Shards, max_replicas_to_use); + Pipes pipes; for (const auto & shard_info : cluster->getShardsInfo()) { if (pipes.size() >= max_replicas_to_use) @@ -688,14 +836,14 @@ void ReadFromCluster::initializePipeline(QueryPipelineBuilder & pipeline, const auto remote_query_executor = std::make_shared( std::vector{try_results.front()}, query_to_send->formatWithSecretsOneLine(), - getOutputHeader(), + output_header, new_context, /*throttler=*/nullptr, scalars, external_tables.has_value() ? *external_tables : Tables(), processed_stage, nullptr, - RemoteQueryExecutor::Extension{.task_iterator = extension->task_iterator, .replica_info = std::move(replica_info)}, + RemoteQueryExecutor::Extension{.task_iterator = extension.task_iterator, .replica_info = std::move(replica_info)}, shard_info.pool); remote_query_executor->setLogger(log); @@ -711,13 +859,73 @@ void ReadFromCluster::initializePipeline(QueryPipelineBuilder & pipeline, const if (pipes.empty()) throw Exception(ErrorCodes::ALL_CONNECTION_TRIES_FAILED, "Cannot connect to any replica for query execution"); - auto pipe = Pipe::unitePipes(std::move(pipes)); + return Pipe::unitePipes(std::move(pipes)); +} + +} + +void ReadFromCluster::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) +{ + auto new_context = updateSettings(context->getSettingsRef()); + + prepareOrdinaryClusterQueryForRemoteExecution(query_to_send); + + createExtension(); + + auto pipe = buildClusterFunctionRemotePipe( + query_to_send, getOutputHeader(), new_context, cluster, processed_stage, *extension, external_tables, log); + + for (const auto & processor : pipe.getProcessors()) + processors.emplace_back(processor); + + pipeline.init(std::move(pipe)); +} + +ContextPtr ReadFromClusterQuery::updateSettings() const +{ + Settings new_settings{context->getSettingsRef()}; + + /// Cluster table functions should always skip unavailable shards. + new_settings[Setting::skip_unavailable_shards] = true; + + /// Every table in the dispatched query, driver included, must be read locally on the worker. Which one + /// takes its files from this step's task iterator is announced separately, per table, by name. + new_settings[Setting::object_storage_cluster] = ""; + + auto new_context = Context::createCopy(context); + new_context->setSettings(new_settings); + return new_context; +} + +void ReadFromClusterQuery::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) +{ + auto new_context = updateSettings(); + prepareWholeQueryDispatchForRemoteExecution(query_to_send); + + /// Not this step's own filter: its output is the whole query's result, so a predicate over that says + /// nothing about which of the driver's files are needed. The planner extracts the driver's own predicate + /// from the query tree instead and hands it over -- see buildDistributedObjectStorageQueryPlan. + const ActionsDAG::Node * predicate = driver_filter ? driver_filter->getOutputs().front() : nullptr; + auto extension = driver_storage->getTaskIteratorExtension( + predicate, driver_filter.get(), new_context, cluster, driver_snapshot->metadata); + + auto pipe = buildClusterFunctionRemotePipe( + query_to_send, getOutputHeader(), new_context, cluster, processed_stage, extension, external_tables, log); + for (const auto & processor : pipe.getProcessors()) processors.emplace_back(processor); pipeline.init(std::move(pipe)); } +void ReadFromClusterQuery::describeActions(FormatSettings & format_settings) const +{ + std::string prefix(format_settings.offset, format_settings.indent_char); + format_settings.out << prefix << "Cluster: " << cluster->getName() << '\n'; + format_settings.out << prefix << "File tasks from: " << driver_storage->getStorageID().getNameForLogs() << '\n'; + format_settings.out << prefix << "Query: " << query_to_send->formatForLogging() << '\n'; +} + IStorageCluster::QueryTreeInfo IStorageCluster::getQueryTreeInfo(QueryTreeNodePtr query_tree, ContextPtr context) { QueryTreeInfo info; @@ -804,6 +1012,17 @@ ContextPtr ReadFromCluster::updateSettings(const Settings & settings) /// Cluster table functions should always skip unavailable shards. new_settings[Setting::skip_unavailable_shards] = true; + /// A ReadFromCluster reached after whole-query dispatch was declined must behave exactly like `allow`. + if (new_settings[Setting::object_storage_cluster_join_mode] == ObjectStorageClusterJoinMode::DISTRIBUTED) + new_settings[Setting::object_storage_cluster_join_mode] = ObjectStorageClusterJoinMode::ALLOW; + + /// Only whole-query dispatch may name a driver, and it sends that name from its own step. This read + /// supplies a task iterator too, so its workers see `collaborate_with_initiator` and would act on a name + /// that reached them: the table so named would read this step's file-task queue instead of listing its + /// own files. Nothing marks these settings internal, so clear them rather than trust no one set them. + new_settings[Setting::object_storage_distributed_driver_database] = ""; + new_settings[Setting::object_storage_distributed_driver_table] = ""; + auto new_context = Context::createCopy(context); new_context->setSettings(new_settings); return new_context; diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index e316b1985330..878ec3704acc 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -3,6 +3,7 @@ #include #include #include +#include #include namespace DB @@ -53,6 +54,21 @@ class IStorageCluster : public IStorage QueryProcessingStage::Enum getQueryProcessingStage(ContextPtr, QueryProcessingStage::Enum, const StorageSnapshotPtr &, SelectQueryInfo &) const override; + /// Dispatches a whole query the caller has already prepared (see + /// Planner/buildDistributedObjectStorageQueryPlan.h) to this storage's cluster, over the same + /// task-iterator transport read() uses. This storage acts only as the cluster handle and the source of + /// file tasks; `query_to_send` is self-contained and its result is not this table's rows, so the read + /// goes through ReadFromClusterQuery rather than ReadFromCluster. + void readPreparedClusterQuery( + QueryPlan & query_plan, + const StorageSnapshotPtr & storage_snapshot, + SelectQueryInfo & query_info, + ContextPtr context, + QueryProcessingStage::Enum processed_stage, + ASTPtr query_to_send, + SharedHeader sample_block, + std::shared_ptr driver_filter); + bool isRemote() const final { return true; } bool supportsSubcolumns() const override { return true; } bool supportsOptimizationToSubcolumns() const override { return false; } @@ -180,4 +196,56 @@ class ReadFromCluster : public SourceStepWithFilter ContextPtr updateSettings(const Settings & settings); }; + +/// Dispatches one already-prepared query to a cluster, giving every node a share of one table's files. +/// +/// Distinct from ReadFromCluster, which reads a single table: here the output is a whole query's result, so +/// there is no table expression to push filters into and no per-table column mapping. `driver_storage` and +/// `driver_snapshot` describe only the table whose files are handed out, never the step's output. +class ReadFromClusterQuery : public ISourceStep +{ +public: + std::string getName() const override { return "ReadFromClusterQuery"; } + void initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override; + void describeActions(FormatSettings & format_settings) const override; + + ReadFromClusterQuery( + SharedHeader output_header_, + std::shared_ptr driver_storage_, + StorageSnapshotPtr driver_snapshot_, + ContextPtr context_, + ASTPtr query_to_send_, + QueryProcessingStage::Enum processed_stage_, + ClusterPtr cluster_, + LoggerPtr log_, + std::optional external_tables_, + std::shared_ptr driver_filter_) + : ISourceStep(std::move(output_header_)) + , driver_storage(std::move(driver_storage_)) + , driver_snapshot(std::move(driver_snapshot_)) + , context(std::move(context_)) + , query_to_send(std::move(query_to_send_)) + , processed_stage(processed_stage_) + , cluster(std::move(cluster_)) + , log(log_) + , external_tables(std::move(external_tables_)) + , driver_filter(std::move(driver_filter_)) + { + } + +private: + std::shared_ptr driver_storage; + StorageSnapshotPtr driver_snapshot; + ContextPtr context; + ASTPtr query_to_send; + QueryProcessingStage::Enum processed_stage; + ClusterPtr cluster; + LoggerPtr log; + std::optional external_tables; + /// Predicate over the driver's own columns, used to prune its file listing. Null means list everything. + std::shared_ptr driver_filter; + + ContextPtr updateSettings() const; +}; + } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index aff028efef01..863d82378e3c 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -45,6 +45,38 @@ namespace Setting extern const SettingsInt64 delta_lake_snapshot_end_version; extern const SettingsUInt64 lock_object_storage_task_distribution_ms; extern const SettingsBool allow_experimental_iceberg_read_optimization; + extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; + extern const SettingsString object_storage_distributed_driver_database; + extern const SettingsString object_storage_distributed_driver_table; +} + +namespace +{ + +/// True on a worker executing a query dispatched by `object_storage_cluster_join_mode='distributed'`. The +/// initiator names the driving table in the settings it sends; their presence is what marks the query, so a +/// worker never has to infer the mode from the shape of its connection. +bool isDistributedJoinDispatchWorker(const ContextPtr & context) +{ + const auto & client_info = context->getClientInfo(); + return client_info.query_kind == ClientInfo::QueryKind::SECONDARY_QUERY + && client_info.collaborate_with_initiator + && !context->getSettingsRef()[Setting::object_storage_distributed_driver_database].value.empty(); +} + +/// True for the one table in a dispatched query whose files the initiator hands out. Every other table in the +/// same query is read in full on every worker, so this must match exactly one table expression -- which the +/// initiator guarantees by declining to dispatch when the name is ambiguous. +bool isAnnouncedDistributedJoinDriver(const ContextPtr & context, const StorageID & table_id) +{ + if (!isDistributedJoinDispatchWorker(context)) + return false; + + const auto & settings = context->getSettingsRef(); + return table_id.getDatabaseName() == settings[Setting::object_storage_distributed_driver_database].value + && table_id.getTableName() == settings[Setting::object_storage_distributed_driver_table].value; +} + } namespace ErrorCodes @@ -254,9 +286,14 @@ StorageObjectStorageCluster::StorageObjectStorageCluster( && context_->canUseTaskBasedParallelReplicas() && !context_->isDistributed(); - bool can_use_distributed_iterator = - context_->getClientInfo().collaborate_with_initiator && - can_use_parallel_replicas; + /// Two mechanisms can put a storage on an initiator's file-task queue: being the announced driver of a + /// whole-query dispatch, or being an ordinary cluster read under parallel replicas. They must not both + /// get a say, because there is one queue and it holds one table's files. A dispatch owns the decision + /// outright while it is in effect: its announced driver consumes the queue and every other table in that + /// query lists its own files, whatever the parallel-replica settings would otherwise allow. + bool can_use_distributed_iterator = isDistributedJoinDispatchWorker(context_) + ? isAnnouncedDistributedJoinDriver(context_, table_id_) + : (context_->getClientInfo().collaborate_with_initiator && can_use_parallel_replicas); pure_storage = std::make_shared( configuration, @@ -649,6 +686,12 @@ RemoteQueryExecutor::Extension StorageObjectStorageCluster::getTaskIteratorExten ClusterPtr cluster, StorageMetadataPtr storage_metadata_snapshot) const { + /// A catalog table is built with lazy_init, so nothing has resolved the configuration -- and with it the + /// path this listing walks -- by the time we get here. `read` reaches it through StorageObjectStorage, + /// which does its own lazy init; whole-query dispatch calls this directly and would otherwise list no + /// files at all, handing every worker an empty queue and silently returning no rows. Idempotent. + configuration->lazyInitializeIfNeeded(object_storage, local_context); + auto iterator = StorageObjectStorageSource::createFileIterator( configuration, configuration->getQuerySettings(local_context), @@ -747,6 +790,12 @@ String StorageObjectStorageCluster::getClusterName(ContextPtr context) const if (!isClusterSupported()) return ""; + /// A worker executing a whole-query dispatch reads every table it resolves locally, including the driver: + /// nothing here may fan out again. The driver still differs from the rest, but by reading the initiator's + /// file-task queue instead of listing its own files -- decided at construction, see the constructor. + if (isDistributedJoinDispatchWorker(context)) + return ""; + auto cluster_name_from_settings = context->getSettingsRef()[Setting::object_storage_cluster].value; if (cluster_name_from_settings.empty()) cluster_name_from_settings = getOriginalClusterName(); diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index 6936fa38b64f..feb11d7e1721 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -682,7 +682,12 @@ Chunk StorageObjectStorageSource::generate() auto query_condition_cache = Context::getGlobalContextInstance()->getQueryConditionCache(); query_condition_cache->write( storage_id.uuid, - object_info->getFileName(), + /// Full path, not the file name: object storage paths are hierarchical and a + /// Hive-partitioned dataset repeats the same file name in every partition + /// (`day=.../part-00000.parquet`). Keying by name alone makes those files share + /// one cache entry, so a partition that matched nothing marks its namesakes as + /// having no matching row groups and they are skipped unread. + object_info->getPath(), *format_filter_info->condition_hash, format_filter_info->filter_actions_dag->dumpNames(), unmatched_ranges, @@ -827,7 +832,7 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade if (query_condition_cache && !object_info->file_bucket_info) { auto matching_marks = query_condition_cache->read( - storage_id.uuid, object_info->getFileName(), *format_filter_info->condition_hash); + storage_id.uuid, object_info->getPath(), *format_filter_info->condition_hash); if (matching_marks.has_value()) { const auto & marks = *matching_marks; diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 7459389e5604..001b4a90f488 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -1064,6 +1064,325 @@ def test_cluster_select(started_cluster): assert node2.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`", settings={"parallel_replicas_for_cluster_engines": 1, "enable_parallel_replicas": 2, "cluster_for_parallel_replicas": "cluster_simple"}) == 'pablo\n' +def _setup_distributed_join_tables(started_cluster, nodes, test_ref): + """Three DataLake-catalog tables: a driver and two dimensions, so a query can have a nested JOIN + on the right-hand side rather than a single partner table.""" + root_namespace = f"{test_ref}_namespace" + fact_table = f"{test_ref}_fact" + dim_table = f"{test_ref}_dim" + weight_table = f"{test_ref}_weight" + + load_catalog_impl(started_cluster) + for node in nodes: + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + node = nodes[0] + create_clickhouse_iceberg_table( + started_cluster, node, root_namespace, fact_table, "(tag Int32, name String)" + ) + create_clickhouse_iceberg_table( + started_cluster, node, root_namespace, dim_table, "(id Int32, city String)" + ) + create_clickhouse_iceberg_table( + started_cluster, node, root_namespace, weight_table, "(city String, weight Int32)" + ) + + insert_settings = {"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1} + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{fact_table}` VALUES (1, 'john'), (2, 'jack'), (3, 'jill');", + settings=insert_settings, + ) + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{dim_table}` VALUES (1, 'berlin'), (2, 'paris'), (3, 'berlin');", + settings=insert_settings, + ) + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{weight_table}` VALUES ('berlin', 10), ('paris', 20), ('berlin', 5);", + settings=insert_settings, + ) + + return ( + f"{CATALOG_NAME}.`{root_namespace}.{fact_table}`", + f"{CATALOG_NAME}.`{root_namespace}.{dim_table}`", + f"{CATALOG_NAME}.`{root_namespace}.{weight_table}`", + ) + + +def _driver_announcement(qualified_name): + """Splits ``catalog.`ns.table` `` into the database/table pair the initiator announces, which is + what `DatabaseDataLake` builds the driver's StorageID from.""" + database, table = qualified_name.split(".", 1) + return database, table.strip("`") + + +def _assert_dispatched_whole(nodes, query_id, driver_database, driver_table): + """Checks the runtime invariants of a whole-query dispatch, from system.query_log: + + 1. the whole query -- JOIN and GROUP BY included -- reached a worker as a secondary query, carrying + the driver announcement the initiator chose; + 2. the driver crossed the wire as the plain catalog table the user wrote. Nothing is rewritten into a + `*Cluster` table function; an earlier design did that, and this pins that it is not coming back; + 3. every secondary query of this dispatch is that whole query. A partner table fanning out again from + a worker would show up as a further secondary query without the JOIN in it. + """ + for node in nodes: + node.query("SYSTEM FLUSH LOGS system.query_log") + + announced_whole_query = 0 + for node in nodes: + announced_whole_query += int( + node.query( + f""" + SELECT count() + FROM system.query_log + WHERE type = 'QueryStart' AND NOT is_initial_query + AND initial_query_id = '{query_id}' + AND positionCaseInsensitive(query, 'join') != 0 + AND positionCaseInsensitive(query, 'group by') != 0 + AND Settings['object_storage_distributed_driver_database'] = '{driver_database}' + AND Settings['object_storage_distributed_driver_table'] = '{driver_table}' + """ + ).strip() + ) + assert announced_whole_query > 0, ( + f"query {query_id} was not dispatched whole to the cluster, or reached a worker without the " + f"driver announcement naming {driver_database}.{driver_table}" + ) + + for node in nodes: + rewritten = int( + node.query( + f""" + SELECT count() + FROM system.query_log + WHERE type = 'QueryStart' AND NOT is_initial_query + AND initial_query_id = '{query_id}' + AND positionCaseInsensitive(query, 'icebergs3cluster') != 0 + """ + ).strip() + ) + assert rewritten == 0, ( + f"query {query_id}: a dispatched query on {node.name} contains a cluster table function -- " + "the driver must travel as the catalog table the user wrote, named in the settings" + ) + + partner_fanout = int( + node.query( + f""" + SELECT count() + FROM system.query_log + WHERE type = 'QueryStart' AND NOT is_initial_query + AND initial_query_id = '{query_id}' + AND positionCaseInsensitive(query, 'join') = 0 + """ + ).strip() + ) + assert partner_fanout == 0, ( + f"query {query_id}: a secondary query without the JOIN was issued on {node.name} -- " + "a table fanned out again instead of being read locally" + ) + + +def test_distributed_join_dispatch(started_cluster): + """object_storage_cluster_join_mode='distributed': a JOIN whose driving table is a DataLake-catalog + table is dispatched whole to the driver's cluster and finalized on the initiator. + + Correctness is checked against the same queries under join_mode='allow' (ordinary local planning), + which is the oracle -- dispatch must not change results, only where the work happens. + """ + node1 = started_cluster.instances["node1"] + node2 = started_cluster.instances["node2"] + nodes = [node1, node2] + + fact, dim, weight = _setup_distributed_join_tables( + started_cluster, nodes, f"test_distributed_join_{uuid.uuid4()}" + ) + + def run(query, join_mode, query_id=None): + return node1.query( + query, + query_id=query_id, + settings={ + "object_storage_cluster": "cluster_simple", + "object_storage_cluster_join_mode": join_mode, + }, + ) + + # The driver is the immediate leftmost table of the JOIN. + direct_driver = f""" + SELECT d.city, count() AS c + FROM {fact} AS f + INNER JOIN {dim} AS d ON f.tag = d.id + WHERE f.tag < 10 + GROUP BY d.city + ORDER BY ALL + """ + + # The driver is buried in a see-through subquery on the JOIN's left, the aggregation sits on the + # enclosing query, and the right-hand side is itself a JOIN over a grouped subquery -- all of which + # every worker recomputes in full. + buried_driver = f""" + SELECT dims.city, count() AS c + FROM + ( + SELECT tag, name + FROM {fact} + WHERE tag < 10 + ) AS f + LEFT JOIN + ( + SELECT d.id AS id, d.city AS city + FROM {dim} AS d + LEFT JOIN + ( + SELECT city, sum(weight) AS w + FROM {weight} + GROUP BY city + ) AS p ON d.city = p.city + ) AS dims ON f.tag = dims.id + GROUP BY dims.city + ORDER BY ALL + """ + + # The same shape written with CTEs rather than derived tables. The setting documents CTE support, + # and the analyzer represents a CTE as a QueryNode just like a derived table, but the serializer has to + # inline the CTE body for the worker to resolve it -- so this is worth exercising directly. + buried_driver_cte = f""" + WITH + f AS + ( + SELECT tag, name + FROM {fact} + WHERE tag < 10 + ), + p AS + ( + SELECT city, sum(weight) AS w + FROM {weight} + GROUP BY city + ), + dims AS + ( + SELECT d.id AS id, d.city AS city + FROM {dim} AS d + LEFT JOIN p ON d.city = p.city + ) + SELECT dims.city, count() AS c + FROM f + LEFT JOIN dims ON f.tag = dims.id + GROUP BY dims.city + ORDER BY ALL + """ + + for name, query in ( + ("direct driver", direct_driver), + ("buried driver", buried_driver), + ("buried driver via CTE", buried_driver_cte), + ): + expected = run(query, "allow") + assert expected == "berlin\t2\nparis\t1\n", f"{name} oracle result changed: {expected!r}" + + query_id = uuid.uuid4().hex + assert run(query, "distributed", query_id=query_id) == expected, f"{name} differs under dispatch" + _assert_dispatched_whole(nodes, query_id, *_driver_announcement(fact)) + + +def test_distributed_join_dispatch_falls_back(started_cluster): + """Tables the dispatch cannot prove safe fall back to ordinary planning instead of failing: a local + Memory JOIN partner is not resolvable on a worker, so the candidate must be rejected.""" + node1 = started_cluster.instances["node1"] + + test_ref = f"test_distributed_join_fallback_{uuid.uuid4()}" + fact, _, _ = _setup_distributed_join_tables(started_cluster, [node1], test_ref) + + local_table = f"{test_ref}_local" + node1.query(f"CREATE TABLE {local_table} (id Int32, city String) ENGINE = Memory()") + node1.query(f"INSERT INTO {local_table} VALUES (1, 'berlin'), (2, 'paris'), (3, 'berlin')") + + query = f""" + SELECT l.city, count() AS c + FROM {fact} AS f + INNER JOIN {local_table} AS l ON f.tag = l.id + GROUP BY l.city + ORDER BY ALL + """ + + def run(join_mode): + return node1.query( + query, + settings={ + "object_storage_cluster": "cluster_simple", + "object_storage_cluster_join_mode": join_mode, + }, + ) + + assert run("distributed") == run("allow") + + +def test_distributed_join_dispatch_ignores_parallel_replicas_settings(started_cluster): + """A whole-query dispatch must not let the parallel-replicas settings change what a worker reads. + + On a worker every DataLake-catalog table is resolved fresh through `DatabaseDataLake`, and + `StorageObjectStorageCluster`'s constructor decides there and then whether its inner plain storage + consumes the initiator's file-task queue. That decision is made from the cluster name plus the + parallel-replicas settings -- it does not consider which table this is. Under dispatch the driver + owns that queue, so a partner answering yes as well would read the driver's files under its own + schema. + + What prevents that is the constructor asking, first, whether this is a dispatched worker at all: if it + is, only the announced driver consumes the queue and the parallel-replica settings get no say. Make + those two conditions independent again -- the earlier `||` -- and a partner becomes a queue consumer. + The settings below are the combination that makes the parallel-replica half true. + """ + node1 = started_cluster.instances["node1"] + node2 = started_cluster.instances["node2"] + nodes = [node1, node2] + + fact, dim, _ = _setup_distributed_join_tables( + started_cluster, nodes, f"test_distributed_join_pr_{uuid.uuid4()}" + ) + + query = f""" + SELECT d.city, count() AS c + FROM {fact} AS f + INNER JOIN {dim} AS d ON f.tag = d.id + WHERE f.tag < 10 + GROUP BY d.city + ORDER BY ALL + """ + + expected = node1.query( + query, + settings={ + "object_storage_cluster": "cluster_simple", + "object_storage_cluster_join_mode": "allow", + }, + ) + assert expected == "berlin\t2\nparis\t1\n", f"oracle result changed: {expected!r}" + + query_id = uuid.uuid4().hex + got = node1.query( + query, + query_id=query_id, + settings={ + "object_storage_cluster": "cluster_simple", + "object_storage_cluster_join_mode": "distributed", + "parallel_replicas_for_cluster_engines": 1, + "enable_parallel_replicas": 1, + "cluster_for_parallel_replicas": "cluster_simple", + "max_parallel_replicas": 2, + }, + ) + assert got == expected, ( + "dispatch with the parallel-replicas settings on returned a different result than ordinary " + f"planning: {got!r} != {expected!r} -- a partner table most likely consumed the driver's " + "file-task queue" + ) + + # Without this the test would silently stop covering anything if the candidate were rejected + # whenever the parallel-replicas settings are set. + _assert_dispatched_whole(nodes, query_id, *_driver_announcement(fact)) + def test_used_storages_in_query_log(started_cluster): node1 = started_cluster.instances["node1"] node2 = started_cluster.instances["node2"] diff --git a/tests/integration/test_storage_iceberg_with_spark/test_cluster_joins.py b/tests/integration/test_storage_iceberg_with_spark/test_cluster_joins.py index 82e9d6c3c572..7c10db45fba3 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_cluster_joins.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_cluster_joins.py @@ -6,7 +6,7 @@ execute_spark_query_general, ) -@pytest.mark.parametrize("join_mode", ["local", "global"]) +@pytest.mark.parametrize("join_mode", ["local", "global", "distributed"]) @pytest.mark.parametrize("storage_type", ["s3", "azure"]) def test_cluster_joins(started_cluster_iceberg_with_spark, storage_type, join_mode): instance = started_cluster_iceberg_with_spark.instances["node1"] diff --git a/tests/queries/0_stateless/05053_query_condition_cache_object_storage_same_file_name.reference b/tests/queries/0_stateless/05053_query_condition_cache_object_storage_same_file_name.reference new file mode 100644 index 000000000000..75d54317d732 --- /dev/null +++ b/tests/queries/0_stateless/05053_query_condition_cache_object_storage_same_file_name.reference @@ -0,0 +1,3 @@ +1000 +1000 +1000 diff --git a/tests/queries/0_stateless/05053_query_condition_cache_object_storage_same_file_name.sql b/tests/queries/0_stateless/05053_query_condition_cache_object_storage_same_file_name.sql new file mode 100644 index 000000000000..c33bfe640d76 --- /dev/null +++ b/tests/queries/0_stateless/05053_query_condition_cache_object_storage_same_file_name.sql @@ -0,0 +1,28 @@ +-- Tags: no-fasttest +-- Tag no-fasttest: Depends on AWS + +-- The query condition cache is keyed per file. A Hive-partitioned dataset repeats the same file name in +-- every partition, so keying by name alone would make all of them share one entry: the first partition +-- that matches nothing marks its namesakes as having no matching row groups, and every later query skips +-- them unread. Reading the same table twice must return the same rows. + +SET s3_truncate_on_insert = 1; +SET use_query_condition_cache = 1; + +INSERT INTO FUNCTION s3(s3_conn, filename='05053_qcc/day=2025-02-27/part-00000.parquet', format=Parquet) + SELECT toDateTime('2025-02-27 00:00:00') + number AS ts, number AS v FROM numbers(1000); + +INSERT INTO FUNCTION s3(s3_conn, filename='05053_qcc/day=2025-02-05/part-00000.parquet', format=Parquet) + SELECT toDateTime('2025-02-05 00:00:00') + number AS ts, number AS v FROM numbers(1000); + +DROP TABLE IF EXISTS t_05053_qcc; + +-- A table, not a table function: the cache entry only outlives a query when the table id is stable. +CREATE TABLE t_05053_qcc (ts DateTime, v UInt32) + ENGINE = S3(s3_conn, filename='05053_qcc/**.parquet', format=Parquet); + +SELECT count() FROM t_05053_qcc WHERE ts >= '2025-02-05 00:00:00' AND ts < '2025-02-06 00:00:00'; +SELECT count() FROM t_05053_qcc WHERE ts >= '2025-02-05 00:00:00' AND ts < '2025-02-06 00:00:00'; +SELECT count() FROM t_05053_qcc WHERE ts >= '2025-02-05 00:00:00' AND ts < '2025-02-06 00:00:00'; + +DROP TABLE t_05053_qcc;