perf(core): wire kernel skip_stats for no-predicate file listing

Signed-off-by: Florian Valeye <florian.valeye@gmail.com>

Author: fvaleye

crates/core/src/kernel/snapshot/iterators/scan_row.rs

@@ -33,21 +33,27 @@ pin_project! {
         stats_schema: KernelSchemaRef,
         partitions_schema: Option<KernelSchemaRef>,
         column_mapping_mode: ColumnMappingMode,
+        skip_stats: bool,
 
         #[pin]
         stream: S,
     }
 }
 
 impl<S> ScanRowOutStream<S> {
-    pub fn try_new(snapshot: Arc<KernelSnapshot>, stream: S) -> DeltaResult<Self> {
+    pub fn try_new(
+        snapshot: Arc<KernelSnapshot>,
+        stream: S,
+        skip_stats: bool,
+    ) -> DeltaResult<Self> {
         let stats_schema = snapshot.stats_schema()?;
         let partitions_schema = snapshot.partitions_schema()?;
         let column_mapping_mode = snapshot.table_configuration().column_mapping_mode();
         Ok(Self {
             stats_schema,
             partitions_schema,
             column_mapping_mode,
+            skip_stats,
             stream,
         })
     }
@@ -68,6 +74,7 @@ where
                     this.stats_schema.clone(),
                     this.partitions_schema.as_ref(),
                     *this.column_mapping_mode,
+                    *this.skip_stats,
                 );
                 Poll::Ready(Some(result))
             }
@@ -121,6 +128,7 @@ pub(crate) fn parse_stats_column_with_schema(
         stats_schema,
         partitions_schema.as_ref(),
         column_mapping_mode,
+        false,
     )
 }
 
