From 303cdaef5f0f0aee734fe0f5fb5b3913d011585e Mon Sep 17 00:00:00 2001 From: Mikhail Kot Date: Tue, 15 Sep 2026 14:27:43 +0100 Subject: [PATCH 1/2] duckdb: return projection_ids handling Signed-off-by: Mikhail Kot --- vortex-duckdb/cpp/include/table_function.h | 21 ++ vortex-duckdb/cpp/multi_file_reader.cpp | 16 ++ vortex-duckdb/src/duckdb/table_init_input.rs | 12 + vortex-duckdb/src/projection.rs | 232 +++++++++++-------- vortex-duckdb/src/table_function.rs | 20 +- 5 files changed, 187 insertions(+), 114 deletions(-) diff --git a/vortex-duckdb/cpp/include/table_function.h b/vortex-duckdb/cpp/include/table_function.h index 82090c26387..5924a209fbc 100644 --- a/vortex-duckdb/cpp/include/table_function.h +++ b/vortex-duckdb/cpp/include/table_function.h @@ -25,8 +25,29 @@ void duckdb_vx_string_map_insert(duckdb_vx_string_map map, const char *key, cons // Input data passed into the init_global and init_local callbacks. typedef struct { const void *bind_data; + + /** + * Projected columns that are requested to be read. These are not + * all columns, only the ones DuckDB optimizer thinks we should read. + */ idx_t *column_ids; size_t column_ids_count; + + /** + * Post filter projected columns. Our table function implements filter + * pushdown so this list is a subset of columns referenced in column_ids + * after filter pushdown and filter pruning. May be empty, in which case + * column_ids should be used. + * Indices in this list reference values from column_ids. I.e. if + * column_ids=[1,5,6], projection_ids=[1], output column should be + * column_ids[1] = 5 + * + * Example usage: + * https://github.com/duckdb/duckdb/blob/dc11eadd8f0a7c600f0034810706605ebe10d5b9/src/include/duckdb/function/table_function.hpp#L147 + */ + const idx_t *projection_ids; + size_t projection_ids_count; + duckdb_vx_table_filter_set filters; duckdb_client_context client_context; } duckdb_vx_tfunc_init_input; diff --git a/vortex-duckdb/cpp/multi_file_reader.cpp b/vortex-duckdb/cpp/multi_file_reader.cpp index 02d4da3caaa..7de57ff907d 100644 --- a/vortex-duckdb/cpp/multi_file_reader.cpp +++ b/vortex-duckdb/cpp/multi_file_reader.cpp @@ -7,6 +7,8 @@ #include "vortex_duckdb.h" #include "vortex.h" +#include "duckdb/execution/operator/scan/physical_table_scan.hpp" + unique_ptr VortexBindData::Copy() const { auto result = make_uniq(); if (ffi_bind_data) { @@ -150,11 +152,25 @@ VortexReaderInterface::InitializeGlobalState(ClientContext &context, column_ids[i] = storage_index; } + // MultiFileGlobalState projection_ids are filled only when this call + // returns. Take these from a physical operator. + const idx_t *projection_ids = nullptr; + size_t projection_ids_count = 0; + if (input.op && input.op->type == PhysicalOperatorType::TABLE_SCAN) { + const PhysicalTableScan &scan = input.op->Cast(); + if (!scan.projection_ids.empty() && scan.projection_ids.size() != column_ids.size()) { + projection_ids = scan.projection_ids.data(); + projection_ids_count = scan.projection_ids.size(); + } + } + void *const ffi_bind = bind.ffi_bind_data->DataPtr(); duckdb_vx_tfunc_init_input ffi_input = { .bind_data = ffi_bind, .column_ids = column_ids.data(), .column_ids_count = column_ids.size(), + .projection_ids = projection_ids, + .projection_ids_count = projection_ids_count, .filters = reinterpret_cast(input.filters.get()), .client_context = reinterpret_cast(&context), }; diff --git a/vortex-duckdb/src/duckdb/table_init_input.rs b/vortex-duckdb/src/duckdb/table_init_input.rs index 1d86cb0fded..f5098c3a347 100644 --- a/vortex-duckdb/src/duckdb/table_init_input.rs +++ b/vortex-duckdb/src/duckdb/table_init_input.rs @@ -17,6 +17,7 @@ impl Debug for TableInitInput<'_> { fn fmt(&self, f: &mut Formatter<'_>) -> Result { f.debug_struct("TableInitInput") .field("column_ids", &self.column_ids()) + .field("projection_ids", &self.projection_ids()) .field("table_filter_set", &self.table_filter_set()) .finish() } @@ -31,6 +32,17 @@ impl<'a> TableInitInput<'a> { unsafe { std::slice::from_raw_parts(self.input.column_ids, self.input.column_ids_count) } } + pub fn projection_ids(&self) -> &[u64] { + if self.input.projection_ids_count == 0 { + // from_raw_parts requires a non-null pointer. C++'s empty vector + // may have a null pointer. + return &[]; + } + unsafe { + std::slice::from_raw_parts(self.input.projection_ids, self.input.projection_ids_count) + } + } + /// Returns the table filter set for the table function. pub fn table_filter_set(&self) -> Option<&TableFilterSetRef> { let ptr = self.input.filters; diff --git a/vortex-duckdb/src/projection.rs b/vortex-duckdb/src/projection.rs index ef91aaae52c..492f12eb3b3 100644 --- a/vortex-duckdb/src/projection.rs +++ b/vortex-duckdb/src/projection.rs @@ -4,6 +4,8 @@ use std::ops::Range; use num_traits::AsPrimitive as _; use vortex::dtype::DType; +use vortex::dtype::Nullability; +use vortex::dtype::PType; use vortex::error::VortexExpect; use vortex::error::VortexResult; use vortex::error::vortex_err; @@ -12,11 +14,12 @@ use vortex::expr::Expression; use vortex::expr::and_collect; use vortex::expr::col; use vortex::expr::get_item; -use vortex::expr::merge; +use vortex::expr::lit; use vortex::expr::pack; use vortex::expr::root; use vortex::expr::select; use vortex::layout::layouts::row_idx::row_idx; +use vortex::scalar::Scalar; use vortex::scan::selection::Selection; use vortex_utils::aliases::hash_set::HashSet; @@ -47,25 +50,55 @@ pub struct DuckdbField { pub projection_expr: Option, } -pub struct Projection(pub Expression); +pub struct Projection { + pub projection: Expression, + pub file_row_number_column_pos: Option, +} impl Projection { - pub fn new(column_ids: &[u64], column_fields: &[DuckdbField]) -> Self { - let mut has_file_row_number = false; + pub fn new(projection_ids: &[u64], column_ids: &[u64], column_fields: &[DuckdbField]) -> Self { + let projection_ids: HashSet = projection_ids.iter().copied().collect(); + // If projection ids are empty, use column_ids. + // See duckdb/src/planner/operator/logical_get.cpp#L168 + let is_projected = + |pos: usize| projection_ids.is_empty() || projection_ids.contains(&(pos as u64)); + + let mut exprs = Vec::with_capacity(column_ids.len() + 1); + let mut file_row_number_column_pos = None; let mut is_star = true; let mut real_column_count = 0; - let mut projected_col_count = 0; // DuckDB uses u64 as column indices but Rust uses usize - for &column_id in column_ids { + for (column_pos, &column_id) in column_ids.iter().enumerate() { if column_id == FILE_ROW_NUMBER_COLUMN_IDX { - has_file_row_number = true; + is_star = false; + if is_projected(column_pos) { + file_row_number_column_pos = Some(exprs.len()); + } else { + // filter-only column needs to be emitted only for output + // vector position match, it will never be read + let dtype = DType::Primitive(PType::U64, Nullability::Nullable); + exprs.push(("file_row_number", lit(Scalar::null(dtype)))); + } continue; } + if is_virtual_column(column_id) { continue; } + let field_idx: usize = column_id.as_(); + let column_field = &column_fields[field_idx]; + let name = column_field.name.as_str(); + + if !is_projected(column_pos) { + is_star = false; + // filter-only column needs to be emitted only for output + // vector position match, it will never be read + exprs.push((name, lit(Scalar::null(column_field.dtype.as_nullable())))); + continue; + } + // In SELECT * DuckDB requests all columns from 0 to column_fields in // increasing order. After removing virtual columns, compare column_id // with (0..column_fields.len()) range. @@ -73,11 +106,14 @@ impl Projection { // Example: if we SELECT len(str), we can't use root() as we try to // pushdown scalar functions. - let column_id: usize = column_id.as_(); - let is_projected_col = column_fields[column_id].projection_expr.is_some(); - projected_col_count += is_projected_col as usize; - is_star &= !is_projected_col; - + let expr = match &column_field.projection_expr { + None => get_item(name, root()), + Some(func) => { + is_star = false; + func.clone() + } + }; + exprs.push((name, expr)); real_column_count += 1; } // Duckdb can request less columns than there are in table i.e. [0, 1] with @@ -85,66 +121,19 @@ impl Projection { is_star &= real_column_count == column_fields.len() as u64; if is_star { - let projection = if has_file_row_number { - // row_idx will be moved to correct position in scan(), prepend here - let row_idx_struct = pack([("file_row_number", row_idx())], false.into()); - merge([row_idx_struct, root()]) - } else { - root() + return Projection { + projection: root(), + file_row_number_column_pos: None, }; - return Projection(projection); } - - let has_columns_with_expr = projected_col_count > 0; - let (mut all_exprs, mut named_fields) = if has_columns_with_expr { - let all = Vec::with_capacity(column_ids.len() + has_file_row_number as usize); - let named = Vec::new(); - (all, named) - } else { - let all = Vec::new(); - let named = Vec::with_capacity(column_ids.len()); - (all, named) - }; - - if has_file_row_number && has_columns_with_expr { + if file_row_number_column_pos.is_some() { // row_idx will be moved to correct position in scan(), prepend here - all_exprs.push(("file_row_number", row_idx())); + exprs.insert(0, ("file_row_number", row_idx())); } - - for &column_id in column_ids { - if is_virtual_column(column_id) { - continue; - } - let column_id: usize = column_id.as_(); - let name = column_fields[column_id].name.as_str(); - if !has_columns_with_expr { - named_fields.push(name); - continue; - } - - let column_field = &column_fields[column_id]; - let expr = match &column_field.projection_expr { - None => get_item(name, root()), - Some(func) => func.clone(), - }; - all_exprs.push((name, expr)); + Self { + projection: pack(exprs, false.into()), + file_row_number_column_pos, } - - let projection = if has_columns_with_expr { - // If has_file_row_number is true, we have already inserted - // file_row_number column to all_exprs (see line 141) - pack(all_exprs, false.into()) - } else if has_file_row_number { - let select = select(named_fields, root()); - // Here we need to prepend file_row_number column manually. - // row_idx will be moved to correct position in scan() - let row_idx_struct = pack([("file_row_number", row_idx())], false.into()); - merge([row_idx_struct, select]) - } else { - select(named_fields, root()) - }; - - Self(projection) } // Create a projection for aggregate scan @@ -176,7 +165,10 @@ impl Projection { let names = exprs.into_iter().map(|(name, _)| name).collect::>(); select(names, root()) }; - Projection(projection) + Projection { + projection, + file_row_number_column_pos: None, + } } } @@ -275,59 +267,99 @@ pub fn extract_schema_from_dtype(dtype: &DType) -> VortexResult mod tests { use vortex::dtype::DType; use vortex::expr::lit; - use vortex::expr::merge; use vortex::expr::pack; use vortex::expr::root; use vortex::layout::layouts::row_idx::row_idx; use super::*; + fn field(name: &str) -> DuckdbField { + DuckdbField { + name: name.to_owned(), + logical_type: LogicalType::null(), + dtype: DType::Null, + projection_expr: None, + } + } + #[test] fn test_select_star() { let ids = [0, 1, 2]; - let mut fields = [ - DuckdbField { - name: "".to_owned(), - logical_type: LogicalType::null(), - dtype: DType::Null, - projection_expr: None, - }, - DuckdbField { - name: "".to_owned(), - logical_type: LogicalType::null(), - dtype: DType::Null, - projection_expr: None, - }, - DuckdbField { - name: "".to_owned(), - logical_type: LogicalType::null(), - dtype: DType::Null, - projection_expr: None, - }, - ]; - - assert_eq!(Projection::new(&ids, &fields).0, root()); + let mut fields = [field("a"), field("b"), field("c")]; - let ids = [FILE_ROW_NUMBER_COLUMN_IDX, 0, 1, 2]; - let exprs = Projection::new(&ids, &fields); - let row_idx_struct = pack([("file_row_number", row_idx())], false.into()); - let root_with_virtual_cols = merge([row_idx_struct, root()]); + assert_eq!(Projection::new(&[], &ids, &fields).projection, root()); - assert_eq!(exprs.0, root_with_virtual_cols); + // file_row_number turns star into an explicit pack with row_idx first + let ids = [FILE_ROW_NUMBER_COLUMN_IDX, 0, 1, 2]; + let result = Projection::new(&[], &ids, &fields); + let expected = pack( + [ + ("file_row_number", row_idx()), + ("a", get_item("a", root())), + ("b", get_item("b", root())), + ("c", get_item("c", root())), + ], + false.into(), + ); + assert_eq!(result.projection, expected); + assert_eq!(result.file_row_number_column_pos, Some(0)); let ids = [0, 1]; - assert_ne!(Projection::new(&ids, &fields).0, root()); + assert_ne!(Projection::new(&[], &ids, &fields).projection, root()); let ids = [0, 2, 2]; - assert_ne!(Projection::new(&ids, &fields).0, root()); + assert_ne!(Projection::new(&[], &ids, &fields).projection, root()); let ids = [2, 1, 0]; - assert_ne!(Projection::new(&ids, &fields).0, root()); + assert_ne!(Projection::new(&[], &ids, &fields).projection, root()); // If any column has a projection expression, we can't use SELECT * fields[0].projection_expr = Some(lit(true)); let ids = [0, 1, 2]; - assert_ne!(Projection::new(&ids, &fields).0, root()); + assert_ne!(Projection::new(&[], &ids, &fields).projection, root()); + } + + #[test] + fn test_projections() { + let fields = [field("a"), field("b"), field("c")]; + + let ids = [0, 1, 2]; + let projection = Projection::new(&[0, 2], &ids, &fields).projection; + let expected = pack( + [ + ("a", get_item("a", root())), + ("b", lit(Scalar::null(DType::Null))), + ("c", get_item("c", root())), + ], + false.into(), + ); + assert_eq!(projection, expected); + + let ids = [FILE_ROW_NUMBER_COLUMN_IDX, 0]; + let result = Projection::new(&[1], &ids, &fields); + let frn_dtype = DType::Primitive(PType::U64, Nullability::Nullable); + let expected = pack( + [ + ("file_row_number", lit(Scalar::null(frn_dtype))), + ("a", get_item("a", root())), + ], + false.into(), + ); + assert_eq!(result.projection, expected); + assert_eq!(result.file_row_number_column_pos, None); + + let ids = [0, FILE_ROW_NUMBER_COLUMN_IDX, 1]; + let result = Projection::new(&[0, 1], &ids, &fields); + let expected = pack( + [ + ("file_row_number", row_idx()), + ("a", get_item("a", root())), + ("b", lit(Scalar::null(DType::Null))), + ], + false.into(), + ); + assert_eq!(result.projection, expected); + assert_eq!(result.file_row_number_column_pos, Some(1)); } #[test] diff --git a/vortex-duckdb/src/table_function.rs b/vortex-duckdb/src/table_function.rs index 88aa9f9bd1c..3f7f9a6964e 100644 --- a/vortex-duckdb/src/table_function.rs +++ b/vortex-duckdb/src/table_function.rs @@ -52,10 +52,8 @@ use crate::duckdb::TableInitInput; use crate::duckdb::Value; use crate::exporter::ArrayExporter; use crate::projection::DuckdbField; -use crate::projection::FILE_ROW_NUMBER_COLUMN_IDX; use crate::projection::Filter; use crate::projection::Projection; -use crate::projection::is_virtual_column; // Duckdb has two state machines for an extension. The outer one is the table // function state machine which calls the file reader state machine. @@ -251,20 +249,14 @@ pub fn init_global(init_input: &TableInitInput) -> VortexResult { .iter() .any(|a| matches!(a, ColumnAggregate::CountStar)); - let mut file_row_number_column_pos = None; let column_ids = init_input.column_ids(); - let mut pos = 0; - for id in column_ids { - if *id == FILE_ROW_NUMBER_COLUMN_IDX { - file_row_number_column_pos = Some(pos); - pos += 1; - } else if !is_virtual_column(*id) { - pos += 1; - } - } + let projection_ids = init_input.projection_ids(); - let Projection(projection) = if bind_data.aggregates.is_empty() { - Projection::new(column_ids, &bind_data.columns) + let Projection { + projection, + file_row_number_column_pos, + } = if bind_data.aggregates.is_empty() { + Projection::new(projection_ids, column_ids, &bind_data.columns) } else { Projection::new_aggregate(&bind_data.aggregates, &bind_data.columns) }; From 515861ae9a688606a60512591577c68d78a259e1 Mon Sep 17 00:00:00 2001 From: Mikhail Kot Date: Tue, 15 Sep 2026 15:02:12 +0100 Subject: [PATCH 2/2] duckdb: report single-column filters as not pushed Signed-off-by: Mikhail Kot --- vortex-duckdb/src/convert/expr.rs | 3 ++ vortex-duckdb/src/table_function.rs | 46 +++++++++++++++-------------- 2 files changed, 27 insertions(+), 22 deletions(-) diff --git a/vortex-duckdb/src/convert/expr.rs b/vortex-duckdb/src/convert/expr.rs index 51aeebebb6d..c027f78113b 100644 --- a/vortex-duckdb/src/convert/expr.rs +++ b/vortex-duckdb/src/convert/expr.rs @@ -394,6 +394,9 @@ fn can_push_cast(cast: &duckdb::BoundCast<'_>, target: &duckdb::LogicalTypeRef) // If we return true here, and expression is in the list for // pushdown_complex_filter, we must handle it, or query engine will break. // +// We also don't have access to scan schema at this point, so we're overly +// restrictive. +// // Example: we don't support substr() expression so we tell Duckdb we can't // push it. // Example: we support CAST but not TRY_CAST. diff --git a/vortex-duckdb/src/table_function.rs b/vortex-duckdb/src/table_function.rs index 3f7f9a6964e..b0157cb007c 100644 --- a/vortex-duckdb/src/table_function.rs +++ b/vortex-duckdb/src/table_function.rs @@ -28,14 +28,14 @@ use vortex::error::VortexResult; use vortex::error::vortex_bail; use vortex::expr::BoundExpression; use vortex::expr::Expression; +use vortex::expr::VortexExprExt; use vortex::extension::uuid::Uuid; use vortex::metrics::tracing::get_global_labels; use vortex::scalar::Scalar; -use vortex::scalar_fn::fns::binary::Binary; -use vortex::scalar_fn::fns::operators::Operator; use vortex_utils::aliases::hash_map::HashMap; use crate::convert::PushedAggregate; +use crate::convert::can_push_expression; use crate::convert::try_from_bound_expression; use crate::convert::try_from_projection_aggregate; use crate::convert::try_from_projection_expression; @@ -399,36 +399,38 @@ fn aggregate_output_value(scalar: Scalar, expected: &LogicalTypeRef) -> VortexRe } } +/// This answers "true" if, given a duckdb table filter and table schema, +/// we can push the filter down to Vortex. pub fn pushdown_complex_filter( bind_data: &mut BindState, expr: &ExpressionRef, ) -> VortexResult { debug!(%expr, "pushing down expression"); - let Some(expr) = try_from_bound_expression(expr, &bind_data.columns)? else { + let Some(vx_expr) = try_from_bound_expression(expr, &bind_data.columns)? else { debug!(%expr, "failed to push down expression"); return Ok(false); }; - // Duckdb calls pushdown_complex_filter during planning phase. - // If all filters are pushed down, duckdb enables a LEFT_DELIM_JOIN -> - // COMPARISON_JOIN (HASH_JOIN) optimization: - // duckdb/src/optimizer/deliminator.cpp: Deliminator::HasSelection, - // Deliminator::Optimize. + // If we report filter as pushed, it disappears from duckdb's plan, so + // Deliminator::HasSelection doesn't see it and rewrites delim joins into + // ordinary joins. This is a bug reported to duckdb: + // https://github.com/duckdb/duckdb/issues/22669. // - // This leads to a massive regression on tpch sf=10 q17 and other - // benchmarks. + // TODO(myrrc): in duckdb 2.0 Deliminator and others see filters which + // have been pushed even if extension claims to process them fully, so this + // is a temporary fix. // - // This bug is reported to Duckdb - // https://github.com/duckdb/duckdb/issues/22669 + // Therefore we report single-column expressions as not pushed. + // FilterCombiner inserts them into get.table_filters and we + // get it back in init_input.filters(). // - // As a hack, report equality filters as not pushed. - // We can also report only the first filter as not pushed, but this - // has a negative performance impact. - let report_pushed = !expr - .as_opt::() - .map(|op| *op == Operator::Eq) - .unwrap_or(false); + // We need can_push_expression for spatial predicates. Duckdb can't insert + // them into get.table_filters so if we reject it here they become a FILTER + // over a Vortex scan. + if can_push_expression(expr) && vx_expr.field_references().len() <= 1 { + return Ok(false); + } // Only table filters may be optional, any complex filter is // non-optional by definition. @@ -436,9 +438,9 @@ pub fn pushdown_complex_filter( .has_non_optional_filter .store(true, Ordering::Relaxed); - debug!(%expr, report_pushed, "pushed down expression"); - bind_data.filters.push(expr); - Ok(report_pushed) + debug!(%vx_expr, "pushed down expression"); + bind_data.filters.push(vx_expr); + Ok(true) } pub fn pushdown_projection_expression(