Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
61 changes: 36 additions & 25 deletions datafusion/functions-nested/benches/array_has.rs
Original file line number Diff line number Diff line change
Expand Up @@ -913,31 +913,42 @@ fn bench_array_has_any_scalar(c: &mut Criterion) {
Field::new("second", list_type_i64.clone(), false).into(),
];

for &scalar_size in &scalar_sizes {
let scalar_arg = create_int64_scalar_list(scalar_size, array_size as i64);
let args = vec![
ColumnarValue::Array(first_arr_i64.clone()),
ColumnarValue::Scalar(scalar_arg),
];
group.bench_with_input(
BenchmarkId::new("i64_no_match", scalar_size),
&scalar_size,
|b, _| {
let udf = ArrayHasAny::new();
b.iter(|| {
black_box(
udf.invoke_with_args(ScalarFunctionArgs {
args: args.clone(),
arg_fields: arg_fields_i64.clone(),
number_rows: NUM_ROWS,
return_field: return_field.clone(),
config_options: config_options.clone(),
})
.unwrap(),
)
})
},
);
// The same number of visible rows, sliced from the middle of a backing
// array that is 100 times larger.
let backing_rows = NUM_ROWS * 100;
let sliced_arr_i64 = create_int64_list_array(backing_rows, array_size, NULL_DENSITY)
.slice(backing_rows / 2, NUM_ROWS);

for (name, arr) in [
("i64_no_match", &first_arr_i64),
("i64_no_match_sliced", &sliced_arr_i64),
] {
for &scalar_size in &scalar_sizes {
let scalar_arg = create_int64_scalar_list(scalar_size, array_size as i64);
let args = vec![
ColumnarValue::Array(arr.clone()),
ColumnarValue::Scalar(scalar_arg),
];
group.bench_with_input(
BenchmarkId::new(name, scalar_size),
&scalar_size,
|b, _| {
let udf = ArrayHasAny::new();
b.iter(|| {
black_box(
udf.invoke_with_args(ScalarFunctionArgs {
args: args.clone(),
arg_fields: arg_fields_i64.clone(),
number_rows: NUM_ROWS,
return_field: return_field.clone(),
config_options: config_options.clone(),
})
.unwrap(),
)
})
},
);
}
}

// String benchmarks
Expand Down
110 changes: 103 additions & 7 deletions datafusion/functions-nested/src/array_has.rs
Original file line number Diff line number Diff line change
Expand Up @@ -966,10 +966,17 @@ fn array_has_any_with_scalar_general(
};

let col_list: ArrayWrapper = col_arr.as_ref().try_into()?;
let col_rows = converter.convert_columns(&[Arc::clone(col_list.values())])?;
let col_offsets: Vec<usize> = col_list.offsets().collect();
let col_nulls = col_list.nulls();

// For efficiency with sliced arrays, only convert the visible elements,
// not the entire underlying buffer. Indices into `col_rows` are therefore
// relative to `elem_start`.
let elem_start = col_offsets[0];
let elem_end = col_offsets[col_list.len()];
let visible_values = col_list.values().slice(elem_start, elem_end - elem_start);
let col_rows = converter.convert_columns(&[visible_values])?;

let mut result = BooleanBufferBuilder::new(col_list.len());
let num_scalar = scalar_rows.num_rows();

Expand All @@ -984,8 +991,8 @@ fn array_has_any_with_scalar_general(
result.append(false);
continue;
}
let start = col_offsets[i];
let end = col_offsets[i + 1];
let start = col_offsets[i] - elem_start;
let end = col_offsets[i + 1] - elem_start;
let found =
(start..end).any(|j| scalar_set.contains(col_rows.row(j).as_ref()));
result.append(found);
Expand All @@ -997,8 +1004,8 @@ fn array_has_any_with_scalar_general(
result.append(false);
continue;
}
let start = col_offsets[i];
let end = col_offsets[i + 1];
let start = col_offsets[i] - elem_start;
let end = col_offsets[i + 1] - elem_start;
let found = (start..end)
.any(|j| (0..num_scalar).any(|k| col_rows.row(j) == scalar_rows.row(k)));
result.append(found);
Expand Down Expand Up @@ -1216,8 +1223,8 @@ mod tests {
use arrow::datatypes::Int32Type;
use arrow::{
array::{
Array, ArrayRef, AsArray, FixedSizeListArray, Int32Array, ListArray,
create_array,
Array, ArrayRef, AsArray, FixedSizeListArray, Int32Array, LargeListArray,
ListArray, create_array,
},
buffer::OffsetBuffer,
datatypes::{DataType, Field, Schema},
Expand Down Expand Up @@ -1683,4 +1690,93 @@ mod tests {
vec![Some(true), Some(false)]
);
}

/// Invoke `array_has_any` with an array haystack and a scalar list of
/// `Int32` elements.
fn invoke_array_has_any_scalar(haystack: ArrayRef, scalar: Vec<i32>) -> ArrayRef {
let num_rows = haystack.len();
let haystack_type = haystack.data_type().clone();
let scalar = SingleRowListArrayBuilder::new(Arc::new(Int32Array::from(scalar)))
.build_list_scalar();
let scalar_type = scalar.data_type();
ArrayHasAny::new()
.invoke_with_args(ScalarFunctionArgs {
args: vec![
ColumnarValue::Array(haystack),
ColumnarValue::Scalar(scalar),
],
arg_fields: vec![
Arc::new(Field::new("haystack", haystack_type, true)),
Arc::new(Field::new("scalar", scalar_type, true)),
],
number_rows: num_rows,
return_field: Arc::new(Field::new("return", DataType::Boolean, true)),
config_options: Arc::new(ConfigOptions::default()),
})
.unwrap()
.into_array(num_rows)
.unwrap()
}

#[test]
fn test_array_has_any_scalar_sliced() {
Comment thread
neilconway marked this conversation as resolved.
// The scalar path only row-converts the visible elements of a sliced
// haystack, so its per-row element ranges must be relative to the
// start of the visible range. 1 and 70 appear only in sliced-away rows.
let full = ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
Some(vec![Some(1), Some(2)]),
Some(vec![Some(10), None, Some(30)]),
None,
Some(vec![]),
Some(vec![Some(50), Some(60)]),
Some(vec![Some(70)]),
]);
let sliced: ArrayRef = Arc::new(full.slice(1, 4));

// At most `SCALAR_SMALL_THRESHOLD` scalar elements: linear scan.
let result = invoke_array_has_any_scalar(Arc::clone(&sliced), vec![1, 70, 60]);
assert_eq!(
result.as_boolean().iter().collect::<Vec<_>>(),
vec![Some(false), None, Some(false), Some(true)]
);
let result = invoke_array_has_any_scalar(Arc::clone(&sliced), vec![30]);
assert_eq!(
result.as_boolean().iter().collect::<Vec<_>>(),
vec![Some(true), None, Some(false), Some(false)]
);

// More than `SCALAR_SMALL_THRESHOLD` scalar elements: HashSet lookup.
let large_scalar = vec![1, 70, 60, 101, 102, 103, 104, 105, 106, 107];
let result = invoke_array_has_any_scalar(sliced, large_scalar);
assert_eq!(
result.as_boolean().iter().collect::<Vec<_>>(),
vec![Some(false), None, Some(false), Some(true)]
);

// Sliced FixedSizeList (width 2; rows 1..=2 of
// [[1,2],[11,12],[21,22],[31,32]] visible).
let field = Arc::new(Field::new("item", DataType::Int32, true));
let fsl_values = Arc::new(Int32Array::from(vec![1, 2, 11, 12, 21, 22, 31, 32]));
let fsl: ArrayRef =
Arc::new(FixedSizeListArray::new(field, 2, fsl_values, None).slice(1, 2));
let result = invoke_array_has_any_scalar(fsl, vec![1, 22, 31]);
assert_eq!(
result.as_boolean().iter().collect::<Vec<_>>(),
vec![Some(false), Some(true)]
);

// Sliced LargeList; 1 and 70 appear only in sliced-away rows.
let large = LargeListArray::from_iter_primitive::<Int32Type, _, _>(vec![
Some(vec![Some(1), Some(2)]),
Some(vec![Some(10), None, Some(30)]),
Some(vec![Some(50), Some(60)]),
Some(vec![Some(70)]),
]);
let large: ArrayRef = Arc::new(large.slice(1, 2));
let result = invoke_array_has_any_scalar(large, vec![1, 70, 60]);
assert_eq!(
result.as_boolean().iter().collect::<Vec<_>>(),
vec![Some(false), Some(true)]
);
}
}
Loading