From 6baf6f885de86cb0a5fbcc43aa26a3d893bbead5 Mon Sep 17 00:00:00 2001 From: Adam Gutglick Date: Thu, 18 Jun 2026 12:48:03 +0100 Subject: [PATCH 1/3] VortexSink display Signed-off-by: Adam Gutglick --- vortex-datafusion/src/persistent/sink.rs | 69 +++++++++++++++++++-- vortex-sqllogictest/slt/datafusion/.keep | 0 vortex-sqllogictest/slt/datafusion/sink.slt | 45 ++++++++++++++ 3 files changed, 110 insertions(+), 4 deletions(-) delete mode 100644 vortex-sqllogictest/slt/datafusion/.keep create mode 100644 vortex-sqllogictest/slt/datafusion/sink.slt diff --git a/vortex-datafusion/src/persistent/sink.rs b/vortex-datafusion/src/persistent/sink.rs index 9c02d72c120..828d55e97e3 100644 --- a/vortex-datafusion/src/persistent/sink.rs +++ b/vortex-datafusion/src/persistent/sink.rs @@ -10,6 +10,7 @@ use datafusion_common::Result as DFResult; use datafusion_common::exec_datafusion_err; use datafusion_common_runtime::JoinSet; use datafusion_common_runtime::SpawnedTask; +use datafusion_datasource::display::FileGroupDisplay; use datafusion_datasource::file_sink_config::FileSink; use datafusion_datasource::file_sink_config::FileSinkConfig; use datafusion_datasource::sink::DataSink; @@ -59,9 +60,12 @@ impl std::fmt::Debug for VortexSink { impl DisplayAs for VortexSink { fn fmt_as(&self, t: DisplayFormatType, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match t { - DisplayFormatType::Default - | DisplayFormatType::Verbose - | DisplayFormatType::TreeRender => { + DisplayFormatType::Default | DisplayFormatType::Verbose => { + write!(f, "VortexSink(file_groups=")?; + FileGroupDisplay(&self.config.file_group).fmt_as(t, f)?; + write!(f, ")") + } + DisplayFormatType::TreeRender => { write!(f, "VortexSink") } } @@ -198,10 +202,22 @@ mod tests { use datafusion::logical_expr::Values; use datafusion::logical_expr::dml::InsertOp; use datafusion_common::ScalarValue; + use datafusion_datasource::TableSchema; use datafusion_datasource::file_format::format_as_file_type; + use datafusion_datasource::file_groups::FileGroup; + use datafusion_datasource::file_sink_config::FileOutputMode; + use datafusion_datasource::sink::DataSinkExec; + use datafusion_execution::object_store::ObjectStoreUrl; + use datafusion_physical_plan::DefaultDisplay; + use datafusion_physical_plan::VerboseDisplay; + use datafusion_physical_plan::display::DisplayableExecutionPlan; + use datafusion_physical_plan::empty::EmptyExec; use futures::TryStreamExt; use rstest::rstest; + use vortex::file::VORTEX_FILE_EXTENSION; + use vortex::session::VortexSession; + use super::*; use crate::common_tests::TestSessionContext; use crate::persistent::VortexFormatFactory; @@ -430,7 +446,7 @@ mod tests { let table = ctx.session.table("my_tbl").await?; assert_eq!(table.count().await?, 3); - let location = object_store::path::Path::parse("table/")?; + let location = Path::parse("table/")?; let file_metas = ctx .store .list(Some(&location)) @@ -447,4 +463,49 @@ mod tests { Ok(()) } + + #[test] + fn test_display_as() { + let session = VortexSession::empty(); + let table_schema = TableSchema::new(Arc::new(Schema::empty()), Vec::new()); + + let config = FileSinkConfig { + original_url: "".to_owned(), + object_store_url: ObjectStoreUrl::local_filesystem(), + file_group: FileGroup::new(Vec::new()), + table_paths: Vec::new(), + output_schema: Arc::new(Schema::empty()), + table_partition_cols: Vec::new(), + insert_op: InsertOp::Overwrite, + keep_partition_by_columns: false, + file_extension: VORTEX_FILE_EXTENSION.to_owned(), + file_output_mode: FileOutputMode::SingleFile, + }; + + let get_sink = || VortexSink { + config: config.clone(), + schema: Arc::clone(table_schema.file_schema()), + session: session.clone(), + }; + + insta::assert_snapshot!(DefaultDisplay(get_sink()).to_string(), @"VortexSink(file_groups=[])"); + insta::assert_snapshot!(VerboseDisplay(get_sink()).to_string(), @"VortexSink(file_groups=[])"); + + let plan = DataSinkExec::new( + Arc::new(EmptyExec::new(Arc::new(Schema::empty()))), + Arc::new(get_sink()), + None, + ); + + insta::assert_snapshot!(DisplayableExecutionPlan::new(&plan).tree_render().to_string(), @r" + ┌───────────────────────────┐ + │ DataSinkExec │ + │ -------------------- │ + │ VortexSink │ + └─────────────┬─────────────┘ + ┌─────────────┴─────────────┐ + │ EmptyExec │ + └───────────────────────────┘ + "); + } } diff --git a/vortex-sqllogictest/slt/datafusion/.keep b/vortex-sqllogictest/slt/datafusion/.keep deleted file mode 100644 index e69de29bb2d..00000000000 diff --git a/vortex-sqllogictest/slt/datafusion/sink.slt b/vortex-sqllogictest/slt/datafusion/sink.slt new file mode 100644 index 00000000000..7ce71590919 --- /dev/null +++ b/vortex-sqllogictest/slt/datafusion/sink.slt @@ -0,0 +1,45 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright the Vortex contributors + +include ../setup.slt.no + +query TT +EXPLAIN +COPY (SELECT * FROM (VALUES (1), (42), (100), (-5), (0)) AS t(num)) +TO '$__TEST_DIR__/datafusion/sink/data1.vortex' +STORED AS VORTEX; +---- +logical_plan +01)CopyTo: format=vortex output_url=/datafusion/sink/data1.vortex options: () +02)--SubqueryAlias: t +03)----Projection: column1 AS num +04)------Values: (Int64(1)), (Int64(42)), (Int64(100)), (Int64(-5)), (Int64(0)) +physical_plan +01)DataSinkExec: sink=VortexSink(file_groups=[]) +02)--ProjectionExec: expr=[column1@0 as num] +03)----DataSourceExec: partitions=1, partition_sizes=[1] + +query TT +EXPLAIN FORMAT TREE +COPY (SELECT * FROM (VALUES (1), (42), (100), (-5), (0)) AS t(num)) +TO '$__TEST_DIR__/datafusion/sink/data2.vortex' +STORED AS VORTEX; +---- +physical_plan +01)┌───────────────────────────┐ +02)│ DataSinkExec │ +03)│ -------------------- │ +04)│ VortexSink │ +05)└─────────────┬─────────────┘ +06)┌─────────────┴─────────────┐ +07)│ ProjectionExec │ +08)│ -------------------- │ +09)│ num: column1 │ +10)└─────────────┬─────────────┘ +11)┌─────────────┴─────────────┐ +12)│ DataSourceExec │ +13)│ -------------------- │ +14)│ bytes: 160 │ +15)│ format: memory │ +16)│ rows: 1 │ +17)└───────────────────────────┘ From f9d182f6dda0c79ddd378bf560850592c9862fe4 Mon Sep 17 00:00:00 2001 From: Adam Gutglick Date: Thu, 18 Jun 2026 13:51:08 +0100 Subject: [PATCH 2/3] Add bytes_length pushdown Signed-off-by: Adam Gutglick --- vortex-datafusion/src/convert/exprs.rs | 139 ++++++++++++++++----- vortex-datafusion/src/persistent/source.rs | 42 +++++++ vortex-datafusion/src/persistent/tests.rs | 48 +++++++ 3 files changed, 198 insertions(+), 31 deletions(-) diff --git a/vortex-datafusion/src/convert/exprs.rs b/vortex-datafusion/src/convert/exprs.rs index c14576c5301..4ba174f572b 100644 --- a/vortex-datafusion/src/convert/exprs.rs +++ b/vortex-datafusion/src/convert/exprs.rs @@ -11,6 +11,7 @@ use datafusion_common::tree_node::TreeNode; use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_expr::Operator as DFOperator; use datafusion_functions::core::getfield::GetFieldFunc; +use datafusion_functions::string::octet_length::OctetLengthFunc; use datafusion_physical_expr::PhysicalExpr; use datafusion_physical_expr::ScalarFunctionExpr; use datafusion_physical_expr::projection::ProjectionExpr; @@ -24,6 +25,7 @@ use vortex::dtype::Nullability; use vortex::dtype::arrow::FromArrowType; use vortex::expr::Expression; use vortex::expr::and_collect; +use vortex::expr::byte_length; use vortex::expr::cast; use vortex::expr::get_item; use vortex::expr::is_not_null; @@ -111,8 +113,28 @@ pub trait ExpressionConvertor: Send + Sync { pub struct DefaultExpressionConvertor {} impl DefaultExpressionConvertor { + /// Attempts to convert DataFusion's `octet_length` function to Vortex `byte_length`. + fn try_convert_octet_length(&self, scalar_fn: &ScalarFunctionExpr) -> DFResult { + let [input] = scalar_fn.args() else { + return Err(exec_datafusion_err!( + "octet_length requires exactly one argument" + )); + }; + + let input = self.convert(input.as_ref())?; + let return_dtype = + DType::from_arrow((scalar_fn.return_type(), scalar_fn.nullable().into())); + Ok(cast(byte_length(input), return_dtype)) + } + /// Attempts to convert a DataFusion ScalarFunctionExpr to a Vortex expression. fn try_convert_scalar_function(&self, scalar_fn: &ScalarFunctionExpr) -> DFResult { + if let Some(octet_length_fn) = + ScalarFunctionExpr::try_downcast_func::(scalar_fn) + { + return self.try_convert_octet_length(octet_length_fn); + } + if let Some(get_field_fn) = ScalarFunctionExpr::try_downcast_func::(scalar_fn) { // DataFusion's GetFieldFunc flattens nested field access into a single call @@ -289,7 +311,7 @@ impl ExpressionConvertor for DefaultExpressionConvertor { let r = projection_expr.expr.apply(|node| { // We only pull column children of scalar functions that we can't push into the scan. if let Some(scalar_fn_expr) = node.downcast_ref::() - && !can_scalar_fn_be_pushed_down(scalar_fn_expr) + && !can_scalar_fn_be_pushed_down(scalar_fn_expr, input_schema) { scan_projection.extend( collect_columns(node) @@ -305,8 +327,8 @@ impl ExpressionConvertor for DefaultExpressionConvertor { // Vortex expects a perfect match so we don't push it down. if let Some(binary_expr) = node.downcast_ref::() && binary_expr.op().is_numerical_operators() - && (is_decimal(&binary_expr.left().data_type(input_schema)?) - && is_decimal(&binary_expr.right().data_type(input_schema)?)) + && binary_expr.left().data_type(input_schema)?.is_decimal() + && binary_expr.right().data_type(input_schema)?.is_decimal() { scan_projection.extend( collect_columns(node) @@ -430,7 +452,7 @@ fn can_be_pushed_down_impl(expr: &Arc, schema: &Schema) -> boo .iter() .all(|e| can_be_pushed_down_impl(e, schema)) } else if let Some(scalar_fn) = expr.downcast_ref::() { - can_scalar_fn_be_pushed_down(scalar_fn) + can_scalar_fn_be_pushed_down(scalar_fn, schema) } else if let Some(case_expr) = expr.downcast_ref::() { can_case_be_pushed_down(case_expr, schema) } else { @@ -454,9 +476,10 @@ fn is_convertible_expr(expr: &Arc) -> bool { || expr.downcast_ref::().is_some() || expr.downcast_ref::().is_some() || expr.downcast_ref::().is_some() - || expr - .downcast_ref::() - .is_some_and(|sf| ScalarFunctionExpr::try_downcast_func::(sf).is_some()) + || expr.downcast_ref::().is_some_and(|sf| { + ScalarFunctionExpr::try_downcast_func::(sf).is_some() + || ScalarFunctionExpr::try_downcast_func::(sf).is_some() + }) } fn can_binary_be_pushed_down(binary: &df_expr::BinaryExpr, schema: &Schema) -> bool { @@ -502,20 +525,11 @@ fn supported_data_types(dt: &DataType) -> bool { let is_supported = dt.is_null() || dt.is_numeric() + || dt.is_binary() + || dt.is_string() || matches!( dt, - Boolean - | Utf8 - | LargeUtf8 - | Utf8View - | Binary - | LargeBinary - | BinaryView - | Date32 - | Date64 - | Timestamp(_, _) - | Time32(_) - | Time64(_) + Boolean | Date32 | Date64 | Timestamp(_, _) | Time32(_) | Time64(_) ); if !is_supported { @@ -526,20 +540,30 @@ fn supported_data_types(dt: &DataType) -> bool { } /// Checks if a scalar function can be pushed down. -/// Currently only GetFieldFunc is supported. -fn can_scalar_fn_be_pushed_down(scalar_fn: &ScalarFunctionExpr) -> bool { - ScalarFunctionExpr::try_downcast_func::(scalar_fn).is_some() +/// Currently GetFieldFunc and OctetLengthFunc are supported. +fn can_scalar_fn_be_pushed_down(scalar_fn: &ScalarFunctionExpr, schema: &Schema) -> bool { + if ScalarFunctionExpr::try_downcast_func::(scalar_fn).is_some() { + return true; + } + + ScalarFunctionExpr::try_downcast_func::(scalar_fn) + .is_some_and(|octet_length| can_octet_length_be_pushed_down(octet_length, schema)) } -// TODO(adam): Replace with `DataType::is_decimal` once its released. -fn is_decimal(dt: &DataType) -> bool { - matches!( - dt, - DataType::Decimal32(_, _) - | DataType::Decimal64(_, _) - | DataType::Decimal128(_, _) - | DataType::Decimal256(_, _) - ) +fn can_octet_length_be_pushed_down(scalar_fn: &ScalarFunctionExpr, schema: &Schema) -> bool { + let [input] = scalar_fn.args() else { + return false; + }; + + input.data_type(schema).as_ref().is_ok_and(|data_type| { + let dt = if let DataType::Dictionary(_, value_type) = data_type { + value_type.as_ref() + } else { + data_type + }; + + dt.is_binary() || dt.is_string() + }) && can_be_pushed_down_impl(input, schema) } #[cfg(test)] @@ -553,7 +577,9 @@ mod tests { use datafusion::arrow::array::AsArray; use datafusion::arrow::datatypes::Int32Type; use datafusion_common::ScalarValue; + use datafusion_common::config::ConfigOptions; use datafusion_expr::Operator as DFOperator; + use datafusion_expr::ScalarUDF; use datafusion_physical_expr::PhysicalExpr; use datafusion_physical_plan::expressions as df_expr; use insta::assert_snapshot; @@ -582,6 +608,18 @@ mod tests { ]) } + fn octet_length_expr(input: Arc, schema: &Schema) -> Arc { + Arc::new( + ScalarFunctionExpr::try_new( + Arc::new(ScalarUDF::from(OctetLengthFunc::new())), + vec![input], + schema, + Arc::new(ConfigOptions::new()), + ) + .unwrap(), + ) + } + #[test] fn test_make_vortex_predicate_empty() { let expr_convertor = DefaultExpressionConvertor::default(); @@ -711,6 +749,23 @@ mod tests { ); } + #[rstest] + fn test_expr_from_df_octet_length(test_schema: Schema) { + let expr = Arc::new(df_expr::Column::new("name", 1)) as Arc; + let octet_length = octet_length_expr(expr, &test_schema); + + let result = DefaultExpressionConvertor::default() + .convert(octet_length.as_ref()) + .unwrap(); + + assert_snapshot!(result.display_tree().to_string(), @r" + vortex.cast(i32?) + └── input: vortex.byte_length() + └── input: vortex.get_item(name) + └── input: vortex.root() + "); + } + #[rstest] // Supported types #[case::null(DataType::Null, true)] @@ -865,6 +920,28 @@ mod tests { assert!(!can_be_pushed_down_impl(&like_expr, &test_schema)); } + #[rstest] + fn test_can_be_pushed_down_octet_length_supported(test_schema: Schema) { + let expr = Arc::new(df_expr::Column::new("name", 1)) as Arc; + let octet_length = octet_length_expr(expr, &test_schema); + + assert!(can_be_pushed_down_impl(&octet_length, &test_schema)); + } + + #[rstest] + fn test_can_be_pushed_down_octet_length_unsupported_operand(test_schema: Schema) { + let expr = Arc::new(df_expr::Column::new("unsupported_list", 5)) as Arc; + let octet_length = Arc::new(ScalarFunctionExpr::new( + "octet_length", + Arc::new(ScalarUDF::from(OctetLengthFunc::new())), + vec![expr], + Arc::new(Field::new("octet_length", DataType::Int32, true)), + Arc::new(ConfigOptions::new()), + )) as Arc; + + assert!(!can_be_pushed_down_impl(&octet_length, &test_schema)); + } + // https://github.com/vortex-data/vortex/issues/6211 #[tokio::test] async fn test_cast_int_to_string() -> anyhow::Result<()> { diff --git a/vortex-datafusion/src/persistent/source.rs b/vortex-datafusion/src/persistent/source.rs index 896f8e39829..876906f26e2 100644 --- a/vortex-datafusion/src/persistent/source.rs +++ b/vortex-datafusion/src/persistent/source.rs @@ -522,8 +522,15 @@ mod tests { use arrow_schema::DataType; use arrow_schema::Field; use arrow_schema::Schema; + use datafusion_common::ScalarValue; + use datafusion_common::config::ConfigOptions; use datafusion_datasource::file_scan_config::FileScanConfigBuilder; use datafusion_execution::object_store::ObjectStoreUrl; + use datafusion_expr::Operator; + use datafusion_expr::ScalarUDF; + use datafusion_functions::string::octet_length::OctetLengthFunc; + use datafusion_physical_expr::ScalarFunctionExpr; + use datafusion_physical_expr::expressions as df_expr; use datafusion_physical_expr::expressions::Column; use object_store::memory::InMemory; use vortex::VortexSessionDefault; @@ -583,6 +590,22 @@ mod tests { ) } + fn octet_length_filter(schema: &Schema) -> PhysicalExprRef { + let name = Arc::new(Column::new("name", 0)) as PhysicalExprRef; + let octet_length = Arc::new( + ScalarFunctionExpr::try_new( + Arc::new(ScalarUDF::from(OctetLengthFunc::new())), + vec![name], + schema, + Arc::new(ConfigOptions::new()), + ) + .unwrap(), + ) as PhysicalExprRef; + let one = Arc::new(df_expr::Literal::new(ScalarValue::Int32(Some(1)))) as PhysicalExprRef; + + Arc::new(df_expr::BinaryExpr::new(octet_length, Operator::Gt, one)) as PhysicalExprRef + } + fn assert_ordered_source(inner: Arc) -> anyhow::Result<()> { let source = inner .downcast_ref::() @@ -643,4 +666,23 @@ mod tests { )); Ok(()) } + + #[test] + fn try_pushdown_filters_accepts_octet_length() -> anyhow::Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("name", DataType::Utf8, false)])); + let source = sort_test_source(Arc::clone(&schema)); + let filter = octet_length_filter(&schema); + + let result = source.try_pushdown_filters(vec![filter], &ConfigOptions::new())?; + + assert!(matches!(result.filters.as_slice(), [PushedDown::Yes])); + let updated_source = result + .updated_node + .ok_or_else(|| anyhow::anyhow!("expected updated VortexSource"))? + .downcast_ref::() + .ok_or_else(|| anyhow::anyhow!("expected VortexSource"))? + .clone(); + assert!(updated_source.vortex_predicate.is_some()); + Ok(()) + } } diff --git a/vortex-datafusion/src/persistent/tests.rs b/vortex-datafusion/src/persistent/tests.rs index 2df4d8c2948..220a1477b13 100644 --- a/vortex-datafusion/src/persistent/tests.rs +++ b/vortex-datafusion/src/persistent/tests.rs @@ -6,6 +6,7 @@ use std::sync::Arc; use anyhow::anyhow; use datafusion::arrow::array::Int32Array; use datafusion::arrow::array::RecordBatch; +use datafusion::arrow::datatypes::DataType; use datafusion::arrow::util::pretty::pretty_format_batches; use datafusion::datasource::provider::DefaultTableFactory; use datafusion::execution::SessionStateBuilder; @@ -185,6 +186,53 @@ async fn test_addition_pushdown() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +async fn test_octet_length_pushdown() -> anyhow::Result<()> { + let ctx = TestSessionContext::new(true); + + ctx.session + .sql( + "CREATE EXTERNAL TABLE written_strings \ + (s VARCHAR NOT NULL) \ + STORED AS vortex \ + LOCATION '/strings/'", + ) + .await?; + + ctx.session + .sql("INSERT INTO written_strings VALUES ('a'), ('é'), ('abcd'), ('')") + .await? + .collect() + .await?; + + let result = ctx + .session + .sql( + "SELECT s, octet_length(s) AS len \ + FROM written_strings \ + WHERE octet_length(s) > 1 \ + ORDER BY s", + ) + .await? + .collect() + .await?; + + assert_eq!( + result[0].schema().field_with_name("len")?.data_type(), + &DataType::Int32 + ); + assert_snapshot!(pretty_format_batches(&result)?, @r" + +------+-----+ + | s | len | + +------+-----+ + | abcd | 4 | + | é | 2 | + +------+-----+ + "); + + Ok(()) +} + #[tokio::test] async fn create_table_ordered_by() -> anyhow::Result<()> { let ctx = TestSessionContext::default(); From 733fe06d6ae60460bf89ead87c65e868d6a44e7e Mon Sep 17 00:00:00 2001 From: Adam Gutglick Date: Thu, 18 Jun 2026 14:33:02 +0100 Subject: [PATCH 3/3] Add config to VortexSource to enable/disable predicate pushdown Signed-off-by: Adam Gutglick --- vortex-datafusion/src/persistent/format.rs | 42 ++++++++++++++----- vortex-datafusion/src/persistent/source.rs | 47 +++++++++++++++++++++- 2 files changed, 79 insertions(+), 10 deletions(-) diff --git a/vortex-datafusion/src/persistent/format.rs b/vortex-datafusion/src/persistent/format.rs index 0381e550893..a0d49e6105a 100644 --- a/vortex-datafusion/src/persistent/format.rs +++ b/vortex-datafusion/src/persistent/format.rs @@ -146,6 +146,7 @@ config_namespace! { /// /// let factory = VortexFormatFactory::new().with_options(VortexTableOptions { /// projection_pushdown: true, + /// predicate_pushdown: true, /// scan_concurrency: Some(8), /// ..Default::default() /// }); @@ -165,6 +166,12 @@ config_namespace! { /// the scan. When disabled, Vortex reads only the referenced columns and /// all expressions are evaluated after the scan. pub projection_pushdown: bool, default = false + /// Whether to enable predicate pushdown into the underlying Vortex scan. + /// + /// When enabled, supported filters are evaluated during the scan. When + /// disabled, DataFusion evaluates filters after the scan, while + /// `VortexSource` can still use the full predicate for file pruning. + pub predicate_pushdown: bool, default = true /// The intra-partition scan concurrency, controlling the number of row splits to process /// concurrently per-thread within each file. /// @@ -198,6 +205,7 @@ impl Eq for VortexTableOptions {} /// /// let factory = Arc::new(VortexFormatFactory::new().with_options(VortexTableOptions { /// projection_pushdown: true, +/// predicate_pushdown: true, /// ..Default::default() /// })); /// @@ -263,6 +271,7 @@ impl VortexFormatFactory { /// /// let factory = VortexFormatFactory::new().with_options(VortexTableOptions { /// projection_pushdown: true, + /// predicate_pushdown: true, /// ..Default::default() /// }); /// # let _ = factory; @@ -617,14 +626,9 @@ impl FileFormat for VortexFormat { } fn file_source(&self, table_schema: TableSchema) -> Arc { - let mut source = VortexSource::new(table_schema, self.session.clone()) - .with_projection_pushdown(self.opts.projection_pushdown); - - if let Some(scan_concurrency) = self.opts.scan_concurrency { - source = source.with_scan_concurrency(scan_concurrency); - } - - Arc::new(source) as _ + Arc::new( + VortexSource::new(table_schema, self.session.clone()).with_options(self.opts.clone()), + ) as _ } } @@ -682,7 +686,7 @@ mod tests { (c1 VARCHAR NOT NULL, c2 INT NOT NULL) \ STORED AS vortex \ LOCATION 'table/' \ - OPTIONS( footer_initial_read_size_bytes '12345', scan_concurrency '3' );", + OPTIONS( footer_initial_read_size_bytes '12345', predicate_pushdown 'false', scan_concurrency '3' );", ) .await? .collect() @@ -699,4 +703,24 @@ mod tests { let format = VortexFormat::new_with_options(VortexSession::default(), opts); assert_eq!(format.options().footer_initial_read_size_bytes, 12345); } + + #[test] + fn format_plumbs_source_options() -> anyhow::Result<()> { + let opts = VortexTableOptions { + projection_pushdown: true, + predicate_pushdown: false, + scan_concurrency: Some(3), + ..Default::default() + }; + let format = VortexFormat::new_with_options(VortexSession::default(), opts.clone()); + let table_schema = TableSchema::from_file_schema(Arc::new(Schema::empty())); + + let source = format.file_source(table_schema); + let source = source + .downcast_ref::() + .ok_or_else(|| anyhow::anyhow!("expected VortexSource"))?; + + assert_eq!(source.options(), &opts); + Ok(()) + } } diff --git a/vortex-datafusion/src/persistent/source.rs b/vortex-datafusion/src/persistent/source.rs index 876906f26e2..2f4888404e8 100644 --- a/vortex-datafusion/src/persistent/source.rs +++ b/vortex-datafusion/src/persistent/source.rs @@ -140,6 +140,13 @@ use crate::persistent::reader::VortexReaderFactory; /// - when enabled, the scan can evaluate a Vortex-native projection and leave /// only unsupported expressions for DataFusion. /// +/// Predicate handling depends on [`VortexTableOptions::predicate_pushdown`]: +/// +/// - when disabled, `VortexSource` still keeps the full predicate for +/// DataFusion file pruning, but reports filters as not pushed down so +/// DataFusion evaluates them after the scan, +/// - when enabled, supported filters are pushed into the Vortex scan. +/// /// # Observability /// /// `VortexSource` owns a Vortex metrics registry for the lifetime of a physical @@ -170,6 +177,7 @@ use crate::persistent::reader::VortexReaderFactory; /// [`VortexAccessPlan`]: crate::VortexAccessPlan /// [`FileMetadataCache`]: datafusion_execution::cache::cache_manager::FileMetadataCache /// [`VortexTableOptions::projection_pushdown`]: crate::VortexTableOptions::projection_pushdown +/// [`VortexTableOptions::predicate_pushdown`]: crate::VortexTableOptions::predicate_pushdown /// [`VortexMetricsFinder`]: crate::metrics::VortexMetricsFinder #[derive(Clone)] pub struct VortexSource { @@ -195,7 +203,7 @@ pub struct VortexSource { pub(crate) ordered: bool, vx_metrics_registry: Arc, file_metadata_cache: Option>, - /// Whether to enable expression pushdown into the underlying Vortex scan. + /// Options controlling scan planning and execution behavior. options: VortexTableOptions, } @@ -242,6 +250,15 @@ impl VortexSource { self } + /// Enables or disables Vortex-native predicate evaluation. + /// + /// When disabled, DataFusion evaluates filters after the scan. The source + /// still records the full predicate for file pruning. + pub fn with_predicate_pushdown(mut self, enabled: bool) -> Self { + self.options.predicate_pushdown = enabled; + self + } + /// Sets the [`ExpressionConvertor`] used to translate DataFusion expressions /// into Vortex expressions. /// @@ -452,6 +469,14 @@ impl FileSource for VortexSource { None => Some(conjunction(filters.clone())), }; + if !source.options.predicate_pushdown { + return Ok(FilterPushdownPropagation::with_parent_pushdown_result(vec![ + PushedDown::No; + filters.len() + ]) + .with_updated_node(Arc::new(source) as _)); + } + let supported_filters = filters .into_iter() .map(|expr| { @@ -685,4 +710,24 @@ mod tests { assert!(updated_source.vortex_predicate.is_some()); Ok(()) } + + #[test] + fn try_pushdown_filters_respects_disabled_predicate_pushdown() -> anyhow::Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("name", DataType::Utf8, false)])); + let source = sort_test_source(Arc::clone(&schema)).with_predicate_pushdown(false); + let filter = octet_length_filter(&schema); + + let result = source.try_pushdown_filters(vec![filter], &ConfigOptions::new())?; + + assert!(matches!(result.filters.as_slice(), [PushedDown::No])); + let updated_source = result + .updated_node + .ok_or_else(|| anyhow::anyhow!("expected updated VortexSource"))? + .downcast_ref::() + .ok_or_else(|| anyhow::anyhow!("expected VortexSource"))? + .clone(); + assert!(updated_source.full_predicate.is_some()); + assert!(updated_source.vortex_predicate.is_none()); + Ok(()) + } }