fix: [branch-54] disable the SortMergeJoinExec broadcast conversion by default

ballista/core/src/config.rs

@@ -103,7 +103,8 @@ pub const BALLISTA_BROADCAST_JOIN_THRESHOLD_BYTES: &str =
 
 /// Configuration key to enable broadcasting a small build side of a
 /// `SortMergeJoinExec` by converting it to a `CollectLeft` hash join in the
-/// static distributed planner. Enabled by default.
+/// static distributed planner. Disabled by default: the conversion can
+/// produce incorrect results, so it is opt-in only.
 pub const BALLISTA_BROADCAST_SORT_MERGE_JOIN_ENABLED: &str =
     "ballista.optimizer.broadcast_sort_merge_join_enabled";
 
@@ -243,9 +244,10 @@ static CONFIG_ENTRIES: LazyLock<HashMap<String, ConfigEntry>> = LazyLock::new(||
         ConfigEntry::new(BALLISTA_BROADCAST_SORT_MERGE_JOIN_ENABLED.to_string(),
                          "Broadcast a small build side of a SortMergeJoinExec by converting it \
                           to a CollectLeft hash join in the static distributed planner. \
-                          The build side must also fit under broadcast_join_threshold_bytes.".to_string(),
+                          The build side must also fit under broadcast_join_threshold_bytes. \
+                          Disabled by default: the conversion can produce incorrect results.".to_string(),
                          DataType::Boolean,
-                         Some(true.to_string())),
+                         Some(false.to_string())),
         ConfigEntry::new(BALLISTA_CLIENT_PULL.to_string(),
                          "Should client employ pull or push job tracking. In pull mode client will make a request to server in the loop, until job finishes. Pull mode is kept for legacy clients.".to_string(),
                          DataType::Boolean,
@@ -570,7 +572,8 @@ impl BallistaConfig {
 
     /// Returns whether broadcasting a small build side of a `SortMergeJoinExec`
     /// (by converting it to a `CollectLeft` hash join) is enabled in the static
-    /// distributed planner.
+    /// distributed planner. Off unless explicitly opted into, because the
+    /// conversion can produce incorrect results.
     pub fn broadcast_sort_merge_join_enabled(&self) -> bool {
         self.get_bool_setting(BALLISTA_BROADCAST_SORT_MERGE_JOIN_ENABLED)
     }
@@ -821,9 +824,11 @@ mod tests {
         Ok(())
     }
 
+    /// The conversion can produce incorrect results, so it must stay off
+    /// unless a user explicitly opts in.
     #[test]
-    fn broadcast_sort_merge_join_enabled_by_default() {
+    fn broadcast_sort_merge_join_disabled_by_default() {
         let config = BallistaConfig::default();
-        assert!(config.broadcast_sort_merge_join_enabled());
+        assert!(!config.broadcast_sort_merge_join_enabled());
     }
 }

ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt

@@ -1,60 +1,57 @@
 === Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
-  StatsExec: rows=25
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
-  UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 3 ===
 SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
   StatsExec: rows=15000000
 
-=== Stage 4 ===
+=== Stage 2 ===
 SortShuffleWriterExec: partitioning=Hash([o_custkey@1], 16)
   FilterExec: o_orderdate@2 >= 1993-10-01 AND o_orderdate@2 < 1994-01-01, projection=[o_orderkey@0, o_custkey@1]
     StatsExec: rows=150000000
 
-=== Stage 5 ===
+=== Stage 3 ===
 SortShuffleWriterExec: partitioning=Hash([o_orderkey@7], 16)
   ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name, c_address@2 as c_address, c_nationkey@3 as c_nationkey, c_phone@4 as c_phone, c_acctbal@5 as c_acctbal, c_comment@6 as c_comment, o_orderkey@7 as o_orderkey]
     SortMergeJoinExec: join_type=Inner, on=[(c_custkey@0, o_custkey@1)]
       SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
-        UnresolvedShuffleExec: stage=3, partitioning: Hash([c_custkey@0], 16)
+        UnresolvedShuffleExec: stage=1, partitioning: Hash([c_custkey@0], 16)
       SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
-        UnresolvedShuffleExec: stage=4, partitioning: Hash([o_custkey@1], 16)
+        UnresolvedShuffleExec: stage=2, partitioning: Hash([o_custkey@1], 16)
 
-=== Stage 6 ===
+=== Stage 4 ===
 SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
   FilterExec: l_returnflag@3 = R, projection=[l_orderkey@0, l_extendedprice@1, l_discount@2]
     StatsExec: rows=600037902
 
-=== Stage 7 ===
+=== Stage 5 ===
 SortShuffleWriterExec: partitioning=Hash([c_nationkey@3], 16)
   ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name, c_address@2 as c_address, c_nationkey@3 as c_nationkey, c_phone@4 as c_phone, c_acctbal@5 as c_acctbal, c_comment@6 as c_comment, l_extendedprice@9 as l_extendedprice, l_discount@10 as l_discount]
     SortMergeJoinExec: join_type=Inner, on=[(o_orderkey@7, l_orderkey@0)]
       SortExec: expr=[o_orderkey@7 ASC], preserve_partitioning=[true]
