chore: harden scan adapter caching and DV mask edge cases (delta-io#4199)

# Description
Follow up to the DF scan adapter integration.

Adds a cached schema adapter path in scan execution to avoid rebuilding
adapters for repeated source schemas. Updates scan finalization to use
the cached path.

Fixes DV mask handling edge cases with perserving remainder across multi
batch file reads, pads short masks with implicit true, errors when mask
length exceeds batch row count.

# Related Issue(s)
<!---
For example:

- closes #106
--->

# Documentation

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

---------

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

Author: ethan-tyler

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

@@ -43,6 +43,18 @@ pub(crate) struct DvMaskResult {
     pub should_remove: bool,
 }
 
+/// Consume the per-file deletion-vector keep-mask for the current batch.
+///
+/// The keep-mask is stored once per file and consumed incrementally as parquet
+/// batches are produced:
+/// - If the mask is shorter than the batch, missing trailing entries are
+///   treated as `true` (keep row).
+/// - If the mask is longer than the batch, the remainder is preserved for the
+///   next batch from the same file.
+///
+/// This function intentionally does not error when `selection_vector.len()` is
+/// greater than `batch_num_rows`; that is expected when one file spans multiple
+/// input batches.
 pub(crate) fn consume_dv_mask(
     selection_vector: &mut Vec<bool>,
     batch_num_rows: usize,
@@ -282,6 +294,7 @@ impl ExecutionPlan for DeltaScanExec {
             file_id_column: self.file_id_column.clone(),
             return_file_ids: self.retain_file_ids,
             pending: VecDeque::new(),
+            schema_adapter: super::SchemaAdapter::new(Arc::clone(&self.scan_plan.result_schema)),
         }))
     }
 
