perf: add skip_stats option to skip reading file statistics (delta-io#1738)

## What changes are proposed in this pull request?
Adds `skip_stats` option to `ScanBuilder` that skips reading the stats
column from checkpoint parquet files using column projection. Useful
when the caller handles data skipping externally or doesn't need file
statistics(e.g., a higher layer like
[delta-rs](https://github.com/delta-io/delta-rs)).

### Changes:
- **Add** `with_skip_stats(bool)` to `ScanBuilder`
- **Add** `CHECKPOINT_READ_SCHEMA_NO_STATS` schema without stats column
- **Transform** outputs null for stats when `skip_stats=true`
- **Disable** `data_skipping_filter` when stats are skipped

## Public API Changes
**New public API, non-breaking change:**
- `ScanBuilder::with_skip_stats(bool)`: opt-in method to skip reading
file statistics

## How was this change tested?
**Unit tests:**
- Added `test_scan_action_iter_with_skip_stats` to verify stats are null
when `skip_stats` is enabled

### Benchmark with synthetic table:
I did a quick local test wit these table characteristics:
- 50,000 files in a single parquet checkpoint
- 20 columns per file
- Stats JSON per file: ~1,000 bytes
- Checkpoint file size: ~20 MB   

**Results (5 iterations, average):**
| Mode | Time | Improvement |
| :--- | :--- | :--- |
| **With stats** | 37 ms | - |
| **Skip stats** | 8 ms | **76%** |

_The improvement comes from parquet column projection, which avoids I/O
and JSON deserialization of the stats column._

---------

Co-authored-by: Drake Lin <drakelin18@gmail.com>

Author: fvaleye

kernel/src/parallel/parallel_phase.rs

@@ -203,6 +203,7 @@ mod tests {
             state_info,
             checkpoint_info,
             seen_file_keys,
+            false,
         )
     }
 

kernel/src/scan/log_replay.rs

@@ -41,6 +41,8 @@ struct InternalScanState {
     physical_stats_schema: Option<SchemaRef>,
     /// Logical stats schema for the file statistics.
     logical_stats_schema: Option<SchemaRef>,
+    #[serde(default)]
+    skip_stats: bool,
 }
 
 /// Serializable processor state for distributed processing. This can be serialized using the
@@ -111,6 +113,8 @@ pub struct ScanLogReplayProcessor {
     /// far in the log. This is used to filter out files with Remove actions as
     /// well as duplicate entries in the log.
     seen_file_keys: HashSet<FileActionKey>,
+    /// Skip reading file statistics.
+    skip_stats: bool,
 }
 
 impl ScanLogReplayProcessor {
@@ -128,8 +132,15 @@ impl ScanLogReplayProcessor {
         engine: &dyn Engine,
         state_info: Arc<StateInfo>,
         checkpoint_info: CheckpointReadInfo,
+        skip_stats: bool,
     ) -> DeltaResult<Self> {
-        Self::new_with_seen_files(engine, state_info, checkpoint_info, Default::default())
+        Self::new_with_seen_files(
+            engine,
+            state_info,
+            checkpoint_info,
+            Default::default(),
+            skip_stats,
+        )
     }
 
     /// Create new [`ScanLogReplayProcessor`] with pre-populated seen_file_keys.
@@ -142,31 +153,42 @@ impl ScanLogReplayProcessor {
     /// - `state_info`: StateInfo containing schemas, transforms, and predicates
     /// - `checkpoint_info`: Information about checkpoint reading for stats optimization
     /// - `seen_file_keys`: Pre-computed set of file action keys that have been seen
+    /// - `skip_stats`: Skip reading file statistics
     pub(crate) fn new_with_seen_files(
         engine: &dyn Engine,
         state_info: Arc<StateInfo>,
         checkpoint_info: CheckpointReadInfo,
         seen_file_keys: HashSet<FileActionKey>,
+        skip_stats: bool,
     ) -> DeltaResult<Self> {
         let CheckpointReadInfo {
             has_stats_parsed,
             checkpoint_read_schema,
         } = checkpoint_info;
 
-        let output_schema =
-            scan_row_schema_with_stats_parsed(state_info.physical_stats_schema.clone());
-
         // Extract the physical predicate for data skipping and partition filtering.
         // DataSkippingFilter expects Option<(PredicateRef, SchemaRef)>.
         let physical_predicate = match &state_info.physical_predicate {
             PhysicalPredicate::Some(predicate, schema) => Some((predicate.clone(), schema.clone())),
             _ => None,
         };
 
+        // When skip_stats is enabled, use None for stats schema to produce null stats output
+        let stats_schema_for_transform = if skip_stats {
+            None
+        } else {
+            state_info.physical_stats_schema.clone()
+        };
+
+        let output_schema = scan_row_schema_with_stats_parsed(stats_schema_for_transform.clone());
+
         // Create data skipping filter that reads stats_parsed from the transformed batch.
         // This avoids double JSON parsing - the transform parses JSON once, then data skipping
         // reads the already-parsed stats_parsed column from the transform output.
-        let data_skipping_filter =
+        // Disabled when skip_stats is enabled.
+        let data_skipping_filter = if skip_stats {
+            None
+        } else {
             state_info
                 .physical_stats_schema
                 .as_ref()
@@ -179,25 +201,27 @@ impl ScanLogReplayProcessor {
                         output_schema.clone(),
                         column_expr_ref!("stats_parsed"),
                     )
-                });
+                })
+        };
 
         Ok(Self {
             partition_filter: physical_predicate.as_ref().map(|(p, _)| p.clone()),
             data_skipping_filter,
             // Log transform: always parse JSON (no stats_parsed in JSON commit files)
             log_transform: engine.evaluation_handler().new_expression_evaluator(
                 checkpoint_read_schema.clone(),
-                get_add_transform_expr(state_info.physical_stats_schema.clone(), false),
+                get_add_transform_expr(stats_schema_for_transform.clone(), false, skip_stats),
                 output_schema.clone().into(),
             )?,
             // Checkpoint transform: read stats_parsed directly when available, otherwise parse JSON
             checkpoint_transform: engine.evaluation_handler().new_expression_evaluator(
                 checkpoint_read_schema,
-                get_add_transform_expr(state_info.physical_stats_schema.clone(), has_stats_parsed),
+                get_add_transform_expr(stats_schema_for_transform, has_stats_parsed, skip_stats),
                 output_schema.into(),
             )?,
             seen_file_keys,
             state_info,
+            skip_stats,
         })
     }
 
