Add config to VortexSource to enable/disable predicate pushdown

Signed-off-by: Adam Gutglick <adam@spiraldb.com>

Author: AdamGS

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<dyn FileSource> {
-        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::<VortexSource>()
+            .ok_or_else(|| anyhow::anyhow!("expected VortexSource"))?;
+
+        assert_eq!(source.options(), &opts);
+        Ok(())
+    }
 }

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<dyn MetricsRegistry>,
     file_metadata_cache: Option<Arc<dyn FileMetadataCache>>,
-    /// 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::<VortexSource>()
+            .ok_or_else(|| anyhow::anyhow!("expected VortexSource"))?
+            .clone();
+        assert!(updated_source.full_predicate.is_some());
+        assert!(updated_source.vortex_predicate.is_none());
+        Ok(())
+    }
 }