-        UnresolvedShuffleExec: stage=5, partitioning: Hash([o_orderkey@7], 16)
+        UnresolvedShuffleExec: stage=3, partitioning: Hash([o_orderkey@7], 16)
       SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
-        UnresolvedShuffleExec: stage=6, partitioning: Hash([l_orderkey@0], 16)
+        UnresolvedShuffleExec: stage=4, partitioning: Hash([l_orderkey@0], 16)
 
-=== Stage 8 ===
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+  StatsExec: rows=25
+
+=== Stage 7 ===
 SortShuffleWriterExec: partitioning=Hash([c_custkey@0, c_name@1, c_acctbal@2, c_phone@3, n_name@4, c_address@5, c_comment@6], 16)
   AggregateExec: mode=Partial, gby=[c_custkey@0 as c_custkey, c_name@1 as c_name, c_acctbal@4 as c_acctbal, c_phone@3 as c_phone, n_name@8 as n_name, c_address@2 as c_address, c_comment@5 as c_comment], aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
     ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name, c_address@2 as c_address, c_phone@4 as c_phone, c_acctbal@5 as c_acctbal, c_comment@6 as c_comment, l_extendedprice@7 as l_extendedprice, l_discount@8 as l_discount, n_name@10 as n_name]
-      ProjectionExec: expr=[c_custkey@2 as c_custkey, c_name@3 as c_name, c_address@4 as c_address, c_nationkey@5 as c_nationkey, c_phone@6 as c_phone, c_acctbal@7 as c_acctbal, c_comment@8 as c_comment, l_extendedprice@9 as l_extendedprice, l_discount@10 as l_discount, n_nationkey@0 as n_nationkey, n_name@1 as n_name]
-        HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, c_nationkey@3)]
-          UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
-          UnresolvedShuffleExec: stage=7, partitioning: Hash([c_nationkey@3], 16)
+      SortMergeJoinExec: join_type=Inner, on=[(c_nationkey@3, n_nationkey@0)]
+        SortExec: expr=[c_nationkey@3 ASC], preserve_partitioning=[true]
+          UnresolvedShuffleExec: stage=5, partitioning: Hash([c_nationkey@3], 16)
+        SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+          UnresolvedShuffleExec: stage=6, partitioning: Hash([n_nationkey@0], 16)
 
-=== Stage 9 ===
+=== Stage 8 ===
 ShuffleWriterExec: partitioning: None
   SortExec: TopK(fetch=20), expr=[revenue@2 DESC], preserve_partitioning=[true]
     ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@7 as revenue, c_acctbal@2 as c_acctbal, n_name@4 as n_name, c_address@5 as c_address, c_phone@3 as c_phone, c_comment@6 as c_comment]
       AggregateExec: mode=FinalPartitioned, gby=[c_custkey@0 as c_custkey, c_name@1 as c_name, c_acctbal@2 as c_acctbal, c_phone@3 as c_phone, n_name@4 as n_name, c_address@5 as c_address, c_comment@6 as c_comment], aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
