Skip to content

perf: Fix NLJ slow join with condition array_has #18161

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alamb merged 2 commits into from
Oct 20, 2025
Merged

perf: Fix NLJ slow join with condition array_has #18161

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e3894fe
fix nlj slow list join
2010YOUY01 Oct 20, 2025
860c47e
Add slt test coverage
alamb Oct 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 29 additions & 6 deletions datafusion/physical-plan/src/joins/nested_loop_join.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,11 +48,15 @@ use crate::{

use arrow::array::{
new_null_array, Array, BooleanArray, BooleanBufferBuilder, RecordBatchOptions,
UInt64Array,
};
use arrow::buffer::BooleanBuffer;
use arrow::compute::{concat_batches, filter, filter_record_batch, not, BatchCoalescer};
use arrow::compute::{
concat_batches, filter, filter_record_batch, not, take, BatchCoalescer,
};
use arrow::datatypes::{Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
use arrow_schema::DataType;
use datafusion_common::cast::as_boolean_array;
use datafusion_common::{
arrow_err, internal_datafusion_err, internal_err, project_schema,
Expand Down Expand Up @@ -1661,11 +1665,30 @@ fn build_row_join_batch(
// Broadcast the single build-side row to match the filtered
// probe-side batch length
let original_left_array = build_side_batch.column(column_index.index);
let scalar_value = ScalarValue::try_from_array(
original_left_array.as_ref(),
build_side_index,
)?;
scalar_value.to_array_of_size(filtered_probe_batch.num_rows())?
// Avoid using `ScalarValue::to_array_of_size()` for `List(Utf8View)` to avoid
// deep copies for buffers inside `Utf8View` array. See below for details.
// https://github.com/apache/datafusion/issues/18159

@alamb alamb Oct 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The root cause is tracked in

//
// In other cases, `to_array_of_size()` is faster.
match original_left_array.data_type() {
DataType::List(field) | DataType::LargeList(field)
if field.data_type() == &DataType::Utf8View =>
{
let indices_iter = std::iter::repeat_n(
build_side_index as u64,
filtered_probe_batch.num_rows(),
);
let indices_array = UInt64Array::from_iter_values(indices_iter);
take(original_left_array.as_ref(), &indices_array, None)?
}
Comment on lines +1674 to +1683

@alamb alamb Oct 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this approach could be ported into ScalarValue::to_array_of_size itself rather than special cased here -- which would improve performance in potentially other places

datafusion/datafusion/common/src/scalar/mod.rs

Lines 3238 to 3245 in 556eb9b

fn list_to_array_of_size(arr: &dyn Array, size: usize) -> Result<ArrayRef> {
let arrays = repeat_n(arr, size).collect::<Vec<_>>();
let ret = match !arrays.is_empty() {
true => arrow::compute::concat(arrays.as_slice())?,
false => arr.slice(0, 0),
};
Ok(ret)
}

That being said, I think this is a nice point fix that we can safely backport to the datafusion 50 branch, so I think we should merge this PR / backport it and I will file a follow on PR to further improve the code

_ => {
let scalar_value = ScalarValue::try_from_array(
original_left_array.as_ref(),
build_side_index,
)?;
scalar_value.to_array_of_size(filtered_probe_batch.num_rows())?
}
}
} else {
// Take the filtered probe-side column using compute::take
Arc::clone(filtered_probe_batch.column(column_index.index))
Expand Down