feat: expose DV metadata and payloads as Arrow streams (delta-io#4168)

# Description
Adds `DeltaTable.deletion_vectors() -> RecordBatchReader` returning one
row per data file with a deletion vector.

Schema: `filepath: utf8`, `selection_vector: list[bool]` (true = keep,
false = deleted).

Reuses the existing DataFusion replay path via
`replay_deletion_vectors(...)`. Results are deterministic and sorted by
filepath.

**Core changes:**
- `DeletionVectorSelection` struct, `DeltaScan::deletion_vectors()`,
shared
  `scan_metadata_stream()` helper to avoid drift between scan paths
- Replaced internal DV `expect(...)` with typed error propagation

**Python binding:**
- `cloned_table_and_state()` to avoid TOCTOU on table + snapshot
- Chunked Arrow batch output with non-nullable list items
- Preserves `without_files` guard behavior

# Related Issue(s)
- Closes delta-io#4159
<!---
For example:

- closes #106
--->

# Documentation

<!---
Share links to useful documentation
--->

cc @ion-elgreco

---------

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,7 +39,7 @@ 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};
 use url::Url;
 
@@ -49,7 +49,7 @@ use crate::DeltaTableError;
 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;
 
@@ -268,6 +268,15 @@ pub struct DeltaScan {
     file_selection: Option<FileSelection>,
 }
 
+/// 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> {
@@ -302,6 +311,52 @@ impl DeltaScan {
         self.file_selection = Some(selection);
         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()
@@ -360,21 +415,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,
@@ -416,8 +457,8 @@ mod tests {
 
     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},
     };
 
@@ -860,4 +901,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

@@ -61,6 +61,7 @@ use crate::{
         DeltaScanConfig,
         engine::{AsObjectStoreUrl as _, to_datafusion_scalar},
         file_id::wrap_file_id_value,
+        table_provider::next::DeletionVectorSelection,
     },
 };
 
@@ -131,6 +132,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(), None, 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

@@ -37,7 +37,7 @@ use pin_project_lite::pin_project;
 use url::Url;
 
 use crate::{
-    DeltaResult,
+    DeltaResult, DeltaTableError,
     delta_datafusion::{DeltaScanConfig, engine::to_datafusion_scalar},
     kernel::{
         LogicalFileView, ReceiverStreamBuilder, Scan, StructDataExt,
@@ -195,8 +195,11 @@ where
                     scan_data
                 };
 
-                let mut ctx = ScanContext::new(this.table_root.clone());
-                ctx = match scan_data.visit_scan_files(ctx, visit_scan_file) {
+                let ctx = match scan_data
+                    .visit_scan_files(ScanContext::new(this.table_root.clone()), visit_scan_file)
+                    .map_err(|err| DataFusionError::from(DeltaTableError::from(err)))
+                    .and_then(ScanContext::error_or)
+                {
                     Ok(ctx) => ctx,
                     Err(err) => return Poll::Ready(Some(Err(err.into()))),
                 };
@@ -430,8 +433,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.
@@ -461,8 +462,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>,
 
@@ -492,6 +491,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> {
@@ -519,13 +534,11 @@ fn apply_file_selection(
             continue;
         }
 
-        let keep = parse_path(
+        let file_url = parse_path(
             table_root,
             LogicalFileView::new(batch.clone(), idx).path_raw(),
-        )
-        .map(|url| file_selection.contains(url.as_str()))
-        .unwrap_or(false);
-        selection_vector[idx] = keep;
+        )?;
+        selection_vector[idx] = file_selection.contains(file_url.as_str());
     }
 
     scan_data.scan_files =
@@ -550,3 +563,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,