-        UnresolvedShuffleExec: stage=8, partitioning: Hash([c_custkey@0, c_name@1, c_acctbal@2, c_phone@3, n_name@4, c_address@5, c_comment@6], 16)
+        UnresolvedShuffleExec: stage=7, partitioning: Hash([c_custkey@0, c_name@1, c_acctbal@2, c_phone@3, n_name@4, c_address@5, c_comment@6], 16)
 
-=== Stage 10 ===
+=== Stage 9 ===
 ShuffleWriterExec: partitioning: None
   SortPreservingMergeExec: [revenue@2 DESC], fetch=20
-    UnresolvedShuffleExec: stage=9, partitioning: Hash([c_custkey@0, c_name@1, c_acctbal@3, c_phone@6, n_name@4, c_address@5, c_comment@7], 16)
+    UnresolvedShuffleExec: stage=8, partitioning: Hash([c_custkey@0, c_name@1, c_acctbal@3, c_phone@6, n_name@4, c_address@5, c_comment@7], 16)

ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt

@@ -1,87 +1,81 @@
 === Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
-  FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
-    StatsExec: rows=25
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
-  UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 3 ===
 SortShuffleWriterExec: partitioning=Hash([ps_suppkey@0], 16)
   StatsExec: rows=80000000
 
-=== Stage 4 ===
+=== Stage 2 ===
 SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
   StatsExec: rows=1000000
 
-=== Stage 5 ===
+=== Stage 3 ===
 SortShuffleWriterExec: partitioning=Hash([s_nationkey@2], 16)
   ProjectionExec: expr=[ps_availqty@1 as ps_availqty, ps_supplycost@2 as ps_supplycost, s_nationkey@4 as s_nationkey]
     SortMergeJoinExec: join_type=Inner, on=[(ps_suppkey@0, s_suppkey@0)]
       SortExec: expr=[ps_suppkey@0 ASC], preserve_partitioning=[true]
-        UnresolvedShuffleExec: stage=3, partitioning: Hash([ps_suppkey@0], 16)
+        UnresolvedShuffleExec: stage=1, partitioning: Hash([ps_suppkey@0], 16)
       SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
-        UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
+        UnresolvedShuffleExec: stage=2, partitioning: Hash([s_suppkey@0], 16)
 
-=== Stage 6 ===
-ShuffleWriterExec: partitioning: None
-  AggregateExec: mode=Partial, gby=[], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
-    ProjectionExec: expr=[ps_availqty@0 as ps_availqty, ps_supplycost@1 as ps_supplycost]
-      ProjectionExec: expr=[ps_availqty@1 as ps_availqty, ps_supplycost@2 as ps_supplycost, s_nationkey@3 as s_nationkey, n_nationkey@0 as n_nationkey]
-        HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@2)]
-          UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
-          UnresolvedShuffleExec: stage=5, partitioning: Hash([s_nationkey@2], 16)
-
-=== Stage 7 ===
+=== Stage 4 ===
 SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
   FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
     StatsExec: rows=25
 
-=== Stage 8 ===
+=== Stage 5 ===
 ShuffleWriterExec: partitioning: None
-  UnresolvedShuffleExec: stage=7, partitioning: Hash([n_nationkey@0], 16)
+  AggregateExec: mode=Partial, gby=[], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
+    ProjectionExec: expr=[ps_availqty@0 as ps_availqty, ps_supplycost@1 as ps_supplycost]
+      SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@2, n_nationkey@0)]
+        SortExec: expr=[s_nationkey@2 ASC], preserve_partitioning=[true]
+          UnresolvedShuffleExec: stage=3, partitioning: Hash([s_nationkey@2], 16)
+        SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+          UnresolvedShuffleExec: stage=4, partitioning: Hash([n_nationkey@0], 16)
 
-=== Stage 9 ===
+=== Stage 6 ===
 SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1], 16)
   StatsExec: rows=80000000
 
-=== Stage 10 ===
+=== Stage 7 ===
 SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
   StatsExec: rows=1000000
 
