feat(python): add deletion_vectors API via DataFusion replay

Expose DeltaTable.deletion_vectors() in Python backed by a new core replay API.

- Add DeltaScan::deletion_vectors() with deterministic filepath ordering and named DeletionVectorSelection output.

- Reuse shared scan metadata stream setup and document replay/drain semantics.

- Build Arrow RecordBatchReader output in Python with filepath URI + selection_vector list[bool], chunked batching, non-null list items, and without_files guard.

- Tighten concurrency by cloning table+state under one lock and using one SessionContext state for registration + scan.

- Strengthen core/Python tests for exact DV values, determinism, eager snapshot path, URI contract, empty results, and error path.

- Replace an internal DV invariant expect() with typed error propagation instead of panic.

Signed-off-by: Ethan Urbanski <ethan@urbanskitech.com>

Author: ethan-tyler

crates/core/src/delta_datafusion/mod.rs

@@ -65,7 +65,7 @@ pub use self::session::{
     DeltaParserOptions, DeltaRuntimeEnvBuilder, DeltaSessionConfig, DeltaSessionContext,
     create_session,
 };
-pub use self::table_provider::next::DeltaScan as DeltaScanNext;
+pub use self::table_provider::next::{DeletionVectorSelection, DeltaScan as DeltaScanNext};
 pub(crate) use self::utils::*;
 pub use cdf::scan::DeltaCdfTableProvider;
 pub(crate) use data_validation::{

crates/core/src/delta_datafusion/table_provider/next/mod.rs

@@ -39,15 +39,15 @@ use datafusion::{
     logical_expr::LogicalPlan,
     physical_plan::ExecutionPlan,
 };
-use delta_kernel::table_configuration::TableConfiguration;
+use delta_kernel::{Engine, table_configuration::TableConfiguration};
 use serde::{Deserialize, Serialize};
 
 pub use self::scan::DeltaScanExec;
 pub(crate) use self::scan::KernelScanPlan;
 use crate::delta_datafusion::DeltaScanConfig;
 use crate::delta_datafusion::engine::DataFusionEngine;
 use crate::delta_datafusion::table_provider::TableProviderBuilder;
-use crate::kernel::{EagerSnapshot, Snapshot};
+use crate::kernel::{EagerSnapshot, SendableScanMetadataStream, Snapshot};
 
 mod scan;
 
@@ -111,6 +111,15 @@ pub struct DeltaScan {
     file_skipping_predicate: Option<Vec<Expr>>,
 }
 
+/// Deletion vector selection for one data file.
+#[derive(Debug, Clone, PartialEq)]
+pub struct DeletionVectorSelection {
+    /// Fully-qualified file URI.
+    pub filepath: String,
+    /// Row-level keep mask where `true` means keep and `false` means deleted.
+    pub keep_mask: Vec<bool>,
+}
+
 impl DeltaScan {
     // create new delta scan
     pub fn new(snapshot: impl Into<SnapshotWrapper>, config: DeltaScanConfig) -> Result<Self> {
@@ -140,6 +149,53 @@ impl DeltaScan {
         self
     }
 
+    /// Materialize deletion vector keep masks for files in this scan.
+    ///
+    /// The result is sorted lexicographically by filepath for deterministic ordering and includes
+    /// only files that have deletion vectors.
+    ///
+    /// This API materializes all deletion vectors in memory.
+    pub async fn deletion_vectors(
+        &self,
+        session: &dyn Session,
+    ) -> Result<Vec<DeletionVectorSelection>> {
+        let engine = DataFusionEngine::new_from_session(session);
+
+        let scan_plan = KernelScanPlan::try_new(
+            self.snapshot.snapshot(),
+            None,
+            &[],
+            &self.config,
+            self.file_skipping_predicate.clone(),
+        )?;
+
+        let stream = self.scan_metadata_stream(&scan_plan, engine.clone());
+
+        scan::replay_deletion_vectors(engine, &scan_plan, &self.config, stream).await
+    }
+
+    fn scan_metadata_stream(
+        &self,
+        scan_plan: &KernelScanPlan,
+        engine: Arc<dyn Engine>,
+    ) -> SendableScanMetadataStream {
+        match &self.snapshot {
+            SnapshotWrapper::Snapshot(_) => scan_plan.scan.scan_metadata(engine),
+            SnapshotWrapper::EagerSnapshot(esn) => {
+                if let Ok(files) = esn.files() {
+                    scan_plan.scan.scan_metadata_from(
+                        engine,
+                        esn.snapshot().version() as u64,
+                        Box::new(files.to_vec().into_iter()),
+                        None,
+                    )
+                } else {
+                    scan_plan.scan.scan_metadata(engine)
+                }
+            }
+        }
+    }
+
     pub fn builder() -> TableProviderBuilder {
         TableProviderBuilder::new()
     }
@@ -197,21 +253,7 @@ impl TableProvider for DeltaScan {
             self.file_skipping_predicate.clone(),
         )?;
 
-        let stream = match &self.snapshot {
-            SnapshotWrapper::Snapshot(_) => scan_plan.scan.scan_metadata(engine.clone()),
-            SnapshotWrapper::EagerSnapshot(esn) => {
-                if let Ok(files) = esn.files() {
-                    scan_plan.scan.scan_metadata_from(
-                        engine.clone(),
-                        esn.snapshot().version() as u64,
-                        Box::new(files.to_vec().into_iter()),
-                        None,
-                    )
-                } else {
-                    scan_plan.scan.scan_metadata(engine.clone())
-                }
-            }
-        };
+        let stream = self.scan_metadata_stream(&scan_plan, engine.clone());
 
         scan::execution_plan(&self.config, session, scan_plan, stream, engine, limit).await
     }
@@ -236,11 +278,12 @@ mod tests {
         physical_plan::{ExecutionPlanVisitor, collect_partitioned, visit_execution_plan},
     };
     use datafusion_datasource::source::DataSourceExec;
+    use url::Url;
 
     use crate::{
         assert_batches_sorted_eq,
-        delta_datafusion::session::create_session,
-        kernel::Snapshot,
+        delta_datafusion::{DeltaScanConfig, session::create_session},
+        kernel::{EagerSnapshot, Snapshot},
         test_utils::{TestResult, TestTables},
     };
 
@@ -358,4 +401,70 @@ mod tests {
 
         Ok(())
     }
+
+    fn expected_dv_small()
+    -> std::result::Result<Vec<DeletionVectorSelection>, Box<dyn std::error::Error>> {
+        let filepath = Url::from_file_path(
+            TestTables::WithDvSmall
+                .as_path()
+                .join("part-00000-fae5310a-a37d-4e51-827b-c3d5516560ca-c000.snappy.parquet"),
+        )
+        .map_err(|_| "failed to convert expected file path to URL")?
+        .to_string();
+        Ok(vec![DeletionVectorSelection {
+            filepath,
+            keep_mask: vec![false, true, true, true, true, true, true, true, true, false],
+        }])
+    }
+
+    #[tokio::test]
+    async fn test_deletion_vectors_with_dv_table() -> TestResult {
+        let log_store = TestTables::WithDvSmall.table_builder()?.build_storage()?;
+        let snapshot = Snapshot::try_new(&log_store, Default::default(), None).await?;
+        let provider = DeltaScan::new(snapshot, DeltaScanConfig::default())?;
+
+        let session = Arc::new(create_session().into_inner());
+        let state = session.state_ref().read().clone();
+        let expected = expected_dv_small()?;
+
+        let deletion_vectors = provider.deletion_vectors(&state).await?;
+        let deletion_vectors_second = provider.deletion_vectors(&state).await?;
+
+        assert_eq!(deletion_vectors, expected);
+        assert_eq!(deletion_vectors_second, expected);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_deletion_vectors_without_dv_table_is_empty() -> TestResult {
+        let log_store = TestTables::Simple.table_builder()?.build_storage()?;
+        let snapshot = Snapshot::try_new(&log_store, Default::default(), None).await?;
+        let provider = DeltaScan::new(snapshot, DeltaScanConfig::default())?;
+
+        let session = Arc::new(create_session().into_inner());
+        let state = session.state_ref().read().clone();
+
+        let deletion_vectors = provider.deletion_vectors(&state).await?;
+
+        assert!(deletion_vectors.is_empty());
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_deletion_vectors_with_eager_snapshot() -> TestResult {
+        let log_store = TestTables::WithDvSmall.table_builder()?.build_storage()?;
+        let eager = EagerSnapshot::try_new(&log_store, Default::default(), None).await?;
+        let provider = DeltaScan::new(eager, DeltaScanConfig::default())?;
+
+        let session = Arc::new(create_session().into_inner());
+        let state = session.state_ref().read().clone();
+        let expected = expected_dv_small()?;
+
+        let deletion_vectors = provider.deletion_vectors(&state).await?;
+        assert_eq!(deletion_vectors, expected);
+
+        Ok(())
+    }
 }

crates/core/src/delta_datafusion/table_provider/next/scan/mod.rs

@@ -60,6 +60,7 @@ use crate::{
         DeltaScanConfig,
         engine::{AsObjectStoreUrl as _, to_datafusion_scalar},
         file_id::wrap_file_id_value,
+        table_provider::next::DeletionVectorSelection,
     },
 };
 
@@ -126,6 +127,52 @@ pub(super) async fn execution_plan(
     .await
 }
 
+/// Materialize deletion vector keep masks for every file in the scan that has one.
+///
+/// Deletion vectors are loaded as a side-effect of consuming [`ScanFileStream`].  We drain the
+/// full stream here (discarding file contexts, stats, and partition values) because the DV
+/// loading tasks are spawned lazily during stream poll.  A dedicated DV-only stream that skips
+/// stats parsing is possible but not yet warranted — this path is not latency-sensitive and the
+/// file-list is typically small.
+///
+/// [`ReceiverStreamBuilder::build`] returns a merged stream that includes a JoinSet checker;
+/// `.try_collect().await` below will not complete until every spawned DV-loading task has
+/// finished, so no results are lost.
+pub(super) async fn replay_deletion_vectors(
+    engine: Arc<dyn Engine>,
+    scan_plan: &KernelScanPlan,
+    config: &DeltaScanConfig,
+    stream: ScanMetadataStream,
+) -> Result<Vec<DeletionVectorSelection>> {
+    let mut stream = ScanFileStream::new(engine, &scan_plan.scan, config.clone(), stream);
+    while stream.try_next().await?.is_some() {}
+
+    let dv_stream = stream.dv_stream.build();
+    // Only files with `dv_info.has_vector()` spawn tasks, so every item should carry a DV.
+    // Guard with a typed error (instead of panic) in case that invariant drifts.
+    let dvs: DashMap<_, _> = dv_stream
+        .and_then(|(url, dv)| {
+            ready(match dv {
+                Some(keep_mask) => Ok((url.to_string(), keep_mask)),
+                None => Err(DeltaTableError::generic(
+                    "Invariant violation: DV task spawned for file without deletion vector",
+                )),
+            })
+        })
+        .try_collect()
+        .await?;
+
+    let mut vectors: Vec<_> = dvs
+        .into_iter()
+        .map(|(filepath, keep_mask)| DeletionVectorSelection {
+            filepath,
+            keep_mask,
+        })
+        .collect();
+    vectors.sort_unstable_by(|left, right| left.filepath.cmp(&right.filepath));
+    Ok(vectors)
+}
+
 async fn replay_files(
     engine: Arc<dyn Engine>,
     scan_plan: &KernelScanPlan,

crates/core/src/delta_datafusion/table_provider/next/scan/replay.rs

@@ -186,6 +186,10 @@ where
                     Ok(ctx) => ctx,
                     Err(err) => return Poll::Ready(Some(Err(err.into()))),
                 };
+                ctx = match ctx.error_or() {
+                    Ok(ctx) => ctx,
+                    Err(err) => return Poll::Ready(Some(Err(err.into()))),
+                };
 
                 // Spawn tasks to read the deletion vectors from disk.
                 for file in &ctx.files {
@@ -416,8 +420,6 @@ pub(crate) struct ScanFileContext {
     pub file_url: Url,
     /// Size of the file on disk.
     pub size: u64,
-    /// Selection vector to filter the data in the file.
-    // pub selection_vector: Option<Vec<bool>>,
     /// Transformations to apply to the data in the file.
     pub transform: Option<ExpressionRef>,
     /// Statistics about the data in the file.
@@ -447,8 +449,6 @@ struct ScanFileContextInner {
     pub file_url: Url,
     /// Size of the file on disk.
     pub size: u64,
-    /// Selection vector to filter the data in the file.
-    // pub selection_vector: Option<Vec<bool>>,
     /// Transformations to apply to the data in the file.
     pub transform: Option<ExpressionRef>,
 
@@ -478,6 +478,22 @@ impl ScanContext {
     fn parse_path(&self, path: &str) -> DeltaResult<Url, DataFusionError> {
         parse_path(&self.table_root, path)
     }
+
+    fn error_or(self) -> DeltaResult<Self, DataFusionError> {
+        let ScanContext {
+            table_root,
+            files,
+            errs,
+            count,
+        } = self;
+        errs.error_or(())?;
+        Ok(ScanContext {
+            table_root,
+            files,
+            errs: DataFusionErrorBuilder::new(),
+            count,
+        })
+    }
 }
 
 fn parse_path(url: &Url, path: &str) -> DeltaResult<Url, DataFusionError> {
@@ -505,3 +521,45 @@ fn visit_scan_file(ctx: &mut ScanContext, scan_file: ScanFile) {
     });
     ctx.count += 1;
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use delta_kernel::scan::state::{DvInfo, ScanFile};
+    use url::Url;
+
+    use super::{ScanContext, visit_scan_file};
+
+    fn scan_file(path: impl Into<String>) -> ScanFile {
+        ScanFile {
+            path: path.into(),
+            size: 1,
+            modification_time: 0,
+            stats: None,
+            dv_info: DvInfo::default(),
+            transform: None,
+            partition_values: HashMap::new(),
+        }
+    }
+
+    #[test]
+    fn test_scan_context_error_or_returns_error_for_invalid_path() {
+        let mut ctx = ScanContext::new(Url::parse("mailto:delta@example.com").unwrap());
+        visit_scan_file(&mut ctx, scan_file("part-000.parquet"));
+        assert!(ctx.error_or().is_err());
+    }
+
+    #[test]
+    fn test_scan_context_error_or_keeps_valid_path() {
+        let mut ctx = ScanContext::new(Url::parse("file:///tmp/delta/").unwrap());
+        visit_scan_file(&mut ctx, scan_file("part-000.parquet"));
+
+        let ctx = ctx.error_or().unwrap();
+        assert_eq!(ctx.files.len(), 1);
+        assert_eq!(
+            ctx.files[0].file_url.as_str(),
+            "file:///tmp/delta/part-000.parquet"
+        );
+    }
+}

python/deltalake/_internal.pyi

@@ -257,6 +257,7 @@ class RawDeltaTable:
         ending_timestamp: str | None = None,
         allow_out_of_range: bool = False,
     ) -> RecordBatchReader: ...
+    def deletion_vectors(self) -> RecordBatchReader: ...
     def transaction_version(self, app_id: str) -> int | None: ...
     def set_column_metadata(
         self,

python/deltalake/table.py

@@ -413,6 +413,21 @@ def load_cdf(
             allow_out_of_range=allow_out_of_range,
         )
 
+    def deletion_vectors(self) -> RecordBatchReader:
+        """
+        Return deletion vectors for data files in this table.
+
+        Returns:
+            RecordBatchReader: A reader with two columns:
+                - filepath (utf8): fully-qualified file URI.
+                - selection_vector (list[bool]): row keep mask where True means keep and False means deleted.
+
+        Notes:
+            Only files that have deletion vectors are returned.
+            Deletion vectors are materialized in memory before being exposed as record batches.
+        """
+        return self._table.deletion_vectors()
+
     @property
     def table_uri(self) -> str:
         return self._table.table_uri()