PERF : modify SMJ shuffle file reader to skip validation

[getChan]

Which issue does this PR close?

Rationale for this change

#14078 #15454 shows when read shuffle file, skipping validation is effective.

What changes are included in this PR?

when SortMergeJoinExec read shuffle file, skipping validation

Are these changes tested?

• already exist cargo test passed.
• Verification completed that the read shuffle file is identical to the written one
• Benchmark test is not included because it was difficult to limit them to shuffle read scope.

Are there any user-facing changes?

no

[alamb]

Benchmark test is not included because it was difficult to limit them to shuffle read scope.

Did you test it locally? Do you have any performance numbers you can share?

• I wonder if we could consolidate the spill file managing into SpillManager that was introduced by @2010YOUY01 in refactor: Use SpillManager for all spilling scenarios #15405

[2010YOUY01]

Benchmark test is not included because it was difficult to limit them to shuffle read scope.

Did you test it locally? Do you have any performance numbers you can share?

• I wonder if we could consolidate the spill file managing into SpillManager that was introduced by @2010YOUY01 in refactor: Use SpillManager for all spilling scenarios #15405

I believe the SMJ reader is the only remaining component that hasn't been migrated to SpillManager (the write path of SMJ has already been refactored to use SpillManager). I agree it would be great to also refactor it to use SpillManager.

[getChan]

SMJExec already use SpillManager at spill write

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

Lines 1439 to 1451 in 11bf46e

	Err(_) if self.runtime_env.disk_manager.tmp_files_enabled() => {
	// Spill buffered batch to disk
	if let Some(batch) = buffered_batch.batch {
	let spill_file = self
	.spill_manager
	.spill_record_batch_and_finish(
	&[batch],
	"sort_merge_join_buffered_spill",
	)?
	.unwrap(); // Operation only return None if no batches are spilled, here we ensure that at least one batch is spilled
	buffered_batch.spill_file = Some(spill_file);
	buffered_batch.batch = None;

but at spill read, direct read spill files

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

Lines 2311 to 2316 in 11bf46e

	(Some(spill_file), None) => {
	let mut buffered_cols: Vec<ArrayRef> =
	Vec::with_capacity(buffered_indices.len());
	let file = BufReader::new(File::open(spill_file.path())?);
	let reader = unsafe {StreamReader::try_new(file, None)?.with_skip_validation(true)};

I will check if using SpillManager is possible. (another issue)

[getChan]

I just add benchmarks for SMJExec spill read execution.
maybe since it is not limited to the read spill scope, the diff is small.

sort_merge_join_spill/SortMergeJoinExec_spill
                        time:   [321.88 µs 325.75 µs 330.51 µs]
                        change: [-2.8665% -1.3757% +0.1785%] (p = 0.09 > 0.05)
                        No change in performance detected.
Found 2 outliers among 100 measurements (2.00%)
  1 (1.00%) high mild
  1 (1.00%) high severe

[getChan]

To use test utilities in the bench as well. Is it okay?

[alamb]

I think you could avoid doing this by using the actual operators -- see comment above

[alamb]

I think a bencmark that has only 5 rows is likely going to only measure the overhead of plan setup rather than the actual performance of a large join that needs to spill

Perhaps we can increase this size to 1M rows or something (is important that b1 and b2 remain sorted)

[getChan]

thanks for review!
now I run benchmark with 1_048_576 rows. with all row spill.
but benchmark result is little performance improvement...

SortMergeJoinExec_spill time:   [79.761 s 79.858 s 79.974 s]
                        change: [-0.7912% -0.5805% -0.3386%] (p = 0.00 < 0.05)
                        Change within noise threshold.
Found 1 outliers among 10 measurements (10.00%)
  1 (10.00%) high mild

It seems that skip validation is small impact on the overall execution.

[alamb]

I think you could avoid doing this by using the actual operators -- see comment above

[alamb]

I worry that using test only structures like this will means the benchmark is not measuring performance that will map directly to query performance. I think you could move the file from

• datafusion/physical-plan/benches/sort_merge_join.rs

to

• datafusion/core/benches/sort_merge_join.rs

And use a SessionContext and actual query to run to be closer.

Here is an example that does something similar: https://github.com/apache/datafusion/blob/main/datafusion/core/benches/filter_query_sql.rs

[getChan]

1. I moved file to /core/benches/
2. try to using actual query by SessionContext, but it is hard to simulate.
   Because It satisfied SMJExec+spill execution, but securing enough memory for operations like RepartitionExec was challenging.