From 6ed5873e5818227b3bd0b54cba398a2e033ac927 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 1 Jul 2026 12:58:09 -0700 Subject: [PATCH] [dbsp] Fix: join hints were not validated correctly. Hints were supposed to get validated before getting installed as balancer constraints. A silly error caused the validation to run on the balancer config without the hint. As a result the pipeline could crash later where it assumes that the current set of constraints is satisfiable. Signed-off-by: Leonid Ryzhyk --- .../src/operator/dynamic/balance/balancer.rs | 48 ++++++++++++------- .../dbsp/src/operator/dynamic/balance/test.rs | 45 +++++++++++++++++ 2 files changed, 75 insertions(+), 18 deletions(-) diff --git a/crates/dbsp/src/operator/dynamic/balance/balancer.rs b/crates/dbsp/src/operator/dynamic/balance/balancer.rs index 4494ef79b29..c6a1d1ac6b0 100644 --- a/crates/dbsp/src/operator/dynamic/balance/balancer.rs +++ b/crates/dbsp/src/operator/dynamic/balance/balancer.rs @@ -778,26 +778,38 @@ impl BalancerInner { // ); let cluster_index = *self.stream_to_cluster.get(&node_id).unwrap(); - let (_exchange_sender, _persistent_id, hints) = self - .integrals - .get_mut(&node_id) - .ok_or(BalancerError::NotRegisteredWithBalancer(node_id))?; - - let mut hints = hints.clone(); - - match hint { - BalancerHint::Policy(policy) => { - if let Some(policy) = policy { - hints.policy_hint = Some(policy); - self.solve_cluster(cluster_index, false)?; - } else { - hints.policy_hint = None; - } + // Apply the hint to the stored hints, remembering the previous value so we + // can roll back if the hint turns out to be unenforceable. The hint must be + // stored *before* validating below, because `solve_cluster` reads the hint + // from `self.integrals`; validating against the pre-hint state would accept + // hints that make the cluster unsatisfiable and crash at the first step. + let old_hints = { + let (_exchange_sender, _persistent_id, hints) = self + .integrals + .get_mut(&node_id) + .ok_or(BalancerError::NotRegisteredWithBalancer(node_id))?; + + let old_hints = hints.clone(); + match hint { + BalancerHint::Policy(policy) => hints.policy_hint = policy, + BalancerHint::Size(size) => hints.size_hint = size, + BalancerHint::Skew(skew) => hints.skew_hint = skew, } - BalancerHint::Size(size) => hints.size_hint = size, - BalancerHint::Skew(skew) => hints.skew_hint = skew, + old_hints + }; + + if let BalancerHint::Policy(Some(policy)) = hint + && self.solve_cluster(cluster_index, false).is_err() + { + self.integrals.get_mut(&node_id).unwrap().2 = old_hints; + return Err(BalancerError::InvalidPolicyHint( + policy, + format!( + "'{policy}' policy is incompatible with the partitioning constraints of the streams it joins with" + ), + )); } - self.integrals.get_mut(&node_id).unwrap().2 = hints.clone(); + self.clusters[cluster_index].refresh_requested = true; Ok(()) diff --git a/crates/dbsp/src/operator/dynamic/balance/test.rs b/crates/dbsp/src/operator/dynamic/balance/test.rs index a1a795b1736..7c6a83e3196 100644 --- a/crates/dbsp/src/operator/dynamic/balance/test.rs +++ b/crates/dbsp/src/operator/dynamic/balance/test.rs @@ -1147,6 +1147,51 @@ fn test_accumulate_trace_with_balancer_policy_returns_to_trace_policy() { assert_eq!(output_trace, expected_output_trace); } +/// A `Broadcast` policy hint on the outer (left) input of a left join cannot be +/// honored: the left side of a left join must be Shard or Balance. The hint must +/// be rejected when it is installed, rather than accepted and then crashing the +/// circuit with `NoSolution` on the first step. +/// +/// Regression test: `set_hint` used to validate the cluster *before* persisting +/// the hint, so it checked a still-satisfiable (hint-free) instance, accepted the +/// hint, and the first step then solved the now-unsatisfiable instance and +/// panicked in `solve_all_clusters().unwrap()`. This is the crash produced by a +/// SQL `/*+ broadcast(pt) */` hint where `pt` is the outer input of a left join. +#[test] +fn test_reject_broadcast_hint_on_left_join_outer_input() { + init_test_logger(); + + let workers = 4; + let ( + mut circuit, + (left_input_handle, right_input_handle, left_input_node_id, _right_input_node_id, _output), + ) = Runtime::init_circuit( + CircuitConfig::from(workers) + .with_splitter_chunk_size_records(2) + .with_balancer_min_absolute_improvement_threshold(0) + .with_mode(Mode::Persistent), + left_join_with_balancer_test_circuit, + ) + .unwrap(); + + // Broadcasting the outer input of a left join is infeasible, so the hint must + // be rejected here instead of being accepted and crashing at the first step. + let result = circuit.set_balancer_hint( + &left_input_node_id, + BalancerHint::Policy(Some(PartitioningPolicy::Broadcast)), + ); + assert!( + result.is_err(), + "broadcast hint on the outer input of a left join must be rejected, got {result:?}" + ); + + // With the hint rejected and rolled back, the circuit runs normally instead of + // crashing with `NoSolution` on the first step. + left_input_handle.push(1, (1, 1)); + right_input_handle.push(1, (1, 1)); + circuit.transaction().unwrap(); +} + /// Join a large left collection with a small right collection. Both are skewed. /// The balancer should balance the left collection using Policy::Balance and the /// right collection using Policy::Broadcast.