Skip to content
Open
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
8 changes: 6 additions & 2 deletions vortex-array/src/expr/exprs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1227,11 +1227,14 @@ pub fn in_list_of(
}

// Ordered elements make equal lists from the same set, so that equal filters deduplicate.
// An engine can build a list from the keys of a hash join, which repeat, and each repeat makes
// the lookup more costly, so equal elements are removed. One null keeps the SQL semantics.
if elements
.windows(2)
.all(|pair| pair[0].partial_cmp(&pair[1]).is_some())
{
elements.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
elements.dedup();
}
let list = Scalar::list(Arc::new(element_dtype), elements, Nullability::NonNullable);
Ok(in_list(needle, lit(list)))
Expand All @@ -1243,10 +1246,11 @@ fn contiguous_integer_bounds(scalars: &[Scalar]) -> Option<(&Scalar, &Scalar)> {
let mut values = scalars
.iter()
.map(|scalar| {
let primitive = scalar.as_primitive_opt()?;
if !primitive.ptype().is_int() {
// Viewing a scalar of another type as a primitive builds an error, which is costly.
if !matches!(scalar.dtype(), DType::Primitive(ptype, _) if ptype.is_int()) {
return None;
}
let primitive = scalar.as_primitive_opt()?;
let value = if primitive.ptype().is_unsigned_int() {
primitive.as_opt::<u64>()??.into()
} else {
Expand Down
16 changes: 15 additions & 1 deletion vortex-datafusion/src/convert/exprs/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -273,7 +273,14 @@ impl DefaultExpressionConverter {
)
.map_err(|_| Unconverted::Unsupported)?;
let converted = self.convert_expr(expr, schema, &input_dtype)?;
let expected_dtype = self.arrow_dtype(expr.return_field(schema)?.as_ref())?;
// `return_field` names the field by formatting the whole expression. For the IN list of a
// hash join filter, with thousands of keys, that costs more than the conversion. A
// predicate has no field metadata, so its type and nullability are enough.
let expected_dtype = if expr.data_type(schema)? == DataType::Boolean {
DType::Bool(expr.nullable(schema)?.into())
} else {
self.arrow_dtype(expr.return_field(schema)?.as_ref())?
};
if !converted_dtype(&converted, &input_dtype)?.eq_ignore_nullability(&expected_dtype) {
return Err(Unconverted::Unsupported);
}
Expand Down Expand Up @@ -442,9 +449,16 @@ impl DefaultExpressionConverter {
}
let value_dtype = converted_dtype(&value, input_dtype)?;

// The list of a hash join filter holds every build-side key, which repeat. Each element
// converts through Arrow, so equal elements are converted once.
let elements = in_list
.list()
.iter()
.unique_by(|element| {
element
.downcast_ref::<df_expr::Literal>()
.map(|literal| literal.value().clone())
})
.map(|element| {
let element = self.convert_expr(element, schema, input_dtype)?;
element
Expand Down
27 changes: 27 additions & 0 deletions vortex-datafusion/src/convert/exprs/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -365,6 +365,33 @@ fn test_in_list_converts_to_set_lookup(#[values(false, true)] negated: bool) ->
Ok(())
}

#[test]
fn test_in_list_drops_repeated_elements() -> DFResult<()> {
let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
let expr = Arc::new(df_expr::InListExpr::try_new(
Arc::new(df_expr::Column::new("a", 0)),
int_literals([Some(7), None, Some(1), Some(7), None, Some(1)]),
false,
&schema,
)?);
let element_dtype = Arc::new(DType::Primitive(PType::I32, Nullability::Nullable));
let list = Scalar::list(
element_dtype,
vec![
Scalar::null(DType::Primitive(PType::I32, Nullability::Nullable)),
Scalar::from(Some(1i32)),
Scalar::from(Some(7i32)),
],
Nullability::NonNullable,
);

assert_eq!(
convert(expr, &schema)?,
vortex::expr::in_list(get_item("a", root()), lit(list))
);
Ok(())
}

#[test]
fn test_contiguous_in_list_converts_to_between() -> DFResult<()> {
let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
Expand Down
Loading