diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index a26761107a115..a98c1b7bcf98b 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -18,7 +18,7 @@ use std::sync::{Arc, LazyLock}; use arrow::{ - array::record_batch, + array::{RecordBatch, record_batch}, datatypes::{DataType, Field, Schema, SchemaRef}, util::pretty::pretty_format_batches, }; @@ -55,7 +55,8 @@ use datafusion_physical_expr::{ utils::conjunction, }; use datafusion_physical_expr::{ - Partitioning, ScalarFunctionExpr, aggregate::AggregateExprBuilder, + Partitioning, RangePartitioning, ScalarFunctionExpr, SplitPoint, + aggregate::AggregateExprBuilder, }; use datafusion_physical_optimizer::{ PhysicalOptimizerRule, filter_pushdown::FilterPushdown, @@ -187,9 +188,6 @@ fn test_pushdown_into_scan_with_config_options() { // distinction this test exercises is not reachable via SQL. #[tokio::test] async fn test_static_filter_pushdown_through_hash_join() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; - // Create build side with limited values let build_batches = vec![ record_batch!( @@ -945,15 +943,73 @@ async fn test_topk_filter_passes_through_coalesce_partitions() { ); } +fn hashjoin_pushdown_scans() -> ( + SchemaRef, + Arc, + SchemaRef, + Arc, +) { + let build_side_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + Field::new("c", DataType::Float64, false), + ])); + let build_scan = TestScanBuilder::new(Arc::clone(&build_side_schema)) + .with_support(true) + .with_batches(vec![ + record_batch!( + ("a", Utf8, ["aa", "ab"]), + ("b", Utf8, ["ba", "bb"]), + ("c", Float64, [1.0, 2.0]) + ) + .unwrap(), + ]) + .build(); + + let probe_side_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + Field::new("e", DataType::Float64, false), + ])); + let probe_scan = TestScanBuilder::new(Arc::clone(&probe_side_schema)) + .with_support(true) + .with_batches(vec![ + record_batch!( + ("a", Utf8, ["aa", "ab", "ac", "ad"]), + ("b", Utf8, ["ba", "bb", "bc", "bd"]), + ("e", Float64, [1.0, 2.0, 3.0, 4.0]) + ) + .unwrap(), + ]) + .build(); + + (build_side_schema, build_scan, probe_side_schema, probe_scan) +} + +async fn optimize_and_collect_pushdown_plan( + plan: Arc, + config: ConfigOptions, +) -> (Arc, Vec) { + let plan = FilterPushdown::new_post_optimization() + .optimize(plan, &config) + .unwrap(); + let session_ctx = + SessionContext::new_with_config(SessionConfig::from(config).with_batch_size(10)); + session_ctx.register_object_store( + ObjectStoreUrl::parse("test://").unwrap().as_ref(), + Arc::new(InMemory::new()), + ); + let task_ctx = session_ctx.state().task_ctx(); + let batches = collect(Arc::clone(&plan), task_ctx).await.unwrap(); + (plan, batches) +} + // Not portable to sqllogictest: this test pins `PartitionMode::Partitioned` // by hand-wiring `RepartitionExec(Hash, 12)` on both join sides. A SQL // INNER JOIN over small parquet inputs plans as `CollectLeft`, so the // per-partition CASE filter this test exercises is not reachable via SQL. #[tokio::test] async fn test_hashjoin_dynamic_filter_pushdown_partitioned() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; - // Rough sketch of the MRE we're trying to recreate: // COPY (select i as k from generate_series(1, 10000000) as t(i)) // TO 'test_files/scratch/push_down_filter/t1.parquet' @@ -994,43 +1050,8 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() { // | | | // +---------------+------------------------------------------------------------+ - // Create build side with limited values - let build_batches = vec![ - record_batch!( - ("a", Utf8, ["aa", "ab"]), - ("b", Utf8, ["ba", "bb"]), - ("c", Float64, [1.0, 2.0]) // Extra column not used in join - ) - .unwrap(), - ]; - let build_side_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - Field::new("c", DataType::Float64, false), - ])); - let build_scan = TestScanBuilder::new(Arc::clone(&build_side_schema)) - .with_support(true) - .with_batches(build_batches) - .build(); - - // Create probe side with more values - let probe_batches = vec![ - record_batch!( - ("a", Utf8, ["aa", "ab", "ac", "ad"]), - ("b", Utf8, ["ba", "bb", "bc", "bd"]), - ("e", Float64, [1.0, 2.0, 3.0, 4.0]) // Extra column not used in join - ) - .unwrap(), - ]; - let probe_side_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - Field::new("e", DataType::Float64, false), - ])); - let probe_scan = TestScanBuilder::new(Arc::clone(&probe_side_schema)) - .with_support(true) - .with_batches(probe_batches) - .build(); + let (build_side_schema, build_scan, probe_side_schema, probe_scan) = + hashjoin_pushdown_scans(); // Create RepartitionExec nodes for both sides with hash partitioning on join keys let partition_count = 12; @@ -1128,20 +1149,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() { let mut config = ConfigOptions::default(); config.execution.parquet.pushdown_filters = true; config.optimizer.enable_dynamic_filter_pushdown = true; - let plan = FilterPushdown::new_post_optimization() - .optimize(plan, &config) - .unwrap(); - let config = SessionConfig::new().with_batch_size(10); - let session_ctx = SessionContext::new_with_config(config); - session_ctx.register_object_store( - ObjectStoreUrl::parse("test://").unwrap().as_ref(), - Arc::new(InMemory::new()), - ); - let state = session_ctx.state(); - let task_ctx = state.task_ctx(); - let batches = collect(Arc::clone(&plan), Arc::clone(&task_ctx)) - .await - .unwrap(); + let (plan, batches) = optimize_and_collect_pushdown_plan(plan, config).await; // Now check what our filter looks like #[cfg(not(feature = "force_hash_collisions"))] @@ -1198,53 +1206,214 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() { ); } -// Not portable to sqllogictest: this test specifically pins a -// `RepartitionExec(Hash, 12)` between `HashJoinExec(CollectLeft)` and the -// probe-side scan to verify the dynamic filter link survives that boundary -// (regression for #17451). The same CollectLeft filter content and -// pushdown counters are already covered by the simpler slt port -// (push_down_filter_parquet.slt::test_hashjoin_dynamic_filter_pushdown). #[tokio::test] -async fn test_hashjoin_dynamic_filter_pushdown_collect_left() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; +async fn test_hashjoin_dynamic_filter_pushdown_range_partitioned() { + // Rough sketch of the Range-partitioned MRE we're trying to recreate. The + // test hand-wires identical Range repartitioning: + // + // EXPLAIN + // SELECT * + // FROM build + // JOIN probe + // ON build.a = probe.a AND build.b = probe.b; + // + // +---------------+------------------------------------------------------------+ + // | plan_type | plan | + // +---------------+------------------------------------------------------------+ + // | physical_plan | ┌───────────────────────────┐ | + // | | │ HashJoinExec │ | + // | | │ -------------------- ├──────────────┐ | + // | | │ on: (a = a), (b = b) │ │ | + // | | └─────────────┬─────────────┘ │ | + // | | ┌─────────────┴─────────────┐┌─────────────┴─────────────┐ | + // | | │ RepartitionExec ││ RepartitionExec │ | + // | | │ -------------------- ││ -------------------- │ | + // | | │ partition_count(in->out): ││ partition_count(in->out): │ | + // | | │ 1 -> 2 ││ 1 -> 2 │ | + // | | │ ││ │ | + // | | │ partitioning_scheme: ││ partitioning_scheme: │ | + // | | │ Range([a ASC, b ASC], 2) ││ Range([a ASC, b ASC], 2) │ | + // | | │ split: (aa, bb) ││ split: (aa, bb) │ | + // | | └─────────────┬─────────────┘└─────────────┬─────────────┘ | + // | | ┌─────────────┴─────────────┐┌─────────────┴─────────────┐ | + // | | │ DataSourceExec (build) ││ DataSourceExec (probe) │ | + // | | │ -------------------- ││ -------------------- │ | + // | | │ rows: (aa,ba), (ab,bb) ││ rows: (aa,ba) ... (ad,bd) │ | + // | | │ ││ predicate: DynamicFilter │ | + // | | │ ││ range CASE -> filter_0/1 │ | + // | | └───────────────────────────┘└───────────────────────────┘ | + // | | | + // +---------------+------------------------------------------------------------+ - let build_batches = vec![ - record_batch!( - ("a", Utf8, ["aa", "ab"]), - ("b", Utf8, ["ba", "bb"]), - ("c", Float64, [1.0, 2.0]) // Extra column not used in join + let (build_side_schema, build_scan, probe_side_schema, probe_scan) = + hashjoin_pushdown_scans(); + + let split_points = vec![SplitPoint::new(vec![ + ScalarValue::Utf8(Some("aa".to_string())), + ScalarValue::Utf8(Some("bb".to_string())), + ])]; + + // Build side: DataSource -> RepartitionExec (Range) + let build_range_ordering = LexOrdering::new(vec![ + PhysicalSortExpr::new( + col("a", &build_side_schema).unwrap(), + SortOptions::default(), + ), + PhysicalSortExpr::new( + col("b", &build_side_schema).unwrap(), + SortOptions::default(), + ), + ]) + .unwrap(); + let build_repartition = Arc::new( + RepartitionExec::try_new( + build_scan, + Partitioning::Range( + RangePartitioning::try_new(build_range_ordering, split_points.clone()) + .unwrap(), + ), ) .unwrap(), - ]; - let build_side_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - Field::new("c", DataType::Float64, false), - ])); - let build_scan = TestScanBuilder::new(Arc::clone(&build_side_schema)) - .with_support(true) - .with_batches(build_batches) - .build(); + ); - // Create probe side with more values - let probe_batches = vec![ - record_batch!( - ("a", Utf8, ["aa", "ab", "ac", "ad"]), - ("b", Utf8, ["ba", "bb", "bc", "bd"]), - ("e", Float64, [1.0, 2.0, 3.0, 4.0]) // Extra column not used in join + // Probe side: DataSource -> RepartitionExec (Range) + let probe_range_ordering = LexOrdering::new(vec![ + PhysicalSortExpr::new( + col("a", &probe_side_schema).unwrap(), + SortOptions::default(), + ), + PhysicalSortExpr::new( + col("b", &probe_side_schema).unwrap(), + SortOptions::default(), + ), + ]) + .unwrap(); + let probe_repartition = Arc::new( + RepartitionExec::try_new( + Arc::clone(&probe_scan), + Partitioning::Range( + RangePartitioning::try_new(probe_range_ordering, split_points).unwrap(), + ), ) .unwrap(), + ); + + // Create HashJoinExec with partitioned inputs + let on = vec![ + ( + col("a", &build_side_schema).unwrap(), + col("a", &probe_side_schema).unwrap(), + ), + ( + col("b", &build_side_schema).unwrap(), + col("b", &probe_side_schema).unwrap(), + ), ]; - let probe_side_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - Field::new("e", DataType::Float64, false), - ])); - let probe_scan = TestScanBuilder::new(Arc::clone(&probe_side_schema)) - .with_support(true) - .with_batches(probe_batches) - .build(); + let hash_join = Arc::new( + HashJoinExec::try_new( + build_repartition, + probe_repartition, + on, + None, + &JoinType::Inner, + None, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + + // Top-level CoalescePartitionsExec + let cp = Arc::new(CoalescePartitionsExec::new(hash_join)) as Arc; + // Add a sort for deterministic output + let plan = Arc::new(SortExec::new( + LexOrdering::new(vec![PhysicalSortExpr::new( + col("a", &probe_side_schema).unwrap(), + SortOptions::new(true, false), // descending, nulls_first + )]) + .unwrap(), + cp, + )) as Arc; + + // expect the predicate to be pushed down into the probe side DataSource + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new_post_optimization(), true), + @r" + OptimizationTest: + input: + - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false] + - CoalescePartitionsExec + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)] + - RepartitionExec: partitioning=Range([a@0 ASC, b@1 ASC], [(aa, bb)], 2), input_partitions=1 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true + - RepartitionExec: partitioning=Range([a@0 ASC, b@1 ASC], [(aa, bb)], 2), input_partitions=1 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true + output: + Ok: + - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false] + - CoalescePartitionsExec + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)] + - RepartitionExec: partitioning=Range([a@0 ASC, b@1 ASC], [(aa, bb)], 2), input_partitions=1 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true + - RepartitionExec: partitioning=Range([a@0 ASC, b@1 ASC], [(aa, bb)], 2), input_partitions=1 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ empty ] + " + ); + + // Actually apply the optimization to the plan and execute to see the filter in action + let mut config = ConfigOptions::default(); + config.execution.parquet.pushdown_filters = true; + config.optimizer.enable_dynamic_filter_pushdown = true; + config.optimizer.preserve_file_partitions = 1; + let (plan, batches) = optimize_and_collect_pushdown_plan(plan, config).await; + + // Now check what our filter looks like + insta::assert_snapshot!( + format!("{}", format_plan_for_test(&plan)), + @r" + - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false] + - CoalescePartitionsExec + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)] + - RepartitionExec: partitioning=Range([a@0 ASC, b@1 ASC], [(aa, bb)], 2), input_partitions=1 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true + - RepartitionExec: partitioning=Range([a@0 ASC, b@1 ASC], [(aa, bb)], 2), input_partitions=1 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ CASE range_partition WHEN 0 THEN a@0 >= aa AND a@0 <= aa AND b@1 >= ba AND b@1 <= ba AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}]) ELSE a@0 >= ab AND a@0 <= ab AND b@1 >= bb AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:ab,c1:bb}]) END ] + " + ); + + let result = format!("{}", pretty_format_batches(&batches).unwrap()); + + let probe_scan_metrics = probe_scan.metrics().unwrap(); + + // The probe side had 4 rows, but after applying the dynamic filter only 2 rows should remain. + // The number of output rows from the probe side scan should stay consistent across executions. + // Issue: https://github.com/apache/datafusion/issues/17451 + assert_eq!(probe_scan_metrics.output_rows().unwrap(), 2); + + insta::assert_snapshot!( + result, + @r" + +----+----+-----+----+----+-----+ + | a | b | c | a | b | e | + +----+----+-----+----+----+-----+ + | ab | bb | 2.0 | ab | bb | 2.0 | + | aa | ba | 1.0 | aa | ba | 1.0 | + +----+----+-----+----+----+-----+ + ", + ); +} + +// Not portable to sqllogictest: this test specifically pins a +// `RepartitionExec(Hash, 12)` between `HashJoinExec(CollectLeft)` and the +// probe-side scan to verify the dynamic filter link survives that boundary +// (regression for #17451). The same CollectLeft filter content and +// pushdown counters are already covered by the simpler slt port +// (push_down_filter_parquet.slt::test_hashjoin_dynamic_filter_pushdown). +#[tokio::test] +async fn test_hashjoin_dynamic_filter_pushdown_collect_left() { + let (build_side_schema, build_scan, probe_side_schema, probe_scan) = + hashjoin_pushdown_scans(); // Create RepartitionExec nodes for both sides with hash partitioning on join keys let partition_count = 12; @@ -1326,20 +1495,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_collect_left() { let mut config = ConfigOptions::default(); config.execution.parquet.pushdown_filters = true; config.optimizer.enable_dynamic_filter_pushdown = true; - let plan = FilterPushdown::new_post_optimization() - .optimize(plan, &config) - .unwrap(); - let config = SessionConfig::new().with_batch_size(10); - let session_ctx = SessionContext::new_with_config(config); - session_ctx.register_object_store( - ObjectStoreUrl::parse("test://").unwrap().as_ref(), - Arc::new(InMemory::new()), - ); - let state = session_ctx.state(); - let task_ctx = state.task_ctx(); - let batches = collect(Arc::clone(&plan), Arc::clone(&task_ctx)) - .await - .unwrap(); + let (plan, batches) = optimize_and_collect_pushdown_plan(plan, config).await; // Now check what our filter looks like insta::assert_snapshot!( @@ -1378,9 +1534,6 @@ async fn test_hashjoin_dynamic_filter_pushdown_collect_left() { #[test] fn test_hashjoin_parent_filter_pushdown_same_column_names() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; - let build_side_schema = Arc::new(Schema::new(vec![ Field::new("id", DataType::Utf8, false), Field::new("build_val", DataType::Utf8, false), @@ -1447,9 +1600,6 @@ fn test_hashjoin_parent_filter_pushdown_same_column_names() { #[test] fn test_hashjoin_parent_filter_pushdown_mark_join() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; - let left_schema = Arc::new(Schema::new(vec![ Field::new("id", DataType::Utf8, false), Field::new("val", DataType::Utf8, false), @@ -1517,9 +1667,6 @@ fn test_hashjoin_parent_filter_pushdown_mark_join() { /// only rely on the output side to preserve their semantics. #[test] fn test_hashjoin_parent_filter_pushdown_semi_anti_join() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; - let left_schema = Arc::new(Schema::new(vec![ Field::new("k", DataType::Utf8, false), Field::new("v", DataType::Utf8, false), @@ -2475,9 +2622,6 @@ fn test_pushdown_with_computed_grouping_key() { // on a hand-wired plan, which does trigger the `false` path. #[tokio::test] async fn test_hashjoin_dynamic_filter_all_partitions_empty() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; - // Test scenario where all build-side partitions are empty // This validates the code path that sets the filter to `false` when no rows can match @@ -2610,46 +2754,8 @@ async fn test_hashjoin_dynamic_filter_all_partitions_empty() { // PartitionMode::Partitioned, which SQL never picks for small parquet inputs. #[tokio::test] async fn test_hashjoin_hash_table_pushdown_partitioned() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; - - // Create build side with limited values - let build_batches = vec![ - record_batch!( - ("a", Utf8, ["aa", "ab"]), - ("b", Utf8, ["ba", "bb"]), - ("c", Float64, [1.0, 2.0]) // Extra column not used in join - ) - .unwrap(), - ]; - let build_side_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - Field::new("c", DataType::Float64, false), - ])); - let build_scan = TestScanBuilder::new(Arc::clone(&build_side_schema)) - .with_support(true) - .with_batches(build_batches) - .build(); - - // Create probe side with more values - let probe_batches = vec![ - record_batch!( - ("a", Utf8, ["aa", "ab", "ac", "ad"]), - ("b", Utf8, ["ba", "bb", "bc", "bd"]), - ("e", Float64, [1.0, 2.0, 3.0, 4.0]) // Extra column not used in join - ) - .unwrap(), - ]; - let probe_side_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - Field::new("e", DataType::Float64, false), - ])); - let probe_scan = TestScanBuilder::new(Arc::clone(&probe_side_schema)) - .with_support(true) - .with_batches(probe_batches) - .build(); + let (build_side_schema, build_scan, probe_side_schema, probe_scan) = + hashjoin_pushdown_scans(); // Create RepartitionExec nodes for both sides with hash partitioning on join keys let partition_count = 12; @@ -2719,24 +2825,11 @@ async fn test_hashjoin_hash_table_pushdown_partitioned() { )) as Arc; // Apply the optimization with config setting that forces HashTable strategy - let session_config = SessionConfig::default() - .with_batch_size(10) - .set_usize("datafusion.optimizer.hash_join_inlist_pushdown_max_size", 1) - .set_bool("datafusion.execution.parquet.pushdown_filters", true) - .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true); - let plan = FilterPushdown::new_post_optimization() - .optimize(plan, session_config.options()) - .unwrap(); - let session_ctx = SessionContext::new_with_config(session_config); - session_ctx.register_object_store( - ObjectStoreUrl::parse("test://").unwrap().as_ref(), - Arc::new(InMemory::new()), - ); - let state = session_ctx.state(); - let task_ctx = state.task_ctx(); - let batches = collect(Arc::clone(&plan), Arc::clone(&task_ctx)) - .await - .unwrap(); + let mut config = ConfigOptions::default(); + config.optimizer.hash_join_inlist_pushdown_max_size = 1; + config.execution.parquet.pushdown_filters = true; + config.optimizer.enable_dynamic_filter_pushdown = true; + let (plan, batches) = optimize_and_collect_pushdown_plan(plan, config).await; // Verify that hash_lookup is used instead of IN (SET) let plan_str = format_plan_for_test(&plan).to_string(); @@ -2776,45 +2869,8 @@ async fn test_hashjoin_hash_table_pushdown_partitioned() { // IN (SET) invariant is captured in the slt port. #[tokio::test] async fn test_hashjoin_hash_table_pushdown_collect_left() { - use datafusion_common::JoinType; - use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode}; - - let build_batches = vec![ - record_batch!( - ("a", Utf8, ["aa", "ab"]), - ("b", Utf8, ["ba", "bb"]), - ("c", Float64, [1.0, 2.0]) // Extra column not used in join - ) - .unwrap(), - ]; - let build_side_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - Field::new("c", DataType::Float64, false), - ])); - let build_scan = TestScanBuilder::new(Arc::clone(&build_side_schema)) - .with_support(true) - .with_batches(build_batches) - .build(); - - // Create probe side with more values - let probe_batches = vec![ - record_batch!( - ("a", Utf8, ["aa", "ab", "ac", "ad"]), - ("b", Utf8, ["ba", "bb", "bc", "bd"]), - ("e", Float64, [1.0, 2.0, 3.0, 4.0]) // Extra column not used in join - ) - .unwrap(), - ]; - let probe_side_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), - Field::new("e", DataType::Float64, false), - ])); - let probe_scan = TestScanBuilder::new(Arc::clone(&probe_side_schema)) - .with_support(true) - .with_batches(probe_batches) - .build(); + let (build_side_schema, build_scan, probe_side_schema, probe_scan) = + hashjoin_pushdown_scans(); // Create RepartitionExec nodes for both sides with hash partitioning on join keys let partition_count = 12; @@ -2870,24 +2926,11 @@ async fn test_hashjoin_hash_table_pushdown_collect_left() { )) as Arc; // Apply the optimization with config setting that forces HashTable strategy - let session_config = SessionConfig::default() - .with_batch_size(10) - .set_usize("datafusion.optimizer.hash_join_inlist_pushdown_max_size", 1) - .set_bool("datafusion.execution.parquet.pushdown_filters", true) - .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true); - let plan = FilterPushdown::new_post_optimization() - .optimize(plan, session_config.options()) - .unwrap(); - let session_ctx = SessionContext::new_with_config(session_config); - session_ctx.register_object_store( - ObjectStoreUrl::parse("test://").unwrap().as_ref(), - Arc::new(InMemory::new()), - ); - let state = session_ctx.state(); - let task_ctx = state.task_ctx(); - let batches = collect(Arc::clone(&plan), Arc::clone(&task_ctx)) - .await - .unwrap(); + let mut config = ConfigOptions::default(); + config.optimizer.hash_join_inlist_pushdown_max_size = 1; + config.execution.parquet.pushdown_filters = true; + config.optimizer.enable_dynamic_filter_pushdown = true; + let (plan, batches) = optimize_and_collect_pushdown_plan(plan, config).await; // Verify that hash_lookup is used instead of IN (SET) let plan_str = format_plan_for_test(&plan).to_string(); @@ -3620,7 +3663,6 @@ fn test_filter_pushdown_through_sort_with_projection() { #[test] fn post_phase_is_idempotent_on_hash_join() { use crate::physical_optimizer::test_utils::{hash_join_exec, parquet_exec, schema}; - use datafusion_common::JoinType; use datafusion_physical_expr::expressions::Column; use datafusion_physical_optimizer::filter_pushdown::FilterPushdown; use datafusion_physical_plan::get_plan_string; diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 4e68d871b81ad..cd9048b58c2ba 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -876,14 +876,24 @@ impl HashJoinExec { return false; } - // `preserve_file_partitions` can report Hash partitioning for Hive-style - // file groups, but those partitions are not actually hash-distributed. - // Partitioned dynamic filters rely on hash routing, so disable them in - // this mode to avoid incorrect results. Follow-up work: enable dynamic - // filtering for preserve_file_partitioned scans (issue #20195). + // `preserve_file_partitions` can report Hive-style file groups as Hash + // partitioned even though their partition indexes do not follow the + // hash router used by partitioned dynamic filters. Reject Hash inputs + // because the metadata cannot distinguish those scans from a real hash + // repartition. Compatible Range inputs remain safe because matching + // ordering and split points align each build filter with its probe + // partition. Other unsupported layouts are rejected. + // Follow-up work: enable dynamic filtering for preserve_file_partitioned scans (issue #20195). // https://github.com/apache/datafusion/issues/20195 if config.optimizer.preserve_file_partitions > 0 && self.mode == PartitionMode::Partitioned + && matches!( + ( + self.left.output_partitioning(), + self.right.output_partitioning() + ), + (Partitioning::Hash(_, _), Partitioning::Hash(_, _)) + ) { return false; } @@ -891,9 +901,6 @@ impl HashJoinExec { if self.mode == PartitionMode::Partitioned && !self.has_partitioned_dynamic_filter_routing() { - // TODO: support partition-routed dynamic filters for compatible - // range co-partitioned joins. - // . return false; } @@ -909,6 +916,14 @@ impl HashJoinExec { Partitioning::Hash(_, left_partition_count), Partitioning::Hash(_, right_partition_count), ) => left_partition_count == right_partition_count, + (Partitioning::Range(_), Partitioning::Range(_)) => { + let children = [self.left.as_ref(), self.right.as_ref()]; + matches!( + self.input_distribution_requirements() + .unsatisfied_co_partitioned_children(self.name(), &children), + Ok(unsatisfied) if unsatisfied.is_empty() + ) + } (left_partitioning, right_partitioning) => { left_partitioning.partition_count() == 1 && right_partitioning.partition_count() == 1 @@ -7089,9 +7104,10 @@ mod tests { Ok(()) } - #[test] - fn test_partitioned_dynamic_filter_pushdown_rejects_range_partitioning() -> Result<()> - { + fn range_partitioned_dynamic_filter_test_join( + left_split: i32, + right_split: i32, + ) -> Result<(HashJoinExec, JoinOn)> { let (left_schema, right_schema, on) = build_schema_and_on()?; let left_partitioning = Partitioning::Range(RangePartitioning::try_new( [PhysicalSortExpr { @@ -7099,7 +7115,7 @@ mod tests { options: Default::default(), }] .into(), - vec![SplitPoint::new(vec![ScalarValue::Int32(Some(10))])], + vec![SplitPoint::new(vec![ScalarValue::Int32(Some(left_split))])], )?); let right_partitioning = Partitioning::Range(RangePartitioning::try_new( [PhysicalSortExpr { @@ -7107,7 +7123,7 @@ mod tests { options: Default::default(), }] .into(), - vec![SplitPoint::new(vec![ScalarValue::Int32(Some(10))])], + vec![SplitPoint::new(vec![ScalarValue::Int32(Some(right_split))])], )?); let left = Arc::new(PartitionedTestExec::try_new( left_schema, @@ -7118,16 +7134,10 @@ mod tests { right_partitioning, )?); - let mut session_config = SessionConfig::default(); - session_config - .options_mut() - .optimizer - .enable_join_dynamic_filter_pushdown = true; - let join = HashJoinExec::try_new( left, right, - on, + on.clone(), None, &JoinType::Inner, None, @@ -7135,8 +7145,73 @@ mod tests { NullEquality::NullEqualsNothing, false, )?; + Ok((join, on)) + } - assert!(!join.allow_join_dynamic_filter_pushdown(session_config.options())); + fn with_hash_partitioned_children( + join: &HashJoinExec, + on: &JoinOn, + ) -> Result { + join.builder() + .with_new_children(vec![ + Arc::new(PartitionedTestExec::try_new( + join.left().schema(), + Partitioning::Hash(vec![Arc::clone(&on[0].0)], 2), + )?), + Arc::new(PartitionedTestExec::try_new( + join.right().schema(), + Partitioning::Hash(vec![Arc::clone(&on[0].1)], 2), + )?), + ])? + .build() + } + + #[test] + fn test_partitioned_dynamic_filter_pushdown_allows_supported_partitioning() + -> Result<()> { + let (range_join, on) = range_partitioned_dynamic_filter_test_join(10, 10)?; + let hash_join = with_hash_partitioned_children(&range_join, &on)?; + let mut session_config = SessionConfig::default(); + session_config + .options_mut() + .optimizer + .enable_join_dynamic_filter_pushdown = true; + + assert!(range_join.allow_join_dynamic_filter_pushdown(session_config.options())); + assert!(hash_join.allow_join_dynamic_filter_pushdown(session_config.options())); + + session_config + .options_mut() + .optimizer + .preserve_file_partitions = 1; + assert!(range_join.allow_join_dynamic_filter_pushdown(session_config.options())); + + Ok(()) + } + + #[test] + fn test_partitioned_dynamic_filter_pushdown_rejects_unsupported_partitioning() + -> Result<()> { + let (range_join, on) = range_partitioned_dynamic_filter_test_join(10, 10)?; + let hash_join = with_hash_partitioned_children(&range_join, &on)?; + let (mismatched_range_join, _) = + range_partitioned_dynamic_filter_test_join(10, 11)?; + let mut session_config = SessionConfig::default(); + session_config + .options_mut() + .optimizer + .enable_join_dynamic_filter_pushdown = true; + + assert!( + !mismatched_range_join + .allow_join_dynamic_filter_pushdown(session_config.options()) + ); + + session_config + .options_mut() + .optimizer + .preserve_file_partitions = 1; + assert!(!hash_join.allow_join_dynamic_filter_pushdown(session_config.options())); Ok(()) } diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 7b58107e93c3f..94ec4565a4cef 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -23,6 +23,7 @@ use std::sync::Arc; use crate::ExecutionPlan; use crate::ExecutionPlanProperties; +use crate::Partitioning; use crate::joins::Map; use crate::joins::PartitionMode; use crate::joins::hash_join::exec::HASH_JOIN_SEED; @@ -30,18 +31,22 @@ use crate::joins::hash_join::inlist_builder::build_struct_fields; use crate::joins::hash_join::partitioned_hash_eval::{ HashExpr, HashTableLookupExpr, SeededRandomState, }; +use crate::repartition::RangeExpr; use arrow::array::ArrayRef; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::config::ConfigOptions; use datafusion_common::{ DataFusionError, NullEquality, Result, ScalarValue, SharedResult, + assert_or_internal_err, }; use datafusion_expr::Operator; use datafusion_functions::core::r#struct as struct_func; use datafusion_physical_expr::expressions::{ BinaryExpr, CaseExpr, DynamicFilterPhysicalExpr, InListExpr, IsNullExpr, lit, }; -use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef, ScalarFunctionExpr}; +use datafusion_physical_expr::{ + PhysicalExpr, PhysicalExprRef, RangePartitioning, ScalarFunctionExpr, +}; use parking_lot::Mutex; use tokio::sync::Notify; @@ -257,6 +262,8 @@ pub(crate) struct SharedBuildAccumulator { repartition_random_state: SeededRandomState, /// Schema of the probe (right) side for evaluating filter expressions probe_schema: Arc, + /// Probe-side Range routing metadata for partitioned dynamic filters. + probe_range_partitioning: Option, /// Null equality of the join. Under `NullEqualsNull` a probe-side NULL can match a /// build-side NULL, so the pushed filter must keep NULL rows here too. null_equality: NullEquality, @@ -410,6 +417,14 @@ impl SharedBuildAccumulator { ), }; + let probe_range_partitioning = + match (partition_mode, right_child.output_partitioning()) { + (PartitionMode::Partitioned, Partitioning::Range(range)) => { + Some(range.clone()) + } + _ => None, + }; + Self { inner: Mutex::new(AccumulatorState { data: mode_data, @@ -420,6 +435,7 @@ impl SharedBuildAccumulator { on_right, repartition_random_state, probe_schema: right_child.schema(), + probe_range_partitioning, null_equality, null_aware, } @@ -628,19 +644,8 @@ impl SharedBuildAccumulator { }, FinalizeInput::Partitioned(partitions) => { let num_partitions = partitions.len(); - let routing_hash_expr = Arc::new(HashExpr::new( - self.on_right.clone(), - self.repartition_random_state.clone(), - "hash_repartition".to_string(), - )) as Arc; - - let modulo_expr = Arc::new(BinaryExpr::new( - routing_hash_expr, - Operator::Modulo, - lit(ScalarValue::UInt64(Some(num_partitions as u64))), - )) as Arc; - - let mut real_branches = Vec::new(); + let mut partition_filters = Vec::with_capacity(num_partitions); + let mut real_partition_ids = Vec::new(); let mut empty_partition_ids = Vec::new(); let mut has_canceled_unknown = false; let mut keys_have_null = false; @@ -651,8 +656,10 @@ impl SharedBuildAccumulator { if matches!(partition.pushdown, PushdownStrategy::Empty) => { empty_partition_ids.push(partition_id); + partition_filters.push(lit(false)); } PartitionStatus::Reported(partition) => { + real_partition_ids.push(partition_id); keys_have_null |= partition.keys_have_null; let membership_expr = create_membership_predicate( &self.on_right, @@ -669,13 +676,11 @@ impl SharedBuildAccumulator { bounds_expr, ) .unwrap_or_else(|| lit(true)); - real_branches.push(( - lit(ScalarValue::UInt64(Some(partition_id as u64))), - then_expr, - )); + partition_filters.push(then_expr); } PartitionStatus::CanceledUnknown => { has_canceled_unknown = true; + partition_filters.push(lit(true)); // A canceled partition's build content is unknown, so it // may hold a NULL key. keys_have_null = true; @@ -688,38 +693,97 @@ impl SharedBuildAccumulator { } } - let filter_expr = if has_canceled_unknown { - let mut when_then_branches = empty_partition_ids + let filter_expr = if has_canceled_unknown + && real_partition_ids.is_empty() + && empty_partition_ids.is_empty() + { + lit(true) + } else if !has_canceled_unknown && real_partition_ids.is_empty() { + lit(false) + } else if !has_canceled_unknown + && real_partition_ids.len() == 1 + && empty_partition_ids.len() + 1 == num_partitions + { + Arc::clone(&partition_filters[real_partition_ids[0]]) + } else if let Some(range_partitioning) = &self.probe_range_partitioning { + // Range partitioning + assert_or_internal_err!( + partition_filters.len() == range_partitioning.partition_count(), + "Dynamic filter partition count {} does not match Range partition count {}", + partition_filters.len(), + range_partitioning.partition_count() + ); + let routing_range_expr = Arc::new(RangeExpr::try_new( + self.on_right.clone(), + range_partitioning, + )?) + as Arc; + let else_expr = partition_filters + .pop() + .expect("Range partitioning always has at least one partition"); + + // CASE range_partition(key) + // WHEN 0 THEN F0 + // WHEN 1 THEN F1 + // ... + // ELSE Fn + // END + let when_then_expr = partition_filters .into_iter() - .map(|partition_id| { + .enumerate() + .map(|(partition_id, then_expr)| { ( lit(ScalarValue::UInt64(Some(partition_id as u64))), - lit(false), + then_expr, ) }) - .collect::>(); - when_then_branches.extend(real_branches); + .collect(); - if when_then_branches.is_empty() { - lit(true) - } else { - Arc::new(CaseExpr::try_new( - Some(modulo_expr), - when_then_branches, - Some(lit(true)), - )?) as Arc - } - } else if real_branches.is_empty() { - lit(false) - } else if real_branches.len() == 1 - && empty_partition_ids.len() + 1 == num_partitions - { - Arc::clone(&real_branches[0].1) + Arc::new(CaseExpr::try_new( + Some(routing_range_expr), + when_then_expr, + Some(else_expr), + )?) as Arc } else { + // Hash partitioning + let routing_hash_expr = Arc::new(HashExpr::new( + self.on_right.clone(), + self.repartition_random_state.clone(), + "hash_repartition".to_string(), + )) + as Arc; + let modulo_expr = Arc::new(BinaryExpr::new( + routing_hash_expr, + Operator::Modulo, + lit(ScalarValue::UInt64(Some(num_partitions as u64))), + )) as Arc; + + let mut when_then_branches = if has_canceled_unknown { + empty_partition_ids + .into_iter() + .map(|partition_id| { + ( + lit(ScalarValue::UInt64(Some(partition_id as u64))), + lit(false), + ) + }) + .collect::>() + } else { + vec![] + }; + when_then_branches.extend(real_partition_ids.into_iter().map( + |partition_id| { + ( + lit(ScalarValue::UInt64(Some(partition_id as u64))), + Arc::clone(&partition_filters[partition_id]), + ) + }, + )); + Arc::new(CaseExpr::try_new( Some(modulo_expr), - real_branches, - Some(lit(false)), + when_then_branches, + Some(lit(has_canceled_unknown)), )?) as Arc }; @@ -809,6 +873,7 @@ pub(super) fn make_partitioned_accumulator_for_test( on_right: vec![], repartition_random_state: SeededRandomState::with_seed(1), probe_schema, + probe_range_partitioning: None, null_equality: NullEquality::NullEqualsNothing, null_aware: false, } @@ -831,8 +896,14 @@ pub(super) fn completed_partitions_for_test(acc: &SharedBuildAccumulator) -> usi mod tests { use super::*; - use arrow::array::{ArrayRef, Int32Array}; - use datafusion_physical_expr::expressions::{Column, Literal}; + use arrow::array::{ArrayRef, BooleanArray, Float64Array, Int32Array}; + use arrow::compute::SortOptions; + use arrow::record_batch::RecordBatch; + use datafusion_common::SplitPoint; + use datafusion_physical_expr::{ + PhysicalSortExpr, + expressions::{Column, Literal}, + }; fn test_on_right() -> Vec { vec![Arc::new(Column::new("probe_key", 0))] @@ -867,6 +938,7 @@ mod tests { on_right, repartition_random_state: SeededRandomState::with_seed(1), probe_schema: test_probe_schema(), + probe_range_partitioning: None, null_equality: NullEquality::NullEqualsNothing, null_aware: false, } @@ -1076,6 +1148,198 @@ mod tests { ); } + #[test] + fn partitioned_range_dynamic_filter_routes_with_range_expr() -> Result<()> { + let mut acc = make_partitioned_expr_accumulator_for_test(4); + acc.probe_range_partitioning = Some(RangePartitioning::try_new( + [PhysicalSortExpr::new( + Arc::clone(&acc.on_right[0]), + Default::default(), + )] + .into(), + vec![ + SplitPoint::new(vec![ScalarValue::Int32(Some(10))]), + SplitPoint::new(vec![ScalarValue::Int32(Some(20))]), + SplitPoint::new(vec![ScalarValue::Int32(Some(30))]), + ], + )?); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(PushdownStrategy::Empty, no_bounds()), + PartitionStatus::CanceledUnknown, + reported(in_list(&[20, 29]), no_bounds()), + reported(in_list(&[30]), no_bounds()), + ]))?; + + let expr = current_expr(&acc); + let case = case_expr(&expr); + assert!( + case.expr() + .and_then(|expr| expr.downcast_ref::()) + .is_some(), + "Range routing must use RangeExpr" + ); + assert_eq!(case.when_then_expr().len(), 3); + + let batch = RecordBatch::try_new( + test_probe_schema(), + vec![Arc::new(Int32Array::from(vec![ + 9, 10, 19, 20, 21, 29, 30, 31, + ]))], + )?; + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = result + .as_any() + .downcast_ref::() + .expect("dynamic filter should evaluate to BooleanArray"); + assert_eq!( + result, + &BooleanArray::from(vec![false, true, true, true, false, true, true, false,]) + ); + + Ok(()) + } + + #[test] + fn partitioned_range_dynamic_filter_routes_compound_nullable_keys() -> Result<()> { + let probe_schema = Arc::new(Schema::new(vec![ + Field::new("probe_key", DataType::Int32, true), + Field::new("probe_tie", DataType::Int32, true), + ])); + let on_right: Vec = vec![ + Arc::new(Column::new("probe_key", 0)), + Arc::new(Column::new("probe_tie", 1)), + ]; + let mut acc = make_accumulator_for_test( + AccumulatedBuildData::Partitioned { + partitions: vec![PartitionStatus::Pending; 4], + completed_partitions: 0, + }, + on_right, + ); + acc.probe_schema = Arc::clone(&probe_schema); + acc.probe_range_partitioning = Some(RangePartitioning::try_new( + [ + PhysicalSortExpr::new( + Arc::clone(&acc.on_right[0]), + SortOptions::new(false, true), + ), + PhysicalSortExpr::new( + Arc::clone(&acc.on_right[1]), + SortOptions::new(false, false), + ), + ] + .into(), + vec![ + SplitPoint::new(vec![ + ScalarValue::Int32(None), + ScalarValue::Int32(Some(10)), + ]), + SplitPoint::new(vec![ScalarValue::Int32(None), ScalarValue::Int32(None)]), + SplitPoint::new(vec![ + ScalarValue::Int32(Some(10)), + ScalarValue::Int32(None), + ]), + ], + )?); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(PushdownStrategy::Empty, no_bounds()), + PartitionStatus::CanceledUnknown, + reported(PushdownStrategy::Empty, no_bounds()), + PartitionStatus::CanceledUnknown, + ]))?; + + let expr = current_expr(&acc); + let case = case_expr(&expr); + assert!(case.expr().is_some()); + assert_eq!(case.when_then_expr().len(), 3); + + let batch = RecordBatch::try_new( + probe_schema, + vec![ + Arc::new(Int32Array::from(vec![ + None, + None, + None, + None, + Some(9), + Some(10), + Some(10), + Some(11), + ])), + Arc::new(Int32Array::from(vec![ + Some(9), + Some(10), + Some(11), + None, + None, + Some(9), + None, + None, + ])), + ], + )?; + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = result + .as_any() + .downcast_ref::() + .expect("dynamic filter should evaluate to BooleanArray"); + assert_eq!( + result, + &BooleanArray::from( + vec![false, true, true, false, false, false, true, true,] + ) + ); + + Ok(()) + } + + #[test] + fn partitioned_range_dynamic_filter_preserves_signed_zero_routing() -> Result<()> { + let probe_schema = Arc::new(Schema::new(vec![Field::new( + "probe_key", + DataType::Float64, + false, + )])); + let on_right: Vec = vec![Arc::new(Column::new("probe_key", 0))]; + let mut acc = make_accumulator_for_test( + AccumulatedBuildData::Partitioned { + partitions: vec![PartitionStatus::Pending; 2], + completed_partitions: 0, + }, + on_right, + ); + acc.probe_schema = Arc::clone(&probe_schema); + acc.probe_range_partitioning = Some(RangePartitioning::try_new( + [PhysicalSortExpr::new( + Arc::clone(&acc.on_right[0]), + SortOptions::default(), + )] + .into(), + vec![SplitPoint::new(vec![ScalarValue::Float64(Some(0.0))])], + )?); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + PartitionStatus::CanceledUnknown, + reported(PushdownStrategy::Empty, no_bounds()), + ]))?; + + let expr = current_expr(&acc); + let batch = RecordBatch::try_new( + probe_schema, + vec![Arc::new(Float64Array::from(vec![-0.0, 0.0]))], + )?; + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = result + .as_any() + .downcast_ref::() + .expect("dynamic filter should evaluate to BooleanArray"); + assert_eq!(result, &BooleanArray::from(vec![true, false])); + + Ok(()) + } + // Regression guard for the build-report lifecycle fix: on `Drop`, a stream // in `BuildReportState::ReportScheduled` still calls `report_canceled_partition` // because it cannot tell whether the coordinator has already observed the @@ -1153,6 +1417,7 @@ mod tests { on_right, repartition_random_state: SeededRandomState::with_seed(1), probe_schema, + probe_range_partitioning: None, null_equality, null_aware, } diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs index 2863524f16cb3..db5c1cff3461b 100644 --- a/datafusion/physical-plan/src/repartition/mod.rs +++ b/datafusion/physical-plan/src/repartition/mod.rs @@ -20,7 +20,7 @@ //! maintaining the order of the input rows in the output. use std::cmp::Ordering; -use std::fmt::{Debug, Formatter}; +use std::fmt::{Debug, Display, Formatter}; use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; @@ -47,9 +47,9 @@ use crate::{ check_if_same_properties, }; -use arrow::array::{Array, PrimitiveArray, RecordBatch, RecordBatchOptions}; +use arrow::array::{Array, PrimitiveArray, RecordBatch, RecordBatchOptions, UInt64Array}; use arrow::compute::take_arrays; -use arrow::datatypes::{SchemaRef, UInt32Type}; +use arrow::datatypes::{DataType, Schema, SchemaRef, UInt32Type}; use arrow_schema::SortOptions; use datafusion_common::config::ConfigOptions; use datafusion_common::stats::Precision; @@ -58,13 +58,22 @@ use datafusion_common::utils::{compare_rows, extract_row_at_idx_to_buf, transpos use datafusion_common::{ ColumnStatistics, DataFusionError, HashMap, ScalarValue, SplitPoint, assert_or_internal_err, internal_datafusion_err, internal_err, + validate_range_split_points, }; use datafusion_common::{Result, not_impl_err}; use datafusion_common_runtime::SpawnedTask; use datafusion_execution::TaskContext; use datafusion_execution::memory_pool::MemoryConsumer; +use datafusion_expr::ColumnarValue; use datafusion_physical_expr::{EquivalenceProperties, PhysicalExpr, RangePartitioning}; +use datafusion_physical_expr_common::physical_expr::PhysicalExprRef; use datafusion_physical_expr_common::sort_expr::LexOrdering; +#[cfg(feature = "proto")] +use datafusion_physical_expr_common::sort_expr::{ + sort_exprs_try_from_proto, sort_exprs_try_to_proto, +}; +#[cfg(feature = "proto")] +use datafusion_proto_models::protobuf; use crate::filter_pushdown::{ ChildPushdownResult, FilterDescription, FilterPushdownPhase, @@ -638,6 +647,219 @@ enum BatchPartitionerState { /// executions and runs. pub const REPARTITION_RANDOM_STATE: SeededRandomState = SeededRandomState::with_seed(0); +/// Physical expression that returns the Range partition for each input row. +/// +/// This uses the same routing function as [`BatchPartitioner`], so dynamic +/// filtering and repartitioning agree for every [`ScalarValue`] comparison. +#[derive(Debug, Hash, PartialEq, Eq)] +pub struct RangeExpr { + on_columns: Vec, + split_points: Vec, + sort_options: Vec, +} + +impl RangeExpr { + /// Creates a Range expression for `on_columns` using the supplied routing + /// metadata. + pub fn try_new( + on_columns: Vec, + range_partitioning: &RangePartitioning, + ) -> Result { + let sort_options = range_partitioning + .ordering() + .iter() + .map(|expr| expr.options) + .collect(); + Self::try_new_parts( + on_columns, + range_partitioning.split_points().to_vec(), + sort_options, + ) + } + + fn try_new_parts( + on_columns: Vec, + split_points: Vec, + sort_options: Vec, + ) -> Result { + assert_or_internal_err!(!on_columns.is_empty(), "RangeExpr requires a key"); + assert_or_internal_err!( + on_columns.len() == sort_options.len(), + "RangeExpr key count must match sort options" + ); + validate_range_split_points(&split_points, &sort_options)?; + Ok(Self { + on_columns, + split_points, + sort_options, + }) + } + + /// Get the columns used to compute Range partition IDs. + pub fn on_columns(&self) -> &[PhysicalExprRef] { + &self.on_columns + } + + /// Returns the Range split points used for routing. + pub fn split_points(&self) -> &[SplitPoint] { + &self.split_points + } + + /// Returns the per-key sort options used for routing. + pub fn sort_options(&self) -> &[SortOptions] { + &self.sort_options + } +} + +impl Display for RangeExpr { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "range_partition") + } +} + +impl PhysicalExpr for RangeExpr { + fn children(&self) -> Vec<&PhysicalExprRef> { + self.on_columns.iter().collect() + } + + fn with_new_children( + self: Arc, + children: Vec, + ) -> Result { + assert_or_internal_err!( + children.len() == self.on_columns.len(), + "RangeExpr expected {} children, got {}", + self.on_columns.len(), + children.len() + ); + Ok(Arc::new(Self::try_new_parts( + children, + self.split_points.clone(), + self.sort_options.clone(), + )?)) + } + + fn data_type(&self, _input_schema: &Schema) -> Result { + Ok(DataType::UInt64) + } + + fn nullable(&self, _input_schema: &Schema) -> Result { + Ok(false) + } + + fn evaluate(&self, batch: &RecordBatch) -> Result { + let arrays = evaluate_expressions_to_arrays(self.on_columns.iter(), batch)?; + let mut row_key_buffer = Vec::with_capacity(arrays.len()); + let mut partition_ids = Vec::with_capacity(batch.num_rows()); + for row_idx in 0..batch.num_rows() { + extract_row_at_idx_to_buf(&arrays, row_idx, &mut row_key_buffer)?; + partition_ids.push(range_partition_id( + &row_key_buffer, + &self.split_points, + &self.sort_options, + )? as u64); + } + Ok(ColumnarValue::Array(Arc::new(UInt64Array::from( + partition_ids, + )))) + } + + fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "range_partition") + } + + #[cfg(feature = "proto")] + fn try_to_proto( + &self, + ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>, + ) -> Result> { + // Encode the raw ordered children: rebuilding a `LexOrdering` would + // deduplicate equivalent children after dynamic-filter remapping. + let sort_exprs = self + .on_columns + .iter() + .zip(&self.sort_options) + .map(|(expr, options)| PhysicalSortExpr::new(Arc::clone(expr), *options)) + .collect::>(); + let sort_expr = sort_exprs_try_to_proto(&sort_exprs, ctx)?; + let split_point = self + .split_points + .iter() + .map(|split_point| { + let value = split_point + .values() + .iter() + .map(|value| value.try_into().map_err(Into::into)) + .collect::>>()?; + Ok(protobuf::PhysicalRangeSplitPoint { value }) + }) + .collect::>>()?; + Ok(Some(protobuf::PhysicalExprNode { + expr_id: None, + expr_type: Some(protobuf::physical_expr_node::ExprType::RangeExpr( + protobuf::PhysicalRangeExprNode { + sort_expr, + split_point, + }, + )), + })) + } +} + +#[cfg(feature = "proto")] +impl RangeExpr { + /// Reconstructs a [`RangeExpr`] from its protobuf representation. + pub fn try_from_proto( + node: &protobuf::PhysicalExprNode, + ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>, + ) -> Result { + // Decode the raw ordered children for the same reason as `try_to_proto`. + let range_expr = match &node.expr_type { + Some(protobuf::physical_expr_node::ExprType::RangeExpr(expr)) => expr, + _ => return internal_err!("PhysicalExprNode is not a RangeExpr"), + }; + let sort_exprs = sort_exprs_try_from_proto(&range_expr.sort_expr, ctx)?; + let (on_columns, sort_options) = sort_exprs + .into_iter() + .map(|sort_expr| (sort_expr.expr, sort_expr.options)) + .unzip(); + let split_points = range_expr + .split_point + .iter() + .map(|split_point| { + let values = split_point + .value + .iter() + .map(|value| ScalarValue::try_from(value).map_err(Into::into)) + .collect::>>()?; + Ok(SplitPoint::new(values)) + }) + .collect::>>()?; + Ok(Arc::new(Self::try_new_parts( + on_columns, + split_points, + sort_options, + )?)) + } +} + +fn range_partition_id( + row_key: &[ScalarValue], + split_points: &[SplitPoint], + sort_options: &[SortOptions], +) -> Result { + let mut low = 0; + let mut high = split_points.len(); + while low < high { + let mid = low + (high - low) / 2; + match compare_rows(row_key, split_points[mid].values(), sort_options)? { + Ordering::Less => high = mid, + Ordering::Equal | Ordering::Greater => low = mid + 1, + } + } + Ok(low) +} + /// Computes `value % divisor` without division in the hot loop when `divisor` /// is fixed for many values. /// @@ -972,22 +1194,9 @@ impl BatchPartitioner { // Note that `extract_row_at_idx_to_buf` clears the `row_key_buffer` on each invocation, creating a new row key for comparison for each row extract_row_at_idx_to_buf(arrays, row_idx, row_key_buffer)?; - let mut low = 0; - let mut high = split_points.len(); - while low < high { - let mid = low + (high - low) / 2; - let comparison = compare_rows( - row_key_buffer, - split_points[mid].values(), - sort_options, - )?; - match comparison { - Ordering::Less => high = mid, - Ordering::Equal | Ordering::Greater => low = mid + 1, - } - } - - indices[low].push(row_idx as u32) + let partition = + range_partition_id(row_key_buffer, split_points, sort_options)?; + indices[partition].push(row_idx as u32) } Ok(()) @@ -1723,9 +1932,7 @@ impl ExecutionPlan for RepartitionExec { fn try_to_proto( &self, ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, - ) -> Result> { - use datafusion_proto_models::protobuf; - + ) -> Result> { let input = ctx.encode_child(self.input())?; let partitioning = self.partitioning().try_to_proto(&ctx.expr_ctx())?; @@ -1748,11 +1955,9 @@ impl ExecutionPlan for RepartitionExec { impl RepartitionExec { /// Reconstruct a [`RepartitionExec`] from its protobuf representation. pub fn try_from_proto( - node: &datafusion_proto_models::protobuf::PhysicalPlanNode, + node: &protobuf::PhysicalPlanNode, ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, ) -> Result> { - use datafusion_proto_models::protobuf; - let repart = crate::expect_plan_variant!( node, protobuf::physical_plan_node::PhysicalPlanType::Repartition, @@ -2263,6 +2468,47 @@ mod tests { use datafusion_physical_expr::{PhysicalSortExpr, RangePartitioning, SplitPoint}; use insta::assert_snapshot; + #[test] + fn range_expr_preserves_duplicate_remapped_children() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::UInt32, false), + Field::new("b", DataType::UInt32, false), + ])); + let sort_options = [SortOptions::new(false, false), SortOptions::new(true, true)]; + let split_points = vec![SplitPoint::new(vec![ + ScalarValue::UInt32(Some(10)), + ScalarValue::UInt32(Some(20)), + ])]; + let range_partitioning = RangePartitioning::try_new( + [ + PhysicalSortExpr::new(col("a", &schema)?, sort_options[0]), + PhysicalSortExpr::new(col("b", &schema)?, sort_options[1]), + ] + .into(), + split_points.clone(), + )?; + let expr = Arc::new(RangeExpr::try_new( + vec![col("a", &schema)?, col("b", &schema)?], + &range_partitioning, + )?); + let remapped = col("a", &schema)?; + let rewritten = + expr.with_new_children(vec![Arc::clone(&remapped), Arc::clone(&remapped)])?; + + let rewritten = rewritten + .downcast_ref::() + .expect("rewritten expression should remain a RangeExpr"); + assert_eq!(rewritten.on_columns().len(), 2); + assert!(Arc::ptr_eq( + &rewritten.on_columns()[0], + &rewritten.on_columns()[1] + )); + assert_eq!(rewritten.sort_options(), sort_options); + assert_eq!(rewritten.split_points(), split_points); + + Ok(()) + } + #[test] fn strength_reduced_u64_remainder_matches_modulo() { let divisors = [ diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index 99b4ef6272b2f..43a90264c2b1f 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -1050,6 +1050,7 @@ message PhysicalExprNode { PhysicalHigherOrderUdfNode higher_order_udf = 24; PhysicalLambdaExprNode lambda = 25; PhysicalLambdaVariableExprNode lambda_variable = 26; + PhysicalRangeExprNode range_expr = 27; } } @@ -1202,6 +1203,11 @@ message PhysicalHashExprNode { string description = 6; } +message PhysicalRangeExprNode { + repeated PhysicalSortExprNode sort_expr = 1; + repeated PhysicalRangeSplitPoint split_point = 2; +} + message FilterExecNode { PhysicalPlanNode input = 1; PhysicalExprNode expr = 2; diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index 61b1ea3ff1043..908f9752b7f18 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -18489,6 +18489,9 @@ impl serde::Serialize for PhysicalExprNode { physical_expr_node::ExprType::LambdaVariable(v) => { struct_ser.serialize_field("lambdaVariable", v)?; } + physical_expr_node::ExprType::RangeExpr(v) => { + struct_ser.serialize_field("rangeExpr", v)?; + } } } struct_ser.end() @@ -18544,6 +18547,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode { "lambda", "lambda_variable", "lambdaVariable", + "range_expr", + "rangeExpr", ]; #[allow(clippy::enum_variant_names)] @@ -18573,6 +18578,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode { HigherOrderUdf, Lambda, LambdaVariable, + RangeExpr, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -18619,6 +18625,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode { "higherOrderUdf" | "higher_order_udf" => Ok(GeneratedField::HigherOrderUdf), "lambda" => Ok(GeneratedField::Lambda), "lambdaVariable" | "lambda_variable" => Ok(GeneratedField::LambdaVariable), + "rangeExpr" | "range_expr" => Ok(GeneratedField::RangeExpr), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -18816,6 +18823,13 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode { return Err(serde::de::Error::duplicate_field("lambdaVariable")); } expr_type__ = map_.next_value::<::std::option::Option<_>>()?.map(physical_expr_node::ExprType::LambdaVariable) +; + } + GeneratedField::RangeExpr => { + if expr_type__.is_some() { + return Err(serde::de::Error::duplicate_field("rangeExpr")); + } + expr_type__ = map_.next_value::<::std::option::Option<_>>()?.map(physical_expr_node::ExprType::RangeExpr) ; } } @@ -20876,6 +20890,116 @@ impl<'de> serde::Deserialize<'de> for PhysicalPlanNode { deserializer.deserialize_struct("datafusion.PhysicalPlanNode", FIELDS, GeneratedVisitor) } } +impl serde::Serialize for PhysicalRangeExprNode { + #[allow(deprecated)] + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + use serde::ser::SerializeStruct; + let mut len = 0; + if !self.sort_expr.is_empty() { + len += 1; + } + if !self.split_point.is_empty() { + len += 1; + } + let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalRangeExprNode", len)?; + if !self.sort_expr.is_empty() { + struct_ser.serialize_field("sortExpr", &self.sort_expr)?; + } + if !self.split_point.is_empty() { + struct_ser.serialize_field("splitPoint", &self.split_point)?; + } + struct_ser.end() + } +} +impl<'de> serde::Deserialize<'de> for PhysicalRangeExprNode { + #[allow(deprecated)] + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + const FIELDS: &[&str] = &[ + "sort_expr", + "sortExpr", + "split_point", + "splitPoint", + ]; + + #[allow(clippy::enum_variant_names)] + enum GeneratedField { + SortExpr, + SplitPoint, + } + impl<'de> serde::Deserialize<'de> for GeneratedField { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + struct GeneratedVisitor; + + impl serde::de::Visitor<'_> for GeneratedVisitor { + type Value = GeneratedField; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "expected one of: {:?}", &FIELDS) + } + + #[allow(unused_variables)] + fn visit_str(self, value: &str) -> std::result::Result + where + E: serde::de::Error, + { + match value { + "sortExpr" | "sort_expr" => Ok(GeneratedField::SortExpr), + "splitPoint" | "split_point" => Ok(GeneratedField::SplitPoint), + _ => Err(serde::de::Error::unknown_field(value, FIELDS)), + } + } + } + deserializer.deserialize_identifier(GeneratedVisitor) + } + } + struct GeneratedVisitor; + impl<'de> serde::de::Visitor<'de> for GeneratedVisitor { + type Value = PhysicalRangeExprNode; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("struct datafusion.PhysicalRangeExprNode") + } + + fn visit_map(self, mut map_: V) -> std::result::Result + where + V: serde::de::MapAccess<'de>, + { + let mut sort_expr__ = None; + let mut split_point__ = None; + while let Some(k) = map_.next_key()? { + match k { + GeneratedField::SortExpr => { + if sort_expr__.is_some() { + return Err(serde::de::Error::duplicate_field("sortExpr")); + } + sort_expr__ = Some(map_.next_value()?); + } + GeneratedField::SplitPoint => { + if split_point__.is_some() { + return Err(serde::de::Error::duplicate_field("splitPoint")); + } + split_point__ = Some(map_.next_value()?); + } + } + } + Ok(PhysicalRangeExprNode { + sort_expr: sort_expr__.unwrap_or_default(), + split_point: split_point__.unwrap_or_default(), + }) + } + } + deserializer.deserialize_struct("datafusion.PhysicalRangeExprNode", FIELDS, GeneratedVisitor) + } +} impl serde::Serialize for PhysicalRangePartitioning { #[allow(deprecated)] fn serialize(&self, serializer: S) -> std::result::Result diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index 233b5fee1b29e..ba00577ab9a1b 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -1557,7 +1557,7 @@ pub struct PhysicalExprNode { pub expr_id: ::core::option::Option, #[prost( oneof = "physical_expr_node::ExprType", - tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15, 16, 18, 19, 20, 21, 22, 23, 24, 25, 26" + tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15, 16, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27" )] pub expr_type: ::core::option::Option, } @@ -1620,6 +1620,8 @@ pub mod physical_expr_node { Lambda(::prost::alloc::boxed::Box), #[prost(message, tag = "26")] LambdaVariable(super::PhysicalLambdaVariableExprNode), + #[prost(message, tag = "27")] + RangeExpr(super::PhysicalRangeExprNode), } } #[derive(Clone, PartialEq, ::prost::Message)] @@ -1861,6 +1863,13 @@ pub struct PhysicalHashExprNode { pub description: ::prost::alloc::string::String, } #[derive(Clone, PartialEq, ::prost::Message)] +pub struct PhysicalRangeExprNode { + #[prost(message, repeated, tag = "1")] + pub sort_expr: ::prost::alloc::vec::Vec, + #[prost(message, repeated, tag = "2")] + pub split_point: ::prost::alloc::vec::Vec, +} +#[derive(Clone, PartialEq, ::prost::Message)] pub struct FilterExecNode { #[prost(message, optional, boxed, tag = "1")] pub input: ::core::option::Option<::prost::alloc::boxed::Box>, diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index b88c3cf28f785..bb1cb26108424 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/datafusion/proto/src/physical_plan/from_proto.rs @@ -46,6 +46,7 @@ use datafusion_physical_plan::expressions::{ }; use datafusion_physical_plan::joins::HashExpr; use datafusion_physical_plan::proto::ExecutionPlanDecodeCtx; +use datafusion_physical_plan::repartition::RangeExpr; use datafusion_physical_plan::windows::{create_window_expr, schema_add_window_field}; use datafusion_physical_plan::{Partitioning, PhysicalExpr, WindowExpr}; use datafusion_proto_common::common::proto_error; @@ -354,6 +355,7 @@ pub fn parse_physical_expr_with_converter( } ExprType::LikeExpr(_) => LikeExpr::try_from_proto(proto, &decode_ctx)?, ExprType::HashExpr(_) => HashExpr::try_from_proto(proto, &decode_ctx)?, + ExprType::RangeExpr(_) => RangeExpr::try_from_proto(proto, &decode_ctx)?, ExprType::ScalarSubquery(_) => { let results = ctx.scalar_subquery_results().ok_or_else(|| { proto_error( diff --git a/datafusion/proto/tests/cases/plans/exprs.rs b/datafusion/proto/tests/cases/plans/exprs.rs index ed9745a4b1294..518b4a62ce072 100644 --- a/datafusion/proto/tests/cases/plans/exprs.rs +++ b/datafusion/proto/tests/cases/plans/exprs.rs @@ -18,18 +18,22 @@ //! Physical expressions embedded in plans, including the binary //! expression linearization. -use super::roundtrip_test; +use super::{roundtrip_test, roundtrip_test_and_return}; use arrow::datatypes::Fields; +use datafusion::arrow::compute::SortOptions; use datafusion::arrow::datatypes::{DataType, Field, IntervalUnit, Schema}; use datafusion::logical_expr::Operator; use datafusion::physical_expr::expressions::Literal; use datafusion::physical_plan::empty::EmptyExec; use datafusion::physical_plan::expressions::{ - BinaryExpr, Column, binary, col, like, lit, + BinaryExpr, Column, PhysicalSortExpr, binary, col, like, lit, }; use datafusion::physical_plan::filter::FilterExec; use datafusion::physical_plan::projection::{ProjectionExec, ProjectionExpr}; -use datafusion::physical_plan::{ExecutionPlan, PhysicalExpr}; +use datafusion::physical_plan::repartition::RangeExpr; +use datafusion::physical_plan::{ + ExecutionPlan, PhysicalExpr, RangePartitioning, SplitPoint, +}; use datafusion::prelude::SessionContext; use datafusion::scalar::ScalarValue; use datafusion_common::Result; @@ -174,6 +178,58 @@ fn roundtrip_hash_expr() -> Result<()> { roundtrip_test(filter) } +#[test] +fn roundtrip_range_expr() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Float64, false), + Field::new("b", DataType::Float64, false), + ])); + let options = [SortOptions::new(true, true), SortOptions::new(false, false)]; + let range_partitioning = RangePartitioning::try_new( + [ + PhysicalSortExpr::new(col("a", &schema)?, options[0]), + PhysicalSortExpr::new(col("b", &schema)?, options[1]), + ] + .into(), + vec![SplitPoint::new(vec![ + ScalarValue::Float64(Some(0.0)), + ScalarValue::Float64(Some(1.0)), + ])], + )?; + let range_expr: Arc = Arc::new(RangeExpr::try_new( + // Expression remapping may produce duplicate children. Preserve both + // so their sort options stay aligned with the split-point values. + vec![col("a", &schema)?, col("a", &schema)?], + &range_partitioning, + )?); + let filter_expr = binary(range_expr, Operator::Eq, lit(0u64), &schema)?; + let plan = Arc::new(FilterExec::try_new( + filter_expr, + Arc::new(EmptyExec::new(Arc::clone(&schema))), + )?); + + let ctx = SessionContext::new(); + let result = roundtrip_test_and_return( + plan, + &ctx, + &DefaultPhysicalExtensionCodec {}, + &DefaultPhysicalProtoConverter {}, + )?; + let filter = result.downcast_ref::().unwrap(); + let binary = filter.predicate().downcast_ref::().unwrap(); + let range_expr = binary.left().downcast_ref::().unwrap(); + assert_eq!(range_expr.split_points(), range_partitioning.split_points()); + assert_eq!(range_expr.sort_options(), &options); + let children = range_expr.on_columns(); + assert_eq!(children.len(), 2); + for child in children { + let column = child.downcast_ref::().unwrap(); + assert_eq!((column.name(), column.index()), ("a", 0)); + } + + Ok(()) +} + #[test] fn roundtrip_call_null_scalar_struct_dict() -> Result<()> { let data_type = DataType::Struct(Fields::from(vec![Field::new( diff --git a/datafusion/sqllogictest/src/test_context/range_partitioning.rs b/datafusion/sqllogictest/src/test_context/range_partitioning.rs index 4141e000145a8..3cde3939f0b7c 100644 --- a/datafusion/sqllogictest/src/test_context/range_partitioning.rs +++ b/datafusion/sqllogictest/src/test_context/range_partitioning.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use std::fs::{create_dir_all, remove_dir_all, write}; +use std::fs::{File, create_dir_all, remove_dir_all}; use std::path::Path; use std::sync::Arc; @@ -25,11 +25,12 @@ use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use arrow::record_batch::RecordBatch; use datafusion::catalog::streaming::StreamingTable; use datafusion::common::{ScalarValue, SplitPoint}; -use datafusion::datasource::file_format::csv::CsvFormat; +use datafusion::datasource::file_format::parquet::ParquetFormat; use datafusion::datasource::listing::{ ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl, }; use datafusion::logical_expr::{Partitioning, RangePartitioning, col}; +use datafusion::parquet::arrow::ArrowWriter; use datafusion::physical_expr::{ Partitioning as PhysicalPartitioning, PhysicalSortExpr, RangePartitioning as PhysicalRangePartitioning, expressions::col as physical_col, @@ -45,6 +46,30 @@ use datafusion::prelude::SessionContext; /// Registers a simple range-partitioned listing table for testing before /// declaring such tables is supported via SQL. pub(super) fn register_range_partitioned_table(ctx: &SessionContext) { + const RANGE_PARTITIONS: [&[(i32, i32, i32)]; 4] = [ + &[(1, 1, 10), (5, 2, 50)], + &[(10, 1, 100), (15, 2, 150)], + &[(20, 1, 200), (25, 2, 250)], + &[(30, 1, 300), (35, 2, 350)], + ]; + const SHIFTED_RANGE_PARTITIONS: [&[(i32, i32, i32)]; 4] = [ + &[(1, 1, 10), (5, 2, 50), (10, 1, 100)], + &[(15, 2, 150)], + &[(20, 1, 200), (25, 2, 250)], + &[(30, 1, 300), (35, 2, 350)], + ]; + const NARROW_RANGE_PARTITIONS: [&[(i32, i32, i32)]; 3] = [ + &[(1, 1, 10), (5, 2, 50)], + &[(10, 1, 100), (15, 2, 150)], + &[(20, 1, 200), (25, 2, 250), (30, 1, 300), (35, 2, 350)], + ]; + const SPARSE_RANGE_PARTITIONS: [&[(i32, i32, i32)]; 4] = [ + &[(5, 2, 50), (8, 3, 80)], + &[(10, 1, 100)], + &[(20, 1, 200)], + &[(30, 1, 300), (40, 4, 400)], + ]; + let schema = Arc::new(Schema::new(vec![ Field::new("range_key", DataType::Int32, false), Field::new("non_range_key", DataType::Int32, false), @@ -65,18 +90,13 @@ pub(super) fn register_range_partitioned_table(ctx: &SessionContext) { let range_table_dir = Path::new(env!("CARGO_MANIFEST_DIR")) .join("test_files/scratch_range_partitioning/range_partitioned"); - register_csv_listing_table( + register_parquet_listing_table( ctx, "range_partitioned", &range_table_dir, Arc::clone(&schema), - [ - "1,1,10\n5,2,50\n", - "10,1,100\n15,2,150\n", - "20,1,200\n25,2,250\n", - "30,1,300\n35,2,350\n", - ], - Some(output_partitioning), + RANGE_PARTITIONS, + output_partitioning, ); register_unbounded_range_stream_table( @@ -84,24 +104,14 @@ pub(super) fn register_range_partitioned_table(ctx: &SessionContext) { "unbounded_range_like", Arc::clone(&schema), [10, 20, 30], - [ - vec![(1, 1, 10), (5, 2, 50)], - vec![(10, 1, 100), (15, 2, 150)], - vec![(20, 1, 200), (25, 2, 250)], - vec![(30, 1, 300), (35, 2, 350)], - ], + RANGE_PARTITIONS.map(|rows| rows.to_vec()), ); register_unbounded_range_stream_table( ctx, "unbounded_range_like_shifted", Arc::clone(&schema), [15, 20, 30], - [ - vec![(1, 1, 10), (5, 2, 50), (10, 1, 100)], - vec![(15, 2, 150)], - vec![(20, 1, 200), (25, 2, 250)], - vec![(30, 1, 300), (35, 2, 350)], - ], + SHIFTED_RANGE_PARTITIONS.map(|rows| rows.to_vec()), ); let shifted_output_partitioning = Partitioning::Range( @@ -116,19 +126,14 @@ pub(super) fn register_range_partitioned_table(ctx: &SessionContext) { .expect("range partitioning should be valid"), ); - register_csv_listing_table( + register_parquet_listing_table( ctx, "range_partitioned_shifted", Path::new(env!("CARGO_MANIFEST_DIR")) .join("test_files/scratch_range_partitioning/range_partitioned_shifted"), Arc::clone(&schema), - [ - "1,1,10\n5,2,50\n10,1,100\n", - "15,2,150\n", - "20,1,200\n25,2,250\n", - "30,1,300\n35,2,350\n", - ], - Some(shifted_output_partitioning), + SHIFTED_RANGE_PARTITIONS, + shifted_output_partitioning, ); // Same rows as `range_partitioned` but split into only three range @@ -145,18 +150,14 @@ pub(super) fn register_range_partitioned_table(ctx: &SessionContext) { .expect("range partitioning should be valid"), ); - register_csv_listing_table( + register_parquet_listing_table( ctx, "range_partitioned_narrow", Path::new(env!("CARGO_MANIFEST_DIR")) .join("test_files/scratch_range_partitioning/range_partitioned_narrow"), Arc::clone(&schema), - [ - "1,1,10\n5,2,50\n", - "10,1,100\n15,2,150\n", - "20,1,200\n25,2,250\n30,1,300\n35,2,350\n", - ], - Some(narrow_output_partitioning), + NARROW_RANGE_PARTITIONS, + narrow_output_partitioning, ); let sparse_output_partitioning = Partitioning::Range( @@ -171,29 +172,24 @@ pub(super) fn register_range_partitioned_table(ctx: &SessionContext) { .expect("range partitioning should be valid"), ); - register_csv_listing_table( + register_parquet_listing_table( ctx, "range_partitioned_sparse", Path::new(env!("CARGO_MANIFEST_DIR")) .join("test_files/scratch_range_partitioning/range_partitioned_sparse"), - schema, - [ - "5,2,50\n8,3,80\n", - "10,1,100\n", - "20,1,200\n", - "30,1,300\n40,4,400\n", - ], - Some(sparse_output_partitioning), + Arc::clone(&schema), + SPARSE_RANGE_PARTITIONS, + sparse_output_partitioning, ); } -fn register_csv_listing_table( +fn register_parquet_listing_table( ctx: &SessionContext, name: &str, table_dir: impl AsRef, - schema: Arc, - partitions: impl IntoIterator, - output_partitioning: Option, + schema: SchemaRef, + partitions: impl IntoIterator, + output_partitioning: Partitioning, ) { let table_dir = table_dir.as_ref(); if table_dir.exists() { @@ -201,8 +197,17 @@ fn register_csv_listing_table( } create_dir_all(table_dir).expect("test table dir should be created"); for (idx, rows) in partitions.into_iter().enumerate() { - write(table_dir.join(format!("part-{idx}.csv")), rows) - .expect("test table csv partition should be written"); + let batch = range_batch(Arc::clone(&schema), rows); + let file = File::create(table_dir.join(format!("part-{idx}.parquet"))) + .expect("test table parquet partition should be created"); + let mut writer = ArrowWriter::try_new(file, Arc::clone(&schema), None) + .expect("test table parquet writer should be created"); + writer + .write(&batch) + .expect("test table parquet partition should be written"); + writer + .close() + .expect("test table parquet writer should close"); } let table_path = format!( @@ -213,9 +218,8 @@ fn register_csv_listing_table( ); let table_url = ListingTableUrl::parse(&table_path).expect("test table url should parse"); - let options = - ListingOptions::new(Arc::new(CsvFormat::default().with_has_header(false))) - .with_output_partitioning(output_partitioning); + let options = ListingOptions::new(Arc::new(ParquetFormat::default())) + .with_output_partitioning(Some(output_partitioning)); let config = ListingTableConfig::new(table_url) .with_listing_options(options) .with_schema(schema); @@ -269,20 +273,22 @@ fn range_stream_partition( schema: SchemaRef, rows: &[(i32, i32, i32)], ) -> Arc { - let range_key: Vec = rows.iter().map(|(range_key, _, _)| *range_key).collect(); - let non_range_key: Vec = rows - .iter() - .map(|(_, non_range_key, _)| *non_range_key) - .collect(); - let value: Vec = rows.iter().map(|(_, _, value)| *value).collect(); - let batch = RecordBatch::try_new( + Arc::new(TestPartitionStream::new_with_batches(vec![range_batch( + schema, rows, + )])) +} + +fn range_batch(schema: SchemaRef, rows: &[(i32, i32, i32)]) -> RecordBatch { + RecordBatch::try_new( schema, vec![ - Arc::new(Int32Array::from(range_key)) as ArrayRef, - Arc::new(Int32Array::from(non_range_key)) as ArrayRef, - Arc::new(Int32Array::from(value)) as ArrayRef, + Arc::new(Int32Array::from_iter_values(rows.iter().map(|row| row.0))) + as ArrayRef, + Arc::new(Int32Array::from_iter_values(rows.iter().map(|row| row.1))) + as ArrayRef, + Arc::new(Int32Array::from_iter_values(rows.iter().map(|row| row.2))) + as ArrayRef, ], ) - .expect("range stream batch should be valid"); - Arc::new(TestPartitionStream::new_with_batches(vec![batch])) + .expect("range batch should be valid") } diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt index 9701c41377ef3..326856a352f36 100644 --- a/datafusion/sqllogictest/test_files/range_partitioning.slt +++ b/datafusion/sqllogictest/test_files/range_partitioning.slt @@ -16,7 +16,7 @@ # under the License. # The sqllogictest harness registers range_partitioned(range_key, non_range_key, value) -# as a CSV ListingTable with four declared range-partitioned file groups: +# as a Parquet ListingTable with four declared range-partitioned file groups: # # partition 0: range_key in [..., 10), rows (1, 1, 10), (5, 2, 50) # partition 1: range_key in [10, 20), rows (10, 1, 100), (15, 2, 150) @@ -26,6 +26,21 @@ statement ok set datafusion.explain.physical_plan_only = true; +statement ok +set datafusion.execution.collect_statistics = false; + +statement ok +set datafusion.optimizer.enable_join_dynamic_filter_pushdown = false; + +statement ok +set datafusion.optimizer.enable_topk_dynamic_filter_pushdown = false; + +statement ok +set datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = false; + +statement ok +set datafusion.execution.parquet.pushdown_filters = false; + ########## # TEST 1: Aggregate on Range Partition Column # With subset threshold met and preserve-file disabled, Range([range_key]) @@ -43,7 +58,7 @@ EXPLAIN SELECT range_key, SUM(value) FROM range_partitioned GROUP BY range_key; ---- physical_plan 01)AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT range_key, SUM(value) FROM range_partitioned GROUP BY range_key ORDER BY range_key; @@ -77,7 +92,7 @@ physical_plan 01)AggregateExec: mode=FinalPartitioned, gby=[non_range_key@0 as non_range_key], aggr=[sum(range_partitioned.value)] 02)--RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4 03)----AggregateExec: mode=Partial, gby=[non_range_key@0 as non_range_key], aggr=[sum(range_partitioned.value)] -04)------DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=parquet query II SELECT non_range_key, SUM(value) FROM range_partitioned GROUP BY non_range_key ORDER BY non_range_key; @@ -103,7 +118,7 @@ EXPLAIN SELECT range_key, non_range_key, SUM(value) FROM range_partitioned GROUP ---- physical_plan 01)AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key, non_range_key@1 as non_range_key], aggr=[sum(range_partitioned.value)] -02)--DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT range_key, non_range_key, SUM(value) FROM range_partitioned GROUP BY range_key, non_range_key ORDER BY range_key, non_range_key; @@ -138,7 +153,7 @@ EXPLAIN SELECT range_key, SUM(value) FROM range_partitioned GROUP BY range_key; ---- physical_plan 01)AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet statement ok set datafusion.execution.target_partitions = 4; @@ -170,7 +185,7 @@ physical_plan 02)--RepartitionExec: partitioning=Hash([range_key@0], 5), input_partitions=5 03)----AggregateExec: mode=Partial, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)] 04)------RepartitionExec: partitioning=RoundRobinBatch(5), input_partitions=4 -05)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet statement ok set datafusion.execution.target_partitions = 4; @@ -202,8 +217,8 @@ JOIN range_partitioned r ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -236,9 +251,9 @@ JOIN range_partitioned_shifted r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] 02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -269,9 +284,9 @@ JOIN range_partitioned r ON l.non_range_key = r.non_range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(non_range_key@0, non_range_key@0)], projection=[non_range_key@0, value@1, value@3] 02)--RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4 -05)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=parquet query III SELECT l.non_range_key, l.value, r.value @@ -326,9 +341,9 @@ ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 03)--FilterExec: value@1 <= 150 -04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -354,9 +369,9 @@ ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(range_key@0, range_key@0)] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 03)--FilterExec: value@1 <= 150, projection=[range_key@0] -04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT l.range_key, l.value @@ -378,9 +393,9 @@ ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=LeftAnti, on=[(range_key@0, range_key@0)] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 03)--FilterExec: value@1 <= 150, projection=[range_key@0] -04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT l.range_key, l.value @@ -411,10 +426,10 @@ ON l.range_key = r.range_key AND l.non_range_key = r.non_range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(range_key@0, range_key@0), (non_range_key@1, non_range_key@1)], projection=[range_key@0, non_range_key@1, value@2, value@5] 02)--RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4), input_partitions=4 05)----FilterExec: value@2 <= 150 -06)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +06)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet # Range([range_key]) does not satisfy a join keyed on non_range_key. query TT @@ -425,9 +440,9 @@ LEFT JOIN range_partitioned r ON l.non_range_key = r.non_range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(non_range_key@1, non_range_key@0)], projection=[range_key@0, non_range_key@1, value@2, value@4] 02)--RepartitionExec: partitioning=Hash([non_range_key@1], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4 -05)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=parquet ########## # TEST 11: Left-Side Range Hash Joins With Incompatible Range Layouts @@ -444,9 +459,9 @@ LEFT JOIN range_partitioned_shifted r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] 02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -473,9 +488,9 @@ LEFT JOIN range_partitioned_narrow r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Left, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] 02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=3 -05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=parquet ########## # TEST 12: LeftMark Subqueries Over Range Hash Joins @@ -492,9 +507,9 @@ WHERE l.non_range_key = 2 OR l.range_key IN ( physical_plan 01)FilterExec: non_range_key@1 = 2 OR mark@3, projection=[range_key@0, value@2] 02)--HashJoinExec: mode=Partitioned, join_type=LeftMark, on=[(range_key@0, range_key@0)] -03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)----FilterExec: value@1 <= 150, projection=[range_key@0] -05)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT l.range_key, l.value @@ -543,9 +558,9 @@ JOIN range_partitioned r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] 02)--RepartitionExec: partitioning=Hash([range_key@0], 5), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([range_key@0], 5), input_partitions=4 -05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -578,8 +593,8 @@ JOIN range_partitioned r ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -617,9 +632,9 @@ JOIN range_partitioned s ON r.range_key = s.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@2, range_key@0)], projection=[range_key@0, value@1, value@3, value@5] 02)--HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)] -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -05)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +05)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query IIII SELECT l.range_key, l.value, r.value, s.value @@ -662,10 +677,10 @@ physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)], projection=[range_key@0, l_sum@1, r_sum@3] 02)--ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value)@1 as l_sum] 03)----AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)] -04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 05)--ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value)@1 as r_sum] 06)----AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)] -07)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +07)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III WITH @@ -707,8 +722,8 @@ GROUP BY l.range_key; physical_plan 01)AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key], aggr=[sum(l.value + r.value)] 02)--HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT l.range_key, SUM(l.value + r.value) @@ -741,8 +756,8 @@ RIGHT JOIN range_partitioned r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0, range_key@0)], projection=[value@1, range_key@2, value@3] 02)--FilterExec: value@1 <= 150 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -04)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=value@2 <= 150, pruning_predicate=value_null_count@1 != row_count@2 AND value_min@0 <= 150, required_guarantees=[] +04)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT l.value, r.range_key, r.value @@ -773,8 +788,8 @@ RIGHT SEMI JOIN range_partitioned r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(range_key@0, range_key@0)] 02)--FilterExec: value@1 <= 150, projection=[range_key@0] -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -04)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=value@2 <= 150, pruning_predicate=value_null_count@1 != row_count@2 AND value_min@0 <= 150, required_guarantees=[] +04)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT r.range_key, r.value @@ -801,8 +816,8 @@ RIGHT ANTI JOIN range_partitioned r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=RightAnti, on=[(range_key@0, range_key@0)] 02)--FilterExec: value@1 <= 150, projection=[range_key@0] -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -04)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=value@2 <= 150, pruning_predicate=value_null_count@1 != row_count@2 AND value_min@0 <= 150, required_guarantees=[] +04)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT r.range_key, r.value @@ -831,9 +846,9 @@ physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0, range_key@0)], projection=[value@1, range_key@2, value@3] 02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 03)----FilterExec: value@1 <= 150 -04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=value@2 <= 150, pruning_predicate=value_null_count@1 != row_count@2 AND value_min@0 <= 150, required_guarantees=[] 05)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -06)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=csv, has_header=false +06)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=parquet query III SELECT l.value, r.range_key, r.value @@ -867,9 +882,9 @@ RIGHT JOIN range_partitioned r ON l.range_key = r.range_key AND l.non_range_key physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0, range_key@0), (non_range_key@1, non_range_key@1)], projection=[range_key@0, non_range_key@1, value@2, value@5] 02)--RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4), input_partitions=4 -05)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query IIII SELECT l.range_key, l.non_range_key, l.value, r.value @@ -905,9 +920,9 @@ RIGHT JOIN range_partitioned_narrow r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(range_key@0, range_key@0)], projection=[value@1, range_key@2, value@3] 02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=3 -05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=parquet query III SELECT l.value, r.range_key, r.value @@ -940,9 +955,9 @@ physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Right, on=[(non_range_key@0, non_range_key@1)], projection=[value@1, range_key@2, value@4] 02)--RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4 03)----FilterExec: range_key@0 < 10, projection=[non_range_key@1, value@2] -04)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=range_key@0 < 10, pruning_predicate=range_key_null_count@1 != row_count@2 AND range_key_min@0 < 10, required_guarantees=[] 05)--RepartitionExec: partitioning=Hash([non_range_key@1], 4), input_partitions=4 -06)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +06)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT l.value, r.range_key, r.value @@ -974,9 +989,9 @@ WHERE r.non_range_key = 2 OR r.range_key IN ( physical_plan 01)FilterExec: non_range_key@1 = 2 OR mark@3, projection=[range_key@0, value@2] 02)--HashJoinExec: mode=Partitioned, join_type=LeftMark, on=[(range_key@0, range_key@0)] -03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)----FilterExec: value@1 <= 150, projection=[range_key@0] -05)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet # Matched rows have mark=true and are returned; unmatched rows have # mark=false and are only returned when non_range_key = 2. @@ -1029,9 +1044,9 @@ physical_plan 01)ProjectionExec: expr=[range_key@0 as range_key, value@1 as value, value@3 as value] 02)--SortMergeJoinExec: join_type=Inner, on=[(range_key@0, range_key@0)] 03)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true] -04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, sort_order_for_reorder=[range_key@0 ASC] 05)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true] -06)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +06)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, sort_order_for_reorder=[range_key@0 ASC] query III SELECT l.range_key, l.value, r.value @@ -1064,10 +1079,10 @@ physical_plan 02)--SortMergeJoinExec: join_type=Inner, on=[(range_key@0, range_key@0)] 03)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true] 04)------RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -05)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 06)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true] 07)------RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -08)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=csv, has_header=false +08)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -1167,8 +1182,8 @@ FULL JOIN range_partitioned r ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Full, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -1199,9 +1214,9 @@ FULL JOIN range_partitioned_shifted r ON l.range_key = r.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Full, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] 02)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet 04)--RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 -05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=csv, has_header=false +05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=parquet query III SELECT l.range_key, l.value, r.value @@ -1234,8 +1249,8 @@ FULL JOIN range_partitioned_sparse r ON l.range_key = r.range_key; ---- physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Full, on=[(range_key@0, range_key@0)], projection=[range_key@0, range_key@2, value@1, value@3] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query IIII SELECT l.range_key, r.range_key, l.value, r.value @@ -1276,8 +1291,8 @@ SELECT range_key, value FROM range_partitioned; ---- physical_plan 01)InterleaveExec -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT range_key, value FROM range_partitioned @@ -1325,7 +1340,7 @@ physical_plan 01)ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW] 02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----SortExec: expr=[range_key@0 ASC NULLS LAST, value@1 ASC NULLS LAST], preserve_partitioning=[true] -04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value) FROM range_partitioned ORDER BY range_key; @@ -1354,7 +1369,7 @@ physical_plan 01)ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@2 as sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING] 02)--WindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Int64, nullable: true }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }] 03)----SortExec: expr=[range_key@0 ASC NULLS LAST, value@1 ASC NULLS LAST], preserve_partitioning=[true] -04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) FROM range_partitioned ORDER BY range_key; @@ -1383,7 +1398,7 @@ physical_plan 02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----SortExec: expr=[non_range_key@0 ASC NULLS LAST, value@1 ASC NULLS LAST], preserve_partitioning=[true] 04)------RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4 -05)--------DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false +05)--------DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=parquet query III SELECT non_range_key, value, SUM(value) OVER (PARTITION BY non_range_key ORDER BY value) FROM range_partitioned ORDER BY non_range_key, value; @@ -1413,7 +1428,7 @@ physical_plan 02)--WindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Int64, nullable: true }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }] 03)----SortExec: expr=[non_range_key@0 ASC NULLS LAST, value@1 ASC NULLS LAST], preserve_partitioning=[true] 04)------RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4 -05)--------DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false +05)--------DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=parquet query III SELECT non_range_key, value, SUM(value) OVER (PARTITION BY non_range_key ORDER BY value ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) FROM range_partitioned ORDER BY non_range_key, value; @@ -1445,7 +1460,7 @@ physical_plan 01)ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@3 as sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW] 02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----SortExec: expr=[range_key@0 ASC NULLS LAST, non_range_key@1 ASC NULLS LAST, value@2 ASC NULLS LAST], preserve_partitioning=[true] -04)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT range_key, SUM(value) OVER (PARTITION BY range_key, non_range_key ORDER BY value) FROM range_partitioned ORDER BY range_key; @@ -1478,7 +1493,7 @@ physical_plan 02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----SortExec: expr=[range_key@0 ASC NULLS LAST, non_range_key@1 ASC NULLS LAST, value@2 ASC NULLS LAST], preserve_partitioning=[true] 04)------RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4), input_partitions=4 -05)--------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)--------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT range_key, SUM(value) OVER (PARTITION BY range_key, non_range_key ORDER BY value) FROM range_partitioned ORDER BY range_key; @@ -1513,7 +1528,7 @@ physical_plan 02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----SortPreservingMergeExec: [value@1 ASC NULLS LAST] 04)------SortExec: expr=[value@1 ASC NULLS LAST], preserve_partitioning=[true] -05)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT range_key, SUM(value) OVER (ORDER BY value) FROM range_partitioned ORDER BY range_key; @@ -1553,7 +1568,7 @@ physical_plan 01)ProjectionExec: expr=[range_key@0 as range_key, value@1 as value, row_number() PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as rn] 02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----PartitionedTopKExec: fn=row_number, fetch=1, partition=[range_key@0], order=[value@1 DESC] -04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query III SELECT * FROM ( @@ -1595,7 +1610,7 @@ physical_plan 02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----PartitionedTopKExec: fn=row_number, fetch=1, partition=[non_range_key@0], order=[value@1 DESC] 04)------RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4 -05)--------DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false +05)--------DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=parquet query III SELECT * FROM ( @@ -1630,7 +1645,7 @@ physical_plan 01)ProjectionExec: expr=[range_key@0 as range_key, non_range_key@1 as non_range_key, value@2 as value, row_number() PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@3 as rn] 02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----PartitionedTopKExec: fn=row_number, fetch=1, partition=[range_key@0, non_range_key@1], order=[value@2 DESC] -04)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +04)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query IIII SELECT * FROM ( @@ -1676,7 +1691,7 @@ physical_plan 02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----PartitionedTopKExec: fn=row_number, fetch=1, partition=[range_key@0, non_range_key@1], order=[value@2 DESC] 04)------RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4), input_partitions=4 -05)--------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +05)--------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet statement ok reset datafusion.optimizer.subset_repartition_threshold; @@ -1706,9 +1721,9 @@ SELECT range_key, value FROM range_partitioned_shifted; ---- physical_plan 01)UnionExec -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -04)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +04)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=parquet query II SELECT range_key, value FROM range_partitioned @@ -1759,8 +1774,8 @@ SELECT range_key, value FROM range_partitioned_shifted; ---- physical_plan 01)UnionExec -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=csv, has_header=false +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(15), (20), (30)], 4), file_type=parquet query II SELECT range_key, value FROM range_partitioned @@ -1802,8 +1817,8 @@ EXPLAIN SELECT range_key, SUM(value) FROM ( physical_plan 01)AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key], aggr=[sum(value)] 02)--InterleaveExec -03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false -04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false +03)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet +04)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet query II SELECT range_key, SUM(value) FROM ( @@ -1821,5 +1836,84 @@ SELECT range_key, SUM(value) FROM ( 30 600 35 700 +########## +# TEST 48: Hash Join Dynamic Filter Pushdown on Compatible Range Inputs +# The Parquet-backed probe accepts the partition-routed dynamic filter. Matching +# Range split points keep build filter i aligned with probe partition i. The +# build has rows only in partitions 0 and 2, so the runtime filter must route +# with a four-way CASE rather than collapse to a single filter. +########## + +statement ok +set datafusion.execution.target_partitions = 4; + +statement ok +set datafusion.optimizer.preserve_file_partitions = 1; + +statement ok +set datafusion.optimizer.enable_join_dynamic_filter_pushdown = true; + +statement ok +set datafusion.execution.parquet.pushdown_filters = true; + +query TT +EXPLAIN SELECT b.range_key, b.value, p.value +FROM ( + SELECT range_key, value + FROM range_partitioned + WHERE range_key IN (5, 20) +) b +JOIN range_partitioned p ON b.range_key = p.range_key; +---- +physical_plan +01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] +02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=range_key@0 = 5 OR range_key@0 = 20, pruning_predicate=range_key_null_count@2 != row_count@3 AND range_key_min@0 <= 5 AND 5 <= range_key_max@1 OR range_key_null_count@2 != row_count@3 AND range_key_min@0 <= 20 AND 20 <= range_key_max@1, required_guarantees=[range_key in (20, 5)] +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + +query TT +EXPLAIN ANALYZE SELECT b.range_key, b.value, p.value +FROM ( + SELECT range_key, value + FROM range_partitioned + WHERE range_key IN (5, 20) +) b +JOIN range_partitioned p ON b.range_key = p.range_key; +---- +Plan with Metrics +01)HashJoinExec: mode=Partitionedmetrics=[output_rows=2,] +02)--DataSourceExec: file_type=parquet, predicate=range_key@0 = 5 OR range_key@0 = 20metrics=[output_rows=2,] +03)--DataSourceExec: file_type=parquet, predicate=DynamicFilter [ CASE range_partition WHEN 0 THEN range_key@0 >= 5 AND range_key@0 <= 5 AND range_key@0 IN (SET) ([5]) WHEN 1 THEN false WHEN 2 THEN range_key@0 >= 20 AND range_key@0 <= 20 AND range_key@0 IN (SET) ([20]) ELSE false END ]metrics=[output_rows=2,] + +query III +SELECT b.range_key, b.value, p.value +FROM ( + SELECT range_key, value + FROM range_partitioned + WHERE range_key IN (5, 20) +) b +JOIN range_partitioned p ON b.range_key = p.range_key +ORDER BY b.range_key; +---- +5 50 50 +20 200 200 + +statement ok +reset datafusion.optimizer.preserve_file_partitions; + +statement ok +reset datafusion.execution.collect_statistics; + +statement ok +reset datafusion.optimizer.enable_join_dynamic_filter_pushdown; + +statement ok +reset datafusion.optimizer.enable_topk_dynamic_filter_pushdown; + +statement ok +reset datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown; + +statement ok +reset datafusion.execution.parquet.pushdown_filters; + statement ok reset datafusion.explain.physical_plan_only;