fix(cubestore): transmit the router's planning flags with the query - #11628
Conversation
A select worker plans its own half of a split plan from the logical plan it receives, and it read `group_by_limit_factor` and `topk_strategy` from its own configuration. When the values at the two ends of a hop disagree, the halves do not fit together and the query returns silently wrong rows instead of failing. Both flags now travel in `WorkerPlanningParams` as `PlanningFlags`, stamped by the router into `ClusterSendExec` and reused by the worker for both hops. The field is optional: a sender that omits it predates the flags and planned from its own configuration, so the receiver falls back to its own configuration rather than to a hardcoded default.
github-actions
Bot
added
cube store
|
Claude finished @waralexrom's task in 4m 21s —— View job Review: sound fix, no correctness blockers — 0 high, 1 medium, 2 low (all posted inline)Full review
VerdictThe diagnosis and the fix both hold up. I independently checked the two claims the PR rests on and they're not just empirical — they're true by construction, which is a stronger statement than the PR makes for itself:
The compatibility machinery is justified. I checked whether Propagation is complete: every
Release ordering as stated is correct and load-bearing: a new router paired with an old worker is exactly the case no fallback can close, so this has to precede #11600. Findings
Finding 1 is the only one I'd want addressed before merge, and it's a test gap rather than a defect in the shipped behavior. Not checkedI did not compile or run the suite — cubestore's build is long and CI covers it. The review is by reading; the PR reports lib/cluster/in-process/migration/multi-process suites green plus the two skew counter-examples, and the flexbuffers behavior the compat story depends on is pinned by the three new wire tests. Security, performance, and docs: nothing to raise. No new inputs cross a trust boundary (the flags travel on the existing intra-cluster channel, which already carries the whole serialized plan), the change is plan-construction only with no runtime cost, and neither env var is documented in |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
Codecov Report✅ All modified and coverable lines are covered by tests.
Additional details and impacted files@@ Coverage Diff @@
## master #11628 +/- ##
===========================================
- Coverage 84.06% 59.34% -24.72%
===========================================
Files 257 228 -29
Lines 81582 18222 -63360
Branches 0 3675 +3675
===========================================
- Hits 68581 10814 -57767
+ Misses 13001 6861 -6140
- Partials 0 547 +547
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Cover the line the whole mechanism turns on: `worker_planning_params()` emitting `Some(flags)`. A regression to `None` would have been papered over by the receiver's configuration fallback, with the silently-wrong-rows failure mode this is meant to close. Pin the strategy names on the wire with explicit `serde(rename)` (the same names `CUBESTORE_TOPK_STRATEGY` accepts), so renaming a variant cannot break a mixed-version cluster, and state the deployment constraint the fallback cannot cover: a value set on the router alone is not reproducible on a receiver that gets no flags.
|
Claude finished @waralexrom's task in 1m 46s —— View job Re-review of
|
| # | Prior finding | Status |
|---|---|---|
| 1 | planning_flags_fall_back_to_the_receiver_configuration tested only from_config; nothing pinned worker_planning_params() emitting Some(flags) |
Addressed |
| 2 | TopKAggregateStrategy wire names were the Rust identifiers, implicitly |
Addressed |
| 3 | PlanningFlags could derive PartialEq, Eq |
Addressed |
1. The test is renamed to planning_flags_from_config with a doc comment saying what it actually covers, and cluster_send_exec_sends_its_planning_flags (query_executor.rs:2556) now covers the load-bearing line: it builds a ClusterSendExec with topk_strategy: FullMerge / group_by_limit_factor: 3 and asserts params.flags == Some(flags). A regression of that line to None now fails a test instead of being papered over by the receiver's config fallback. Using an exhaustive struct literal rather than a constructor is the right choice here — a future field that forgets to travel makes this test stop compiling.
I checked the pieces the new test leans on rather than assuming: PreSerializedPlan::try_new is the 3-arg (LogicalPlan, PlanningMeta, Option<String>) form (serialized_plan.rs:876), the PlanningMeta literal covers all three fields (planning.rs:93-105), MockCluster is generated unconditionally by #[automock] (cluster/mod.rs:87) so it's available here, and the struct literal matches all ten ClusterSendExec fields (query_executor.rs:1473-1492).
2. #[serde(rename = "streaming" | "vectorized_streaming" | "full_merge")], and planning_flags_strategy_wire_names pins each one by deserializing from a hand-built struct with the strategy as a &str — so a rename that changes the wire name fails the test. The three names are all accepted by env_topk_strategy (config/mod.rs:1353-1357), so the doc claim that these are "the same names CUBESTORE_TOPK_STRATEGY accepts" holds (env parsing is lenient and also takes vectorized, v2, full-merge, …; the serde names are a subset, which is the right direction).
3. Derived, and the round-trip test now asserts params.flags == Some(flags) in one line.
Two doc changes beyond what was asked, both good: group_by_limit_per_partition now says at its definition why it stays out of PlanningFlags, and the fallback comment in worker_context (query_executor.rs:582-586) now names the one deployment it cannot reproduce — a value set on the router alone — instead of implying the fallback is always exact. That was the sharper version of the claim; the earlier wording overstated it slightly.
New issues
Nothing substantive. One cosmetic nit: config/mod.rs:1330 runs to ~150 chars where the rest of that doc block wraps at 100 —
/// the query rather than fall back to a value the sender did not plan with. The receiving node fails to deserialize the message and drops the connection, so
rustfmt doesn't reflow doc comments, so CI won't catch it; a manual rewrap is all it needs.
Not checked
cargo check --tests -p cubestore isn't in this job's allowlist, so the new test code is reviewed by reading, not compiled — CI covers it. Everything the two new tests reference is verified against the current source (signatures, field lists, MockCluster availability), so I'd expect it to build, but that's inference rather than a green compile.
Keep the three facts that matter -- both halves must be planned from one value, the wire names are the env names, an unknown strategy fails the deserialize -- and drop the restatements.
|
Claude finished @waralexrom's task in 1m 17s —— View job Re-review of
|
| /// | ||
| /// Both halves of a split plan must be planned from the same value, or the router combines a worker | ||
| /// stream whose ordering it does not have and returns wrong rows instead of failing. So it rides in | ||
| /// [`PlanningFlags`], and a worker uses its own value only when the sender sent no flags. |
There was a problem hiding this comment.
Nit (cosmetic, applies to all three sites — this line plus :590 and :600): PlanningFlags lives in crate::cluster and isn't imported into config/mod.rs (only Cluster, ClusterImpl, ClusterMetaStoreClient are, at :14), so [PlanningFlags] doesn't resolve and cargo doc reports broken_intra_doc_links. The reference reads fine to a human either way; [crate::cluster::PlanningFlags] would also make it clickable.
Otherwise the rewrap is good — the ~150-char line from the previous head is gone, and the shortened text keeps the load-bearing claims (both halves must plan from the same value; fallback only when no flags were sent; no catch-all variant).
2 participants
A select worker does not execute the physical plan the router sends it -- it plans its own
half from the logical plan it receives (
worker_plan->worker_context->CubeQueryPlanner::new_on_worker), and it read the planning flags from its ownconfiguration. Two hops are involved: router -> worker (
NetworkMessage::Select) andworker -> select subprocess (
WorkerMessage::Select, where the subprocess is respawned withConfig::default(), i.e. from env).When the values at the two ends disagree, the two halves of a split plan do not fit
together and the query returns silently wrong rows rather than failing. Measured by
skewing the router and the select subprocess:
topk_strategy: FullMerge / Streamingurl10instead ofurl2)group_by_limit_factor: 0 / 2limit_pushdown_group_nonprefix_order: sum100instead of1110group_by_limit_factor: 2 / 0coalesce_under_hash_aggregate,push_partial_aggregate_below_mergeSo the bundle holds exactly
topk_strategyandgroup_by_limit_factor.group_by_limit_per_partitionis deliberately left out: it only changes the layout insidethe worker (both
resort_worker_subtreepaths end in aCoalescePartitionsExec, so theschema and partition count the router sees are the same either way).
What this does
PlanningFlags { group_by_limit_factor, topk_strategy }, carried as a field ofWorkerPlanningParams-- a struct that already travelled on both hops.ClusterSendExec; the worker plans from what itreceived.
Option<PlanningFlags>with#[serde(default)]. A sender that does not sendit is an older binary, which planned its half from its own configuration -- so the
receiver falls back to its own configuration (
worker_context), not to a hardcodeddefault. Since this branch does not change any default, that fallback is exact in both
cases: the value is set explicitly on every node of a cluster, or it is unset and both
binaries default to the same thing.
WorkerExec::newtakesworker_partition_count: usizeinstead of the wholeWorkerPlanningParams-- it only ever read that field, and this way the struct can growwithout touching plan construction.
TopKAggregateStrategygets serde derives deliberately without#[serde(other)]: astrategy from a newer node that this binary does not know must fail the query loudly.
cluster::testspin the wire contract (message without the flags, olderreceiver, round trip) and one pins the configuration fallback. flexbuffers+serde
compatibility was checked empirically in both directions.
Release ordering -- important
This must ship before the default flip (
cubestore-change-perf-defaults, #11600). Theflags only help when both nodes are new: an older worker does not know the field and plans
from its own configuration. If the default flip lands at the same time or earlier, there is
a "new router + old worker" window with silently wrong results that no fallback can close.
Testing
cargo test --no-fail-fastinrust/cubestore-- lib, cluster, in-process, migration andmulti-process suites green.
Counter-examples that pass because of this branch (the subprocess with a skewed env now
obeys the router):
and the reverse skew (
group_by_limit_factor: 0inConfig::test, run withCUBESTORE_GROUP_BY_LIMIT_FACTOR=2):limit_pushdown_group_nonprefix_orderpasses, wherewithout the branch it returned 100 instead of 1110.