SubhamSinghal commented on code in PR #24955:
URL: https://github.com/apache/datafusion/pull/24955#discussion_r3941151101
##########
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:
Addressed in 89a1400f2ea1ad3e35d8be19133fade2e716a871
--
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]