@@ -245,6 +269,7 @@ impl ScanLogReplayProcessor {
             column_mapping_mode,
             physical_stats_schema,
             logical_stats_schema,
+            skip_stats: self.skip_stats,
         };
         let internal_state_blob = serde_json::to_vec(&internal_state)
             .map_err(|e| Error::generic(format!("Failed to serialize internal state: {}", e)))?;
@@ -308,6 +333,7 @@ impl ScanLogReplayProcessor {
             state_info,
             state.checkpoint_info,
             state.seen_file_keys,
+            internal_state.skip_stats,
         )?;
 
         Ok(Arc::new(processor))
@@ -542,19 +568,27 @@ fn scan_row_schema_with_stats_parsed(stats_schema: Option<SchemaRef>) -> SchemaR
 /// - `physical_stats_schema`: Schema for parsing stats from JSON and for output (physical column
 ///   names), or None if stats should not be included in output.
 /// - `has_stats_parsed`: Whether checkpoint has pre-parsed stats_parsed column.
+/// - `skip_stats`: When true, replaces the stats column with a null literal, avoiding reads of the
+///   raw stats JSON string from checkpoint parquet files.
 ///
 /// The transform includes `stats_parsed` only when `physical_stats_schema` is Some.
 /// Stats are output using physical column names. Engines can use `Scan::logical_stats_schema()`
 /// to map physical names back to logical names when column mapping is enabled.
 fn get_add_transform_expr(
     physical_stats_schema: Option<SchemaRef>,
     has_stats_parsed: bool,
+    skip_stats: bool,
 ) -> ExpressionRef {
+    let stats_expr = if skip_stats {
+        Arc::new(Expression::Literal(Scalar::Null(DataType::STRING)))
+    } else {
+        column_expr_ref!("add.stats")
+    };
     let mut fields = vec![
         column_expr_ref!("add.path"),
         column_expr_ref!("add.size"),
         column_expr_ref!("add.modificationTime"),
-        column_expr_ref!("add.stats"),
+        stats_expr,
         column_expr_ref!("add.deletionVector"),
         Arc::new(Expression::Struct(vec![
             column_expr_ref!("add.partitionValues"),
@@ -756,16 +790,20 @@ impl LogReplayProcessor for ScanLogReplayProcessor {
 /// that is selected in the returned `engine_data` _must_ be processed to complete the scan.
 /// Non-selected rows _must_ be ignored.
 ///
+/// When `skip_stats` is true, file statistics are not read from checkpoint parquet files and
+/// data skipping is disabled.
+///
 /// Note: The iterator of [`ActionsBatch`]s ('action_iter' parameter) must be sorted by the order of
 /// the actions in the log from most recent to least recent.
 pub(crate) fn scan_action_iter(
     engine: &dyn Engine,
     action_iter: impl Iterator<Item = DeltaResult<ActionsBatch>>,
     state_info: Arc<StateInfo>,
     checkpoint_info: CheckpointReadInfo,
+    skip_stats: bool,
 ) -> DeltaResult<impl Iterator<Item = DeltaResult<ScanMetadata>>> {
     Ok(
-        ScanLogReplayProcessor::new(engine, state_info, checkpoint_info)?
+        ScanLogReplayProcessor::new(engine, state_info, checkpoint_info, skip_stats)?
             .process_actions_iter(action_iter),
     )
 }
@@ -914,6 +952,7 @@ mod tests {
                 .map(|batch| Ok(ActionsBatch::new(batch as _, true))),
             state_info,
             test_checkpoint_info(),
+            false,
         )
         .unwrap();
         for res in iter {
@@ -941,6 +980,7 @@ mod tests {
                 .map(|batch| Ok(ActionsBatch::new(batch as _, true))),
             Arc::new(state_info),
             test_checkpoint_info(),
+            false,
         )
         .unwrap();
 
@@ -1025,6 +1065,7 @@ mod tests {
                 .map(|batch| Ok(ActionsBatch::new(batch as _, true))),
             Arc::new(state_info),
             test_checkpoint_info(),
+            false,
         )
         .unwrap();
 
@@ -1069,6 +1110,7 @@ mod tests {
             &engine,
             Arc::new(get_simple_state_info(schema.clone(), vec![]).unwrap()),
             checkpoint_info.clone(),
+            false,
         )
         .unwrap();
 
@@ -1136,9 +1178,13 @@ mod tests {
             _ => panic!("Expected predicate"),
         };
         let checkpoint_info = test_checkpoint_info();
-        let processor =
-            ScanLogReplayProcessor::new(&engine, state_info.clone(), checkpoint_info.clone())
-                .unwrap();
+        let processor = ScanLogReplayProcessor::new(
+            &engine,
+            state_info.clone(),
+            checkpoint_info.clone(),
+            false,
+        )
+        .unwrap();
         let deserialized = ScanLogReplayProcessor::from_serializable_state(
             &engine,
             processor.into_serializable_state(checkpoint_info).unwrap(),
@@ -1189,9 +1235,13 @@ mod tests {
         let original_transform = state_info.transform_spec.clone();
         assert!(original_transform.is_some());
         let checkpoint_info = test_checkpoint_info();
-        let processor =
-            ScanLogReplayProcessor::new(&engine, state_info.clone(), checkpoint_info.clone())
-                .unwrap();
+        let processor = ScanLogReplayProcessor::new(
+            &engine,
+            state_info.clone(),
+            checkpoint_info.clone(),
+            false,
+        )
+        .unwrap();
         let deserialized = ScanLogReplayProcessor::from_serializable_state(
             &engine,
             processor.into_serializable_state(checkpoint_info).unwrap(),
@@ -1225,7 +1275,8 @@ mod tests {
             });
             let checkpoint_info = test_checkpoint_info();
             let processor =
-                ScanLogReplayProcessor::new(&engine, state_info, checkpoint_info.clone()).unwrap();
+                ScanLogReplayProcessor::new(&engine, state_info, checkpoint_info.clone(), false)
+                    .unwrap();
             let deserialized = ScanLogReplayProcessor::from_serializable_state(
                 &engine,
                 processor.into_serializable_state(checkpoint_info).unwrap(),
@@ -1255,7 +1306,8 @@ mod tests {
             logical_stats_schema: None,
         });
         let processor =
-            ScanLogReplayProcessor::new(&engine, state_info, checkpoint_info.clone()).unwrap();
+            ScanLogReplayProcessor::new(&engine, state_info, checkpoint_info.clone(), false)
+                .unwrap();
         let serialized = processor.into_serializable_state(checkpoint_info).unwrap();
         assert!(serialized.predicate.is_none());
         let deserialized =
@@ -1296,6 +1348,7 @@ mod tests {
             column_mapping_mode: ColumnMappingMode::None,
             physical_stats_schema: None,
             logical_stats_schema: None,
+            skip_stats: false,
         };
         let predicate = Arc::new(crate::expressions::Predicate::column(["id"]));
         let invalid_blob = serde_json::to_vec(&invalid_internal_state).unwrap();
@@ -1327,6 +1380,7 @@ mod tests {
             column_mapping_mode: ColumnMappingMode::None,
             physical_stats_schema: None,
             logical_stats_schema: None,
+            skip_stats: false,
         };
         let blob = serde_json::to_string(&invalid_internal_state).unwrap();
         let mut obj: serde_json::Value = serde_json::from_str(&blob).unwrap();
@@ -1374,4 +1428,37 @@ mod tests {
         let result = serde_json::to_string(&state);
         assert_result_error_with_message(result, "Cannot serialize an Opaque Predicate");
     }
+
+    #[test]
+    fn test_scan_action_iter_with_skip_stats() {
+        let batch = vec![add_batch_simple(get_commit_schema().clone())];
+        let schema: SchemaRef = Arc::new(StructType::new_unchecked([
+            StructField::new("value", DataType::INTEGER, true),
+            StructField::new("date", DataType::DATE, true),
+        ]));
+        let state_info = get_simple_state_info(schema, vec!["date".to_string()]).unwrap();
+
+        let iter = scan_action_iter(
+            &SyncEngine::new(),
+            batch
+                .into_iter()
+                .map(|batch| Ok(ActionsBatch::new(batch as _, true))),
+            Arc::new(state_info),
+            test_checkpoint_info(),
+            true,
+        )
+        .unwrap();
+
+        let mut found_add = false;
+        for res in iter {
+            let scan_metadata = res.unwrap();
+            scan_metadata
+                .visit_scan_files((), |_: &mut (), scan_file: ScanFile| {
+                    assert!(scan_file.stats.is_none());
+                })
+                .unwrap();
+            found_add = true;
+        }
+        assert!(found_add);
+    }
 }