diff --git a/Cargo.lock b/Cargo.lock index 7c3a95148fa..7dc188cd414 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11735,6 +11735,7 @@ name = "vortex-datafusion" version = "0.1.0" dependencies = [ "anyhow", + "arc-swap", "arrow-array 59.3.0", "arrow-schema 59.3.0", "async-trait", diff --git a/vortex-datafusion/Cargo.toml b/vortex-datafusion/Cargo.toml index 5335af982d0..0006a03a9c9 100644 --- a/vortex-datafusion/Cargo.toml +++ b/vortex-datafusion/Cargo.toml @@ -13,6 +13,7 @@ rust-version = { workspace = true } version = { workspace = true } [dependencies] +arc-swap = { workspace = true } arrow-array = { workspace = true } arrow-schema = { workspace = true } async-trait = { workspace = true } diff --git a/vortex-datafusion/src/convert/dynamic.rs b/vortex-datafusion/src/convert/dynamic.rs new file mode 100644 index 00000000000..cbe40a23333 --- /dev/null +++ b/vortex-datafusion/src/convert/dynamic.rs @@ -0,0 +1,1050 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Conversion of DataFusion [`DynamicFilterPhysicalExpr`]s into live Vortex dynamic comparisons. +//! +//! DataFusion operators such as TopK and hash joins push a [`DynamicFilterPhysicalExpr`] into the +//! scan and tighten it while the query runs. Its column children are fixed, so for each child we +//! emit [`dynamic`] comparisons whose right-hand sides are re-read from the DataFusion filter +//! whenever its generation changes. If the filter already has bounds when a file is opened, only +//! the comparisons those bounds use are emitted, since a TopK filter keeps its shape as it +//! tightens. Before its first update (it starts as `true`) the shape is unknown, so every column +//! gets the full template (`<`, `<=`, `>`, `>=`). +//! +//! Every comparison defaults to `true` when the current DataFusion predicate has no matching +//! bound, so the Vortex filter only ever keeps a superset of the rows DataFusion's filter keeps. +//! That is sufficient because the operators that produce dynamic filters still enforce their own +//! semantics; the filter only lets the scan skip data. +//! +//! Filters that are already complete when the file is opened (typically hash-join filters, whose +//! build side has finished) can never change again. Their bounds are read once and only applied +//! when the file's statistics suggest they will skip a meaningful share of rows, because +//! evaluating a non-selective filter costs more than it saves. +//! When range bounds are too broad, a small, completed integer membership filter can still skip +//! most rows. Such filters are retained when their size is small relative to the file's range. +//! Completed, null-rejecting comparisons also discard nulls; live filters retain them because a +//! later generation may admit them. + +use std::cmp::Ordering; +use std::sync::Arc; + +use arc_swap::ArcSwapOption; +use datafusion_common::ScalarValue as DFScalarValue; +use datafusion_expr::Operator as DFOperator; +use datafusion_physical_expr::DynamicFilterTracking; +use datafusion_physical_expr::PhysicalExpr; +use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr; +use datafusion_physical_expr::split_conjunction; +use datafusion_physical_plan::expressions as df_expr; +use vortex::dtype::DType; +use vortex::dtype::StructFields; +use vortex::expr::Expression; +use vortex::expr::and_collect; +use vortex::expr::binary; +use vortex::expr::dynamic; +use vortex::expr::get_item; +use vortex::expr::is_null; +use vortex::expr::lit; +use vortex::expr::or; +use vortex::expr::root; +use vortex::expr::stats::Stat; +use vortex::file::FileStatistics; +use vortex::scalar::Scalar; +use vortex::scalar::ScalarValue; +use vortex::scalar_fn::fns::operators::CompareOperator; +use vortex::session::VortexSession; + +use crate::convert::exprs::DefaultExpressionConvertor; +use crate::convert::exprs::ExpressionConvertor; +use crate::convert::scalar_from_df; + +/// Complete filters are only applied when they are estimated to keep at most this fraction of a +/// file's value range for some column. +const MAX_KEPT_FRACTION: f64 = 0.5; + +// Constant IN lists currently expand into a disjunction of comparisons. Bound that work while +// retaining the small dimension-key sets whose broad min/max bounds miss most of their pruning. +const MAX_MEMBERSHIP_VALUES: usize = 16; +// Each comparison can decode the same compressed column again. Require more pruning than a +// range filter to offset that work, including for compact encodings such as PCO. +const MAX_MEMBERSHIP_KEPT_FRACTION: f64 = 0.1; + +const TEMPLATE_OPS: [CompareOperator; 4] = [ + CompareOperator::Lt, + CompareOperator::Lte, + CompareOperator::Gt, + CompareOperator::Gte, +]; + +/// The column referenced by `expr`, looking through casts the expression adapter adds when a +/// file's type differs from the table's (e.g. a different timestamp unit). +/// +/// Bounds on the cast column are converted back to the file type and only kept when that +/// conversion is exact, see [`LiveBounds::parse_comparison`]. +fn column_of(expr: &Arc) -> Option<&df_expr::Column> { + expr.downcast_ref::().or_else(|| { + expr.downcast_ref::()? + .expr() + .downcast_ref::() + }) +} + +/// Returns the dynamic filter if `expr` is one whose children are all (possibly cast) columns. +pub(crate) fn as_column_dynamic_filter( + expr: &Arc, +) -> Option<&DynamicFilterPhysicalExpr> { + let dynamic = expr.downcast_ref::()?; + let children = dynamic.children(); + (!children.is_empty() && children.iter().all(|child| column_of(child).is_some())) + .then_some(dynamic) +} + +/// Converts a DataFusion dynamic filter into a Vortex expression over a file with `file_fields`. +/// +/// Returns `None` when the filter is not worth applying to this file, in which case the scan +/// simply ignores it. +pub(crate) fn dynamic_filter_to_vortex( + expr: &Arc, + file_fields: &StructFields, + file_stats: Option<&FileStatistics>, + session: &VortexSession, +) -> Option { + let dynamic_filter = as_column_dynamic_filter(expr)?; + + let columns: Vec<(String, DType)> = dynamic_filter + .children() + .into_iter() + .filter_map(|child| column_of(child)) + .filter_map(|col| { + let dtype = file_fields.field(col.name())?; + Some((col.name().to_owned(), dtype)) + }) + .collect(); + if columns.is_empty() { + return None; + } + + let bounds = Arc::new(LiveBounds { + filter: Arc::clone(expr), + columns: columns.clone(), + session: session.clone(), + cache: ArcSwapOption::empty(), + }); + + if matches!( + DynamicFilterTracking::classify(expr), + DynamicFilterTracking::AllComplete + ) { + return bounds.selective_static_filter(file_fields, file_stats?); + } + + let current = bounds.current_bounds(); + let per_column = columns + .into_iter() + .enumerate() + .filter_map(|(idx, (name, dtype))| { + // Once the filter has bounds, columns and operators it doesn't bound are skipped. + let ops: Vec<_> = if current.is_empty() { + TEMPLATE_OPS.to_vec() + } else { + TEMPLATE_OPS + .into_iter() + .filter(|op| current.iter().any(|(c, o, _)| *c == idx && o == op)) + .collect() + }; + let lhs = get_item(name, root()); + let comparisons = and_collect(ops.into_iter().map(|op| { + let bounds = Arc::clone(&bounds); + dynamic( + op, + move || bounds.value(idx, op), + dtype.clone(), + true, + lhs.clone(), + ) + }))?; + // Nulls always pass: the DataFusion filter may keep them (e.g. TopK with NULLS FIRST), + // and keeping extra rows is always safe. + Some(if dtype.is_nullable() { + or(is_null(lhs), comparisons) + } else { + comparisons + }) + }); + + and_collect(per_column) +} + +/// `(column index, operator, value)`, with the value already cast to the column's file dtype. +type Bound = (usize, CompareOperator, ScalarValue); + +/// The bounds extracted from one generation of a dynamic filter. +struct Snapshot { + generation: u64, + bounds: Arc<[Bound]>, +} + +struct LiveBounds { + /// Always a [`DynamicFilterPhysicalExpr`]; kept type-erased because that type is not `Clone`. + filter: Arc, + columns: Vec<(String, DType)>, + session: VortexSession, + /// Read on every evaluation of the Vortex filter but replaced only when the DataFusion filter + /// changes, a few dozen times per query for TopK, so reads must not contend. + cache: ArcSwapOption, +} + +impl LiveBounds { + /// Builds static comparisons from the filter's current bounds, keeping only columns whose + /// bounds are estimated to exclude a meaningful share of the file. + fn selective_static_filter( + &self, + file_fields: &StructFields, + file_stats: &FileStatistics, + ) -> Option { + let current = self + .filter + .downcast_ref::()? + .current() + .ok()?; + let bounds = self.bounds_of(¤t).unwrap_or_default(); + let per_column = self + .columns + .iter() + .enumerate() + .filter_map(|(idx, (name, dtype))| { + let column_bounds: Vec<_> = bounds + .iter() + .filter(|(c, ..)| *c == idx) + .map(|(_, op, value)| { + (*op, Scalar::try_new(dtype.clone(), Some(value.clone()))) + }) + .map(|(op, scalar)| Some((op, scalar.ok()?))) + .collect::>()?; + let (stats, _) = file_stats.get(file_fields.find(name)?); + let stat = |stat| Scalar::try_new(dtype.clone(), stats.get(stat).into_inner()).ok(); + let min = stat(Stat::Min)?; + let max = stat(Stat::Max)?; + let kept = kept_fraction(&column_bounds, &min, &max)?; + if kept > MAX_KEPT_FRACTION { + return self.selective_membership(¤t, name, dtype, &min, &max); + } + + let lhs = get_item(name.clone(), root()); + let comparisons = and_collect( + column_bounds + .into_iter() + .map(|(op, value)| binary(op.into(), lhs.clone(), lit(value))), + )?; + // A top-level comparison cannot keep a null column. Bounds extracted from an + // OR or CASE may still admit nulls, so retain the escape for those filters. + let rejects_null = split_conjunction(¤t).into_iter().any(|expr| { + expr.downcast_ref::() + .is_some_and(|binary| { + compare_op(binary.op()).is_some() + && [binary.left(), binary.right()].into_iter().any(|child| { + column_of(child).is_some_and(|column| column.name() == name) + }) + }) + }); + Some(if dtype.is_nullable() && !rejects_null { + or(is_null(lhs), comparisons) + } else { + comparisons + }) + }); + and_collect(per_column) + } + + /// A sparse key set can be selective even when its min/max covers the entire file. Only + /// inspect conjuncts: an IN list inside an OR or CASE need not hold for every matching row. + fn selective_membership( + &self, + current: &Arc, + name: &str, + dtype: &DType, + min: &Scalar, + max: &Scalar, + ) -> Option { + if !dtype.is_int() { + return None; + } + let range = + max.as_primitive_opt()?.as_::()? - min.as_primitive_opt()?.as_::()? + 1.0; + let convertor = DefaultExpressionConvertor::new(self.session.clone()); + split_conjunction(current) + .into_iter() + .filter_map(|expr| expr.downcast_ref::()) + .filter(|list| { + !list.negated() + && !list.list().is_empty() + && list.list().len() <= MAX_MEMBERSHIP_VALUES + && list + .expr() + .downcast_ref::() + .is_some_and(|column| column.name() == name) + && list.list().len() as f64 / range <= MAX_MEMBERSHIP_KEPT_FRACTION + }) + .find_map(|list| convertor.convert(list).ok()) + } + + fn value(&self, column: usize, op: CompareOperator) -> Option { + let generation = self.filter.snapshot_generation(); + if let Some(snapshot) = self.cache.load().as_ref() + && snapshot.generation == generation + { + return find_bound(&snapshot.bounds, column, op); + } + find_bound(&self.refresh(generation), column, op) + } + + /// The bounds of the filter's current generation, re-read only when the generation changes. + fn current_bounds(&self) -> Arc<[Bound]> { + let generation = self.filter.snapshot_generation(); + match self.cache.load().as_ref() { + Some(snapshot) if snapshot.generation == generation => Arc::clone(&snapshot.bounds), + _ => self.refresh(generation), + } + } + + /// Re-reads the bounds for `generation` and caches them, unless a concurrent refresh has + /// already cached a newer generation. + fn refresh(&self, generation: u64) -> Arc<[Bound]> { + let bounds: Arc<[Bound]> = self.extract_bounds().into(); + let snapshot = Arc::new(Snapshot { + generation, + bounds: Arc::clone(&bounds), + }); + self.cache.rcu(|cached| match cached { + Some(cached) if cached.generation > generation => Some(Arc::clone(cached)), + _ => Some(Arc::clone(&snapshot)), + }); + bounds + } + + /// Reads `col literal` bounds from the filter's current predicate. + fn extract_bounds(&self) -> Vec { + self.filter + .downcast_ref::() + .and_then(|filter| filter.current().ok()) + .and_then(|current| self.bounds_of(¤t)) + .unwrap_or_default() + } + + /// Bounds that every row satisfying `expr` also satisfies, unless the bound's column is null, + /// or `None` if no row can satisfy `expr` (it is `false` or `NULL`). + /// + /// Anything not understood yields no bounds, which can only make the Vortex filter keep more + /// rows. + fn bounds_of(&self, expr: &Arc) -> Option> { + if let Some(literal) = expr.downcast_ref::() { + return match literal.value() { + DFScalarValue::Boolean(Some(true)) => Some(vec![]), + DFScalarValue::Boolean(Some(false) | None) | DFScalarValue::Null => None, + _ => Some(vec![]), + }; + } + if let Some(case) = expr.downcast_ref::() { + // Rows the CASE keeps satisfy one of its results. Without an ELSE, unmatched rows are + // NULL and so not kept. Partitioned hash joins route each key to its partition's + // bounds this way. + return case + .when_then_expr() + .iter() + .map(|(_, then)| then) + .chain(case.else_expr()) + .map(|branch| self.bounds_of(branch)) + .reduce(|a, b| self.hull(a, b)) + .flatten(); + } + let Some(binary) = expr.downcast_ref::() else { + return Some(vec![]); + }; + match binary.op() { + DFOperator::And => { + let mut bounds = self.bounds_of(binary.left())?; + bounds.extend(self.bounds_of(binary.right())?); + Some(bounds) + } + DFOperator::Or => { + // `col IS NULL OR `: nulls always pass our template, so bounds `` + // places on the same column still hold. + if let Some(null_col) = binary + .left() + .downcast_ref::() + .and_then(|is_null| column_of(is_null.arg())) + { + let idx = self.column_index(null_col.name()); + return Some( + self.bounds_of(binary.right()) + .unwrap_or_default() + .into_iter() + .filter(|(c, ..)| Some(*c) == idx) + .collect(), + ); + } + // E.g. multi-column TopK, `a < v OR (a = v AND b < w)`, implies `a <= v`. + self.hull( + self.bounds_of(binary.left()), + self.bounds_of(binary.right()), + ) + } + _ => Some(self.comparison(binary)), + } + } + + /// Bounds implied by either side holding: for each column and direction bounded on both + /// sides, the looser of the two. A side no row can satisfy (`None`) contributes nothing. + fn hull(&self, left: Option>, right: Option>) -> Option> { + let (left, right) = match (left, right) { + (Some(left), Some(right)) => (left, right), + (side, None) | (None, side) => return side, + }; + let mut bounds = vec![]; + for (idx, (_, dtype)) in self.columns.iter().enumerate() { + for upper in [true, false] { + let tightest = |side: &[Bound]| { + side.iter() + .filter(|(c, op, _)| *c == idx && is_upper(*op) == Some(upper)) + .filter_map(|(_, op, value)| { + Some(( + *op, + Scalar::try_new(dtype.clone(), Some(value.clone())).ok()?, + )) + }) + // Every bound on one side holds, so any of them is safe to keep. + .reduce(|a, b| { + if looser(&a, &b, upper) == Some(false) { + b + } else { + a + } + }) + }; + if let (Some(l), Some(r)) = (tightest(&left), tightest(&right)) + && let Some(r_is_looser) = looser(&l, &r, upper) + && let Some(value) = if r_is_looser { r.1 } else { l.1 }.into_value() + { + let op = if r_is_looser { r.0 } else { l.0 }; + bounds.push((idx, op, value)); + } + } + } + Some(bounds) + } + + fn column_index(&self, name: &str) -> Option { + self.columns.iter().position(|(column, _)| column == name) + } + + /// Parses `col literal` or `literal col` into bounds on a tracked column, reading + /// `col = literal` as both `<=` and `>=`. + fn comparison(&self, binary: &df_expr::BinaryExpr) -> Vec { + let Some(op) = compare_op(binary.op()) else { + return vec![]; + }; + let Some((idx, op, value)) = self.parse_comparison(binary, op) else { + return vec![]; + }; + match op { + CompareOperator::Eq => vec![ + (idx, CompareOperator::Lte, value.clone()), + (idx, CompareOperator::Gte, value), + ], + op => vec![(idx, op, value)], + } + } + + fn parse_comparison(&self, binary: &df_expr::BinaryExpr, op: CompareOperator) -> Option { + let (col, literal, op) = match ( + column_of(binary.left()), + binary.right().downcast_ref::(), + ) { + (Some(col), Some(literal)) => (col, literal, op), + _ => ( + column_of(binary.right())?, + binary.left().downcast_ref::()?, + op.swap(), + ), + }; + + let idx = self.column_index(col.name())?; + let dtype = &self.columns[idx].1; + + let scalar = scalar_from_df(literal.value(), &self.session).ok()?; + let scalar = if scalar.dtype().eq_ignore_nullability(dtype) { + scalar + } else { + // A lossy cast (e.g. truncating a timestamp to a coarser unit) could make the bound + // stricter than the original, so only accept casts that round-trip exactly. + let cast = scalar.cast(&dtype.as_nullable()).ok()?; + if cast.cast(scalar.dtype()).ok()? != scalar { + return None; + } + cast + }; + Some((idx, op, scalar.into_value()?)) + } +} + +/// Estimates the fraction of the `[min, max]` range kept by `bounds`, assuming values are spread +/// uniformly. Returns `None` for non-numeric columns. +fn kept_fraction(bounds: &[(CompareOperator, Scalar)], min: &Scalar, max: &Scalar) -> Option { + let as_f64 = |s: &Scalar| s.as_primitive_opt()?.as_::(); + let (min, max) = (as_f64(min)?, as_f64(max)?); + let mut lower = min; + let mut upper = max; + for (op, value) in bounds { + let value = as_f64(value)?; + match op { + CompareOperator::Gt | CompareOperator::Gte => lower = lower.max(value), + CompareOperator::Lt | CompareOperator::Lte => upper = upper.min(value), + CompareOperator::Eq | CompareOperator::NotEq => {} + } + } + if max <= min { + return Some(if lower <= upper { 1.0 } else { 0.0 }); + } + Some(((upper - lower) / (max - min)).clamp(0.0, 1.0)) +} + +fn find_bound(bounds: &[Bound], column: usize, op: CompareOperator) -> Option { + bounds + .iter() + .find(|(c, o, _)| *c == column && *o == op) + .map(|(_, _, value)| value.clone()) +} + +/// `Some(true)` for upper bounds (`<`, `<=`), `Some(false)` for lower bounds (`>`, `>=`). +fn is_upper(op: CompareOperator) -> Option { + match op { + CompareOperator::Lt | CompareOperator::Lte => Some(true), + CompareOperator::Gt | CompareOperator::Gte => Some(false), + CompareOperator::Eq | CompareOperator::NotEq => None, + } +} + +/// Whether bound `b` keeps more values than bound `a` (both upper or both lower), or `None` when +/// their values can't be compared. +fn looser( + a: &(CompareOperator, Scalar), + b: &(CompareOperator, Scalar), + upper: bool, +) -> Option { + Some(match a.1.partial_cmp(&b.1)? { + Ordering::Equal => { + matches!(a.0, CompareOperator::Lt | CompareOperator::Gt) + && matches!(b.0, CompareOperator::Lte | CompareOperator::Gte) + } + Ordering::Less => upper, + Ordering::Greater => !upper, + }) +} + +fn compare_op(op: &DFOperator) -> Option { + Some(match op { + DFOperator::Lt => CompareOperator::Lt, + DFOperator::LtEq => CompareOperator::Lte, + DFOperator::Gt => CompareOperator::Gt, + DFOperator::GtEq => CompareOperator::Gte, + DFOperator::Eq => CompareOperator::Eq, + _ => return None, + }) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use arrow_schema::DataType; + use arrow_schema::Field; + use arrow_schema::Schema; + use datafusion_common::ScalarValue; + use datafusion_expr::Operator as DFOperator; + use datafusion_physical_expr::PhysicalExpr; + use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr; + use datafusion_physical_plan::expressions as df_expr; + use rstest::rstest; + use vortex::VortexSessionDefault; + use vortex::array::ArrayRef; + use vortex::array::IntoArray; + use vortex::array::VortexSessionExecute; + use vortex::array::arrays::BoolArray; + use vortex::array::arrays::PrimitiveArray; + use vortex::array::arrays::StructArray; + use vortex::array::assert_arrays_eq; + use vortex::array::stats::StatsSet; + use vortex::buffer::buffer; + use vortex::error::VortexResult; + use vortex::expr::stats::Precision; + use vortex::expr::stats::Stat; + use vortex::file::FileStatistics; + use vortex::scalar::ScalarValue as VxScalarValue; + use vortex::session::VortexSession; + + use super::dynamic_filter_to_vortex; + + type PhysicalExprRef = Arc; + + fn col_a() -> PhysicalExprRef { + Arc::new(df_expr::Column::new("a", 0)) + } + + fn lit_i32(v: i32) -> PhysicalExprRef { + Arc::new(df_expr::Literal::new(ScalarValue::Int32(Some(v)))) + } + + fn binary(l: PhysicalExprRef, op: DFOperator, r: PhysicalExprRef) -> PhysicalExprRef { + Arc::new(df_expr::BinaryExpr::new(l, op, r)) + } + + fn assert_filter( + session: &VortexSession, + input: &ArrayRef, + filter: &PhysicalExprRef, + expected: BoolArray, + ) -> VortexResult<()> { + let fields = input.dtype().as_struct_fields_opt().expect("struct input"); + let converted = + dynamic_filter_to_vortex(filter, fields, None, session).expect("filter should convert"); + let actual = input.clone().apply(&converted)?; + assert_arrays_eq!(actual, expected, &mut session.create_execution_ctx()); + Ok(()) + } + + /// The Vortex filter tracks updates made to the DataFusion filter after conversion. + #[test] + fn tracks_updates() -> anyhow::Result<()> { + let session = VortexSession::default(); + let input = + StructArray::from_fields(&[("a", buffer![1i32, 5, 10].into_array())])?.into_array(); + let dynamic = Arc::new(DynamicFilterPhysicalExpr::new( + vec![col_a()], + Arc::new(df_expr::Literal::new(ScalarValue::Boolean(Some(true)))), + )); + let filter: PhysicalExprRef = Arc::clone(&dynamic) as _; + let fields = input.dtype().as_struct_fields_opt().expect("struct input"); + let converted = dynamic_filter_to_vortex(&filter, fields, None, &session) + .expect("filter should convert"); + let mut ctx = session.create_execution_ctx(); + + assert_arrays_eq!( + input.clone().apply(&converted)?, + BoolArray::from_iter([true, true, true]), + &mut ctx + ); + + dynamic.update(binary(col_a(), DFOperator::Lt, lit_i32(5)))?; + assert_arrays_eq!( + input.clone().apply(&converted)?, + BoolArray::from_iter([true, false, false]), + &mut ctx + ); + + dynamic.update(binary(col_a(), DFOperator::Lt, lit_i32(2)))?; + assert_arrays_eq!( + input.apply(&converted)?, + BoolArray::from_iter([true, false, false]), + &mut ctx + ); + Ok(()) + } + + /// A filter that already has bounds when the file opens only gets the comparisons it uses, + /// which keep tracking later updates, and stay safe if the filter's shape changes. + #[test] + fn narrowed_template_after_first_update() -> anyhow::Result<()> { + let session = VortexSession::default(); + let input = StructArray::from_fields(&[ + ("a", buffer![1i32, 5, 10].into_array()), + ("b", buffer![7i32, 0, 0].into_array()), + ])? + .into_array(); + let dynamic = Arc::new(DynamicFilterPhysicalExpr::new( + vec![col_a(), col_b()], + binary(col_a(), DFOperator::Lt, lit_i32(10)), + )); + let filter: PhysicalExprRef = Arc::clone(&dynamic) as _; + let fields = input.dtype().as_struct_fields_opt().expect("struct input"); + let converted = dynamic_filter_to_vortex(&filter, fields, None, &session) + .expect("filter should convert"); + assert_eq!(converted.to_string().matches("dynamic(").count(), 1); + let mut ctx = session.create_execution_ctx(); + + dynamic.update(binary(col_a(), DFOperator::Lt, lit_i32(5)))?; + assert_arrays_eq!( + input.clone().apply(&converted)?, + BoolArray::from_iter([true, false, false]), + &mut ctx + ); + + // `b` was not part of the filter when the file opened, so it is not tracked, and `a` no + // longer has a bound: every row passes. + dynamic.update(binary(col_b(), DFOperator::Lt, lit_i32(3)))?; + assert_arrays_eq!( + input.apply(&converted)?, + BoolArray::from_iter([true, true, true]), + &mut ctx + ); + Ok(()) + } + + #[rstest] + // Literal on the left flips the operator. + #[case(binary(lit_i32(10), DFOperator::Gt, col_a()), [true, true, false])] + // Hash join min/max bounds. + #[case( + binary( + binary(col_a(), DFOperator::GtEq, lit_i32(5)), + DFOperator::And, + binary(col_a(), DFOperator::LtEq, lit_i32(10)), + ), + [false, true, true], + )] + // Disjunctions across bounds (aggregate min/max) are not tracked, so every row passes. + #[case( + binary( + binary(col_a(), DFOperator::Lt, lit_i32(2)), + DFOperator::Or, + binary(col_a(), DFOperator::Gt, lit_i32(8)), + ), + [true, true, true], + )] + // Unsupported conjuncts are dropped while supported ones still apply. + #[case( + binary( + binary(col_a(), DFOperator::Gt, lit_i32(1)), + DFOperator::And, + binary(col_a(), DFOperator::NotEq, lit_i32(5)), + ), + [false, true, true], + )] + fn converts_filter_shapes( + #[case] current: PhysicalExprRef, + #[case] expected: [bool; 3], + ) -> VortexResult<()> { + let session = VortexSession::default(); + let input = + StructArray::from_fields(&[("a", buffer![1i32, 5, 10].into_array())])?.into_array(); + let filter: PhysicalExprRef = + Arc::new(DynamicFilterPhysicalExpr::new(vec![col_a()], current)); + assert_filter(&session, &input, &filter, BoolArray::from_iter(expected)) + } + + fn col_b() -> PhysicalExprRef { + Arc::new(df_expr::Column::new("b", 1)) + } + + /// Multi-column TopK filters are disjunctions over the sort keys; the leading key still gets + /// a bound from the loosest side of each disjunction. + #[rstest] + // ORDER BY a, b: `a < 5 OR (a = 5 AND b < 3)` implies `a <= 5`. + #[case( + binary( + binary(col_a(), DFOperator::Lt, lit_i32(5)), + DFOperator::Or, + binary( + binary(col_a(), DFOperator::Eq, lit_i32(5)), + DFOperator::And, + binary(col_b(), DFOperator::Lt, lit_i32(3)), + ), + ), + [true, true, false], + )] + // ORDER BY a DESC, b DESC: `a > 5 OR (a = 5 AND b > 3)` implies `a >= 5`. + #[case( + binary( + binary(col_a(), DFOperator::Gt, lit_i32(5)), + DFOperator::Or, + binary( + binary(col_a(), DFOperator::Eq, lit_i32(5)), + DFOperator::And, + binary(col_b(), DFOperator::Gt, lit_i32(3)), + ), + ), + [false, true, true], + )] + // The hull keeps the looser bound: `a < 2 OR a <= 5` implies `a <= 5`. + #[case( + binary( + binary(col_a(), DFOperator::Lt, lit_i32(2)), + DFOperator::Or, + binary(col_a(), DFOperator::LtEq, lit_i32(5)), + ), + [true, true, false], + )] + fn disjunction_bounds( + #[case] current: PhysicalExprRef, + #[case] expected: [bool; 3], + ) -> VortexResult<()> { + let session = VortexSession::default(); + let input = StructArray::from_fields(&[ + ("a", buffer![1i32, 5, 10].into_array()), + ("b", buffer![0i32, 0, 0].into_array()), + ])? + .into_array(); + let filter: PhysicalExprRef = Arc::new(DynamicFilterPhysicalExpr::new( + vec![col_a(), col_b()], + current, + )); + assert_filter(&session, &input, &filter, BoolArray::from_iter(expected)) + } + + fn range(lower: i32, upper: i32) -> PhysicalExprRef { + binary( + binary(col_a(), DFOperator::GtEq, lit_i32(lower)), + DFOperator::And, + binary(col_a(), DFOperator::LtEq, lit_i32(upper)), + ) + } + + fn lit_bool(value: bool) -> PhysicalExprRef { + Arc::new(df_expr::Literal::new(ScalarValue::Boolean(Some(value)))) + } + + /// Partitioned hash joins route each key to its partition's bounds with a CASE; every kept + /// row lies within the hull of the partitions' bounds. + #[rstest] + // Partitions [2, 4] and [5, 6], plus an empty partition (`false`): `2 <= a <= 6`. + #[case(vec![range(2, 4), lit_bool(false)], Some(range(5, 6)), [false, true, false])] + // Without an ELSE, unmatched rows are NULL and not kept. + #[case(vec![range(2, 4), range(5, 6)], None, [false, true, false])] + // A partition with unknown contents (`true`) keeps every row. + #[case(vec![range(2, 4), lit_bool(true)], None, [true, true, true])] + fn partitioned_join_case_bounds( + #[case] thens: Vec, + #[case] else_expr: Option, + #[case] expected: [bool; 3], + ) -> anyhow::Result<()> { + let session = VortexSession::default(); + let input = + StructArray::from_fields(&[("a", buffer![1i32, 5, 10].into_array())])?.into_array(); + let when_then = thens + .into_iter() + .enumerate() + .map(|(partition, then)| { + let partition = u64::try_from(partition).expect("small partition index"); + let when: PhysicalExprRef = + Arc::new(df_expr::Literal::new(ScalarValue::UInt64(Some(partition)))); + (when, then) + }) + .collect(); + let current: PhysicalExprRef = Arc::new(df_expr::CaseExpr::try_new( + Some(col_a()), + when_then, + else_expr, + )?); + let filter: PhysicalExprRef = + Arc::new(DynamicFilterPhysicalExpr::new(vec![col_a()], current)); + assert_filter(&session, &input, &filter, BoolArray::from_iter(expected))?; + Ok(()) + } + + /// Nulls always pass, matching TopK's `a IS NULL OR a > v` for NULLS FIRST and staying safe + /// for NULLS LAST. + #[rstest] + #[case(binary(col_a(), DFOperator::Gt, lit_i32(4)))] + #[case(binary( + Arc::new(df_expr::IsNullExpr::new(col_a())), + DFOperator::Or, + binary(col_a(), DFOperator::Gt, lit_i32(4)), + ))] + fn nulls_pass(#[case] current: PhysicalExprRef) -> VortexResult<()> { + let session = VortexSession::default(); + let a = PrimitiveArray::from_option_iter([Some(1i32), None, Some(10)]).into_array(); + let input = StructArray::from_fields(&[("a", a)])?.into_array(); + let filter: PhysicalExprRef = + Arc::new(DynamicFilterPhysicalExpr::new(vec![col_a()], current)); + // The comparison is nullable, but the `IS NULL` branch makes every row non-null. + let expected = BoolArray::from_iter([Some(false), Some(true), Some(true)]); + assert_filter(&session, &input, &filter, expected) + } + + /// Bounds on a cast column are applied only when the literal converts exactly to the file type. + #[rstest] + #[case(DataType::Int64, ScalarValue::Int64(Some(5)), [true, false, false])] + // 4.5 has no exact INT equivalent; truncating to `a < 4` would wrongly drop `a = 4`. + #[case(DataType::Float64, ScalarValue::Float64(Some(4.5)), [true, true, true])] + fn cast_column_bounds( + #[case] cast_to: DataType, + #[case] value: ScalarValue, + #[case] expected: [bool; 3], + ) -> VortexResult<()> { + let session = VortexSession::default(); + let input = + StructArray::from_fields(&[("a", buffer![1i32, 5, 10].into_array())])?.into_array(); + let cast_a: PhysicalExprRef = Arc::new(df_expr::CastExpr::new(col_a(), cast_to, None)); + let current = binary( + Arc::clone(&cast_a), + DFOperator::Lt, + Arc::new(df_expr::Literal::new(value)), + ); + let filter: PhysicalExprRef = + Arc::new(DynamicFilterPhysicalExpr::new(vec![cast_a], current)); + assert_filter(&session, &input, &filter, BoolArray::from_iter(expected)) + } + + fn complete_filter(current: PhysicalExprRef) -> PhysicalExprRef { + let filter = DynamicFilterPhysicalExpr::new(vec![col_a()], current); + filter.mark_complete(); + Arc::new(filter) + } + + /// Complete filters are applied as static bounds only when they skip enough of the file. + #[rstest] + // Keeps [4, 5] of [1, 10]: applied. + #[case(4, 5, Some([false, true, false]))] + // Keeps [2, 10] of [1, 10]: not worth evaluating. + #[case(2, 10, None)] + fn complete_filter_applied_when_selective( + #[case] lower: i32, + #[case] upper: i32, + #[case] expected: Option<[bool; 3]>, + ) -> VortexResult<()> { + let session = VortexSession::default(); + let input = + StructArray::from_fields(&[("a", buffer![1i32, 5, 10].into_array())])?.into_array(); + let stats = StatsSet::from_iter([ + (Stat::Min, Precision::exact(VxScalarValue::from(1i32))), + (Stat::Max, Precision::exact(VxScalarValue::from(10i32))), + ]); + let file_stats = FileStatistics::new_with_dtype(Arc::from([stats]), input.dtype()); + let filter = complete_filter(binary( + binary(col_a(), DFOperator::GtEq, lit_i32(lower)), + DFOperator::And, + binary(col_a(), DFOperator::LtEq, lit_i32(upper)), + )); + let fields = input.dtype().as_struct_fields_opt().expect("struct input"); + + let converted = dynamic_filter_to_vortex(&filter, fields, Some(&file_stats), &session); + match (converted, expected) { + (Some(converted), Some(expected)) => assert_arrays_eq!( + input.apply(&converted)?, + BoolArray::from_iter(expected), + &mut session.create_execution_ctx() + ), + (None, None) => {} + (converted, expected) => panic!("expected {expected:?}, got {converted:?}"), + } + Ok(()) + } + + #[test] + fn complete_filter_without_stats_is_not_converted() -> VortexResult<()> { + let session = VortexSession::default(); + let input = StructArray::from_fields(&[("a", buffer![1i32].into_array())])?.into_array(); + let filter = complete_filter(binary(col_a(), DFOperator::Lt, lit_i32(5))); + let fields = input.dtype().as_struct_fields_opt().expect("struct input"); + assert!(dynamic_filter_to_vortex(&filter, fields, None, &session).is_none()); + Ok(()) + } + + #[rstest] + #[case::comparison(false, None)] + #[case::null_escape(true, Some(true))] + fn completed_range_nulls( + #[case] allow_null: bool, + #[case] expected_null: Option, + ) -> VortexResult<()> { + let session = VortexSession::default(); + let input = StructArray::from_fields(&[( + "a", + PrimitiveArray::from_option_iter([Some(1i32), Some(5), Some(10), None]).into_array(), + )])? + .into_array(); + let range = binary(col_a(), DFOperator::LtEq, lit_i32(5)); + let current = if allow_null { + binary( + Arc::new(df_expr::IsNullExpr::new(col_a())), + DFOperator::Or, + range, + ) + } else { + range + }; + let filter = complete_filter(current); + let stats = StatsSet::from_iter([ + (Stat::Min, Precision::exact(VxScalarValue::from(1i32))), + (Stat::Max, Precision::exact(VxScalarValue::from(10i32))), + ]); + let file_stats = FileStatistics::new_with_dtype(Arc::from([stats]), input.dtype()); + let fields = input.dtype().as_struct_fields_opt().expect("struct input"); + let converted = dynamic_filter_to_vortex(&filter, fields, Some(&file_stats), &session) + .expect("selective range should be applied"); + assert_arrays_eq!( + input.apply(&converted)?, + BoolArray::from_iter([Some(true), Some(true), Some(false), expected_null]), + &mut session.create_execution_ctx() + ); + Ok(()) + } + + #[rstest] + #[case::sparse(vec![1, 100], false, false, 100, true)] + #[case::moderately_selective(vec![1, 10], false, false, 10, false)] + #[case::dense((1..=10).collect(), false, false, 10, false)] + #[case::large_sparse((1..=17).collect(), false, false, 1000, false)] + #[case::negated(vec![1, 100], true, false, 100, false)] + #[case::disjunction(vec![1, 100], false, true, 100, false)] + fn completed_membership_filter( + #[case] values: Vec, + #[case] negated: bool, + #[case] disjunction: bool, + #[case] max: i32, + #[case] applied: bool, + ) -> anyhow::Result<()> { + let session = VortexSession::default(); + let input = StructArray::from_fields(&[( + "a", + PrimitiveArray::from_option_iter([Some(1i32), Some(5), Some(max), None]).into_array(), + )])? + .into_array(); + let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); + let membership: PhysicalExprRef = Arc::new(df_expr::InListExpr::try_new( + col_a(), + values.into_iter().map(lit_i32).collect(), + negated, + &schema, + )?); + let current = if disjunction { + binary( + membership, + DFOperator::Or, + binary(col_a(), DFOperator::Eq, lit_i32(5)), + ) + } else { + membership + }; + let filter = complete_filter(current); + let stats = StatsSet::from_iter([ + (Stat::Min, Precision::exact(VxScalarValue::from(1i32))), + (Stat::Max, Precision::exact(VxScalarValue::from(max))), + ]); + let file_stats = FileStatistics::new_with_dtype(Arc::from([stats]), input.dtype()); + let fields = input.dtype().as_struct_fields_opt().expect("struct input"); + let converted = dynamic_filter_to_vortex(&filter, fields, Some(&file_stats), &session); + if applied { + assert_arrays_eq!( + input.apply(&converted.expect("sparse membership should be applied"))?, + BoolArray::from_iter([Some(true), Some(false), Some(true), None]), + &mut session.create_execution_ctx() + ); + } else { + assert!(converted.is_none()); + } + Ok(()) + } + + #[test] + fn missing_column_is_not_converted() -> VortexResult<()> { + let session = VortexSession::default(); + let input = StructArray::from_fields(&[("b", buffer![1i32].into_array())])?.into_array(); + let filter: PhysicalExprRef = Arc::new(DynamicFilterPhysicalExpr::new( + vec![col_a()], + binary(col_a(), DFOperator::Lt, lit_i32(5)), + )); + let fields = input.dtype().as_struct_fields_opt().expect("struct input"); + assert!(dynamic_filter_to_vortex(&filter, fields, None, &session).is_none()); + Ok(()) + } +} diff --git a/vortex-datafusion/src/convert/mod.rs b/vortex-datafusion/src/convert/mod.rs index 476f32afa49..3237204044f 100644 --- a/vortex-datafusion/src/convert/mod.rs +++ b/vortex-datafusion/src/convert/mod.rs @@ -11,6 +11,7 @@ use vortex::error::VortexResult; +pub(crate) mod dynamic; pub(crate) mod exprs; mod scalars; pub(crate) mod schema; diff --git a/vortex-datafusion/src/persistent/opener.rs b/vortex-datafusion/src/persistent/opener.rs index 9858aedc809..f482578bba1 100644 --- a/vortex-datafusion/src/persistent/opener.rs +++ b/vortex-datafusion/src/persistent/opener.rs @@ -22,6 +22,8 @@ use datafusion_datasource::file_stream::FileOpener; use datafusion_execution::cache::cache_manager::CachedFileMetadataEntry; use datafusion_execution::cache::cache_manager::FileMetadataCache; use datafusion_physical_expr::PhysicalExprRef; +use datafusion_physical_expr::conjunction; +use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr; use datafusion_physical_expr::expressions::LambdaExpr; use datafusion_physical_expr::projection::ProjectionExprs; use datafusion_physical_expr::simplifier::PhysicalExprSimplifier; @@ -42,6 +44,8 @@ use tracing::Instrument; use vortex::array::VortexSessionExecute; use vortex::error::VortexError; use vortex::error::VortexExpect; +use vortex::expr::and_collect; +use vortex::expr::bound; use vortex::file::OpenOptionsSessionExt; use vortex::io::InstrumentedReadAt; use vortex::layout::LayoutReader; @@ -54,6 +58,7 @@ use vortex_utils::aliases::dash_map::DashMap; use vortex_utils::aliases::dash_map::Entry; use crate::VortexAccessPlan; +use crate::convert::dynamic::dynamic_filter_to_vortex; use crate::convert::exprs::ExpressionConvertor; use crate::convert::exprs::ProcessedProjection; use crate::convert::exprs::make_vortex_predicate; @@ -346,6 +351,21 @@ impl FileOpener for VortexOpener { scan_builder = vortex_plan.apply_to_builder(scan_builder); } + // Dynamic filters (TopK, hash join, ...) become Vortex comparisons that track the + // DataFusion filter. They are only an optimization, so any we can't convert (e.g. + // after the expression adapter rewrote their columns) are dropped. + let (dynamic_filters, filter): (Vec, Vec) = filter + .iter() + .flat_map(split_conjunction) + .cloned() + .partition(|expr| expr.downcast_ref::().is_some()); + let filter = (!filter.is_empty()).then(|| conjunction(filter)); + let dynamic_filter = vxf.dtype().as_struct_fields_opt().and_then(|fields| { + and_collect(dynamic_filters.iter().filter_map(|expr| { + dynamic_filter_to_vortex(expr, fields, vxf.file_stats(), &session) + })) + }); + let filter = filter .and_then(|f| { // Verify that all filters we've accepted from DataFusion get pushed down. @@ -382,11 +402,25 @@ impl FileOpener for VortexOpener { .transpose() .map_err(|e| exec_datafusion_err!("Couldn't bind Vortex scan filter: {e}"))?; - if let Some(limit) = limit + // The scan limit counts rows before filtering, so it is only valid without a filter. + // Prefer it over a dynamic filter, which is purely an optimization. + let filter = if let Some(limit) = limit && filter.is_none() { scan_builder = scan_builder.with_limit(limit); - } + None + } else { + let dynamic_filter = dynamic_filter + .map(|f| f.bind(vxf.dtype())?.optimize_recursive()) + .transpose() + .map_err(|e| { + exec_datafusion_err!("Couldn't bind Vortex dynamic filter: {e}") + })?; + match (filter, dynamic_filter) { + (Some(filter), Some(dynamic)) => Some(bound::and(filter, dynamic)), + (filter, dynamic) => filter.or(dynamic), + } + }; if let Some(concurrency) = scan_concurrency { scan_builder = scan_builder.with_concurrency(concurrency); diff --git a/vortex-datafusion/src/persistent/source.rs b/vortex-datafusion/src/persistent/source.rs index fbd37f04ded..317bba93e44 100644 --- a/vortex-datafusion/src/persistent/source.rs +++ b/vortex-datafusion/src/persistent/source.rs @@ -39,6 +39,7 @@ use vortex_utils::aliases::dash_map::DashMap; use super::opener::NaturalSplits; use super::opener::VortexOpener; use crate::VortexTableOptions; +use crate::convert::dynamic::as_column_dynamic_filter; use crate::convert::exprs::DefaultExpressionConvertor; use crate::convert::exprs::ExpressionConvertor; use crate::persistent::reader::DefaultVortexReaderFactory; @@ -325,6 +326,21 @@ impl VortexSource { self.vortex_predicate.as_ref() } + /// Whether `expr` is a dynamic filter (TopK, hash join, ...) that the opener can track live. + /// + /// The scan applies a relaxed form of the filter that may keep extra rows. That is safe to + /// report as pushed down because the producing operator enforces its own semantics and + /// ignores the pushdown result for its dynamic filters. + fn is_supported_dynamic_filter(&self, expr: &Arc) -> bool { + let file_schema = self.table_schema.file_schema(); + as_column_dynamic_filter(expr).is_some_and(|dynamic| { + dynamic.children().into_iter().all(|child| { + self.expression_convertor + .can_be_pushed_down(child, file_schema) + }) + }) + } + fn create_vortex_opener( &self, object_store: Arc, @@ -472,9 +488,11 @@ impl FileSource for VortexSource { let supported_filters = filters .into_iter() .map(|expr| { + let file_schema = self.table_schema.file_schema(); if self .expression_convertor - .can_be_pushed_down(&expr, self.table_schema.file_schema()) + .can_be_pushed_down(&expr, file_schema) + || self.is_supported_dynamic_filter(&expr) { PushedDownPredicate::supported(expr) } else { diff --git a/vortex-datafusion/src/persistent/tests.rs b/vortex-datafusion/src/persistent/tests.rs index a1b63e77b8d..aae07d810d2 100644 --- a/vortex-datafusion/src/persistent/tests.rs +++ b/vortex-datafusion/src/persistent/tests.rs @@ -4,9 +4,12 @@ use std::sync::Arc; use anyhow::anyhow; +use arrow_schema::Field; +use arrow_schema::Schema; use datafusion::arrow::array::Int32Array; use datafusion::arrow::array::RecordBatch; use datafusion::arrow::datatypes::DataType; +use datafusion::arrow::util::display::array_value_to_string; use datafusion::arrow::util::pretty::pretty_format_batches; use datafusion::datasource::provider::DefaultTableFactory; use datafusion::execution::SessionStateBuilder; @@ -678,3 +681,116 @@ async fn arrow_uuid_extension_roundtrip_nested_struct() -> anyhow::Result<()> { Ok(()) } + +/// Writes `files` Vortex files of a nullable `a` column under `/dyn/` and registers them as `t`. +/// +/// File `f` holds `f * 100 + i` for `i` in `0..100`, with nulls wherever `i % 17 == 0`. +async fn register_dynamic_filter_table(ctx: &TestSessionContext, files: i32) -> anyhow::Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); + for file in 0..files { + let values = (0..100).map(|i| (i % 17 != 0).then_some(file * 100 + i)); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int32Array::from_iter(values))], + )?; + ctx.write_arrow_batch(format!("dyn/{file}.vortex"), &batch) + .await?; + } + ctx.session + .sql("CREATE EXTERNAL TABLE t (a INT) STORED AS vortex LOCATION '/dyn/'") + .await?; + Ok(()) +} + +/// Runs `sql` and renders its single output column as comma-separated values. +async fn query_values(ctx: &TestSessionContext, sql: &str) -> anyhow::Result { + let batches = ctx.session.sql(sql).await?.collect().await?; + let mut values = vec![]; + for batch in &batches { + let column = batch.column(0); + for row in 0..column.len() { + values.push(if column.is_null(row) { + "NULL".to_owned() + } else { + array_value_to_string(column, row)? + }); + } + } + Ok(values.join(", ")) +} + +/// The scan accepts TopK dynamic filters and still returns the right rows, including nulls. +#[rstest] +#[case("ORDER BY a DESC NULLS LAST LIMIT 3", "499, 498, 497")] +#[case("ORDER BY a ASC NULLS LAST LIMIT 3", "1, 2, 3")] +#[case("ORDER BY a DESC NULLS FIRST LIMIT 3", "NULL, NULL, NULL")] +#[case("ORDER BY a ASC NULLS FIRST OFFSET 29 LIMIT 3", "NULL, 1, 2")] +#[tokio::test] +async fn topk_dynamic_filter_pushdown( + #[case] order: &str, + #[case] expected: &str, +) -> anyhow::Result<()> { + let ctx = TestSessionContext::default(); + register_dynamic_filter_table(&ctx, 5).await?; + + let query = format!("SELECT a FROM t {order}"); + let plan = ctx + .session + .sql(&query) + .await? + .create_physical_plan() + .await?; + let plan_str = DisplayableExecutionPlan::new(plan.as_ref()) + .indent(true) + .to_string(); + assert!( + plan_str.contains("predicate: DynamicFilter"), + "expected the scan to accept the dynamic filter:\n{plan_str}" + ); + + assert_eq!(query_values(&ctx, &query).await?, expected); + Ok(()) +} + +/// Hash join dynamic filters on the probe side keep the join result correct, whether or not +/// their bounds are selective enough to be applied. +#[rstest] +// 151 and 17 are null in the table, and the bounds [17, 352] span most files. +#[case("(150), (151), (352), (17)", "150, 352")] +// Bounds [150, 152] skip all but one file's worth of values, so they are applied. +#[case("(150), (152)", "150, 152")] +#[tokio::test] +async fn hash_join_dynamic_filter_pushdown( + #[case] build_values: &str, + #[case] expected: &str, +) -> anyhow::Result<()> { + let ctx = TestSessionContext::default(); + register_dynamic_filter_table(&ctx, 5).await?; + + let query = + format!("SELECT t.a FROM t JOIN (VALUES {build_values}) AS b(x) ON t.a = b.x ORDER BY t.a"); + assert_eq!(query_values(&ctx, &query).await?, expected); + Ok(()) +} + +/// When the file's type differs from the table's, the expression adapter wraps the dynamic +/// filter's columns in casts. The filter must still be handled rather than rejected by the scan. +#[rstest] +#[case("ORDER BY a DESC LIMIT 3", "499, 498, 497")] +#[case("ORDER BY a ASC LIMIT 3", "1, 2, 3")] +#[tokio::test] +async fn topk_dynamic_filter_with_cast_column( + #[case] order: &str, + #[case] expected: &str, +) -> anyhow::Result<()> { + let ctx = TestSessionContext::default(); + // Files store `a` as INT while this table declares BIGINT. + register_dynamic_filter_table(&ctx, 5).await?; + ctx.session + .sql("CREATE EXTERNAL TABLE t_wide (a BIGINT) STORED AS vortex LOCATION '/dyn/'") + .await?; + + let query = format!("SELECT a FROM t_wide WHERE a IS NOT NULL {order}"); + assert_eq!(query_values(&ctx, &query).await?, expected); + Ok(()) +}