-=== Stage 11 ===
+=== Stage 8 ===
 SortShuffleWriterExec: partitioning=Hash([s_nationkey@3], 16)
   ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_availqty@2 as ps_availqty, ps_supplycost@3 as ps_supplycost, s_nationkey@5 as s_nationkey]
     SortMergeJoinExec: join_type=Inner, on=[(ps_suppkey@1, s_suppkey@0)]
       SortExec: expr=[ps_suppkey@1 ASC], preserve_partitioning=[true]
-        UnresolvedShuffleExec: stage=9, partitioning: Hash([ps_suppkey@1], 16)
+        UnresolvedShuffleExec: stage=6, partitioning: Hash([ps_suppkey@1], 16)
       SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
-        UnresolvedShuffleExec: stage=10, partitioning: Hash([s_suppkey@0], 16)
+        UnresolvedShuffleExec: stage=7, partitioning: Hash([s_suppkey@0], 16)
 
-=== Stage 12 ===
+=== Stage 9 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+  FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
+    StatsExec: rows=25
+
+=== Stage 10 ===
 SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
   AggregateExec: mode=Partial, gby=[ps_partkey@0 as ps_partkey], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
     ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_availqty@1 as ps_availqty, ps_supplycost@2 as ps_supplycost]
-      ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_availqty@2 as ps_availqty, ps_supplycost@3 as ps_supplycost, s_nationkey@4 as s_nationkey, n_nationkey@0 as n_nationkey]
-        HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@3)]
-          UnresolvedShuffleExec: stage=8, broadcast=true, upstream_partitions: 16
-          UnresolvedShuffleExec: stage=11, partitioning: Hash([s_nationkey@3], 16)
+      SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@3, n_nationkey@0)]
+        SortExec: expr=[s_nationkey@3 ASC], preserve_partitioning=[true]
+          UnresolvedShuffleExec: stage=8, partitioning: Hash([s_nationkey@3], 16)
+        SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+          UnresolvedShuffleExec: stage=9, partitioning: Hash([n_nationkey@0], 16)
 
-=== Stage 13 ===
+=== Stage 11 ===
 ShuffleWriterExec: partitioning: None
   SortExec: expr=[value@1 DESC], preserve_partitioning=[true]
     ProjectionExec: expr=[ps_partkey@1 as ps_partkey, sum(partsupp.ps_supplycost * partsupp.ps_availqty)@2 as value]
       NestedLoopJoinExec: join_type=Inner, filter=join_proj_push_down_1@1 > sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)@0, projection=[sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)@0, ps_partkey@1, sum(partsupp.ps_supplycost * partsupp.ps_availqty)@2]
         ProjectionExec: expr=[CAST(CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty)@0 AS Float64) * 0.0001 AS Decimal128(38, 15)) as sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)]
           AggregateExec: mode=Final, gby=[], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
             CoalescePartitionsExec
-              UnresolvedShuffleExec: stage=6, partitioning: Hash([s_nationkey@2], 16)
+              UnresolvedShuffleExec: stage=5, partitioning: Hash([n_nationkey@3], 16)
         ProjectionExec: expr=[ps_partkey@0 as ps_partkey, sum(partsupp.ps_supplycost * partsupp.ps_availqty)@1 as sum(partsupp.ps_supplycost * partsupp.ps_availqty), CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty)@1 AS Decimal128(38, 15)) as join_proj_push_down_1]
           AggregateExec: mode=FinalPartitioned, gby=[ps_partkey@0 as ps_partkey], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
-            UnresolvedShuffleExec: stage=12, partitioning: Hash([ps_partkey@0], 16)
+            UnresolvedShuffleExec: stage=10, partitioning: Hash([ps_partkey@0], 16)
 
-=== Stage 14 ===
+=== Stage 12 ===
 ShuffleWriterExec: partitioning: None
   SortPreservingMergeExec: [value@1 DESC]
-    UnresolvedShuffleExec: stage=13, partitioning: Hash([ps_partkey@0], 16)
+    UnresolvedShuffleExec: stage=11, partitioning: Hash([ps_partkey@0], 16)