disable HashJoinExec sideways information passing for partitioned queries

Author: adriangb

datafusion/physical-plan/src/joins/hash_join.rs

@@ -905,8 +905,9 @@ impl ExecutionPlan for HashJoinExec {
                     reservation,
                     need_produce_result_in_final(self.join_type),
                     1,
-                    enable_dynamic_filter_pushdown
-                        .then_some(Arc::clone(&self.dynamic_filter)),
+                    // Disable dynamic filter bounds for partitioned mode to avoid race conditions
+                    // where multiple partitions overwrite each other's bounds
+                    None,
                     on_right.clone(),
                 ))
             }

datafusion/sqllogictest/test_files/push_down_filter.slt

@@ -288,3 +288,33 @@ physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/
 
 statement ok
 drop table t;
+
+# Regression test for https://github.com/apache/datafusion/issues/17188
+query I
+COPY (select i as k from generate_series(1, 10000000) as t(i))
+TO 'test_files/scratch/push_down_filter/t1.parquet'
+STORED AS PARQUET;
+----
+10000000
+
+query I
+COPY (select i as k, i as v from generate_series(1, 10000000) as t(i))
+TO 'test_files/scratch/push_down_filter/t2.parquet'
+STORED AS PARQUET;
+----
+10000000
+
+statement ok
+create external table t1 stored as parquet location 'test_files/scratch/push_down_filter/t1.parquet';
+
+statement ok
+create external table t2 stored as parquet location 'test_files/scratch/push_down_filter/t2.parquet';
+
+query III
+select *
+from t1
+join t2 on t1.k = t2.k
+where v = 1 or v = 10000000;
+----
+1 1 1
+10000000 10000000 10000000