Skip to content

Commit 4a933ec

Browse files
authored
perf(arrow): evaluate In/NotIn with a hash set for byte, string, and integer columns (#619)
1 parent 885dd29 commit 4a933ec

1 file changed

Lines changed: 211 additions & 0 deletions

File tree

crates/paimon/src/arrow/residual.rs

Lines changed: 211 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -540,6 +540,10 @@ fn evaluate_set_membership_predicate(
540540
});
541541
}
542542

543+
if let Some(mask) = set_membership_hash_mask(array, op, literals) {
544+
return Ok(mask);
545+
}
546+
543547
let mut combined = match op {
544548
PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
545549
PredicateOperator::NotIn => {
@@ -566,6 +570,132 @@ fn evaluate_set_membership_predicate(
566570
Ok(combined)
567571
}
568572

573+
/// One-pass hash-set evaluation of `In`/`NotIn` for byte-like, string, and
574+
/// integer columns. The general path above OR-combines one comparison kernel
575+
/// per literal — O(rows × literals) — which turns a large pushed-down literal
576+
/// set (an engine probing a batch of keys as `In`) quadratic. Returns `None`
577+
/// for column/literal shapes outside the fast path, or when any literal would
578+
/// not convert for the column, so the general path keeps its exact semantics —
579+
/// including its error behavior for unconvertible literals.
580+
fn set_membership_hash_mask(
581+
array: &ArrayRef,
582+
op: PredicateOperator,
583+
literals: &[Datum],
584+
) -> Option<BooleanArray> {
585+
use arrow_schema::DataType as ArrowType;
586+
use std::collections::HashSet;
587+
588+
let keep = matches!(op, PredicateOperator::In);
589+
590+
fn byte_mask<'a>(
591+
values: impl Iterator<Item = Option<&'a [u8]>>,
592+
literals: &[Datum],
593+
keep: bool,
594+
) -> Option<BooleanArray> {
595+
let set = literals
596+
.iter()
597+
.map(|literal| match literal {
598+
Datum::Bytes(bytes) => Some(bytes.as_slice()),
599+
_ => None,
600+
})
601+
.collect::<Option<HashSet<_>>>()?;
602+
Some(
603+
values
604+
.map(|value| Some(value.is_some_and(|v| set.contains(v) == keep)))
605+
.collect(),
606+
)
607+
}
608+
609+
fn str_mask<'a>(
610+
values: impl Iterator<Item = Option<&'a str>>,
611+
literals: &[Datum],
612+
keep: bool,
613+
) -> Option<BooleanArray> {
614+
let set = literals
615+
.iter()
616+
.map(|literal| match literal {
617+
Datum::String(value) => Some(value.as_str()),
618+
_ => None,
619+
})
620+
.collect::<Option<HashSet<_>>>()?;
621+
Some(
622+
values
623+
.map(|value| Some(value.is_some_and(|v| set.contains(v) == keep)))
624+
.collect(),
625+
)
626+
}
627+
628+
fn int_mask<T>(array: &ArrayRef, literals: &[Datum], keep: bool) -> Option<BooleanArray>
629+
where
630+
T: arrow_array::types::ArrowPrimitiveType,
631+
T::Native: TryFrom<i128> + std::hash::Hash + Eq,
632+
{
633+
let array = array
634+
.as_any()
635+
.downcast_ref::<arrow_array::PrimitiveArray<T>>()?;
636+
let set = literals
637+
.iter()
638+
.map(|literal| integer_literal(literal).and_then(|v| T::Native::try_from(v).ok()))
639+
.collect::<Option<HashSet<_>>>()?;
640+
Some(
641+
array
642+
.iter()
643+
.map(|value| Some(value.is_some_and(|v| set.contains(&v) == keep)))
644+
.collect(),
645+
)
646+
}
647+
648+
match array.data_type() {
649+
ArrowType::Binary => byte_mask(
650+
array.as_any().downcast_ref::<BinaryArray>()?.iter(),
651+
literals,
652+
keep,
653+
),
654+
ArrowType::LargeBinary => byte_mask(
655+
array
656+
.as_any()
657+
.downcast_ref::<arrow_array::LargeBinaryArray>()?
658+
.iter(),
659+
literals,
660+
keep,
661+
),
662+
ArrowType::BinaryView => byte_mask(
663+
array
664+
.as_any()
665+
.downcast_ref::<arrow_array::BinaryViewArray>()?
666+
.iter(),
667+
literals,
668+
keep,
669+
),
670+
ArrowType::Utf8 => str_mask(
671+
array.as_any().downcast_ref::<StringArray>()?.iter(),
672+
literals,
673+
keep,
674+
),
675+
ArrowType::LargeUtf8 => str_mask(
676+
array
677+
.as_any()
678+
.downcast_ref::<arrow_array::LargeStringArray>()?
679+
.iter(),
680+
literals,
681+
keep,
682+
),
683+
ArrowType::Utf8View => str_mask(
684+
array
685+
.as_any()
686+
.downcast_ref::<arrow_array::StringViewArray>()?
687+
.iter(),
688+
literals,
689+
keep,
690+
),
691+
ArrowType::Int8 => int_mask::<arrow_array::types::Int8Type>(array, literals, keep),
692+
ArrowType::Int16 => int_mask::<arrow_array::types::Int16Type>(array, literals, keep),
693+
ArrowType::Int32 => int_mask::<arrow_array::types::Int32Type>(array, literals, keep),
694+
ArrowType::Int64 => int_mask::<arrow_array::types::Int64Type>(array, literals, keep),
695+
_ => None,
696+
}
697+
}
698+
569699
fn evaluate_column_predicate(
570700
column: &ArrayRef,
571701
scalar: &Scalar<ArrayRef>,
@@ -985,6 +1115,87 @@ mod tests {
9851115
(0..col.len()).map(|i| col.value(i).to_string()).collect()
9861116
}
9871117

1118+
#[test]
1119+
fn test_in_hash_path_filters_exactly_with_nulls() {
1120+
let f = int_field(0, "age");
1121+
let b = int_batch("age", vec![Some(10), None, Some(20), Some(40), Some(50)]);
1122+
let pred = leaf(
1123+
0,
1124+
DataType::Int(IntType::new()),
1125+
PredicateOperator::In,
1126+
vec![Datum::Int(20), Datum::Int(40), Datum::Int(999)],
1127+
);
1128+
let fp = file_predicates(vec![pred], vec![f.clone()]);
1129+
let out = filter_record_batch_by_predicates(b, &fp, &[f]).unwrap();
1130+
assert_eq!(int_values(&out), vec![20, 40]);
1131+
}
1132+
1133+
#[test]
1134+
fn test_not_in_hash_path_excludes_nulls() {
1135+
let f = str_field(0, "name");
1136+
let b = str_batch(
1137+
"name",
1138+
vec![Some("apple"), None, Some("banana"), Some("cherry")],
1139+
);
1140+
let pred = leaf(
1141+
0,
1142+
DataType::VarChar(VarCharType::string_type()),
1143+
PredicateOperator::NotIn,
1144+
vec![Datum::String("banana".to_string())],
1145+
);
1146+
let fp = file_predicates(vec![pred], vec![f.clone()]);
1147+
let out = filter_record_batch_by_predicates(b, &fp, &[f]).unwrap();
1148+
assert_eq!(str_values(&out), vec!["apple", "cherry"]);
1149+
}
1150+
1151+
#[test]
1152+
fn test_in_hash_path_on_binary_column() {
1153+
use crate::spec::VarBinaryType;
1154+
let array: ArrayRef = Arc::new(BinaryArray::from(vec![
1155+
Some(b"aa".as_slice()),
1156+
None,
1157+
Some(b"bb".as_slice()),
1158+
Some(b"cc".as_slice()),
1159+
]));
1160+
let data_type =
1161+
DataType::VarBinary(VarBinaryType::try_new(true, VarBinaryType::MAX_LENGTH).unwrap());
1162+
let literals = vec![Datum::Bytes(b"bb".to_vec()), Datum::Bytes(b"zz".to_vec())];
1163+
let mask =
1164+
evaluate_set_membership_predicate(&array, &data_type, PredicateOperator::In, &literals)
1165+
.unwrap();
1166+
assert_eq!(
1167+
mask.iter().collect::<Vec<_>>(),
1168+
vec![Some(false), Some(false), Some(true), Some(false)]
1169+
);
1170+
let mask = evaluate_set_membership_predicate(
1171+
&array,
1172+
&data_type,
1173+
PredicateOperator::NotIn,
1174+
&literals,
1175+
)
1176+
.unwrap();
1177+
assert_eq!(
1178+
mask.iter().collect::<Vec<_>>(),
1179+
vec![Some(true), Some(false), Some(false), Some(true)]
1180+
);
1181+
}
1182+
1183+
#[test]
1184+
fn test_in_unconvertible_literal_still_errors() {
1185+
// An out-of-range literal must keep the general path's error behavior:
1186+
// the hash path declines the literal set and the per-literal loop
1187+
// raises the unconvertible-literal error it always has.
1188+
use crate::spec::TinyIntType;
1189+
let array: ArrayRef = Arc::new(arrow_array::Int8Array::from(vec![Some(1i8), Some(2)]));
1190+
let result = evaluate_set_membership_predicate(
1191+
&array,
1192+
&DataType::TinyInt(TinyIntType::new()),
1193+
PredicateOperator::In,
1194+
&[Datum::Long(300)],
1195+
);
1196+
assert!(result.is_err());
1197+
}
1198+
9881199
#[test]
9891200
fn test_gt_filters_exactly() {
9901201
let f = int_field(0, "age");

0 commit comments

Comments
 (0)