kumarUjjawal commented on code in PR #24955:
URL: https://github.com/apache/datafusion/pull/24955#discussion_r3940435490
##########
datafusion/core/benches/pwmj_semi_anti_sql.rs:
##########
@@ -472,5 +518,180 @@ fn bench_pwmj_semi_anti_sql(c: &mut Criterion) {
group.finish();
}
-criterion_group!(benches, bench_pwmj_semi_anti_sql);
+/// A `JoinFilter` for `lhs.key < rhs.key`, the same range relation every case
in this file
+/// uses, for the `NestedLoopJoinExec` arm below -- which needs one built by
hand since it is
+/// constructed directly rather than planned from SQL.
+fn key_lt_key_filter(s: &SchemaRef) -> JoinFilter {
+ let expr = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("key", 0)),
+ Operator::Lt,
+ Arc::new(Column::new("key", 1)),
+ )) as _;
+ let column_indices = vec![
+ ColumnIndex {
+ index: 0,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 0,
+ side: JoinSide::Right,
+ },
+ ];
+ let key_field = s.field_with_name("key").unwrap().clone();
+ let intermediate_schema = Schema::new(vec![key_field.clone(), key_field]);
+ JoinFilter::new(expr, column_indices, Arc::new(intermediate_schema))
+}
+
+/// `RightMark` has no SQL surface (see the module doc's "Mark joins"
section), so both arms
+/// are hand-built here instead of planned from SQL text:
`PiecewiseMergeJoinExec` and
+/// `NestedLoopJoinExec`, over the same data, the same `lhs.key < rhs.key`
relation, and the
+/// same join type -- with no risk of comparing an operator against itself,
since which
+/// operator each arm uses is fixed by construction rather than read back off
a plan.
+///
+/// Neither arm needs a `SortExec`: `RightMark`, like `RightSemi`/`RightAnti`,
folds the
+/// buffered side to one key regardless of its order, and `NestedLoopJoinExec`
never needs
+/// either side ordered.
+fn bench_pwmj_right_mark_hand_built(c: &mut Criterion) {
+ let rt = Runtime::new().unwrap();
+ let s = schema();
+ let ctx = SessionContext::new();
+
+ // Every other case in this file falls back to `NestedLoopJoinExec`
through the planner
+ // itself when a build does not yet support the join type -- there is no
such fallback
+ // here, since `RightMark` has no SQL surface to plan through in the first
place. Probe
+ // `try_new` directly and skip the whole group rather than panic, so this
benchmark stays
+ // runnable (as a no-op) against a build that has not merged `RightMark`
support yet, and
+ // starts measuring on its own once that support lands.
+ if let Err(err) = PiecewiseMergeJoinExec::try_new(
+ Arc::new(EmptyExec::new(Arc::clone(&s))),
+ Arc::new(EmptyExec::new(Arc::clone(&s))),
+ (
+ Arc::new(Column::new("key", 0)) as _,
+ Arc::new(Column::new("key", 0)) as _,
+ ),
+ Operator::Lt,
+ JoinType::RightMark,
+ 1,
+ ) {
+ println!(
+ "note: pwmj_vs_nlj_right_mark_hand_built skipped -- this build's \
+ PiecewiseMergeJoinExec does not support RightMark yet: {err}"
+ );
+ return;
+ }
+
+ let mut group = c.benchmark_group("pwmj_vs_nlj_right_mark_hand_built");
+ group.sample_size(10);
+
+ for (regime, right_offset, _fraction) in REGIMES {
+ let lhs_batches = build_batches(LEFT_ROWS, 0, &s);
+ let rhs_batches = build_batches(RIGHT_ROWS, right_offset, &s);
+
+ let pwmj_plan = {
+ let (lhs_batches, rhs_batches, s) =
+ (lhs_batches.clone(), rhs_batches.clone(), Arc::clone(&s));
+ move || -> Arc<dyn ExecutionPlan> {
+ let lhs = MemorySourceConfig::try_new_exec(
+ std::slice::from_ref(&lhs_batches),
+ Arc::clone(&s),
+ None,
+ )
+ .unwrap();
+ let rhs = MemorySourceConfig::try_new_exec(
+ std::slice::from_ref(&rhs_batches),
+ Arc::clone(&s),
+ None,
+ )
+ .unwrap();
+ Arc::new(
+ PiecewiseMergeJoinExec::try_new(
+ lhs,
+ rhs,
+ (
+ Arc::new(Column::new("key", 0)) as _,
+ Arc::new(Column::new("key", 0)) as _,
+ ),
+ Operator::Lt,
+ JoinType::RightMark,
+ 1,
+ )
+ .unwrap(),
+ )
+ }
+ };
+ let nlj_plan = {
+ let (lhs_batches, rhs_batches, s) =
+ (lhs_batches.clone(), rhs_batches.clone(), Arc::clone(&s));
+ move || -> Arc<dyn ExecutionPlan> {
+ let lhs = MemorySourceConfig::try_new_exec(
+ std::slice::from_ref(&lhs_batches),
+ Arc::clone(&s),
+ None,
+ )
+ .unwrap();
+ let rhs = MemorySourceConfig::try_new_exec(
+ std::slice::from_ref(&rhs_batches),
+ Arc::clone(&s),
+ None,
+ )
+ .unwrap();
+ Arc::new(
+ NestedLoopJoinExec::try_new(
+ lhs,
+ rhs,
+ Some(key_lt_key_filter(&s)),
+ &JoinType::RightMark,
+ None,
+ )
+ .unwrap(),
+ )
+ }
+ };
+
+ // `RightMark` keeps every streamed row, matched or not, so both arms
must return
+ // exactly `RIGHT_ROWS` regardless of the regime -- unlike
`RightSemi`/`RightAnti`,
+ // where the regime changes the row count. The regime still matters to
what is timed
+ // below: it changes how much of the comparison work each arm actually
does (`mark`
+ // true vs false), even though the row count it returns cannot show
that.
+ let pwmj_rows = run(pwmj_plan(), &ctx, &rt);
+ let nlj_rows = run(nlj_plan(), &ctx, &rt);
+ assert_eq!(
Review Comment:
`run` returns only the number of rows, but `RightMark` always returns every
row from the right input regardless of the match regime. Because `LEFT_ROWS ==
RIGHT_ROWS`, these assertions cannot even detect returning the wrong side. A
PWMJ implementation that produces incorrect or uniformly false mark values
would therefore pass these guards and could report misleading timings.
Please inspect the materialized mark column before timing. Ideally compare
PWMJ and NLJ results per input row; at minimum, compare their true/false/null
counts and assert the expected all-match, no-match, and half-match
distributions.
##########
datafusion/core/benches/pwmj_semi_anti_sql.rs:
##########
@@ -65,6 +66,26 @@
//! The **same three key offsets** serve both halves, each read against the
other side's
//! extreme: a left row survives `EXISTS` iff `lhs.key < max(rhs.key)`, a
right row survives
//! `RIGHT SEMI` iff `min(lhs.key) < rhs.key`.
+//!
+//! ## Mark joins
+//! `LeftMark` reuses the exact same watermark-marking path as `LeftSemi` --
the only
+//! difference is the final pass, which appends a `mark` column instead of
slicing the batch
+//! -- so it is folded into the same SQL sweep above via an `EXISTS` wrapped
in an
+//! always-false `OR`, the one shape that decorrelates to it (see
`Kind::LeftMark`'s doc). It
+//! needs no build-dependent handling beyond what `Kind::LeftMark` already
gets from
+//! `PWMJ_OR_NLJ`: on a build that does not yet route `LeftMark` to PWMJ, the
planner itself
+//! falls back to `NestedLoopJoinExec`, same as every other case here before
its dependency
+//! landed.
+//!
+//! `RightMark` has no SQL surface at all: no keyword parses to it, and no
optimizer rule
Review Comment:
There is no direct SQL syntax for `RightMark`, but the physical optimizer
can construct one indirectly: `JoinSelection` swaps a statistics-selected
nested-loop join, and `NestedLoopJoinExec::swap_inputs` maps `LeftMark` to
`RightMark` through `JoinType::swap`.
Please narrow this to “no SQL syntax or logical optimizer rule emits
RightMark directly” and describe this group as a deliberate direct-operator
microbenchmark.
##########
datafusion/core/benches/pwmj_semi_anti_sql.rs:
##########
@@ -472,5 +518,180 @@ fn bench_pwmj_semi_anti_sql(c: &mut Criterion) {
group.finish();
}
-criterion_group!(benches, bench_pwmj_semi_anti_sql);
+/// A `JoinFilter` for `lhs.key < rhs.key`, the same range relation every case
in this file
+/// uses, for the `NestedLoopJoinExec` arm below -- which needs one built by
hand since it is
+/// constructed directly rather than planned from SQL.
+fn key_lt_key_filter(s: &SchemaRef) -> JoinFilter {
+ let expr = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("key", 0)),
+ Operator::Lt,
+ Arc::new(Column::new("key", 1)),
+ )) as _;
+ let column_indices = vec![
+ ColumnIndex {
+ index: 0,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 0,
+ side: JoinSide::Right,
+ },
+ ];
+ let key_field = s.field_with_name("key").unwrap().clone();
+ let intermediate_schema = Schema::new(vec![key_field.clone(), key_field]);
+ JoinFilter::new(expr, column_indices, Arc::new(intermediate_schema))
+}
+
+/// `RightMark` has no SQL surface (see the module doc's "Mark joins"
section), so both arms
+/// are hand-built here instead of planned from SQL text:
`PiecewiseMergeJoinExec` and
+/// `NestedLoopJoinExec`, over the same data, the same `lhs.key < rhs.key`
relation, and the
+/// same join type -- with no risk of comparing an operator against itself,
since which
+/// operator each arm uses is fixed by construction rather than read back off
a plan.
+///
+/// Neither arm needs a `SortExec`: `RightMark`, like `RightSemi`/`RightAnti`,
folds the
+/// buffered side to one key regardless of its order, and `NestedLoopJoinExec`
never needs
+/// either side ordered.
+fn bench_pwmj_right_mark_hand_built(c: &mut Criterion) {
+ let rt = Runtime::new().unwrap();
+ let s = schema();
+ let ctx = SessionContext::new();
+
+ // Every other case in this file falls back to `NestedLoopJoinExec`
through the planner
+ // itself when a build does not yet support the join type -- there is no
such fallback
+ // here, since `RightMark` has no SQL surface to plan through in the first
place. Probe
+ // `try_new` directly and skip the whole group rather than panic, so this
benchmark stays
+ // runnable (as a no-op) against a build that has not merged `RightMark`
support yet, and
+ // starts measuring on its own once that support lands.
+ if let Err(err) = PiecewiseMergeJoinExec::try_new(
Review Comment:
This branch converts every `try_new` error into a successful benchmark skip.
Once RightMark support lands, an unrelated schema, operator, or invariant
regression would silently remove the benchmark while reporting that the build
merely lacks support.
Please skip only the specific expected `NotImplemented` error and fail on
every unexpected error. The compatibility skip can then be removed when the
implementation lands.
--
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]
