kosiew commented on code in PR #25550:
URL: https://github.com/apache/datafusion/pull/25550#discussion_r4063434322
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -980,20 +980,44 @@ impl HashJoinExec {
Arc::new(DynamicFilterPhysicalExpr::new(right_keys, lit(true)))
}
- /// Join types whose output rows all carry a matching key on both sides.
+ /// How this join may transfer a parent filter over one side's join keys to
+ /// the other side's input, see [`KeyTransfer`].
///
- /// For these a parent filter over one side's join keys can be transferred
- /// to the other side's input: an input row that fails the transferred
- /// filter can only pair with rows that fail the original, so pruning it
- /// changes nothing, and once the transferred filter is applied exactly on
- /// one side every output row satisfies the original. Outer, anti and mark
- /// joins also emit unmatched rows, whose key on the other side is absent,
- /// so the transferred filter is not exact for them.
- fn supports_key_transfer(join_type: JoinType) -> bool {
- matches!(
- join_type,
- JoinType::Inner | JoinType::LeftSemi | JoinType::RightSemi
- )
+ /// The common ground: an input row that fails the transferred filter can
+ /// only pair with rows that fail the original (their keys are equal, also
+ /// under [`NullEquality::NullEqualsNull`], and a join filter only removes
+ /// pairs), so every row that passes the original keeps the same matches.
+ fn key_transfer(&self) -> KeyTransfer {
+ match self.join_type {
+ // Every output row carries a matching key on both sides, so once
+ // the transferred filter is applied exactly on one side every
+ // output row satisfies the original.
+ JoinType::Inner | JoinType::LeftSemi | JoinType::RightSemi => {
+ KeyTransfer::Exact
+ }
+ // A NULL probe key decides a null-aware join's whole result, and a
+ // transferred filter is not true for NULL, so it would prune it.
+ _ if self.null_aware => KeyTransfer::None,
+ // The preserved side also emits unmatched rows. Pruning the other
+ // side can turn a row that fails the filter from matched into
+ // unmatched (NULL-extended, or mark = false), but every row
+ // derived from it still fails the filter, which stays above the
+ // join. It never adds output rows, so a `fetch` on this join or
Review Comment:
I think we need to guard `PruneOnly` for outer joins when this join has
`fetch`. Pruning matching rows from the non-preserved side can remove rows that
would otherwise fill the fetch, which can expose a later row that passes the
parent filter and change the query result.
For example, consider left rows `(bb, aa)`, right rows `(bb, bb, aa)`, a
parent filter `lk = 'aa'`, and `fetch = 2`. Without transferring `rk = 'aa'`,
the two `bb` matches can fill the fetch, so the parent filter returns no rows.
With the transferred filter, those right-side `bb` rows are pruned. The `aa`
match can then reach the fetch, so the parent filter can return it.
Could we avoid prune-only transfer for `Left` and `Right` when
`self.fetch.is_some()`? Mark joins should still be safe because they do not
have this match-multiplicity issue. It would also be good to update the fetch
test to compare supported and unsupported plans using the same fetch, with a
failing-side key ordered first so this case is exercised directly.
##########
datafusion/core/tests/physical_optimizer/filter_pushdown.rs:
##########
@@ -2861,6 +2862,622 @@ async fn
test_hashjoin_dynamic_filter_transferred_through_nested_join() {
);
}
+// ==== Prune-only key transfer: Left, Right, LeftMark and RightMark joins ====
+//
+// These joins also emit unmatched rows of their preserved side, so a filter
+// over the preserved side's key, transferred to the other side, may only prune
+// that side's input. It never makes the filter hold for the join's output.
+
+/// The side of a [`prune_only_join`] whose rows the join type preserves.
+fn prune_only_preserved_side(join_type: JoinType) ->
datafusion_common::JoinSide {
+ match join_type {
+ JoinType::Left | JoinType::LeftMark | JoinType::LeftAnti => {
+ datafusion_common::JoinSide::Left
+ }
+ JoinType::Right | JoinType::RightMark | JoinType::RightAnti => {
+ datafusion_common::JoinSide::Right
+ }
+ other => panic!("{other} has no single preserved side"),
+ }
+}
+
+/// Options of a [`prune_only_join`].
+#[derive(Clone, Copy)]
+struct PruneOnlyJoin {
+ join_type: JoinType,
+ null_equality: datafusion_common::NullEquality,
+ null_aware: bool,
+ /// Whether the left / right scan accepts pushed down filters.
+ left_support: bool,
+ right_support: bool,
+ /// Matching pairs must also satisfy `rv != 'r2'`.
+ join_filter: bool,
+}
+
+impl PruneOnlyJoin {
+ fn new(join_type: JoinType) -> Self {
+ Self {
+ join_type,
+ null_equality: datafusion_common::NullEquality::NullEqualsNothing,
+ null_aware: false,
+ left_support: false,
+ right_support: false,
+ join_filter: false,
+ }
+ }
+
+ fn with_support(mut self, left_support: bool, right_support: bool) -> Self
{
+ self.left_support = left_support;
+ self.right_support = right_support;
+ self
+ }
+}
+
+/// `left(lk, lv) JOIN right(rk, rv) ON lk = rk` over NULL, duplicate and
+/// one-sided keys.
+fn prune_only_join(options: PruneOnlyJoin) -> Arc<HashJoinExec> {
+ let PruneOnlyJoin {
+ join_type,
+ null_equality,
+ null_aware,
+ left_support,
+ right_support,
+ join_filter,
+ } = options;
+ use datafusion_common::JoinSide;
+ use datafusion_physical_plan::joins::utils::{ColumnIndex, JoinFilter};
+
+ let left_schema = Arc::new(Schema::new(vec![
+ Field::new("lk", DataType::Utf8, true),
+ Field::new("lv", DataType::Utf8, false),
+ ]));
+ let left_scan = TestScanBuilder::new(Arc::clone(&left_schema))
+ .with_support(left_support)
+ .with_batches(vec![
+ record_batch!(
+ (
+ "lk",
+ Utf8,
+ [Some("aa"), Some("aa"), Some("bb"), None, Some("cc")]
+ ),
+ ("lv", Utf8, ["l1", "l2", "l3", "l4", "l5"])
+ )
+ .unwrap(),
+ ])
+ .build();
+
+ let right_schema = Arc::new(Schema::new(vec![
+ Field::new("rk", DataType::Utf8, true),
+ Field::new("rv", DataType::Utf8, false),
+ ]));
+ let right_scan = TestScanBuilder::new(Arc::clone(&right_schema))
+ .with_support(right_support)
+ .with_batches(vec![
+ record_batch!(
+ (
+ "rk",
+ Utf8,
+ [Some("aa"), Some("bb"), Some("bb"), None, Some("dd"),
None]
+ ),
+ ("rv", Utf8, ["r1", "r2", "r3", "r4", "r5", "r6"])
+ )
+ .unwrap(),
+ ])
+ .build();
+
+ let filter = join_filter.then(|| {
+ let intermediate =
+ Arc::new(Schema::new(vec![Field::new("rv", DataType::Utf8,
false)]));
+ JoinFilter::new(
+ Arc::new(BinaryExpr::new(
+ col("rv", &intermediate).unwrap(),
+ Operator::NotEq,
+ Arc::new(Literal::new(ScalarValue::from("r2"))),
+ )),
+ vec![ColumnIndex {
+ index: 1,
+ side: JoinSide::Right,
+ }],
+ intermediate,
+ )
+ });
+
+ Arc::new(
+ HashJoinExec::try_new(
+ left_scan,
+ right_scan,
+ vec![(
+ col("lk", &left_schema).unwrap(),
+ col("rk", &right_schema).unwrap(),
+ )],
+ filter,
+ &join_type,
+ None,
+ PartitionMode::CollectLeft,
+ null_equality,
+ null_aware,
+ )
+ .unwrap(),
+ )
+}
+
+/// Runs `FilterPushdown` with row-level pushdown on and collects the rows,
+/// one formatted string per row, sorted.
+async fn prune_only_run(
+ plan: Arc<dyn ExecutionPlan>,
+) -> (Arc<dyn ExecutionPlan>, Vec<String>) {
+ let mut config = ConfigOptions::default();
+ config.execution.parquet.pushdown_filters = true;
+ let optimized = FilterPushdown::new().optimize(plan, &config).unwrap();
+ let session_ctx = SessionContext::new();
+ session_ctx.register_object_store(
+ ObjectStoreUrl::parse("test://").unwrap().as_ref(),
+ Arc::new(InMemory::new()),
+ );
+ let batches = collect(Arc::clone(&optimized), session_ctx.task_ctx())
+ .await
+ .unwrap();
+ let formatted = pretty_format_batches(&batches).unwrap().to_string();
+ let mut rows: Vec<String> = formatted
+ .lines()
+ .filter(|line| line.starts_with('|'))
+ .skip(1) // header
+ .map(str::to_string)
+ .collect();
+ rows.sort();
+ (optimized, rows)
+}
+
+/// The `predicate=` of each scan in `plan`, left scan first.
+fn scan_predicates(plan: &Arc<dyn ExecutionPlan>) -> Vec<Option<String>> {
+ format_plan_for_test(plan)
+ .lines()
+ .filter(|line| line.contains("DataSourceExec"))
+ .map(|line| {
+ line.split_once("predicate=")
+ .map(|(_, predicate)| predicate.to_string())
+ })
+ .collect()
+}
+
+/// Predicates over the preserved side's key `k` that behave differently on
+/// NULL, which is where a transferred copy could go wrong.
+fn prune_only_predicates(
+ key: &str,
+ schema: &Schema,
+) -> Vec<(&'static str, Arc<dyn PhysicalExpr>)> {
+ use datafusion_physical_expr::expressions::NotExpr;
+ let eq = || col_lit_predicate(key, "aa", schema);
+ let is_null =
+ || Arc::new(IsNullExpr::new(col(key, schema).unwrap())) as Arc<dyn
PhysicalExpr>;
+ vec![
+ ("k = aa", eq()),
+ ("k IS NULL", is_null()),
+ ("NOT (k = aa)", Arc::new(NotExpr::new(eq()))),
+ (
+ "k = aa OR k IS NULL",
+ Arc::new(BinaryExpr::new(eq(), Operator::Or, is_null())),
+ ),
+ ]
+}
+
+/// Whatever the scans accept, the rows are those of the plan in which no scan
+/// accepts anything: a transferred copy only prunes rows that cannot pair with
+/// a row passing the filter, and never stands in for the filter itself.
+#[tokio::test]
+async fn test_hashjoin_prune_only_transfer_differential() {
+ use datafusion_common::{JoinSide, NullEquality};
+
+ for join_type in [
+ JoinType::Left,
+ JoinType::Right,
+ JoinType::LeftMark,
+ JoinType::RightMark,
+ ] {
+ let key = match prune_only_preserved_side(join_type) {
+ JoinSide::Left => "lk",
+ _ => "rk",
+ };
+ for null_equality in [
+ NullEquality::NullEqualsNothing,
+ NullEquality::NullEqualsNull,
+ ] {
+ for join_filter in [false, true] {
+ let options = PruneOnlyJoin {
+ null_equality,
+ join_filter,
+ ..PruneOnlyJoin::new(join_type)
+ };
+ let schema = prune_only_join(options).schema();
+ for (name, _) in prune_only_predicates(key, &schema) {
+ let run = |left_support: bool, right_support: bool| {
+ let join = prune_only_join(
+ options.with_support(left_support, right_support),
+ );
+ let predicate = prune_only_predicates(key,
&join.schema())
+ .into_iter()
+ .find(|(n, _)| *n == name)
+ .unwrap()
+ .1;
+ prune_only_run(Arc::new(
+ FilterExec::try_new(predicate, join).unwrap(),
+ ))
+ };
+ let (_, expected) = run(false, false).await;
+ for (left_support, right_support) in
+ [(true, false), (false, true), (true, true)]
+ {
+ let (plan, rows) = run(left_support,
right_support).await;
+ assert_eq!(
+ rows,
+ expected,
+ "{join_type} {null_equality:?}
join_filter={join_filter} \
+ `{name}` left_support={left_support} \
+ right_support={right_support}\n{}",
+ format_plan_for_test(&plan)
+ );
+ }
+ }
+ }
+ }
+ }
+}
+
+/// The copy reaches the non-preserved scan, and it alone does not remove the
+/// filter: only the preserved side's scan accepting it does.
+#[tokio::test]
+async fn test_hashjoin_prune_only_transfer_keeps_parent_filter() {
+ use datafusion_common::JoinSide;
+
+ for join_type in [
+ JoinType::Left,
+ JoinType::Right,
+ JoinType::LeftMark,
+ JoinType::RightMark,
+ ] {
+ let preserved = prune_only_preserved_side(join_type);
+ let (key, other_key, preserved_idx) = match preserved {
+ JoinSide::Left => ("lk", "rk", 0),
+ _ => ("rk", "lk", 1),
+ };
+ let plan = |preserved_support: bool| {
+ let (left_support, right_support) = match preserved {
+ JoinSide::Left => (preserved_support, true),
+ _ => (true, preserved_support),
+ };
+ let join = prune_only_join(
+ PruneOnlyJoin::new(join_type).with_support(left_support,
right_support),
+ );
+ let predicate = col_lit_predicate(key, "aa", &join.schema());
+ Arc::new(FilterExec::try_new(predicate, join).unwrap())
+ as Arc<dyn ExecutionPlan>
+ };
+
+ // Only the non-preserved scan accepts filters: it gets the copy, the
+ // filter stays.
+ let (optimized, _) = prune_only_run(plan(false)).await;
+ assert!(
+ optimized.downcast_ref::<FilterExec>().is_some(),
+ "{join_type}: the transferred copy must not remove the filter\n{}",
+ format_plan_for_test(&optimized)
+ );
+ let predicates = scan_predicates(&optimized);
+ assert_eq!(predicates[preserved_idx], None, "{join_type}");
+ assert_eq!(
+ predicates[1 - preserved_idx],
+ Some(format!("{other_key}@0 = aa")),
+ "{join_type}"
+ );
+
+ // Both scans accept: the preserved side applies the filter exactly,
+ // so it is gone, and the other side is pruned too.
+ let (optimized, _) = prune_only_run(plan(true)).await;
+ assert!(
+ optimized.downcast_ref::<HashJoinExec>().is_some(),
+ "{join_type}: the preserved side accepted the filter\n{}",
+ format_plan_for_test(&optimized)
+ );
+ let predicates = scan_predicates(&optimized);
+ assert_eq!(
+ predicates[preserved_idx],
+ Some(format!("{key}@0 = aa")),
+ "{join_type}"
+ );
+ assert_eq!(
+ predicates[1 - preserved_idx],
+ Some(format!("{other_key}@0 = aa")),
+ "{join_type}"
+ );
+ }
+}
+
+#[test]
+fn test_hashjoin_prune_only_transfer_left_join_plan() {
+ let join =
+ prune_only_join(PruneOnlyJoin::new(JoinType::Left).with_support(false,
true));
+ let predicate = col_lit_predicate("lk", "aa", &join.schema());
+ let plan = Arc::new(FilterExec::try_new(predicate, join).unwrap());
+ insta::assert_snapshot!(
+ OptimizationTest::new(plan, FilterPushdown::new(), true),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: lk@0 = aa
+ - HashJoinExec: mode=CollectLeft, join_type=Left, on=[(lk@0, rk@0)]
+ - DataSourceExec: file_groups={1 group: [[test.parquet]]},
projection=[lk, lv], file_type=test, pushdown_supported=false
+ - DataSourceExec: file_groups={1 group: [[test.parquet]]},
projection=[rk, rv], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - FilterExec: lk@0 = aa
+ - HashJoinExec: mode=CollectLeft, join_type=Left, on=[(lk@0, rk@0)]
+ - DataSourceExec: file_groups={1 group: [[test.parquet]]},
projection=[lk, lv], file_type=test, pushdown_supported=false
+ - DataSourceExec: file_groups={1 group: [[test.parquet]]},
projection=[rk, rv], file_type=test, pushdown_supported=true, predicate=rk@0 =
aa
+ "
+ );
+}
+
+/// Nothing is transferred where it would be wrong: from the non-preserved
+/// side (its key is also NULL for unmatched rows, so `rk IS NULL` above a left
+/// join says nothing about `lk`), for non-key and mark columns, and for full,
+/// anti and null-aware joins.
+#[tokio::test]
+async fn test_hashjoin_prune_only_transfer_negative_cases() {
+ let is_null = |name: &str, schema: &Schema| {
+ Arc::new(IsNullExpr::new(col(name, schema).unwrap())) as Arc<dyn
PhysicalExpr>
+ };
+ type Predicate = Box<dyn Fn(&Schema) -> Arc<dyn PhysicalExpr>>;
+ let cases: Vec<(&str, JoinType, bool, Predicate)> = vec![
+ (
+ "left join, non-preserved key",
+ JoinType::Left,
+ false,
+ Box::new(move |s| is_null("rk", s)),
+ ),
+ (
+ "right join, non-preserved key",
+ JoinType::Right,
+ false,
+ Box::new(move |s| is_null("lk", s)),
+ ),
+ (
+ "left join, non-key column",
+ JoinType::Left,
+ false,
+ Box::new(|s| col_lit_predicate("lv", "l1", s)),
+ ),
+ (
+ "left mark join, mark column",
+ JoinType::LeftMark,
+ false,
+ Box::new(|s| col_lit_predicate("mark", true, s)),
+ ),
+ (
+ "full join",
+ JoinType::Full,
+ false,
+ Box::new(|s| col_lit_predicate("lk", "aa", s)),
+ ),
+ (
+ "left anti join",
+ JoinType::LeftAnti,
+ false,
+ Box::new(|s| col_lit_predicate("lk", "aa", s)),
+ ),
+ (
+ "right anti join",
+ JoinType::RightAnti,
+ false,
+ Box::new(|s| col_lit_predicate("rk", "aa", s)),
+ ),
+ (
+ "null-aware left mark join",
+ JoinType::LeftMark,
+ true,
+ Box::new(|s| col_lit_predicate("lk", "aa", s)),
+ ),
+ (
+ "null-aware left anti join",
+ JoinType::LeftAnti,
+ true,
+ Box::new(|s| col_lit_predicate("lk", "aa", s)),
+ ),
+ ];
+
+ for (name, join_type, null_aware, predicate) in cases {
+ let run = |support: bool| {
+ let join = prune_only_join(
+ PruneOnlyJoin {
+ null_aware,
+ ..PruneOnlyJoin::new(join_type)
+ }
+ .with_support(support, support),
+ );
+ let predicate = predicate(&join.schema());
+ prune_only_run(Arc::new(FilterExec::try_new(predicate,
join).unwrap()))
+ };
+ let (_, expected) = run(false).await;
+ let (optimized, rows) = run(true).await;
+ assert_eq!(
+ rows,
+ expected,
+ "{name}\n{}",
+ format_plan_for_test(&optimized)
+ );
+
+ // At most one scan holds the predicate: the side that owns its
+ // columns, never a transferred copy on the other side.
+ let holders = scan_predicates(&optimized)
+ .iter()
+ .filter(|predicate| predicate.is_some())
+ .count();
+ assert!(
+ holders <= 1,
+ "{name}: unexpected transfer\n{}",
+ format_plan_for_test(&optimized)
+ );
+ }
+}
+
+/// With a `fetch` on the join the rows are not unique, but the transferred
+/// copy never adds output rows, so every row still comes from the join
+/// without a fetch.
+#[tokio::test]
+async fn test_hashjoin_prune_only_transfer_with_fetch() {
+ use datafusion_common::JoinSide;
+
+ for join_type in [JoinType::Left, JoinType::Right] {
Review Comment:
Once the outer-join fetch case is guarded, could we also add `LeftMark` and
`RightMark` to a same-fetch differential test? Mark joins keep one output row
per preserved-side input row, so this would document why prune-only transfer is
still safe for mark joins when fetch is present.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
