This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new ce115795f fix: preserve null_equality so distributed INTERSECT/EXCEPT
match NULLs (#2064)
ce115795f is described below
commit ce115795f153f4056c3c35eda5ade7b12c881213
Author: Andy Grove <[email protected]>
AuthorDate: Wed Jul 22 12:31:05 2026 -0600
fix: preserve null_equality so distributed INTERSECT/EXCEPT match NULLs
(#2064)
`INTERSECT` and `EXCEPT` lower to semi/anti joins carrying
`NullEquality::NullEqualsNull` so that NULL matches NULL. Ballista sends the
client's logical plan to the scheduler as protobuf, and datafusion-proto's
logical `JoinNode` decode rebuilt the join through `LogicalPlanBuilder`,
whose
`join_with_expr_keys` / `join_using` hardcode `NullEqualsNothing`. The
encoded
`null_equality` was therefore dropped on decode and the scheduler planned a
different query than the client asked for: NULL stopped matching itself, so
INTERSECT dropped NULL rows and EXCEPT retained them.
This reproduces with a single partition, so it is not a shuffle or
join-strategy artifact, and it affects both hash and sort-merge joins:
select count(*) from (select distinct c_last_name from customer
intersect
select distinct c_last_name from customer);
-- 4972, expected 4973
Fixed upstream by apache/datafusion#22104 and backported to `branch-54` as
apache/datafusion#22785, which is not in the released 54.0.0 tag. Pin
DataFusion to the `branch-54` tip to pick it up; the pin can be dropped
once a
54.x patch release containing it is published.
TPC-DS q38 (INTERSECT) and q87 (EXCEPT) now match single-process DataFusion
and
join the correctness gate, taking it from 86 to 88 queries. q4 and q78
remain
skipped: neither uses a set operation and both are unaffected by this fix,
so
they have a separate root cause.
---
ballista/core/src/serde/mod.rs | 56 ++++++++++++++++++++++++++++++++++++++++++
benchmarks/src/bin/tpcds.rs | 11 ---------
2 files changed, 56 insertions(+), 11 deletions(-)
diff --git a/ballista/core/src/serde/mod.rs b/ballista/core/src/serde/mod.rs
index 40df51bd4..9b826ee2f 100644
--- a/ballista/core/src/serde/mod.rs
+++ b/ballista/core/src/serde/mod.rs
@@ -1432,6 +1432,62 @@ mod test {
assert_eq!(decoded_exec.partition.len(), 1);
}
+ /// `INTERSECT` / `EXCEPT` lower to semi/anti joins with
+ /// `NullEquality::NullEqualsNull` so that NULL matches NULL. Ballista
ships
+ /// the client's *logical* plan to the scheduler as protobuf, so that flag
+ /// has to survive the roundtrip: if it decodes back as `NullEqualsNothing`
+ /// the scheduler silently plans a different query than the client asked
for
+ /// (NULL stops matching itself, so INTERSECT drops rows and EXCEPT keeps
+ /// them). See https://github.com/apache/datafusion-ballista/issues/2046
+ ///
+ /// `null_equality` is not part of the rendered plan, so this inspects the
+ /// decoded `Join` directly rather than asserting on a plan string.
+ #[tokio::test]
+ async fn logical_join_roundtrip_preserves_null_equality() {
+ use datafusion::common::NullEquality;
+ use datafusion::logical_expr::LogicalPlan;
+
+ fn join_null_equality(plan: &LogicalPlan) -> Option<NullEquality> {
+ if let LogicalPlan::Join(join) = plan {
+ return Some(join.null_equality);
+ }
+ plan.inputs().into_iter().find_map(join_null_equality)
+ }
+
+ let ctx = SessionContext::new();
+ ctx.register_parquet(
+ "t",
+ "../../examples/testdata/alltypes_plain.parquet",
+ datafusion::prelude::ParquetReadOptions::default(),
+ )
+ .await
+ .unwrap();
+
+ let plan = ctx
+ .state()
+ .create_logical_plan(
+ "select string_col from t intersect select string_col from t",
+ )
+ .await
+ .unwrap();
+ let plan = ctx.state().optimize(&plan).unwrap();
+ assert_eq!(
+ join_null_equality(&plan),
+ Some(NullEquality::NullEqualsNull),
+ "precondition: INTERSECT should plan a NullEqualsNull join"
+ );
+
+ let codec = BallistaLogicalExtensionCodec::default();
+ let node = LogicalPlanNode::try_from_logical_plan(&plan,
&codec).unwrap();
+ let roundtripped = node.try_into_logical_plan(&ctx.task_ctx(),
&codec).unwrap();
+
+ assert_eq!(
+ join_null_equality(&roundtripped),
+ Some(NullEquality::NullEqualsNull),
+ "null_equality must survive the logical plan protobuf roundtrip"
+ );
+ }
+
/// `RuntimeStatsExec` in sketching mode round-trips its ORDER BY
/// and keeps the sketch accessor available. Row-count accessor is
/// always available (and empty on a fresh operator).
diff --git a/benchmarks/src/bin/tpcds.rs b/benchmarks/src/bin/tpcds.rs
index b55cdfeac..630e6f2b4 100644
--- a/benchmarks/src/bin/tpcds.rs
+++ b/benchmarks/src/bin/tpcds.rs
@@ -62,17 +62,6 @@ const TABLES: &[&str] = &[
/// under the default (static) planner; this list reflects that configuration.
/// Remove an entry as the underlying cause is fixed.
const SKIP: &[(usize, &str)] = &[
- // Distributed execution diverges from single-process DataFusion on the
same
- // data (confirmed reproducible; DataFusion is correct under both join
modes).
- // See https://github.com/apache/datafusion-ballista/issues/2046
- (
- 38,
- "distributed INTERSECT diverges from DataFusion (issue #2046)",
- ),
- (
- 87,
- "distributed EXCEPT diverges from DataFusion (issue #2046)",
- ),
// Non-deterministic: LIMIT/ORDER BY ties without a total order make the
// result vary run-to-run in both engines, so a row-by-row diff is
unstable.
(31, "non-deterministic (ORDER BY ties; varies run-to-run)"),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
