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 { + 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)"),