You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/02/22 16:18:47 UTC
[PR] fix: propagate errors from take in sort merge join spill path [datafusion]
andygrove opened a new pull request, #20480:
URL: https://github.com/apache/datafusion/pull/20480
## Summary
- Fix silent error swallowing in `fetch_right_columns_from_batch_by_idxs` spill path where `vec.extend(take(...))` silently dropped errors (since `Result<T>` implements `IntoIterator`, yielding 0 items on `Err`)
- Replace with `vec.push(take(...)?)` to properly propagate errors
- Fix incorrect `Vec::with_capacity(buffered_indices.len())` which pre-allocated based on row count instead of column count
- Add unit test verifying spill path produces identical results to in-memory path
## Test plan
- [x] `cargo test -p datafusion-physical-plan sort_merge_join` — all 49 tests pass
- [x] `cargo clippy -p datafusion-physical-plan -- -D warnings` — clean
- [x] New `spill_path_matches_in_memory_path` test verifies column-for-column parity between spill and in-memory paths
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: propagate errors from take in sort merge join spill path [datafusion]
Posted by "jonathanc-n (via GitHub)" <gi...@apache.org>.
jonathanc-n commented on code in PR #20480:
URL: https://github.com/apache/datafusion/pull/20480#discussion_r2839369130
##########
datafusion/physical-plan/src/joins/sort_merge_join/stream.rs:
##########
Review Comment:
Expanded upwards and found this duplicate comment. If possible can you quickly remove 😆
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: propagate errors from take in sort merge join spill path [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #20480:
URL: https://github.com/apache/datafusion/pull/20480#discussion_r2944817890
##########
datafusion/physical-plan/src/joins/sort_merge_join/stream.rs:
##########
@@ -1807,3 +1805,93 @@ fn is_join_arrays_equal(
}
Ok(true)
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::datatypes::{Field, Schema};
+ use arrow::ipc::writer::StreamWriter;
+ use datafusion_execution::disk_manager::DiskManager;
+
+ /// Creates a `BufferedBatch` with `BufferedBatchState::Spilled` by writing
+ /// the given batch to an IPC temp file.
+ fn make_spilled_batch(batch: &RecordBatch) -> Result<BufferedBatch> {
Review Comment:
`make_spilled_batch` writes IPC files directly with `StreamWriter`, which is fine for this test, but using the existing spill machinery (SpillManager) would keep the test closer to the production path.
##########
datafusion/physical-plan/src/joins/sort_merge_join/stream.rs:
##########
@@ -1807,3 +1805,93 @@ fn is_join_arrays_equal(
}
Ok(true)
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::datatypes::{Field, Schema};
+ use arrow::ipc::writer::StreamWriter;
+ use datafusion_execution::disk_manager::DiskManager;
+
+ /// Creates a `BufferedBatch` with `BufferedBatchState::Spilled` by writing
+ /// the given batch to an IPC temp file.
+ fn make_spilled_batch(batch: &RecordBatch) -> Result<BufferedBatch> {
+ let disk_manager = Arc::new(DiskManager::builder().build()?);
+ let spill_file = disk_manager.create_tmp_file("test_spill")?;
+ {
+ let file = File::create(spill_file.path())?;
+ let mut writer = StreamWriter::try_new(file, &batch.schema())?;
+ writer.write(batch)?;
+ writer.finish()?;
+ }
+ Ok(BufferedBatch {
+ batch: BufferedBatchState::Spilled(spill_file),
+ range: 0..batch.num_rows(),
+ join_arrays: vec![],
+ null_joined: vec![],
+ size_estimation: 0,
+ join_filter_not_matched_map: HashMap::new(),
+ num_rows: batch.num_rows(),
+ })
+ }
+
+ /// Verifies the spill path in `fetch_right_columns_from_batch_by_idxs`
+ /// produces identical results to the in-memory path.
+ ///
+ /// This catches a prior bug where the spill path used
+ /// `vec.extend(take(...))`. Since `Result` implements `IntoIterator`
+ /// (yielding 0 items on `Err`), any `take` error would silently drop
+ /// that column from the output instead of propagating the error.
+ #[test]
+ fn spill_path_matches_in_memory_path() -> Result<()> {
Review Comment:
The new regression test only covers the happy path.
It does not exercise the silent error-swallowing bug that this patch fixes.
The previous bug appeared when take(...) returned Err. Because Result<T, E> implements IntoIterator, vec.extend(result) would quietly add zero items instead of failing.
In this test, all indices are in bounds, so every take(...) succeeds. The old buggy implementation would also pass this test.
Please add a failure-path case. Use an out-of-bounds index (or another way to make take return an error).
The test should verify that the spilled path returns an error rather than a truncated column list.
##########
datafusion/physical-plan/src/joins/sort_merge_join/stream.rs:
##########
@@ -1637,16 +1637,14 @@ fn fetch_right_columns_from_batch_by_idxs(
}
// If the batch was spilled to disk, less likely
BufferedBatchState::Spilled(spill_file) => {
- let mut buffered_cols: Vec<ArrayRef> =
- Vec::with_capacity(buffered_indices.len());
-
let file = BufReader::new(File::open(spill_file.path())?);
let reader = StreamReader::try_new(file, None)?;
+ let mut buffered_cols: Vec<ArrayRef> = Vec::new();
for batch in reader {
- batch?.columns().iter().for_each(|column| {
- buffered_cols.extend(take(column, &buffered_indices, None))
- });
+ for column in batch?.columns() {
+ buffered_cols.push(take(column, buffered_indices, None)?);
Review Comment:
The spilled branch still reimplements the column-by-column take logic.
It would be clearer if both branches shared a small helper over &[ArrayRef].
Alternatively, the spilled branch could call take_arrays(batch.columns(), buffered_indices, None) after reading the spilled batch.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: propagate errors from take in sort merge join spill path [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #20480: fix: propagate errors from take in sort merge join spill path
URL: https://github.com/apache/datafusion/pull/20480
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: propagate errors from take in sort merge join spill path [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20480:
URL: https://github.com/apache/datafusion/pull/20480#issuecomment-4468883805
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org