Workflow [PR], commit [10cdc0b]

Summary: ❌

• Performance Comparison: Performance dashboard

AI Review

Summary

This PR adds an experimental Cascades cost-based optimizer for distributed query plans. The overall design is substantial, but the current head still has several live contract violations in plan reconstruction, distributed read consistency, and task routing that can either change query results or reject otherwise valid queries late in execution, so this is not ready to merge.

Findings

❌ Blockers

• [src/Processors/QueryPlan/Optimizations/Cascades/Rules/DefaultImplementation.cpp:30-46] DefaultImplementation still admits steps that the Cascades plan builder cannot reconstruct. buildBestPlan unconditionally calls clone on the chosen step, so plans containing ReadFromPreparedSource / ReadFromStorageStep can still die with NOT_IMPLEMENTED during optimization instead of either staying local or being rejected up front. Suggested fix: reject non-cloneable steps before memo construction, or restrict DefaultImplementation to step types with a working clone.
• [src/Processors/QueryPlan/Optimizations/Cascades/Rules/JoinCommutativity.cpp:24-47] JoinCommutativity still ignores JoinStepLogical::isOptimized. That lets Cascades swap USING joins whose changed_types contract has already been baked into actions_after_join, but swapInputs does not rebuild that projection/cast contract, so the swapped plan can return different semantics from the original join. Suggested fix: skip isOptimized joins here and in any other transformation that reconstructs a logical join.
• [src/Processors/QueryPlan/Optimizations/Cascades/Rules/ParallelReadImplementation.cpp:101-136] ReplicatedReadImplementation still advertises a replicated full copy for any ReadFromMergeTree, but it neither proves that every worker sees the same complete table nor pins the coordinator snapshot. In a broadcast join, different workers can therefore read different or partial right-hand sides, changing query results. Suggested fix: gate ReplicatedRead to storage layouts with a guaranteed shared full snapshot, and serialize the coordinator-selected part list for the replicated path too; otherwise use a single read plus BroadcastExchangeStep.
• [src/Processors/QueryPlan/Optimizations/makeDistributed.cpp:1010-1039] reconciling 1 shard with N shards still clones the first single-shard task into every destination shard and then merges parameters with insert. If the first child is single-shard and the second child is distributed, the copied bucket_id / total_buckets stay at 0 / 1, so every combined task reads bucket 0 from the distributed side and the other buckets are dropped. Suggested fix: build the expanded task from the N-shard side, or overwrite routing parameters after merging child task parameters.
• [src/Processors/QueryPlan/ReadFromMergeTree.cpp:3797-3815, 5102-5113] the serialized pinned-part contract is still wrong in two ways: an empty coordinator snapshot is indistinguishable from “no pinned list”, and non-empty part lists are re-sorted lexicographically before worker bucketing. That lets workers read parts the coordinator did not see and can also break the execution order required by distributed requestReadingInOrder. Suggested fix: serialize an explicit “pinned list present” flag and preserve the coordinator’s original parts_with_ranges order exactly.
• [src/QueryPipeline/DistributedPlanExecutor.cpp:970-975] chooseTaskSerializationVersion still decides whether version 1 is safe by comparing producer ports with the initiator’s distributed_query.streaming_exchange_port. With per-node exchange ports, a consumer whose local fallback port differs can still receive a version-1 task and dial producer:<consumer_port> instead of the producer’s real port. Suggested fix: keep version 2 whenever producer endpoints may differ from the destination worker’s fallback port, or base the decision on the destination worker port rather than the initiator’s port.

⚠️ Majors

• [src/Processors/QueryPlan/Optimizations/QueryPlanOptimizationSettings.cpp:213-215] projection rewrites still stay enabled under make_distributed_plan, while ReadFromMergeTree::isSerializable at [src/Processors/QueryPlan/ReadFromMergeTree.h:432] still reports every bucketed read as serializable. Unsupported distributed reads from projections, deferred FINAL filters, or direct text-index tasks therefore slip past planning-time validation and fail only later when serializeQueryPlan walks the fragment. Suggested fix: either disable normal / forced projection rewrites for distributed plans, or make isSerializable reflect the bucketed-read constraints and rerun the distributed-read support check after projection selection.
• [src/Processors/QueryPlan/Optimizations/Cascades/Rules/AggregationImplementations.cpp:266-276] the two-phase merge path still hardcodes memory_efficient_aggregation to false, so enabling distributed_aggregation_memory_efficient = 1 no longer affects Cascades-built distributed aggregation even though that setting defaults to enabled in the legacy path. Suggested fix: thread the optimization setting into Cascades and pass the same distributed_aggregation_memory_efficient && !has_grouping_sets contract used by the rule-based planner.
• [src/Processors/QueryPlan/AggregatingStep.h:87-90] the new in-order aggregation serialization payload is still unreachable because AggregatingStep::isSerializable rejects every step with a non-empty sort_description_for_merging. A remote fragment that legitimately contains such an aggregation still fails the pre-check with make_distributed_plan cannot distribute this query instead of using the serialized state. Suggested fix: make isSerializable match the serializer, or keep rejecting / planning away from in-order aggregation consistently before those steps enter a distributed fragment.
• [src/Processors/QueryPlan/Optimizations/Cascades/Rules/TopNImplementation.cpp:136-137] the internal TopN merge cap is still a real LimitStep built with always_read_till_end = false. Under exact_rows_before_limit = 1, rows_before_limit_at_least then stops at that internal cap and under-reports the true pre-limit row count for distributed top-N queries. Suggested fix: either skip TwoStageTopN when the user-visible limit must read till end, or replace the internal cap with an implementation that does not block the upstream counter.
• [src/Processors/QueryPlan/Optimizations/Cascades/StatisticsDerivation.cpp:366-370] deriveFilterStatistics still models every FilterStep as selectivity 1: it remaps NDVs but leaves row counts unchanged. Real post-join or computed filters can therefore be treated as if they removed no rows at all, which can flip broadcast-vs-shuffle and other downstream cost choices. Suggested fix: apply at least a heuristic row-count reduction here, or reuse the existing selectivity estimator when the filter remains close enough to a table-read path.

Tests

• ⚠️ Add a worker-backed regression for ReplicatedRead where the build-side table is either local/sharded or changes between task starts; that is the smallest proof that every worker sees the same right-hand-side snapshot.
• ⚠️ Add a distributed ORDER BY ... LIMIT regression with exact_rows_before_limit = 1 and Cascades enabled to prove that rows_before_limit_at_least survives the internal top-N merge cap.
• ⚠️ Add a focused distributed-read regression that selects a projection or deferred-FINAL / direct text-index path under make_distributed_plan and proves the query is rejected during support checks rather than later during fragment serialization.

Final Verdict

Status: ❌ Block

Minimum required actions:

• Fix the live wrong-results / execution-contract bugs in ReplicatedReadImplementation, task-shard reconciliation, pinned distributed reads, JoinCommutativity, and exchange-port serialization.
• Make unsupported distributed-read shapes fail closed during planning instead of during fragment serialization.
• Add focused regressions for the distributed snapshot, exact_rows_before_limit, and fail-close validation paths above.

LLVM Coverage Report

Changed lines: Changed C/C++ lines covered: 2410/2681 (89.89%) · Uncovered code

Full report · Diff report