@@ -360,10 +373,12 @@ struct DeltaScanStream {
     /// Denotes if file ids should be returned as part of the output
     return_file_ids: bool,
     pending: VecDeque<RecordBatch>,
+    /// Cached schema adapter for efficient batch adaptation across batches
+    schema_adapter: super::SchemaAdapter,
 }
 
 impl DeltaScanStream {
-    fn batch_project(&self, batch: RecordBatch) -> Result<Vec<RecordBatch>> {
+    fn batch_project(&mut self, batch: RecordBatch) -> Result<Vec<RecordBatch>> {
         let _timer = self.baseline_metrics.elapsed_compute().timer();
 
         if batch.num_rows() == 0 {
@@ -377,18 +392,34 @@ impl DeltaScanStream {
 
         file_runs
             .into_iter()
-            .map(|(file_id, slice)| self.batch_project_single_file(slice, file_id, file_id_idx))
+            .map(|(file_id, slice)| {
+                Self::batch_project_single_file_inner(
+                    &self.scan_plan,
+                    &self.kernel_type,
+                    &self.transforms,
+                    &self.selection_vectors,
+                    self.return_file_ids,
+                    &mut self.schema_adapter,
+                    slice,
+                    file_id,
+                    file_id_idx,
+                )
+            })
             .collect::<Result<Vec<_>>>()
     }
 
-    fn batch_project_single_file(
-        &self,
+    fn batch_project_single_file_inner(
+        scan_plan: &KernelScanPlan,
+        kernel_type: &KernelDataType,
+        transforms: &HashMap<String, ExpressionRef>,
+        selection_vectors: &DashMap<String, Vec<bool>>,
+        return_file_ids: bool,
+        schema_adapter: &mut super::SchemaAdapter,
         batch: RecordBatch,
         file_id: String,
         file_id_idx: usize,
     ) -> Result<RecordBatch> {
-        let dv_result = if let Some(mut selection_vector) = self.selection_vectors.get_mut(&file_id)
-        {
+        let dv_result = if let Some(mut selection_vector) = selection_vectors.get_mut(&file_id) {
             consume_dv_mask(&mut selection_vector, batch.num_rows())
         } else {
             DvMaskResult {
@@ -398,7 +429,7 @@ impl DeltaScanStream {
         };
 
         if dv_result.should_remove {
-            self.selection_vectors.remove(&file_id);
+            selection_vectors.remove(&file_id);
         }
 
         let mut batch = if let Some(selection) = dv_result.selection {
@@ -410,12 +441,12 @@ impl DeltaScanStream {
         let file_id_field = batch.schema_ref().field(file_id_idx).clone();
         let file_id_col = batch.remove_column(file_id_idx);
 
-        let result = if let Some(transform) = self.transforms.get(&file_id) {
+        let result = if let Some(transform) = transforms.get(&file_id) {
             let evaluator = ARROW_HANDLER
                 .new_expression_evaluator(
-                    self.scan_plan.scan.physical_schema().clone(),
+                    scan_plan.scan.physical_schema().clone(),
                     transform.clone(),
-                    self.kernel_type.clone(),
+                    kernel_type.clone(),
                 )
                 .map_err(|e| DataFusionError::External(Box::new(e)))?;
 
@@ -426,14 +457,15 @@ impl DeltaScanStream {
             batch
         };
 
-        if self.return_file_ids {
+        if return_file_ids {
             super::finalize_transformed_batch(
                 result,
-                &self.scan_plan,
+                scan_plan,
                 Some((file_id_col, Arc::new(file_id_field))),
+                schema_adapter,
             )
         } else {
-            super::finalize_transformed_batch(result, &self.scan_plan, None)
+            super::finalize_transformed_batch(result, scan_plan, None, schema_adapter)
         }
     }
 }
@@ -1162,6 +1194,7 @@ mod tests {
             futures::stream::iter(input_batches.into_iter().map(Ok)),
         ));
 
+        let schema_adapter = super::super::SchemaAdapter::new(Arc::clone(&scan_plan.result_schema));
         DeltaScanStream {
             scan_plan,
             kernel_type,
@@ -1172,6 +1205,7 @@ mod tests {
             file_id_column: FILE_ID_COLUMN_DEFAULT.to_string(),
             return_file_ids,
             pending: VecDeque::new(),
+            schema_adapter,
         }
     }
 
@@ -1194,7 +1228,7 @@ mod tests {
         assert!(selection_vectors.contains_key("f1"));
         assert!(selection_vectors.contains_key("f2"));
 
-        let stream = test_scan_stream(
+        let mut stream = test_scan_stream(
             Arc::clone(&scan_plan),
             kernel_type,
             selection_vectors,
@@ -1265,7 +1299,7 @@ mod tests {
             false,
         )?;
 
-        let stream = test_scan_stream(
+        let mut stream = test_scan_stream(
             Arc::clone(&scan_plan),
             kernel_type,
             selection_vectors,
@@ -1478,4 +1512,21 @@ mod tests {
             }
         );
     }
+
+    #[test]
+    fn test_dv_long_mask_retains_remainder_for_next_batch() {
+        use super::{DvMaskResult, consume_dv_mask};
+
+        let mut sv = vec![true, false, false, true];
+        let result = consume_dv_mask(&mut sv, 2);
+
+        assert_eq!(
+            result,
+            DvMaskResult {
+                selection: Some(vec![true, false]),
+                should_remove: false,
+            }
+        );
+        assert_eq!(sv, vec![false, true]);
+    }
 }

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

@@ -27,14 +27,17 @@ use datafusion::physical_plan::execution_plan::{
     Boundedness, CardinalityEffect, EmissionType, PlanProperties,
 };
 use datafusion::physical_plan::filter_pushdown::{FilterDescription, FilterPushdownPhase};
-use datafusion::physical_plan::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
+use datafusion::physical_plan::metrics::{
+    BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricsSet,
+};
 use datafusion::physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PhysicalExpr, Statistics,
 };
 use delta_kernel::schema::{Schema as KernelSchema, SchemaRef as KernelSchemaRef};
 use delta_kernel::{EvaluationHandler, ExpressionRef};
 use futures::stream::Stream;
 use itertools::Itertools as _;
+use tracing::debug;
 
 use crate::delta_datafusion::table_provider::next::KernelScanPlan;
 use crate::kernel::ARROW_HANDLER;
@@ -191,9 +194,12 @@ impl ExecutionPlan for DeltaScanMetaExec {
             scan_plan: Arc::clone(&self.scan_plan),
             input: self.input[partition].clone(),
             baseline_metrics: BaselineMetrics::new(&self.metrics, partition),
+            dv_short_mask_padded_files_total: MetricBuilder::new(&self.metrics)
+                .counter("dv_short_mask_padded_files_total", partition),
             transforms: Arc::clone(&self.transforms),
             selection_vectors: Arc::clone(&self.selection_vectors),
             file_id_field: self.file_id_field.clone(),
+            schema_adapter: super::SchemaAdapter::new(Arc::clone(&self.scan_plan.result_schema)),
         }))
     }
 
@@ -258,17 +264,21 @@ struct DeltaScanMetaStream {
     input: VecDeque<(String, usize)>,
     /// Execution metrics
     baseline_metrics: BaselineMetrics,
+    /// Count of file batches where short deletion-vector masks required padding.
+    dv_short_mask_padded_files_total: Count,
     /// Transforms to be applied to data read from individual files
     transforms: Arc<HashMap<String, ExpressionRef>>,
     /// Selection vectors to be applied to data read from individual files
     selection_vectors: Arc<DashMap<String, Vec<bool>>>,
     /// Column name for the file id
     file_id_field: Option<FieldRef>,
+    /// Cached schema adapter for efficient batch adaptation across batches
+    schema_adapter: super::SchemaAdapter,
 }
 
 impl DeltaScanMetaStream {
     /// Apply the per-file transformation to a RecordBatch.
-    fn batch_project(&self, file_id: String, row_count: usize) -> Result<RecordBatch> {
+    fn batch_project(&mut self, file_id: String, row_count: usize) -> Result<RecordBatch> {
         static EMPTY_SCHEMA: LazyLock<SchemaRef> =
             LazyLock::new(|| Arc::new(Schema::new(Fields::empty())));
         static EMPTY_KERNEL_SCHEMA: LazyLock<KernelSchemaRef> =
@@ -283,17 +293,19 @@ impl DeltaScanMetaStream {
         )?;
 
         let batch = if let Some(selection) = self.selection_vectors.get(&file_id) {
-            if selection.len() != batch.num_rows() {
-                return Err(internal_datafusion_err!(
-                    "Selection vector length ({}) does not match row count ({}) for file '{}'. \
-                     This indicates a bug in deletion vector processing.",
-                    selection.len(),
-                    batch.num_rows(),
-                    file_id
-                ));
+            let mask_len = selection.len();
+            let (batch, padded_rows) = apply_selection_vector(batch, selection.value(), &file_id)?;
+            if padded_rows > 0 {
+                self.dv_short_mask_padded_files_total.add(1);
+                debug!(
+                    file_id = file_id.as_str(),
+                    mask_len,
+                    row_count,
+                    padded_rows,
+                    "Padded short deletion-vector keep-mask in metadata scan"
+                );
             }
-            let filter = BooleanArray::from_iter(selection.iter());
-            filter_record_batch(&batch, &filter)?
+            batch
         } else {
             batch
         };
@@ -343,13 +355,46 @@ impl DeltaScanMetaStream {
                 result,
                 &self.scan_plan,
                 Some((Arc::new(file_id_array), file_id_field.clone())),
+                &mut self.schema_adapter,
             )
         } else {
-            super::finalize_transformed_batch(result, &self.scan_plan, None)
+            super::finalize_transformed_batch(
+                result,
+                &self.scan_plan,
+                None,
+                &mut self.schema_adapter,
+            )
         }
     }
 }
 
+fn apply_selection_vector(
+    batch: RecordBatch,
+    selection: &[bool],
+    file_id: &str,
+) -> Result<(RecordBatch, usize)> {
+    if selection.len() > batch.num_rows() {
+        return Err(internal_datafusion_err!(
+            "Selection vector length ({}) exceeds row count ({}) for file '{}'. \
+             This indicates a bug in deletion vector processing.",
+            selection.len(),
+            batch.num_rows(),
+            file_id
+        ));
+    }
+
+    let n_rows_to_pad = batch.num_rows() - selection.len();
+    // Delta Kernel may emit short keep-masks; missing trailing entries are
+    // implicitly `true` (row is kept).
+    let filter = BooleanArray::from_iter(
+        selection
+            .iter()
+            .copied()
+            .chain(std::iter::repeat_n(true, n_rows_to_pad)),
+    );
+    Ok((filter_record_batch(&batch, &filter)?, n_rows_to_pad))
+}
+
 impl Stream for DeltaScanMetaStream {
     type Item = Result<RecordBatch>;
 
@@ -377,6 +422,9 @@ impl RecordBatchStream for DeltaScanMetaStream {
 mod tests {
     use std::sync::Arc;
 
+    use arrow::array::RecordBatch;
+    use arrow_array::RecordBatchOptions;
+    use arrow_schema::{Fields, Schema};
     use datafusion::{
         physical_plan::collect_partitioned,
         prelude::{col, lit},
@@ -389,6 +437,54 @@ mod tests {
         test_utils::{TestResult, open_fs_path},
     };
 
+    #[test]
+    fn test_apply_selection_vector_short_mask_pads_with_true() {
+        let batch = RecordBatch::try_new_with_options(
+            Arc::new(Schema::new(Fields::empty())),
+            vec![],
+            &RecordBatchOptions::new().with_row_count(Some(4)),
+        )
+        .unwrap();
+
+        // Short masks are valid: missing trailing entries are implicitly true.
+        let (filtered, padded_rows) =
+            apply_selection_vector(batch, &[false, true], "file:///f.parquet").unwrap();
+        assert_eq!(filtered.num_rows(), 3);
+        assert_eq!(padded_rows, 2);
+    }
+
+    #[test]
+    fn test_apply_selection_vector_no_padding_when_lengths_match() {
+        let batch = RecordBatch::try_new_with_options(
+            Arc::new(Schema::new(Fields::empty())),
+            vec![],
+            &RecordBatchOptions::new().with_row_count(Some(2)),
+        )
+        .unwrap();
+
+        let (filtered, padded_rows) =
+            apply_selection_vector(batch, &[true, false], "file:///f.parquet").unwrap();
+        assert_eq!(filtered.num_rows(), 1);
+        assert_eq!(padded_rows, 0);
+    }
+
+    #[test]
+    fn test_apply_selection_vector_longer_than_batch_errors() {
+        let batch = RecordBatch::try_new_with_options(
+            Arc::new(Schema::new(Fields::empty())),
+            vec![],
+            &RecordBatchOptions::new().with_row_count(Some(2)),
+        )
+        .unwrap();
+
+        let err = apply_selection_vector(batch, &[true, true, true], "file:///f.parquet")
+            .expect_err("selection vector longer than row count must error");
+        assert!(
+            err.to_string().contains("Selection vector length"),
+            "unexpected error: {err}"
+        );
+    }
+
     #[tokio::test]
     async fn test_meta_only_scan() -> TestResult {
         let table = open_fs_path("../../dat/v0.0.3/reader_tests/generated/multi_partitioned/delta");