@@ -129,6 +137,7 @@ fn parse_stats_column_impl(
     stats_schema: KernelSchemaRef,
     partitions_schema: Option<&KernelSchemaRef>,
     column_mapping_mode: ColumnMappingMode,
+    skip_stats: bool,
 ) -> DeltaResult<RecordBatch> {
     let Some((stats_idx, _)) = batch.schema_ref().column_with_name("stats") else {
         return Err(DeltaTableError::SchemaMismatch {
@@ -139,13 +148,24 @@ fn parse_stats_column_impl(
     let mut columns = batch.columns().to_vec();
     let mut fields = batch.schema().fields().to_vec();
 
-    let stats_batch = batch.project(&[stats_idx])?;
-    let stats_data = Box::new(ArrowEngineData::new(stats_batch));
-
-    let parsed = parse_json(stats_data, stats_schema)?;
-    let parsed: RecordBatch = ArrowEngineData::try_from_engine_data(parsed)?.into();
-
-    let stats_array: Arc<StructArray> = Arc::new(parsed.into());
+    let stats_array: Arc<StructArray> = if skip_stats {
+        // `parse_json` on a null `stats` column still runs the full JSON
+        // machinery and produces `{}` structs, not nulls: cancels the
+        // skip_stats win. Build the fully-null `StructArray` directly instead.
+        let arrow_struct: arrow_schema::Schema = stats_schema.as_ref().try_into_arrow()?;
+        Arc::new(StructArray::new_null(
+            arrow_struct.fields().clone(),
+            batch.num_rows(),
+        ))
+    } else {
+        let stats_batch = batch.project(&[stats_idx])?;
+        let stats_data = Box::new(ArrowEngineData::new(stats_batch));
+
+        let parsed = parse_json(stats_data, stats_schema)?;
+        let parsed: RecordBatch = ArrowEngineData::try_from_engine_data(parsed)?.into();
+
+        Arc::new(parsed.into())
+    };
     fields[stats_idx] = Arc::new(Field::new(
         "stats_parsed",
         stats_array.data_type().to_owned(),

crates/core/src/kernel/snapshot/mod.rs

@@ -394,7 +394,14 @@ impl Snapshot {
         engine: Arc<dyn Engine>,
         predicate: Option<PredicateRef>,
     ) -> SendableRBStream {
-        let scan = match self.scan_builder().with_predicate(predicate).build() {
+        self.warn_if_skip_stats_with_predicate(&predicate);
+        let skip_stats = predicate.is_none() && self.config.skip_stats;
+        let scan = match self
+            .scan_builder()
+            .with_predicate(predicate)
+            .with_skip_stats(skip_stats)
+            .build()
+        {
             Ok(scan) => scan,
             Err(err) => return Box::pin(once(ready(Err(err)))),
         };
@@ -403,12 +410,23 @@ impl Snapshot {
             .scan_metadata(engine)
             .map(|d| Ok(rb_from_scan_meta(d?)?));
 
-        match ScanRowOutStream::try_new(self.inner.clone(), stream) {
+        match ScanRowOutStream::try_new(self.inner.clone(), stream, skip_stats) {
             Ok(s) => s.boxed(),
             Err(err) => Box::pin(once(ready(Err(err)))),
         }
     }
 
+    fn warn_if_skip_stats_with_predicate(&self, predicate: &Option<PredicateRef>) {
+        if self.config.skip_stats && predicate.is_some() {
+            tracing::warn!(
+                "`DeltaTable` was opened with `skip_stats=true`, but this query has \
+                 a predicate. Every file in the table will be scanned. To avoid \
+                 this, open a separate `DeltaTable` without `skip_stats=true` for \
+                 query workloads."
+            );
+        }
+    }
+
     pub(crate) fn files_from<T: Iterator<Item = RecordBatch> + Send + 'static>(
         &self,
         engine: Arc<dyn Engine>,
@@ -417,7 +435,14 @@ impl Snapshot {
         existing_data: Box<T>,
         existing_predicate: Option<PredicateRef>,
     ) -> SendableRBStream {
-        let scan = match self.scan_builder().with_predicate(predicate).build() {
+        self.warn_if_skip_stats_with_predicate(&predicate);
+        let skip_stats = predicate.is_none() && self.config.skip_stats;
+        let scan = match self
+            .scan_builder()
+            .with_predicate(predicate)
+            .with_skip_stats(skip_stats)
+            .build()
+        {
             Ok(scan) => scan,
             Err(err) => return Box::pin(once(ready(Err(err)))),
         };
@@ -426,7 +451,7 @@ impl Snapshot {
             .scan_metadata_from(engine, existing_version, existing_data, existing_predicate)
             .map(|d| Ok(rb_from_scan_meta(d?)?));
 
-        match ScanRowOutStream::try_new(self.inner.clone(), stream) {
+        match ScanRowOutStream::try_new(self.inner.clone(), stream, skip_stats) {
             Ok(s) => s.boxed(),
             Err(err) => Box::pin(once(ready(Err(err)))),
         }
@@ -1046,7 +1071,7 @@ mod tests {
     // use super::replay::tests::test_log_replay;
     use super::*;
     use crate::{
-        DeltaTable, checkpoints,
+        DeltaTable, DeltaTableConfig, checkpoints,
         kernel::transaction::CommitData,
         kernel::transaction::{CommitBuilder, TableReference},
         kernel::{Action, DataType, PrimitiveType, StructField, StructType},
@@ -1432,6 +1457,82 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_file_views_skip_stats_same_paths() -> TestResult {
+        let base = TestTables::Checkpoints.table_builder()?.build_storage()?;
+        let mut skip_cfg = DeltaTableConfig::default();
+        skip_cfg.skip_stats = true;
+        let with_skip = EagerSnapshot::try_new(base.as_ref(), skip_cfg, Some(12)).await?;
+        let full = EagerSnapshot::try_new(base.as_ref(), Default::default(), Some(12)).await?;
+        let mut paths_skip: Vec<String> = with_skip
+            .file_views(base.as_ref(), None)
+            .map_ok(|v| v.path().to_string())
+            .try_collect()
+            .await?;
+        let mut paths_full: Vec<String> = full
+            .file_views(base.as_ref(), None)
+            .map_ok(|v| v.path().to_string())
+            .try_collect()
+            .await?;
+        paths_skip.sort();
+        paths_full.sort();
+        assert_eq!(paths_skip, paths_full);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_skip_stats_leaves_stats_parsed_null() -> TestResult {
+        let base = TestTables::Checkpoints.table_builder()?.build_storage()?;
+
+        let default_eager =
+            EagerSnapshot::try_new(base.as_ref(), Default::default(), Some(12)).await?;
+        let default_stats: Vec<bool> = default_eager
+            .file_views(base.as_ref(), None)
+            .map_ok(|view| view.stats().is_some())
+            .try_collect()
+            .await?;
+        assert!(!default_stats.is_empty());
+        assert!(default_stats.iter().any(|b| *b));
+
+        let mut skip_cfg = DeltaTableConfig::default();
+        skip_cfg.skip_stats = true;
+        let skip_eager = EagerSnapshot::try_new(base.as_ref(), skip_cfg, Some(12)).await?;
+        let skip_stats: Vec<Option<String>> = skip_eager
+            .file_views(base.as_ref(), None)
+            .map_ok(|view| view.stats())
+            .try_collect()
+            .await?;
+        assert!(!skip_stats.is_empty());
+        assert!(skip_stats.iter().all(|s| s.is_none()));
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_skip_stats_bypassed_when_predicate_present() -> TestResult {
+        use delta_kernel::expressions::Scalar;
+
+        let base = TestTables::Checkpoints.table_builder()?.build_storage()?;
+
+        let mut skip_cfg = DeltaTableConfig::default();
+        skip_cfg.skip_stats = true;
+        let snapshot = Snapshot::try_new(base.as_ref(), skip_cfg, Some(12)).await?;
+
+        let predicate: PredicateRef =
+            Arc::new(Expression::column(["value"]).gt(Scalar::String("".to_string())));
+
+        let has_stats: Vec<bool> = snapshot
+            .file_views(base.as_ref(), Some(predicate))
+            .map_ok(|view| view.stats().is_some())
+            .try_collect()
+            .await?;
+
+        assert!(!has_stats.is_empty());
+        assert!(has_stats.iter().any(|b| *b));
+
+        Ok(())
+    }
+
     #[test]
     fn test_materialized_files_full_table_seed_shares_batches() {
         let batch = RecordBatch::new_empty(Arc::new(arrow_schema::Schema::empty()));

crates/core/src/kernel/snapshot/scan.rs

@@ -65,6 +65,17 @@ impl ScanBuilder {
         self
     }
 
+    /// Skip parsing file-level statistics during kernel log replay.
+    ///
+    /// When `true`, per-file min/max/null stats are not parsed; `stats_parsed` in scan
+    /// output may be null. Partition-based filtering still applies. When combined with a
+    /// non-empty predicate, the kernel cannot use stats for data skipping; prefer `false`
+    /// when you need predicate-based file pruning from statistics.
+    pub fn with_skip_stats(mut self, skip_stats: bool) -> Self {
+        self.inner = self.inner.with_skip_stats(skip_stats);
+        self
+    }
+
     pub fn build(self) -> DeltaResult<Scan> {
         Ok(Scan::from(self.inner.build()?))
     }

crates/core/src/table/builder.rs

@@ -53,6 +53,15 @@ pub struct DeltaTableConfig {
     /// when processing record batches.
     pub log_batch_size: usize,
 
+    /// Skip parsing per-file statistics while opening the table.
+    /// This defaults to `false`.
+    ///
+    /// Use for workflows that never need file pruning (vacuum, filesystem check,
+    /// append-only writes). Any predicated query on this instance will scan every
+    /// file because the cache has no stats. Partition pruning is unaffected.
+    #[serde(default)]
+    pub skip_stats: bool,
+
     #[serde(skip_serializing, skip_deserializing)]
     #[delta(skip)]
     /// When a runtime handler is provided, all IO tasks are spawn in that handle
@@ -65,6 +74,7 @@ impl Default for DeltaTableConfig {
             require_files: true,
             log_buffer_size: num_cpus::get() * 4,
             log_batch_size: 1024,
+            skip_stats: false,
             io_runtime: None,
         }
     }
@@ -75,6 +85,7 @@ impl PartialEq for DeltaTableConfig {
         self.require_files == other.require_files
             && self.log_buffer_size == other.log_buffer_size
             && self.log_batch_size == other.log_batch_size
+            && self.skip_stats == other.skip_stats
     }
 }
 
@@ -129,6 +140,13 @@ impl DeltaTableBuilder {
         self
     }
 
+    /// Sets `skip_stats` to the builder. See [`DeltaTableConfig::skip_stats`]
+    /// for the impact on predicated queries.
+    pub fn with_skip_stats(mut self, skip_stats: bool) -> Self {
+        self.table_config.skip_stats = skip_stats;
+        self
+    }
+
     /// Sets `version` to the builder
     pub fn with_version(mut self, version: Version) -> Self {
         self.version = DeltaVersion::Version(version);