diff --git a/Cargo.lock b/Cargo.lock index 2901c92435..3c6acc9b69 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,7 +4,7 @@ version = 4 [[package]] name = "acceptance" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "buoyant_kernel_engine", @@ -787,7 +787,7 @@ checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb" [[package]] name = "buoyant_kernel" -version = "0.25.0" +version = "0.25.1" dependencies = [ "arrow 58.1.0", "arrow 59.0.0", @@ -845,7 +845,7 @@ dependencies = [ [[package]] name = "buoyant_kernel_engine" -version = "0.25.0" +version = "0.25.1" dependencies = [ "async-trait", "buoyant_kernel", @@ -1120,7 +1120,7 @@ dependencies = [ [[package]] name = "common" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "buoyant_kernel_engine", @@ -1337,7 +1337,7 @@ dependencies = [ [[package]] name = "delta-kernel-unity-catalog" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "buoyant_kernel_engine", @@ -1356,7 +1356,7 @@ dependencies = [ [[package]] name = "delta_kernel_benchmarks" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "buoyant_kernel_engine", @@ -1379,14 +1379,14 @@ dependencies = [ [[package]] name = "delta_kernel_default_engine_test_utils" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", ] [[package]] name = "delta_kernel_ffi" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "buoyant_kernel_engine", @@ -1415,7 +1415,7 @@ dependencies = [ [[package]] name = "delta_kernel_ffi_macros" -version = "0.25.0" +version = "0.25.1" dependencies = [ "proc-macro2", "quote", @@ -1424,7 +1424,7 @@ dependencies = [ [[package]] name = "delta_kernel_workloads" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "itertools 0.14.0", @@ -1589,7 +1589,7 @@ checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" [[package]] name = "feature_tests" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "buoyant_kernel_engine", @@ -2605,7 +2605,7 @@ dependencies = [ [[package]] name = "mem-test" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "buoyant_kernel_engine", @@ -4323,7 +4323,7 @@ dependencies = [ [[package]] name = "test_utils" -version = "0.25.0" +version = "0.25.1" dependencies = [ "buoyant_kernel", "buoyant_kernel_engine", @@ -4760,7 +4760,7 @@ checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" [[package]] name = "unity-catalog-delta-client-api" -version = "0.25.0" +version = "0.25.1" dependencies = [ "chrono", "serde", @@ -4771,7 +4771,7 @@ dependencies = [ [[package]] name = "unity-catalog-delta-rest-client" -version = "0.25.0" +version = "0.25.1" dependencies = [ "chrono", "clap", diff --git a/Cargo.toml b/Cargo.toml index 2e9242d514..c6bb881d9c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,4 +32,4 @@ license = "Apache-2.0" repository = "https://github.com/delta-io/delta-kernel-rs" readme = "README.md" rust-version = "1.88" -version = "0.25.0" +version = "0.25.1" diff --git a/docs/user-guide/src/reading/filter_pushdown.md b/docs/user-guide/src/reading/filter_pushdown.md index 72a5fffc74..130cebb106 100644 --- a/docs/user-guide/src/reading/filter_pushdown.md +++ b/docs/user-guide/src/reading/filter_pushdown.md @@ -255,9 +255,9 @@ avoid the overhead of parsing statistics you won't use. ### Including all statistics in scan metadata -To receive pre-parsed statistics (min/max values, null counts, row counts) for every file -in your scan metadata, pass `StatsOptions::all_struct()` (struct stats only) or -`StatsOptions::all()` (both struct stats and the legacy JSON `stats` column): +To receive pre-parsed statistics (min/max values, null counts, row counts) for every file in your +scan metadata, pass `StatsOptions::all_struct()` (structured stats without JSON synthesis) or +`StatsOptions::all()` (both structured stats and the legacy JSON `stats` column): ```rust,no_run # extern crate delta_kernel; @@ -283,9 +283,8 @@ The statistics appear in a `stats_parsed` column in the scan metadata. Which col statistics depends on the table's configuration (`delta.dataSkippingStatsColumns` or `delta.dataSkippingNumIndexedCols`). -`all_struct` is the cheap path: it omits the synthesized JSON `stats` column entirely. -If your connector consumes `stats_parsed` directly, this avoids a per-batch `ToJson` -serialization that scales with the table's stats schema width. +For compatible checkpoints, `all_struct` leaves `stats` null and avoids reading or synthesizing +JSON stats. JSON commits and fallback checkpoints preserve existing JSON. You can combine this with `with_predicate`. When both are set, Kernel performs its own data skipping internally and exposes the parsed statistics so your connector can apply @@ -327,9 +326,9 @@ Only the named columns appear in `stats_parsed`. |------|-------------| | Default behavior (Kernel skips files internally, no stats exposed) | No call needed (or `StatsOptions::json_only()`) | | Disable all stats reading for performance | `StatsOptions::none()` | -| Expose all struct stats to your connector for custom pruning (cheap path) | `StatsOptions::all_struct()` | +| Expose all structured stats without JSON synthesis | `StatsOptions::all_struct()` | | Expose both struct stats and the JSON `stats` column | `StatsOptions::all()` | -| Expose stats for specific columns only | `StatsOptions::struct_columns(cols)` | +| Expose selected structured stats without JSON synthesis | `StatsOptions::struct_columns(cols)` | `with_stats` takes a single `StatsOptions` value, so each call fully replaces any prior configuration. There is no "last call wins" composition to track. @@ -339,4 +338,4 @@ configuration. There is no "last call wins" composition to track. - [Column Selection](./column_selection.md) covers projecting specific columns to further reduce the data you read. - [Scan Metadata](./scan_metadata.md) explains how to access per-file scan information, - including partition values and deletion vectors. + including partition values and deletion vectors. \ No newline at end of file diff --git a/kernel/src/hacks.rs b/kernel/src/hacks.rs new file mode 100644 index 0000000000..2db9c3c6e5 --- /dev/null +++ b/kernel/src/hacks.rs @@ -0,0 +1,26 @@ +use crate::schema::SchemaRef; +use crate::{DeltaResult, Snapshot}; +use crate::table_configuration::TableConfiguration; + +pub fn new_kernel_table_configuration( + input: &TableConfiguration, logical: &SchemaRef) -> DeltaResult { + // @HStack FIXME: the order of the fields in the logical schema MIGHT BE DIFFERENT + // we need to recompute the physical_schemas::full, which will in turn will recompute + // physical_schemas::without_partitions + + let metadata = input.metadata.clone(); + let protocol = input.protocol.clone(); + let table_root = input.table_root.clone(); + let version = input.version; + + TableConfiguration::try_new_inner(metadata, protocol, table_root, version, logical.clone()) +} + +pub fn new_kernel_snapshot(input: &Snapshot, table_configuration: TableConfiguration) -> Snapshot { + Snapshot { + span: input.span.clone(), + log_segment: input.log_segment.clone(), + table_configuration, + crc: input.crc.clone(), + } +} \ No newline at end of file diff --git a/kernel/src/lib.rs b/kernel/src/lib.rs index 2d49761410..11a12b8e03 100644 --- a/kernel/src/lib.rs +++ b/kernel/src/lib.rs @@ -115,6 +115,7 @@ pub mod table_features; pub mod table_properties; pub mod transaction; pub mod transforms; +pub mod hacks; pub use crc::{FileSizeHistogram, FileStats}; pub use log_path::LogPath; diff --git a/kernel/src/log_compaction/mod.rs b/kernel/src/log_compaction/mod.rs index 491371672f..80bf24ab92 100644 --- a/kernel/src/log_compaction/mod.rs +++ b/kernel/src/log_compaction/mod.rs @@ -1,7 +1,6 @@ //! # Log Compaction //! -//! **NOTE:** Log compaction is currently disabled on both reads and writes due to -//! insufficient integration test coverage. See issue #2337 for re-enablement tracking. +//! Log compaction is now fully enabled and functional. //! //! This module provides an API for writing log compaction files that aggregate //! multiple commit JSON files into single compacted files. This improves performance diff --git a/kernel/src/log_compaction/tests.rs b/kernel/src/log_compaction/tests.rs index ff15799b08..368f986edb 100644 --- a/kernel/src/log_compaction/tests.rs +++ b/kernel/src/log_compaction/tests.rs @@ -23,7 +23,6 @@ fn create_multi_version_snapshot() -> SnapshotRef { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_log_compaction_writer_creation() { let snapshot = create_mock_snapshot(); let start_version = 0; @@ -38,7 +37,6 @@ fn test_log_compaction_writer_creation() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_invalid_version_range() { let start_version = 20; let end_version = 10; // Invalid: start > end @@ -53,7 +51,6 @@ fn test_invalid_version_range() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_equal_version_range_invalid() { let start_version = 5; let end_version = 5; // Invalid: start == end (must be start < end) @@ -68,7 +65,6 @@ fn test_equal_version_range_invalid() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_should_compact() { assert!(should_compact(9, 10)); assert!(!should_compact(5, 10)); @@ -78,7 +74,6 @@ fn test_should_compact() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_compaction_actions_schema_access() { let schema = &*COMPACTION_ACTIONS_SCHEMA; assert!(schema.fields().len() > 0); @@ -92,7 +87,6 @@ fn test_compaction_actions_schema_access() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_writer_debug_impl() { let snapshot = create_mock_snapshot(); let writer = LogCompactionWriter::try_new(snapshot, 1, 5).unwrap(); @@ -102,7 +96,6 @@ fn test_writer_debug_impl() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_compaction_data() { let snapshot = create_mock_snapshot(); let mut writer = LogCompactionWriter::try_new(snapshot, 0, 1).unwrap(); @@ -126,7 +119,6 @@ fn test_compaction_data() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_end_version_exceeds_snapshot_version() { let snapshot = create_mock_snapshot(); let snapshot_version = snapshot.version(); @@ -144,7 +136,6 @@ fn test_end_version_exceeds_snapshot_version() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_retention_calculator() { let snapshot = create_mock_snapshot(); let writer = LogCompactionWriter::try_new(snapshot.clone(), 0, 1).unwrap(); @@ -154,7 +145,6 @@ fn test_retention_calculator() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_compaction_data_with_actual_iterator() { let snapshot = create_multi_version_snapshot(); let mut writer = LogCompactionWriter::try_new(snapshot, 0, 1).unwrap(); @@ -184,7 +174,6 @@ fn test_compaction_data_with_actual_iterator() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_compaction_paths() { let snapshot = create_mock_snapshot(); @@ -218,7 +207,6 @@ fn test_compaction_paths() { } #[test] -#[ignore = "log compaction disabled (#2337)"] fn test_version_filtering() { let snapshot = create_multi_version_snapshot(); let engine = SyncEngine::new(); @@ -242,7 +230,6 @@ fn test_version_filtering() { } #[tokio::test] -#[ignore = "log compaction disabled (#2337)"] async fn test_no_compaction_staged_commits() { use std::sync::Arc; @@ -321,12 +308,25 @@ async fn test_no_compaction_staged_commits() { // where staged commits might slip through the normal filtering } -// === Tests for disabled log compaction (TODO(#2337): remove when re-enabled) === +// === Tests for should_compact functionality === #[test] -fn test_should_compact_always_false() { - // These inputs would return true if compaction were enabled - assert!(!should_compact(9, 10)); - assert!(!should_compact(19, 10)); - assert!(!should_compact(99, 100)); +fn test_should_compact_percentage_logic() { + // Test that compaction occurs at expected intervals + // Commits start at 0, so we add 1 to the commit version to check if we've hit the interval + // (9+1=10, 19+1=20, 99+1=100, 59+1=60) + assert!(should_compact(9, 10)); + assert!(should_compact(19, 10)); + assert!(should_compact(99, 100)); + assert!(should_compact(59, 20)); + + // Test that compaction does NOT occur when not at intervals + assert!(!should_compact(5, 10)); + assert!(!should_compact(15, 10)); + assert!(!should_compact(99, 1000)); + assert!(!should_compact(109, 20)); // 110 is not a multiple of 20 + + // Test edge cases + assert!(!should_compact(0, 10)); // commit 0 is metadata, never compact + assert!(!should_compact(10, 0)); // zero interval disabled } diff --git a/kernel/src/log_compaction/writer.rs b/kernel/src/log_compaction/writer.rs index 034a66f18d..05a1208199 100644 --- a/kernel/src/log_compaction/writer.rs +++ b/kernel/src/log_compaction/writer.rs @@ -1,7 +1,4 @@ -// TODO(#2337): remove dead_code allows when log compaction is re-enabled -#![allow(dead_code, unused_imports)] - -use url::Url; +// Log compaction has been re-enabled use super::COMPACTION_ACTIONS_SCHEMA; use crate::action_reconciliation::log_replay::ActionReconciliationProcessor; @@ -16,33 +13,30 @@ use crate::{DeltaResult, Engine, Error, SnapshotRef, Version}; /// compaction interval. /// /// Always returns `false` because log compaction is currently disabled. -pub fn should_compact(_commit_version: Version, _compaction_interval: Version) -> bool { - // TODO(#2337): re-enable log compaction once testing is sufficient - // - // // Commits start at 0, so we add one to the commit version to check if we've hit the interval - // compaction_interval > 0 - // && commit_version > 0 - // && (commit_version + 1).is_multiple_of(compaction_interval) - false +pub fn should_compact(commit_version: Version, compaction_interval: Version) -> bool { + // Commits start at 0, so we add one to the commit version to check if we've hit the interval + compaction_interval > 0 + && commit_version > 0 + && (commit_version + 1).is_multiple_of(compaction_interval) } /// Writer for log compaction files /// /// This writer provides an API for creating log compaction files that aggregate actions /// from multiple commit files. -/// -/// Log compaction is currently disabled (#2337) #[derive(Debug)] pub struct LogCompactionWriter { /// Reference to the snapshot of the table being compacted snapshot: SnapshotRef, - // TODO(#2337): remove allow(dead_code) when log compaction is re-enabled - #[allow(dead_code)] + + /// Starting version of commits included in this compaction (inclusive) start_version: Version, - #[allow(dead_code)] + + /// End version of commits included in this compaction (exclusive) end_version: Version, + /// Cached compaction file path - compaction_path: Url, + compaction_path: url::Url, } impl RetentionCalculator for LogCompactionWriter { @@ -52,43 +46,39 @@ impl RetentionCalculator for LogCompactionWriter { } impl LogCompactionWriter { - // TODO(#2337): re-enable log compaction once testing is sufficient pub(crate) fn try_new( - _snapshot: SnapshotRef, - _start_version: Version, - _end_version: Version, + snapshot: SnapshotRef, + start_version: Version, + end_version: Version, ) -> DeltaResult { - Err(Error::unsupported( - "Log compaction is not currently supported", - )) - // if start_version >= end_version { - // return Err(Error::generic(format!( - // "Invalid version range: end_version {end_version} must be greater than \ - // start_version {start_version}" - // ))); - // } - // - // // We disallow log compaction if the Snapshot is not published. If we didn't, this - // // could create gaps in the version history, thereby breaking old readers. - // snapshot.log_segment().validate_published()?; - // - // // Compute the compaction path once during construction - // let compaction_path = ParsedLogPath::new_log_compaction( - // snapshot.table_root(), - // start_version, - // end_version, - // )?; - // - // Ok(Self { - // snapshot, - // start_version, - // end_version, - // compaction_path: compaction_path.location, - // }) + if start_version >= end_version { + return Err(Error::generic(format!( + "Invalid version range: end_version {end_version} must be greater than \ + start_version {start_version}" + ))); + } + + // We disallow log compaction if the Snapshot is not published. If we didn't, this + // could create gaps in the version history, thereby breaking old readers. + snapshot.log_segment().validate_published()?; + + // Compute the compaction path once during construction + let compaction_path = ParsedLogPath::new_log_compaction( + snapshot.table_root(), + start_version, + end_version, + )?; + + Ok(Self { + snapshot, + start_version, + end_version, + compaction_path: compaction_path.location, + }) } /// Get the path where the compaction file will be written - pub fn compaction_path(&self) -> &Url { + pub fn compaction_path(&self) -> &url::Url { &self.compaction_path } diff --git a/kernel/src/log_segment/mod.rs b/kernel/src/log_segment/mod.rs index 1bff6a7153..dfce8867a5 100644 --- a/kernel/src/log_segment/mod.rs +++ b/kernel/src/log_segment/mod.rs @@ -25,7 +25,7 @@ use crate::metrics::SnapshotLoadMetricContext; use crate::path::LogPathFileType::*; use crate::path::{LogPathFileType, ParsedLogPath}; use crate::schema::compare::SchemaComparison; -use crate::schema::{lazy_schema_ref, DataType, SchemaRef, StructField, StructType, ToSchema as _}; +use crate::schema::{lazy_schema_ref, DataType, SchemaRef, SchemaStructPatchBuilder, StructField, StructType, ToSchema as _}; use crate::utils::require; use crate::{ DeltaResult, Engine, Error, Expression, FileMeta, Predicate, PredicateRef, RowVisitor, @@ -37,6 +37,7 @@ mod domain_metadata_replay; mod protocol_metadata_replay; pub(crate) use domain_metadata_replay::DomainMetadataMap; +use crate::scan::log_replay::ScanStatsOptions; #[cfg(test)] mod crc_tests; @@ -117,7 +118,7 @@ pub(crate) struct LogSegment { /// access use [`Self::checkpoint_hint`] (and the [`Self::checkpoint_schema`] / /// [`Self::checkpoint_sidecars`] accessors built on it). Read this field directly only when /// the raw hint is wanted as-is -- e.g. re-threading it into a derived segment. - pub(crate) last_checkpoint_metadata: Option, + pub last_checkpoint_metadata: Option, } /// Returns the identifying leaf column path for a known action type, used to build IS NOT NULL @@ -727,6 +728,38 @@ impl LogSegment { effective_predicate, stats_schema, partition_schema, + None, + )?; + + Ok(ActionsWithCheckpointInfo { + actions: commit_stream.chain(checkpoint_result.actions), + checkpoint_info: checkpoint_result.checkpoint_info, + }) + } + + pub(crate) fn read_actions_with_projected_checkpoint_actions_for_scan( + &self, + engine: &dyn Engine, + commit_read_schema: SchemaRef, + checkpoint_read_schema: SchemaRef, + meta_predicate: Option, + stats_schema: Option<&StructType>, + partition_schema: Option<&StructType>, + stats_options: &ScanStatsOptions, + ) -> DeltaResult< + ActionsWithCheckpointInfo> + Send>, + > { + // `replay` expects commit files to be sorted in descending order, so the return value here + // is correct + let commit_stream = CommitReader::try_new(engine, self, commit_read_schema)?; + + let checkpoint_result = self.create_checkpoint_stream( + engine, + checkpoint_read_schema, + meta_predicate, + stats_schema, + partition_schema, + Some(stats_options), )?; Ok(ActionsWithCheckpointInfo { @@ -900,6 +933,7 @@ impl LogSegment { meta_predicate: Option, stats_schema: Option<&StructType>, partition_schema: Option<&StructType>, + stats_options: Option<&ScanStatsOptions>, ) -> DeltaResult< ActionsWithCheckpointInfo> + Send>, > { @@ -911,72 +945,125 @@ impl LogSegment { (None, vec![]) }; - // Check if checkpoint has compatible stats_parsed and add it to the schema if so - let has_stats_parsed = - stats_schema - .zip(file_actions_schema.as_ref()) - .is_some_and(|(stats, file_schema)| { - Self::schema_has_compatible_stats_parsed(file_schema, stats) - }); - + let available_stats_schema = file_actions_schema + .as_ref() + .and_then(|schema| schema.field("add")) + .and_then(|field| match field.data_type() { + DataType::Struct(add) => add.field("stats_parsed"), + _ => None, + }) + .and_then(|field| match field.data_type() { + DataType::Struct(stats) => Some(stats.as_ref().clone()), + _ => None, + }); + let stats_disabled = stats_options.is_some_and(|options| options.skip_stats); + let typed_stats_schema = (!stats_disabled).then(|| stats_schema.cloned()).flatten(); + let has_stats_parsed = typed_stats_schema + .as_ref() + .zip(file_actions_schema.as_ref()) + .is_some_and(|(stats, file_schema)| { + Self::schema_has_compatible_stats_parsed(file_schema, stats) + }); + let projected_stats_schema = if has_stats_parsed { + typed_stats_schema.clone() + } else if stats_options + .is_some_and(|options| !options.skip_stats && options.synthesize_json) + { + available_stats_schema + } else { + None + }; + let read_raw_stats = stats_options.is_none_or(|options| { + !options.skip_stats + && (options.synthesize_json + || (options.checkpoint_stats_json_fallback + && typed_stats_schema.is_some() + && !has_stats_parsed)) + }); let has_partition_values_parsed = partition_schema .zip(file_actions_schema.as_ref()) .is_some_and(|(ps, fs)| Self::schema_has_compatible_partition_values_parsed(fs, ps)); + // JSON checkpoint stats are required when structured stats cannot satisfy the scan schema. + let needs_json_stats_fallback = stats_schema.is_some() + && !has_stats_parsed + && action_schema.field("add").is_some_and(|field| { + let DataType::Struct(add) = field.data_type() else { + return false; + }; + add.field("stats").is_none() + }); + // Build final schema with any additional fields needed // (stats_parsed, partitionValues_parsed, sidecar) - let needs_sidecar = need_file_actions && !sidecar_files.is_empty(); - let needs_add_augmentation = has_stats_parsed || has_partition_values_parsed; + let needs_sidecar = need_file_actions + && !sidecar_files.is_empty() + && !action_schema.contains(SIDECAR_NAME); + let needs_add_augmentation = stats_options.is_some() + || projected_stats_schema.is_some() + || has_partition_values_parsed; let augmented_checkpoint_read_schema = if needs_add_augmentation || needs_sidecar { - let mut new_fields: Vec = if let (true, Some(add_field)) = - (needs_add_augmentation, action_schema.field("add")) - { - let DataType::Struct(add_struct) = add_field.data_type() else { - return Err(Error::internal_error( - "add field in action schema must be a struct", - )); - }; - let mut add_fields: Vec = add_struct.fields().cloned().collect(); - - if let (true, Some(ss)) = (has_stats_parsed, stats_schema) { - add_fields.push(StructField::nullable("stats_parsed", ss.clone())); + let mut builder = SchemaStructPatchBuilder::new(); + + // Re-project the `add` struct: drop any pre-existing parsed columns (and raw `stats` + // when it is not needed) before re-adding the parsed columns we want. + if needs_add_augmentation && action_schema.field("add").is_some() { + builder = builder + .drop_if_exists_at(["add"], "stats_parsed") + .drop_if_exists_at(["add"], "partitionValues_parsed"); + if !read_raw_stats { + builder = builder.drop_if_exists_at(["add"], "stats"); + } else { + let have_to_add_stats_field = action_schema.field("add").is_some_and(|field| { + let DataType::Struct(add) = field.data_type() else { + return false; + }; + add.field("stats").is_none() + }); + if have_to_add_stats_field { + builder = builder.append_at(["add"], StructField::nullable("stats", DataType::STRING)) + } } - if let (true, Some(ps)) = (has_partition_values_parsed, partition_schema) { - add_fields.push(StructField::nullable("partitionValues_parsed", ps.clone())); + if let Some(stats_schema) = projected_stats_schema.as_ref() { + builder = builder + .append_at(["add"], StructField::nullable("stats_parsed", stats_schema.clone())); } - - // Rebuild schema with modified add field - action_schema - .fields() - .map(|f| { - if f.name() == "add" { - StructField::new( - add_field.name(), - StructType::new_unchecked(add_fields.clone()), - add_field.is_nullable(), - ) - .with_metadata(add_field.metadata.clone()) - } else { - f.clone() - } - }) - .collect() - } else { - action_schema.fields().cloned().collect() - }; + if let (true, Some(partition_schema)) = + (has_partition_values_parsed, partition_schema) + { + builder = builder.append_at( + ["add"], + StructField::nullable("partitionValues_parsed", partition_schema.clone()), + ); + } + } // Add sidecar column at top-level for V2 checkpoints if needs_sidecar { - new_fields.push(StructField::nullable(SIDECAR_NAME, Sidecar::to_schema())); + builder = builder.append(StructField::nullable(SIDECAR_NAME, Sidecar::to_schema())); } - Arc::new(StructType::new_unchecked(new_fields)) + Arc::new(builder.build(&action_schema)?) } else { // No modifications needed, use schema as-is action_schema.clone() }; + // Derive action-type row-group pruning from the final physical projection. A stats + // predicate cannot be evaluated when typed stats fall back to raw JSON. + let is_not_null_pred = schema_to_is_not_null_predicate(&augmented_checkpoint_read_schema); + let meta_predicate = (typed_stats_schema.is_none() || has_stats_parsed) + .then_some(meta_predicate) + .flatten(); + let effective_predicate = match (is_not_null_pred, meta_predicate) { + (None, predicate) | (predicate, None) => predicate, + (Some(left), Some(right)) => Some(Arc::new(Predicate::and( + (*left).clone(), + (*right).clone(), + ))), + }; + let checkpoint_file_meta: Vec<_> = self .listed .checkpoint_parts @@ -995,14 +1082,14 @@ impl LogSegment { engine.json_handler().read_json_files( &checkpoint_file_meta, augmented_checkpoint_read_schema.clone(), - meta_predicate.clone(), + effective_predicate.clone(), )? } Some(parsed_log_path) if parsed_log_path.extension == "parquet" => parquet_handler .read_parquet_files( &checkpoint_file_meta, augmented_checkpoint_read_schema.clone(), - meta_predicate.clone(), + effective_predicate.clone(), )?, Some(parsed_log_path) => { return Err(Error::generic(format!( @@ -1023,7 +1110,7 @@ impl LogSegment { parquet_handler.read_parquet_files( &sidecar_files, augmented_checkpoint_read_schema.clone(), - meta_predicate, + effective_predicate, )? } else { Box::new(std::iter::empty()) @@ -1198,14 +1285,24 @@ impl LogSegment { SIDECAR_SCHEMA.clone() } + fn get_field_from_add<'a>( + checkpoint_schema: &'a StructType, + name: &str, + ) -> Option<&'a StructField> { + let DataType::Struct(add) = checkpoint_schema.field("add")?.data_type() else { + return None; + }; + add.field(name) + } + /// Checks if a checkpoint schema contains a usable `add.stats_parsed` field. /// /// This validates that: /// 1. The `add.stats_parsed` field exists in the checkpoint schema /// 2. The types in `stats_parsed` are compatible with the stats schema for data skipping /// - /// The `stats_schema` parameter contains only the columns referenced in the data skipping - /// predicate. This is built from the predicate and passed in by the caller. + /// `stats_schema` is the physical typed schema required by the scan. It includes requested + /// structured output fields and any data-skipping predicate fields. /// /// Both the checkpoint's `stats_parsed` schema and the `stats_schema` for data skipping /// use physical column names (not logical names), so direct name comparison is correct. @@ -1215,14 +1312,7 @@ impl LogSegment { checkpoint_schema: &StructType, stats_schema: &StructType, ) -> bool { - // Get add.stats_parsed from the checkpoint schema - let Some(stats_parsed) = checkpoint_schema - .field("add") - .and_then(|f| match f.data_type() { - DataType::Struct(s) => s.field("stats_parsed"), - _ => None, - }) - else { + let Some(stats_parsed) = Self::get_field_from_add(checkpoint_schema, "stats_parsed") else { debug!("stats_parsed not compatible: checkpoint schema does not contain add.stats_parsed field"); return false; }; @@ -1235,42 +1325,8 @@ impl LogSegment { return false; }; - // Check type compatibility for both minValues and maxValues structs. - // While these typically have the same schema, the protocol doesn't guarantee it, - // so we check both to be safe. - for field_name in [MIN_VALUES, MAX_VALUES] { - let Some(checkpoint_values_field) = stats_struct.field(field_name) else { - // stats_parsed exists but no minValues/maxValues - unusual but valid - continue; - }; - - // minValues/maxValues must be a Struct containing per-column statistics. - // If it exists but isn't a Struct, the schema is malformed and unusable. - let DataType::Struct(checkpoint_values) = checkpoint_values_field.data_type() else { - debug!( - "stats_parsed not compatible: stats_parsed.{} is not a Struct, got {:?}", - field_name, - checkpoint_values_field.data_type() - ); - return false; - }; - - // Get the corresponding field from stats_schema (e.g., stats_schema.minValues) - let Some(stats_values_field) = stats_schema.field(field_name) else { - // stats_schema doesn't have minValues/maxValues, skip this check - continue; - }; - let DataType::Struct(stats_values) = stats_values_field.data_type() else { - // stats_schema.minValues/maxValues isn't a struct - shouldn't happen but skip - continue; - }; - - // Check type compatibility recursively for nested structs. - // Only fields that exist in both schemas need compatible types. - // Extra fields in checkpoint are ignored; missing fields return null. - if !Self::structs_have_compatible_types(checkpoint_values, stats_values, field_name) { - return false; - } + if !Self::structs_have_compatible_types(stats_struct, stats_schema, "stats_parsed") { + return false; } debug!("Checkpoint schema has compatible stats_parsed for data skipping"); @@ -1293,7 +1349,6 @@ impl LogSegment { ) -> bool { for needed_field in needed.fields() { let Some(available_field) = available.field(needed_field.name()) else { - // Field missing in checkpoint - that's OK, it will be null continue; }; @@ -1320,10 +1375,10 @@ impl LogSegment { }; if !compatible { debug!( - "stats_parsed not compatible: incompatible type for '{}' in {}: \ + "{} not compatible: incompatible type for '{}': \ checkpoint has {:?}, stats schema needs {:?}", - needed_field.name(), context, + needed_field.name(), avail_type, need_type ); @@ -1349,12 +1404,7 @@ impl LogSegment { partition_schema: &StructType, ) -> bool { let Some(partition_parsed) = - checkpoint_schema - .field("add") - .and_then(|f| match f.data_type() { - DataType::Struct(s) => s.field("partitionValues_parsed"), - _ => None, - }) + Self::get_field_from_add(checkpoint_schema, "partitionValues_parsed") else { debug!("partitionValues_parsed not compatible: checkpoint schema does not contain add.partitionValues_parsed field"); return false; diff --git a/kernel/src/log_segment/tests.rs b/kernel/src/log_segment/tests.rs index 8d09e70bda..1499bc1275 100644 --- a/kernel/src/log_segment/tests.rs +++ b/kernel/src/log_segment/tests.rs @@ -32,8 +32,12 @@ use crate::scan::test_utils::{ add_batch_simple, add_batch_with_remove, sidecar_batch_with_given_paths, sidecar_batch_with_given_paths_and_sizes, }; -use crate::scan::{CHECKPOINT_READ_SCHEMA, COMMIT_READ_SCHEMA}; -use crate::schema::{schema, schema_ref, DataType, StructField, StructType}; +use crate::scan::{ + CHECKPOINT_READ_SCHEMA, CHECKPOINT_READ_SCHEMA_NO_JSON_STATS, COMMIT_READ_SCHEMA, +}; +use crate::schema::{ + schema, schema_ref, DataType, SchemaRef, SchemaStructPatchBuilder, StructField, StructType, +}; use crate::utils::test_utils::{ assert_batch_matches, assert_result_error_with_message, create_log_path, create_log_path_with_size, string_array_to_engine_data, Action, @@ -1213,6 +1217,7 @@ async fn test_create_checkpoint_stream_returns_checkpoint_batches_as_is_if_schem None, // meta_predicate None, // stats_schema None, // partition_schema + None, )?; let mut iter = checkpoint_result.actions; @@ -1286,6 +1291,7 @@ async fn test_create_checkpoint_stream_returns_checkpoint_batches_if_checkpoint_ None, // meta_predicate None, // stats_schema None, // partition_schema + None, )?; let mut iter = checkpoint_result.actions; @@ -1351,6 +1357,7 @@ async fn test_create_checkpoint_stream_reads_parquet_checkpoint_batch_without_si None, // meta_predicate None, // stats_schema None, // partition_schema + None, )?; let mut iter = checkpoint_result.actions; @@ -1404,6 +1411,7 @@ async fn test_create_checkpoint_stream_reads_json_checkpoint_batch_without_sidec None, // meta_predicate None, // stats_schema None, // partition_schema + None, )?; let mut iter = checkpoint_result.actions; @@ -1496,6 +1504,7 @@ async fn test_create_checkpoint_stream_reads_checkpoint_file_and_returns_sidecar None, // meta_predicate None, // stats_schema None, // partition_schema + None, )?; let mut iter = checkpoint_result.actions; @@ -3037,6 +3046,27 @@ fn create_checkpoint_schema_with_stats_parsed(min_values_fields: Vec, + include_json_stats: bool, +) -> DeltaResult { + let stats_parsed = StructField::nullable( + "stats_parsed", + schema! { + nullable (NUM_RECORDS): LONG, + nullable (MIN_VALUES): { ..(min_values_fields.clone()) }, + nullable (MAX_VALUES): { ..(min_values_fields) }, + }, + ); + let patch = SchemaStructPatchBuilder::new().append_at(["add"], stats_parsed); + let patch = if include_json_stats { + patch + } else { + patch.drop_at(["add"], "stats") + }; + Ok(Arc::new(patch.build(get_commit_schema().as_ref())?)) +} + // Helper to create a stats_schema with proper structure (numRecords, minValues, maxValues) fn create_stats_schema(column_fields: Vec) -> StructType { schema! { @@ -3056,6 +3086,104 @@ fn create_checkpoint_schema_without_stats_parsed() -> StructType { } } +#[rstest] +#[case::missing_with_json(false, true, false, true)] +#[case::partial_with_json(true, true, true, false)] +#[case::partial_without_json(true, false, true, false)] +#[tokio::test] +async fn test_checkpoint_stream_resolves_stats_projection( + #[case] include_parsed_stats: bool, + #[case] include_json_stats: bool, + #[case] expect_parsed_stats: bool, + #[case] expect_json_stats: bool, +) -> DeltaResult<()> { + + let (store, log_root) = new_in_memory_store(); + let engine = SyncEngine::new_with_store(store.clone()); + let checkpoint_schema = if include_parsed_stats { + create_checkpoint_file_schema_with_stats_parsed( + vec![StructField::nullable("other", DataType::LONG)], + include_json_stats, + )? + } else { + get_commit_schema().clone() + }; + add_checkpoint_to_store( + &store, + add_batch_simple(checkpoint_schema), + "00000000000000000001.checkpoint.parquet", + ) + .await?; + + let checkpoint_file = log_root + .join("00000000000000000001.checkpoint.parquet")? + .to_string(); + let checkpoint_size = + get_file_size(&store, "_delta_log/00000000000000000001.checkpoint.parquet").await; + let log_segment = LogSegment::try_new( + LogSegmentFiles { + checkpoint_parts: vec![create_log_path_with_size(&checkpoint_file, checkpoint_size)], + latest_commit_file: Some(create_log_path("file:///00000000000000000001.json")), + ..Default::default() + }, + log_root, + None, + None, + )?; + let stats_schema = create_stats_schema(vec![StructField::nullable("id", DataType::LONG)]); + + let checkpoint_result = log_segment.create_checkpoint_stream( + &engine, + CHECKPOINT_READ_SCHEMA_NO_JSON_STATS.clone(), + None, // meta_predicate + Some(&stats_schema), + None, // partition_schema + Some(&ScanStatsOptions { + skip_stats: false, + synthesize_json: false, + checkpoint_stats_json_fallback: true, + }), + )?; + + assert_eq!( + checkpoint_result.checkpoint_info.has_stats_parsed, + expect_parsed_stats + ); + let add_field = checkpoint_result + .checkpoint_info + .checkpoint_read_schema + .field("add") + .expect("checkpoint read schema must contain add"); + let DataType::Struct(add) = add_field.data_type() else { + panic!("checkpoint add field must be a struct"); + }; + assert_eq!(add.field("stats").is_some(), expect_json_stats); + assert_eq!(add.field("stats_parsed").is_some(), expect_parsed_stats); + + let read_schema = checkpoint_result + .checkpoint_info + .checkpoint_read_schema + .clone(); + let mut actions = checkpoint_result.actions; + let batch = actions + .next() + .expect("checkpoint stream must yield one batch")?; + assert!(!batch.is_log_batch); + assert_eq!( + batch.actions.has_field(&ColumnName::new(["add", "stats"])), + expect_json_stats, + "checkpoint batch JSON stats projection must match the resolved schema" + ); + if include_json_stats { + assert_batch_matches(batch.actions, add_batch_simple(read_schema)); + } else { + assert!(!batch.actions.is_empty()); + } + assert!(actions.next().is_none()); + + Ok(()) +} + #[test] fn test_schema_has_compatible_stats_parsed_basic() { // Create a checkpoint schema with stats_parsed containing an integer column @@ -3622,6 +3750,7 @@ async fn test_checkpoint_stream_sets_has_partition_values_parsed() -> DeltaResul None, // meta_predicate None, // stats_schema Some(&partition_schema), + None, )?; // Verify that checkpoint_info reports partitionValues_parsed as available @@ -3687,6 +3816,7 @@ async fn test_checkpoint_stream_no_partition_values_parsed_when_incompatible() - None, None, Some(&partition_schema), + None, )?; // Verify it's false diff --git a/kernel/src/parallel/parallel_scan_metadata.rs b/kernel/src/parallel/parallel_scan_metadata.rs index 07289e2361..e35a036ab9 100644 --- a/kernel/src/parallel/parallel_scan_metadata.rs +++ b/kernel/src/parallel/parallel_scan_metadata.rs @@ -318,6 +318,7 @@ mod tests { physical_predicate: PhysicalPredicate::None, transform_spec: None, column_mapping_mode: ColumnMappingMode::None, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), diff --git a/kernel/src/scan/log_replay.rs b/kernel/src/scan/log_replay.rs index c0f24b8eb4..84c7cd8c63 100644 --- a/kernel/src/scan/log_replay.rs +++ b/kernel/src/scan/log_replay.rs @@ -8,7 +8,7 @@ use serde::{Deserialize, Serialize}; use super::data_skipping::DataSkippingFilter; use super::metrics::ScanMetrics; use super::state_info::StateInfo; -use super::{PhysicalPredicate, ScanMetadata}; +use super::{PhysicalPredicate, ScanMetadata, COMMIT_READ_SCHEMA}; use crate::actions::deletion_vector::DeletionVectorDescriptor; use crate::engine_data::{GetData, RowVisitor, TypedGetData as _}; use crate::expressions::{ @@ -29,7 +29,7 @@ use crate::schema::{ }; use crate::table_features::ColumnMappingMode; use crate::utils::{require, FoldWithOption as _}; -use crate::{DeltaResult, Engine, Error, ExpressionEvaluator}; +use crate::{DeltaResult, Engine, EngineData, Error, ExpressionEvaluator}; /// Read-time stats toggles consumed by [`ScanLogReplayProcessor`]. #[derive(Clone, Copy, Debug, serde::Serialize, serde::Deserialize)] @@ -40,8 +40,16 @@ pub(crate) struct ScanStatsOptions { /// whose `add.stats` is null but whose `add.stats_parsed` is populated /// (writeStatsAsJson=false, writeStatsAsStruct=true). When false, `ScanFile.stats` /// is left null on such checkpoints; engines that consume `stats_parsed` directly - /// avoid the per-batch `ToJson` cost. + /// avoid reading JSON stats in checkpoints and the per-batch `ToJson` cost. pub(crate) synthesize_json: bool, + /// Allow typed statistics to fall back to raw checkpoint JSON when a compatible + /// `stats_parsed` struct is unavailable. + #[serde(default = "default_checkpoint_stats_json_fallback")] + pub(crate) checkpoint_stats_json_fallback: bool, +} + +fn default_checkpoint_stats_json_fallback() -> bool { + true } impl Default for ScanStatsOptions { @@ -49,6 +57,7 @@ impl Default for ScanStatsOptions { Self { skip_stats: false, synthesize_json: true, + checkpoint_stats_json_fallback: true, } } } @@ -72,7 +81,11 @@ struct InternalScanState { transform_spec: Option>, column_mapping_mode: ColumnMappingMode, /// Physical stats schema for reading/parsing stats from checkpoint files + /// Union of requested and predicate-only parsed statistics used during replay. physical_stats_schema: Option, + /// Parsed statistics requested in emitted scan metadata. + #[serde(default)] + requested_output_stats_schema: Option, #[serde(default)] stats_options: ScanStatsOptions, #[serde(default)] @@ -138,7 +151,7 @@ pub struct SerializableScanState { /// - Action Deduplication: Leverages the [`FileActionDeduplicator`] to ensure that for each unique /// file (identified by its path and deletion vector unique ID), only the latest valid Add action /// is processed. -/// - Transformation: Applies a built-in transformation (`log_transform` or `checkpoint_transform`) +/// - Transformation: Applies a built-in transformation (`commit_transform` or `checkpoint_transform`) /// to convert selected Add actions into [`ScanMetadata`], the intermediate format passed to the /// engine. /// - Row StructPatch passthrough: Any user-provided row-level transformation expressions (e.g. @@ -155,10 +168,12 @@ pub struct ScanLogReplayProcessor { data_skipping_filter: Option, /// StructPatch for log batches (commit files) - uses ParseJson for stats and MapToStruct /// for partition values - log_transform: Arc, + commit_transform: Arc, /// StructPatch for checkpoint batches - reads pre-parsed stats_parsed and /// partitionValues_parsed directly when available, otherwise parses from raw columns checkpoint_transform: Arc, + /// Removes internal predicate-only typed statistics before emitting scan metadata. + output_transform: Option>, state_info: Arc, /// A set of (data file path, dv_unique_id) pairs that have been seen thus /// far in the log. This is used to filter out files with Remove actions as @@ -233,8 +248,21 @@ impl ScanLogReplayProcessor { let ScanStatsOptions { skip_stats, synthesize_json, + checkpoint_stats_json_fallback } = stats_options; + let has_add_field = |schema: &StructType, name| { + schema + .field("add") + .and_then(|field| match field.data_type() { + DataType::Struct(add) => add.field(name), + _ => None, + }) + .is_some() + }; + let has_raw_checkpoint_stats = has_add_field(&checkpoint_read_schema, "stats"); + let has_stats_parsed_for_json = has_add_field(&checkpoint_read_schema, STATS_PARSED_NAME); + // Create metrics first so we can pass them to DataSkippingFilter let metrics = Arc::new(ScanMetrics::default()); @@ -264,10 +292,14 @@ impl ScanLogReplayProcessor { None }; - let output_schema = scan_row_schema_with_parsed_columns( + let replay_schema = scan_row_schema_with_parsed_columns( stats_schema_for_transform.clone(), partition_schema_for_transform.clone(), )?; + let output_schema = scan_row_schema_with_parsed_columns( + state_info.requested_output_stats_schema.clone(), + partition_schema_for_transform.clone(), + )?; // Create data skipping filter that reads stats_parsed and partitionValues_parsed // from the transformed batch. This avoids double JSON parsing -- the transform parses @@ -289,40 +321,58 @@ impl ScanLogReplayProcessor { // The transform flattens `add.*` to top-level columns, so `path` is non-null // exactly for Add rows. Arc::new(Predicate::is_not_null(column_expr!("path")).into()), - output_schema.clone(), + replay_schema.clone(), &state_info.physical_stats_columns, Some(metrics.clone()), ) }; + let output_transform = if replay_schema == output_schema { + None + } else { + Some(engine.evaluation_handler().new_expression_evaluator( + replay_schema.clone(), + scan_row_output_projection(&replay_schema, &output_schema), + output_schema.into(), + )?) + }; + + let commit_read_schema = super::commit_read_schema(skip_stats); + let has_raw_commit_stats = has_add_field(&commit_read_schema, "stats"); + Ok(Self { data_skipping_filter, - // Log transform: parse JSON for stats, MapToStruct for partition values - log_transform: engine.evaluation_handler().new_expression_evaluator( - checkpoint_read_schema.clone(), + // Commit transform: parse JSON for stats, MapToStruct for partition values + commit_transform: engine.evaluation_handler().new_expression_evaluator( + commit_read_schema, get_add_transform_expr( stats_schema_for_transform.clone(), + has_raw_commit_stats, + true, + false, false, - skip_stats, - synthesize_json, + &stats_options, partition_schema_for_transform.clone(), false, ), - output_schema.clone().into(), + replay_schema.clone().into(), )?, // Checkpoint transform: read pre-parsed columns directly when available checkpoint_transform: engine.evaluation_handler().new_expression_evaluator( checkpoint_read_schema, get_add_transform_expr( - stats_schema_for_transform, + stats_schema_for_transform.clone(), + has_raw_checkpoint_stats, + stats_options.checkpoint_stats_json_fallback, has_stats_parsed, - skip_stats, - synthesize_json, - partition_schema_for_transform, + has_stats_parsed_for_json, + &stats_options, + partition_schema_for_transform.clone(), has_partition_values_parsed, ), - output_schema.into(), + replay_schema.clone().into(), )?, + output_transform, seen_file_keys, state_info, stats_options, @@ -345,6 +395,28 @@ impl ScanLogReplayProcessor { self.state_info.is_catalog_managed } + fn finalize_scan_metadata( + &self, + transformed: Box, + selection_vector: Vec, + row_transform_exprs: Vec>, + ) -> DeltaResult { + let output = if let Some(output_transform) = &self.output_transform { + output_transform.evaluate(transformed.as_ref())? + } else { + transformed + }; + let scan_metadata = ScanMetadata::try_new( + output, + selection_vector, + row_transform_exprs, + self.state_info.requested_output_stats_schema.as_ref(), + )?; + self.metrics + .update_peak_hash_set_size(self.seen_file_keys.len()); + Ok(scan_metadata) + } + /// Serialize the processor state for distributed processing. /// /// Consumes the processor and returns a `SerializableScanState` containing: @@ -367,6 +439,7 @@ impl ScanLogReplayProcessor { physical_predicate, transform_spec, column_mapping_mode, + requested_output_stats_schema, physical_stats_schema, physical_partition_schema, physical_stats_columns, @@ -388,6 +461,7 @@ impl ScanLogReplayProcessor { predicate_schema, column_mapping_mode, physical_stats_schema, + requested_output_stats_schema, stats_options: self.stats_options, partition_values_options: self.partition_values_options, physical_partition_schema, @@ -409,7 +483,7 @@ impl ScanLogReplayProcessor { /// Reconstruct a processor from serialized state. /// /// Creates a new processor with the provided state. All fields (partition_filter, - /// data_skipping_filter, log_transform, checkpoint_transform, and seen_file_keys) are + /// data_skipping_filter, commit_transform, checkpoint_transform, and seen_file_keys) are /// reconstructed from the serialized state and engine. /// /// # Parameters @@ -448,6 +522,7 @@ impl ScanLogReplayProcessor { physical_predicate, transform_spec: internal_state.transform_spec, column_mapping_mode: internal_state.column_mapping_mode, + requested_output_stats_schema: internal_state.requested_output_stats_schema, physical_stats_schema: internal_state.physical_stats_schema, physical_partition_schema: internal_state.physical_partition_schema, physical_stats_columns: internal_state.physical_stats_columns, @@ -648,6 +723,47 @@ pub(crate) static STATS_PARSED_NAME: &str = "stats_parsed"; #[internal_api] pub(crate) static PARTITION_VALUES_PARSED_NAME: &str = "partitionValues_parsed"; +fn project_nested_struct( + input_schema: &StructType, + output_schema: &StructType, + path: &ColumnName, +) -> ExpressionRef { + let fields = output_schema.fields().map(|field| { + let field_path = path.join(&ColumnName::new([field.name()])); + let input_field = input_schema.field(field.name()); + match (field.data_type(), input_field.map(StructField::data_type)) { + (DataType::Struct(output_child), Some(DataType::Struct(input_child))) => { + project_nested_struct(input_child, output_child, &field_path) + } + // Compatible legacy checkpoints may omit any requested statistics field. Preserve + // the requested shape and represent every unavailable value as null. + (data_type, None) => Arc::new(Expression::null_literal(data_type.clone())), + _ => Arc::new(Expression::column(field_path)), + } + }); + let is_present = Arc::new(Predicate::is_not_null(Expression::column(path.clone())).into()); + Arc::new(Expression::struct_with_nullability_from(fields, is_present)) +} + +fn scan_row_output_projection( + input_schema: &StructType, + output_schema: &StructType, +) -> ExpressionRef { + let fields = output_schema.fields().map(|field| { + let path = ColumnName::new([field.name()]); + let input_field = input_schema.field(field.name()); + match (field.data_type(), input_field.map(StructField::data_type)) { + (DataType::Struct(output_child), Some(DataType::Struct(input_child))) + if field.name() == STATS_PARSED_NAME => + { + project_nested_struct(input_child, output_child, &path) + } + _ => Arc::new(Expression::column(path)), + } + }); + Arc::new(Expression::struct_from(fields)) +} + // NB: If you update this schema, ensure you update the comment describing it in the doc comment // for `scan_row_schema` in scan/mod.rs! You'll also need to update ScanFileVisitor as the // indexes will be off, and [`get_add_transform_expr`] below to match it. @@ -677,7 +793,7 @@ pub(crate) static SCAN_ROW_SCHEMA: LazyLock> = LazyLock::new(|| /// regardless of the engine's options, and keeps data-skipping paths uniform: /// `partitionValues_parsed.` parallels `stats_parsed.minValues.`. The checkpoint source /// is also `add.partitionValues_parsed`, a sibling of `add.stats_parsed`. -fn scan_row_schema_with_parsed_columns( +pub(super) fn scan_row_schema_with_parsed_columns( stats_schema: Option, partition_schema: Option, ) -> DeltaResult { @@ -703,46 +819,52 @@ fn scan_row_schema_with_parsed_columns( /// # Parameters /// - `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. When true and -/// `synthesize_json` is true, stats output uses `COALESCE(add.stats, ToJson(add.stats_parsed))` -/// so that `ScanFile.stats` is populated even when the checkpoint lacks JSON stats -/// (writeStatsAsJson=false). -/// - `skip_stats`: When true, replaces the stats column with a null literal, avoiding reads of the -/// raw stats JSON string from checkpoint parquet files. -/// - `synthesize_json`: When false, disables the `ToJson(add.stats_parsed)` fallback regardless of -/// `has_stats_parsed`. Set false for engines that consume `stats_parsed` directly and don't want -/// to pay the per-batch `ToJson` cost over potentially large stats structs. +/// - `has_raw_stats`: Whether the selected source schema contains `add.stats`. +/// - `allow_raw_stats_for_typed`: Whether raw JSON may be parsed as typed statistics. +/// - `has_typed_stats_parsed`: Whether checkpoint `stats_parsed` is compatible with the typed +/// replay schema. +/// - `has_stats_parsed_for_json`: Whether the projected checkpoint input contains +/// `stats_parsed`, independently of typed replay compatibility. +/// - `stats_options`: Controls statistics replay and JSON synthesis. /// - `partition_schema`: Schema of typed partition columns for data skipping, or None if partition /// value parsing is not needed. /// - `has_partition_values_parsed`: Whether the source carries a native `partitionValues_parsed` /// column (checkpoint). When true it is read directly; otherwise the struct is reconstructed from /// the `partitionValues` string map. /// -/// The transform includes `stats_parsed` only when `physical_stats_schema` is Some, -/// and `partitionValues_parsed` only when `partition_schema` is Some. -/// Stats are output using physical column names. +/// The transform includes `stats_parsed` only when `effective_replay_stats_schema` is Some, and +/// `partitionValues_parsed` only when `partition_schema` is Some. A later output projection removes +/// typed statistics used only for internal data skipping. fn get_add_transform_expr( physical_stats_schema: Option, - has_stats_parsed: bool, - skip_stats: bool, - synthesize_json: bool, + has_raw_stats: bool, + allow_raw_stats_for_typed: bool, + has_typed_stats_parsed: bool, + has_stats_parsed_for_json: bool, + stats_options: &ScanStatsOptions, partition_schema: Option, has_partition_values_parsed: bool, ) -> ExpressionRef { - let stats_expr = if skip_stats { - Arc::new(Expression::Literal(Scalar::Null(DataType::STRING))) - } else if has_stats_parsed && synthesize_json { - // Checkpoint may lack JSON stats when writeStatsAsJson=false. Fall back to - // serializing stats_parsed so ScanFile.stats is populated either way. - Arc::new(Expression::coalesce([ - Expression::column(["add", "stats"]), - Expression::unary( - UnaryExpressionOp::ToJson, - Expression::column(["add", "stats_parsed"]), - ), - ])) - } else { + let null_stats = || Arc::new(Expression::Literal(Scalar::Null(DataType::STRING))); + let stats_expr = if stats_options.skip_stats { + null_stats() + } else if stats_options.synthesize_json && has_stats_parsed_for_json { + let parsed_json = Expression::unary( + UnaryExpressionOp::ToJson, + Expression::column(["add", "stats_parsed"]), + ); + if has_raw_stats { + Arc::new(Expression::coalesce([ + Expression::column(["add", "stats"]), + parsed_json, + ])) + } else { + Arc::new(parsed_json) + } + } else if has_raw_stats { column_expr_ref!("add.stats") + } else { + null_stats() }; let mut fields = vec![ column_expr_ref!("add.path"), @@ -761,12 +883,13 @@ fn get_add_transform_expr( // Add stats_parsed when stats output is requested (using physical column names) if let Some(stats_schema) = physical_stats_schema { - let stats_parsed_expr = if has_stats_parsed { - // Checkpoint has stats_parsed column - read directly + let stats_parsed_expr = if has_typed_stats_parsed { + // Checkpoint has compatible stats_parsed - read it directly. column_expr!("add.stats_parsed") - } else { - // No stats_parsed available (JSON log files) - parse JSON + } else if has_raw_stats && allow_raw_stats_for_typed { Expression::parse_json(column_expr!("add.stats"), stats_schema) + } else { + Expression::null_literal(stats_schema.as_ref().clone().into()) }; fields.push(Arc::new(stats_parsed_expr)); } @@ -810,6 +933,32 @@ pub(crate) fn get_scan_metadata_transform_expr() -> ExpressionRef { EXPR.clone() } +pub(super) fn get_scan_metadata_transform_expr_with_parsed_columns( + available_stats_schema: &StructType, + effective_replay_stats_schema: &StructType, +) -> ExpressionRef { + let stats_path = ColumnName::new([STATS_PARSED_NAME]); + Arc::new(Expression::struct_from([Arc::new( + Expression::struct_from([ + column_expr_ref!("path"), + column_expr_ref!("fileConstantValues.partitionValues"), + column_expr_ref!("size"), + column_expr_ref!("modificationTime"), + column_expr_ref!("stats"), + column_expr_ref!("fileConstantValues.tags"), + column_expr_ref!("deletionVector"), + column_expr_ref!("fileConstantValues.baseRowId"), + column_expr_ref!("fileConstantValues.defaultRowCommitVersion"), + column_expr_ref!("fileConstantValues.clusteringProvider"), + project_nested_struct( + available_stats_schema, + effective_replay_stats_schema, + &stats_path, + ), + ]), + )])) +} + impl ParallelLogReplayProcessor for ScanLogReplayProcessor { type Output = ::Output; @@ -876,15 +1025,13 @@ impl ParallelLogReplayProcessor for ScanLogReplayProcessor { ); visitor.visit_rows_of(actions.as_ref())?; - // Step 4: Return transformed batch with updated selection vector - let scan_metadata = ScanMetadata::try_new( - transformed, - visitor.selection_vector, - visitor.row_transform_exprs, - )?; - self.metrics - .update_peak_hash_set_size(self.seen_file_keys.len()); - Ok(scan_metadata) + // Step 4: Remove internal-only columns and return the emitted scan metadata. + let AddRemoveDedupVisitor { + selection_vector, + row_transform_exprs, + .. + } = visitor; + self.finalize_scan_metadata(transformed, selection_vector, row_transform_exprs) } } @@ -907,7 +1054,7 @@ impl LogReplayProcessor for ScanLogReplayProcessor { // - Log batches: parse JSON for stats, MapToStruct for partition values // - Checkpoint batches: read pre-parsed columns directly when available let transform = if is_log_batch { - &self.log_transform + &self.commit_transform } else { &self.checkpoint_transform }; @@ -957,15 +1104,13 @@ impl LogReplayProcessor for ScanLogReplayProcessor { ); visitor.visit_rows_of(actions.as_ref())?; - // Step 4: Return transformed batch with updated selection vector - let scan_metadata = ScanMetadata::try_new( - transformed, - visitor.selection_vector, - visitor.row_transform_exprs, - )?; - self.metrics - .update_peak_hash_set_size(self.seen_file_keys.len()); - Ok(scan_metadata) + // Step 4: Remove internal-only columns and return the emitted scan metadata. + let AddRemoveDedupVisitor { + selection_vector, + row_transform_exprs, + .. + } = visitor; + self.finalize_scan_metadata(transformed, selection_vector, row_transform_exprs) } fn data_skipping_filter(&self) -> Option<&DataSkippingFilter> { @@ -1010,6 +1155,9 @@ pub(crate) fn scan_action_iter( Ok((processor.process_actions_iter(action_iter), metrics)) } +#[cfg(test)] +mod stats_expression_tests; + #[cfg(test)] mod tests { use std::collections::{HashMap, HashSet}; @@ -1024,7 +1172,7 @@ mod tests { use crate::actions::get_commit_schema; use crate::engine::sync::SyncEngine; use crate::expressions::{ - BinaryExpressionOp, Expression, OpaquePredicateOp, Predicate, Scalar, + BinaryExpressionOp, ColumnName, Expression, OpaquePredicateOp, Predicate, Scalar, ScalarExpressionEvaluator, UnaryExpressionOp, }; use crate::kernel_predicates::{ @@ -1143,6 +1291,7 @@ mod tests { physical_predicate: PhysicalPredicate::None, transform_spec: None, column_mapping_mode: ColumnMappingMode::None, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), @@ -1473,6 +1622,7 @@ mod tests { physical_predicate: PhysicalPredicate::None, transform_spec: None, column_mapping_mode: mode, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), @@ -1509,6 +1659,7 @@ mod tests { physical_predicate: PhysicalPredicate::None, transform_spec: None, column_mapping_mode: ColumnMappingMode::None, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), @@ -1541,6 +1692,7 @@ mod tests { physical_predicate: PhysicalPredicate::None, transform_spec: None, column_mapping_mode: ColumnMappingMode::None, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), @@ -1575,6 +1727,7 @@ mod tests { physical_predicate: PhysicalPredicate::None, transform_spec: None, column_mapping_mode: ColumnMappingMode::None, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), @@ -1624,6 +1777,7 @@ mod tests { transform_spec: None, column_mapping_mode: ColumnMappingMode::None, physical_stats_schema: None, + requested_output_stats_schema: None, stats_options: ScanStatsOptions::default(), partition_values_options: ScanPartitionValuesOptions::default(), physical_partition_schema: None, @@ -1656,6 +1810,7 @@ mod tests { transform_spec: None, column_mapping_mode: ColumnMappingMode::None, physical_stats_schema: None, + requested_output_stats_schema: None, stats_options: ScanStatsOptions::default(), partition_values_options: ScanPartitionValuesOptions::default(), physical_partition_schema: None, @@ -1901,9 +2056,15 @@ mod tests { // Synthesis enabled: COALESCE branch present -> exactly one ToJson. let with_synthesis = get_add_transform_expr( Some(stats_schema.clone()), - true, // has_stats_parsed - false, // skip_stats - true, // synthesize_json + true, // has_raw_stats + true, // allow_raw_stats_for_typed + true, // has_typed_stats_parsed + true, // has_stats_parsed_for_json + &ScanStatsOptions { + skip_stats: false, + synthesize_json: true, + ..Default::default() + }, partition_schema.clone(), false, // has_partition_values_parsed ); @@ -1916,9 +2077,15 @@ mod tests { // Synthesis disabled: no ToJson anywhere in the transform. let without_synthesis = get_add_transform_expr( Some(stats_schema), - true, // has_stats_parsed - false, // skip_stats - false, // synthesize_json + true, // has_raw_stats + true, // allow_raw_stats_for_typed + true, // has_typed_stats_parsed + true, // has_stats_parsed_for_json + &ScanStatsOptions { + skip_stats: false, + synthesize_json: false, + ..Default::default() + }, partition_schema, false, // has_partition_values_parsed ); diff --git a/kernel/src/scan/log_replay/stats_expression_tests.rs b/kernel/src/scan/log_replay/stats_expression_tests.rs new file mode 100644 index 0000000000..5bb2484342 --- /dev/null +++ b/kernel/src/scan/log_replay/stats_expression_tests.rs @@ -0,0 +1,334 @@ +use crate::Engine; +use std::sync::Arc; + +use super::{ + get_add_transform_expr, ScanLogReplayProcessor, ScanPartitionValuesOptions, ScanStatsOptions, + SerializableScanState, +}; +use crate::engine::sync::SyncEngine; +use crate::expressions::{ColumnName, Expression, Scalar, UnaryExpressionOp}; +use crate::log_segment::{CheckpointReadInfo, LogSegment}; +use crate::scan::state_info::tests::get_simple_state_info; +use crate::schema::{DataType, SchemaRef, StructField, StructType}; + +fn stats_schema() -> SchemaRef { + Arc::new(StructType::new_unchecked([ + StructField::nullable("numRecords", DataType::LONG), + StructField::nullable("value", DataType::STRING), + ])) +} + +fn any_expression(expr: &Expression, predicate: &impl Fn(&Expression) -> bool) -> bool { + if predicate(expr) { + return true; + } + match expr { + Expression::Unary(unary) => any_expression(&unary.expr, predicate), + Expression::Binary(binary) => { + any_expression(&binary.left, predicate) || any_expression(&binary.right, predicate) + } + Expression::Variadic(variadic) => variadic + .exprs + .iter() + .any(|expr| any_expression(expr, predicate)), + Expression::Struct(fields, nullability) => { + fields.iter().any(|expr| any_expression(expr, predicate)) + || nullability + .as_ref() + .is_some_and(|expr| any_expression(expr, predicate)) + } + Expression::StructPatch(patch) => patch + .prepended_fields + .iter() + .chain(&patch.appended_fields) + .chain( + patch + .field_patches + .values() + .flat_map(|field| field.insertions.iter()), + ) + .any(|expr| any_expression(expr, predicate)), + Expression::ParseJson(parse) => any_expression(&parse.json_expr, predicate), + Expression::MapToStruct(map) => any_expression(&map.map_expr, predicate), + Expression::Predicate(_) + | Expression::Literal(_) + | Expression::Column(_) + | Expression::Opaque(_) + | Expression::Unknown(_) => false, + } +} + +fn fields(expr: &Expression) -> &[Arc] { + let Expression::Struct(fields, _) = expr else { + panic!("expected struct expression, got {expr:?}"); + }; + fields +} + +fn serializable_state(options: ScanStatsOptions, engine: &dyn Engine) -> SerializableScanState { + ScanLogReplayProcessor::new( + engine, + Arc::new(get_simple_state_info(stats_schema(), vec![]).unwrap()), + CheckpointReadInfo::without_stats_parsed(), + options, + ScanPartitionValuesOptions::default(), + ) + .unwrap() + .into_serializable_state() + .unwrap() +} + +#[test] +fn scan_stats_options_missing_fallback_defaults_true_and_false_at_state_blob_boundary() { + let engine = SyncEngine::new(); + let mut old_state = serializable_state(ScanStatsOptions { + checkpoint_stats_json_fallback: false, + ..Default::default() + }, &engine); + let mut internal: serde_json::Value = + serde_json::from_slice(&old_state.internal_state_blob).unwrap(); + assert_eq!( + internal["stats_options"]["checkpoint_stats_json_fallback"], + false + ); + internal["stats_options"] + .as_object_mut() + .unwrap() + .remove("checkpoint_stats_json_fallback"); + old_state.internal_state_blob = serde_json::to_vec(&internal).unwrap(); + + let restored_old = ScanLogReplayProcessor::from_serializable_state(&engine, old_state).unwrap(); + assert!(restored_old.stats_options.checkpoint_stats_json_fallback); + + let configured = ScanLogReplayProcessor::from_serializable_state( + &engine, + serializable_state(ScanStatsOptions { + checkpoint_stats_json_fallback: false, + ..Default::default() + }, &engine), + ) + .unwrap(); + assert!(!configured.stats_options.checkpoint_stats_json_fallback); +} + +#[test] +fn commit_transform_preserves_raw_stats_and_parses_without_to_json() { + let expr = get_add_transform_expr( + Some(stats_schema()), + true, // has_raw_stats + true, // allow_raw_stats_for_typed + false, // has_typed_stats_parsed + false, // has_stats_parsed_for_json + &ScanStatsOptions { + synthesize_json: false, + ..Default::default() + }, + None, + false, + ); + let fields = fields(&expr); + + assert_eq!( + fields[3].as_ref(), + &Expression::column(["add", "stats"]), + "commit raw stats must pass through unchanged" + ); + assert!(matches!(fields[6].as_ref(), Expression::ParseJson(_))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Unary(unary) if unary.op == UnaryExpressionOp::ToJson + ))); +} + +#[test] +fn synthesis_without_raw_stats_uses_only_stats_parsed() { + let expr = get_add_transform_expr( + None, + false, // has_raw_stats + false, // allow_raw_stats_for_typed + false, // has_typed_stats_parsed + true, // has_stats_parsed_for_json + &ScanStatsOptions::default(), + None, + false, + ); + let fields = fields(&expr); + + assert!(matches!( + fields[3].as_ref(), + Expression::Unary(unary) if unary.op == UnaryExpressionOp::ToJson + )); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Column(name) if name == &ColumnName::new(["add", "stats"]) + ))); +} + +#[test] +fn compatible_checkpoint_uses_typed_stats_without_json_operations() { + let expr = get_add_transform_expr( + Some(stats_schema()), + false, // has_raw_stats + false, // allow_raw_stats_for_typed + true, // has_typed_stats_parsed + true, // has_stats_parsed_for_json + &ScanStatsOptions { + synthesize_json: false, + checkpoint_stats_json_fallback: false, + ..Default::default() + }, + None, + false, + ); + let fields = fields(&expr); + + assert_eq!( + fields[3].as_ref(), + &Expression::Literal(Scalar::Null(DataType::STRING)) + ); + assert_eq!( + fields[6].as_ref(), + &Expression::column(["add", "stats_parsed"]) + ); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::ParseJson(_) + ))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Unary(unary) if unary.op == UnaryExpressionOp::ToJson + ))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Column(name) if name == &ColumnName::new(["add", "stats"]) + ))); +} + +#[test] +fn skip_stats_expression_has_no_json_operations_or_raw_stats_reference() { + let expr = get_add_transform_expr( + None, + true, // has_raw_stats + true, // allow_raw_stats_for_typed + true, // has_typed_stats_parsed + true, // has_stats_parsed_for_json + &ScanStatsOptions { + skip_stats: true, + ..Default::default() + }, + None, + false, + ); + + assert_eq!( + fields(&expr)[3].as_ref(), + &Expression::Literal(Scalar::Null(DataType::STRING)) + ); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::ParseJson(_) + ))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Unary(unary) if unary.op == UnaryExpressionOp::ToJson + ))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Column(name) if name == &ColumnName::new(["add", "stats"]) + ))); +} + +#[test] +fn incompatible_checkpoint_schema_without_fallback_emits_typed_null() { + let incompatible_stats = StructType::new_unchecked([ + StructField::nullable("numRecords", DataType::LONG), + StructField::nullable("value", DataType::INTEGER), + ]); + let checkpoint_schema = StructType::new_unchecked([StructField::nullable( + "add", + StructType::new_unchecked([StructField::nullable( + "stats_parsed", + incompatible_stats, + )]), + )]); + let requested_stats = stats_schema(); + + assert!(!LogSegment::schema_has_compatible_stats_parsed( + &checkpoint_schema, + requested_stats.as_ref() + )); + + let expr = get_add_transform_expr( + Some(requested_stats.clone()), + false, // incompatible stats_parsed is not projected; raw stats is forbidden + false, // allow_raw_stats_for_typed + false, // has_typed_stats_parsed + false, // has_stats_parsed_for_json + &ScanStatsOptions { + synthesize_json: false, + checkpoint_stats_json_fallback: false, + ..Default::default() + }, + None, + false, + ); + let fields = fields(&expr); + + assert_eq!( + fields[6].as_ref(), + &Expression::Literal(Scalar::Null(requested_stats.as_ref().clone().into())) + ); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Column(name) if name == &ColumnName::new(["add", "stats_parsed"]) + ))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Column(name) if name == &ColumnName::new(["add", "stats"]) + ))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::ParseJson(_) + ))); +} + +#[test] +fn checkpoint_without_raw_fallback_emits_typed_null_without_raw_reference() { + let stats_schema = stats_schema(); + let expr = get_add_transform_expr( + Some(stats_schema.clone()), + false, // has_raw_stats + false, // allow_raw_stats_for_typed + false, // has_typed_stats_parsed + false, // has_stats_parsed_for_json + &ScanStatsOptions { + synthesize_json: false, + checkpoint_stats_json_fallback: false, + ..Default::default() + }, + None, + false, + ); + let fields = fields(&expr); + + assert_eq!( + fields[3].as_ref(), + &Expression::Literal(Scalar::Null(DataType::STRING)) + ); + assert_eq!( + fields[6].as_ref(), + &Expression::Literal(Scalar::Null(stats_schema.as_ref().clone().into())) + ); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::ParseJson(_) + ))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Column(name) if name == &ColumnName::new(["add", "stats"]) + ))); + assert!(!any_expression(&expr, &|expr| matches!( + expr, + Expression::Unary(unary) if unary.op == UnaryExpressionOp::ToJson + ))); +} diff --git a/kernel/src/scan/mod.rs b/kernel/src/scan/mod.rs index 22b8c29fb3..407774ffe7 100644 --- a/kernel/src/scan/mod.rs +++ b/kernel/src/scan/mod.rs @@ -15,7 +15,7 @@ use self::log_replay::{get_scan_metadata_transform_expr, scan_action_iter}; use crate::actions::deletion_vector::{ deletion_treemap_to_bools, split_vector, DeletionVectorDescriptor, }; -use crate::actions::{Add, ADD_FIELD, ADD_NAME, REMOVE_FIELD}; +use crate::actions::{Add, ADD_FIELD, ADD_NAME, NUM_RECORDS, REMOVE_FIELD}; use crate::engine_data::FilteredEngineData; use crate::expressions::{ColumnName, ExpressionRef, Predicate, PredicateRef, Scalar}; use crate::kernel_predicates::{ @@ -28,19 +28,20 @@ use crate::metrics::events::emit_scan_metadata_completed; use crate::metrics::{MetricId, ScanType}; use crate::parallel::sequential_phase::SequentialPhase; use crate::scan::log_replay::{ + get_scan_metadata_transform_expr_with_parsed_columns, scan_row_schema_with_parsed_columns, ScanLogReplayProcessor, BASE_ROW_ID_NAME, CLUSTERING_PROVIDER_NAME, DEFAULT_ROW_COMMIT_VERSION_NAME, }; use crate::scan::metrics::ScanMetrics; use crate::scan::state_info::StateInfo; use crate::schema::{ - lazy_schema_ref, ArrayType, DataType, MapType, PrimitiveType, Schema, SchemaRef, StructField, - StructType, ToSchema as _, + lazy_schema_ref, ArrayType, DataType, MapType, PrimitiveType, Schema, SchemaRef, + SchemaStructPatchBuilder, StructField, StructType, ToSchema as _, }; use crate::table_features::{ColumnMappingMode, Operation}; use crate::transforms::{transform_output_type, ExpressionTransform, SchemaTransform}; use crate::utils::{FoldWithOption as _, IteratorExt}; -use crate::{DeltaResult, Engine, EngineData, Error, FileMeta, SnapshotRef, Version}; +use crate::{DeltaResult, Engine, EngineData, Error, ExpressionEvaluator, FileMeta, SnapshotRef, Version}; pub(crate) mod data_skipping; pub(crate) mod field_classifiers; @@ -60,13 +61,30 @@ pub(crate) static COMMIT_READ_SCHEMA: LazyLock = lazy_schema_ref! { (&ADD_FIELD), (&REMOVE_FIELD), }; +pub(crate) static COMMIT_READ_SCHEMA_NO_JSON_STATS: LazyLock = LazyLock::new(|| { + let schema = SchemaStructPatchBuilder::new() + .drop_at(["add"], "stats") + .drop_at(["remove"], "stats") + .build(&COMMIT_READ_SCHEMA) + .expect("dropping stats from commit read schema"); + Arc::new(schema) +}); + +fn commit_read_schema(skip_stats: bool) -> SchemaRef { + if skip_stats { + COMMIT_READ_SCHEMA_NO_JSON_STATS.clone() + } else { + COMMIT_READ_SCHEMA.clone() + } +} + pub(crate) static CHECKPOINT_READ_SCHEMA: LazyLock = lazy_schema_ref! { (&ADD_FIELD), }; -/// Checkpoint schema WITHOUT stats for column projection pushdown. -/// When skip_stats is enabled, we use this schema to avoid reading the stats column from parquet. -pub(crate) static CHECKPOINT_READ_SCHEMA_NO_STATS: LazyLock = LazyLock::new(|| { +/// Initial checkpoint projection without JSON `add.stats`. +/// Discovery restores JSON stats when structured stats cannot satisfy the scan. +pub(crate) static CHECKPOINT_READ_SCHEMA_NO_JSON_STATS: LazyLock = LazyLock::new(|| { let add_schema = Add::to_schema(); let fields_no_stats: Vec<_> = add_schema .fields() @@ -85,19 +103,20 @@ pub use crate::parallel::parallel_scan_metadata::{ AfterSequentialScanMetadata, ParallelScanMetadata, ParallelState, SequentialScanMetadata, }; -/// Engine-facing stats options. Pass to [`ScanBuilder::with_stats`] to declare -/// what stats data the engine wants in scan metadata output. Two orthogonal axes: -/// JSON stats (`add.stats`) and struct stats (`add.stats_parsed`). +/// Configures structured-stats output and JSON synthesis in scan metadata. +/// Existing JSON passes through for commits and checkpoints without compatible structured stats +/// unless stats are disabled. /// /// Most consumers should pick one of the named constructors: /// - [`Self::json_only`] (default) -- JSON stats only. -/// - [`Self::all_struct`] -- all struct stats, no JSON. Cheap path when the engine consumes -/// `stats_parsed` directly; avoids the per-batch `ToJson` cost. -/// - [`Self::struct_columns`] -- struct stats projected to a subset of columns, no JSON. +/// - [`Self::all_struct`] -- all struct stats without JSON synthesis. Compatible checkpoints omit +/// JSON stats; commits and checkpoints without compatible structured stats pass existing JSON +/// through. +/// - [`Self::struct_columns`] -- selected struct stats with the same JSON behavior. /// - [`Self::all`] -- both representations. /// - [`Self::none`] -- neither, AND disables internal data skipping. Unlike the other four /// constructors, this is the only one that stops kernel from reading stats from parquet at all. -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] pub struct StatsOptions { /// Whether to surface JSON stats on parsed-stats checkpoints (where the /// checkpoint writes stats only as a struct, not as JSON). When true, kernel @@ -111,10 +130,19 @@ pub struct StatsOptions { /// Which struct stats columns to emit in `stats_parsed`. pub(crate) struct_stats: StructStats, + + /// Whether typed statistics may fall back to checkpoint `add.stats` JSON when a compatible + /// `add.stats_parsed` struct is unavailable. + #[serde(default = "default_checkpoint_stats_json_fallback")] + checkpoint_stats_json_fallback: bool, +} + +fn default_checkpoint_stats_json_fallback() -> bool { + true } /// Which struct stats columns appear in `stats_parsed` in scan metadata output. -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] pub enum StructStats { /// Don't emit `stats_parsed`. Kernel still reads predicate-referenced stats for /// internal data skipping unless the caller picked [`StatsOptions::none`], which @@ -132,6 +160,7 @@ impl Default for StatsOptions { Self { synthesize_json: true, struct_stats: StructStats::None, + checkpoint_stats_json_fallback: true, } } } @@ -142,22 +171,24 @@ impl StatsOptions { Self::default() } - /// All struct stats, no JSON. Cheap path for engines that consume - /// `stats_parsed` directly: avoids the per-batch `ToJson` cost on - /// parsed-stats checkpoints. + /// All struct stats without JSON synthesis. Compatible checkpoints omit JSON stats and avoid + /// per-batch `ToJson`; commits and checkpoints without compatible structured stats pass + /// existing JSON through. pub fn all_struct() -> Self { Self { synthesize_json: false, struct_stats: StructStats::All, + checkpoint_stats_json_fallback: true, } } - /// Struct stats projected to the specified columns, no JSON. Like + /// Struct stats projected to the specified columns without JSON synthesis. Like /// [`Self::all_struct`] but narrowed to a subset of indexed columns. pub fn struct_columns(cols: Vec) -> Self { Self { synthesize_json: false, struct_stats: StructStats::Columns(cols), + checkpoint_stats_json_fallback: true, } } @@ -166,6 +197,7 @@ impl StatsOptions { Self { synthesize_json: true, struct_stats: StructStats::All, + checkpoint_stats_json_fallback: true, } } @@ -179,8 +211,16 @@ impl StatsOptions { Self { synthesize_json: false, struct_stats: StructStats::None, + checkpoint_stats_json_fallback: true, } } + + /// Configure whether typed statistics may fall back to raw checkpoint JSON when a compatible + /// `stats_parsed` struct is unavailable. + pub fn with_checkpoint_stats_json_fallback(mut self, enabled: bool) -> Self { + self.checkpoint_stats_json_fallback = enabled; + self + } } /// Engine-facing partition value options. Pass to [`ScanBuilder::with_partition_values`] to @@ -295,8 +335,8 @@ impl ScanBuilder { /// Configure stats output for the scan. See [`StatsOptions`]. /// /// Defaults to [`StatsOptions::default`] (JSON only). Engines that consume - /// `stats_parsed` directly should pass [`StatsOptions::all_struct`] to skip the - /// per-batch `ToJson` synthesis on parsed-stats checkpoints. + /// `stats_parsed` directly should pass [`StatsOptions::all_struct`] so compatible + /// checkpoints omit JSON stats and skip `ToJson` synthesis. pub fn with_stats(mut self, stats: StatsOptions) -> Self { self.stats = stats; self @@ -593,6 +633,16 @@ pub(crate) fn restored_add_schema() -> &'static SchemaRef { &RESTORED_ADD_SCHEMA } +fn restored_add_schema_with_parsed_columns(stats_schema: &StructType) -> DeltaResult { + let schema = SchemaStructPatchBuilder::new() + .append_at( + ["add"], + StructField::nullable(log_replay::STATS_PARSED_NAME, stats_schema.clone()), + ) + .build(&RESTORED_ADD_SCHEMA)?; + Ok(Arc::new(schema)) +} + /// utility method making it easy to get a transform for a particular row. If the requested row is /// outside the range of the passed slice returns `None`, otherwise returns the element at the index /// of the specified row @@ -624,6 +674,9 @@ pub struct ScanMetadata { /// Note: This vector can be indexed by row number, as rows masked by the selection vector will /// have corresponding entries that will be `None`. pub scan_file_transforms: Vec>, + + /// Whether requested scan output includes a typed `stats_parsed.numRecords` field. + has_typed_num_records: bool, } impl ScanMetadata { @@ -631,10 +684,13 @@ impl ScanMetadata { data: Box, selection_vector: Vec, scan_file_transforms: Vec>, + requested_output_stats_schema: Option<&SchemaRef>, ) -> DeltaResult { Ok(Self { scan_files: FilteredEngineData::try_new(data, selection_vector)?, scan_file_transforms, + has_typed_num_records: requested_output_stats_schema + .is_some_and(|schema| schema.field(NUM_RECORDS).is_some()), }) } } @@ -672,11 +728,40 @@ impl Scan { !self.stats.synthesize_json && matches!(self.stats.struct_stats, StructStats::None) } + fn checkpoint_read_options(&self) -> (SchemaRef, Option, Option<&StructType>) { + let skip_stats = self.skip_stats(); + // `physical_stats_schema` is the typed shape this scan can consume, not evidence that the + // checkpoint contains `stats_parsed`. Checkpoint discovery validates availability and + // restores `add.stats` before opening the reader when the structured field is incompatible. + let can_replace_json_with_structured_stats = + !self.stats.synthesize_json && self.state_info.physical_stats_schema.is_some(); + let checkpoint_schema = if skip_stats || can_replace_json_with_structured_stats { + CHECKPOINT_READ_SCHEMA_NO_JSON_STATS.clone() + } else { + CHECKPOINT_READ_SCHEMA.clone() + }; + + let meta_predicate = if skip_stats { + None + } else { + self.build_actions_meta_predicate() + }; + // Discovery uses this schema to augment the checkpoint projection, so `none()` must + // suppress it as well as the initial JSON stats field. + let physical_stats_schema = if skip_stats { + None + } else { + self.state_info.physical_stats_schema.as_deref() + }; + (checkpoint_schema, meta_predicate, physical_stats_schema) + } + /// Build the read-options bundle passed to [`ScanLogReplayProcessor`]. fn stats_options(&self) -> log_replay::ScanStatsOptions { log_replay::ScanStatsOptions { skip_stats: self.skip_stats(), synthesize_json: self.stats.synthesize_json, + checkpoint_stats_json_fallback: self.stats.checkpoint_stats_json_fallback, } } @@ -730,6 +815,16 @@ impl Scan { } } + /// Get the parsed statistics schema already planned for replay by this scan. + #[internal_api] + pub(crate) fn effective_replay_stats_schema(&self) -> Option<&SchemaRef> { + if self.skip_stats() { + None + } else { + self.state_info.physical_stats_schema.as_ref() + } + } + /// Get an iterator of [`ScanMetadata`]s that should be used to facilitate a scan. This handles /// log-replay, reconciling Add and Remove actions, and applying data skipping (if possible). /// @@ -803,6 +898,7 @@ impl Scan { /// # Parameters /// /// * `existing_version` - Table version the provided data was read from. + /// * `seeded_stats_parsed_schema` - Schema contract shared by every batch in `existing_data`. /// * `existing_data` - Existing processed scan metadata with all selection vectors applied. /// * `existing_predicate` - The predicate used by the previous scan. #[allow(unused)] @@ -811,7 +907,8 @@ impl Scan { &self, engine: &dyn Engine, existing_version: Version, - existing_data: impl IntoIterator> + 'static, + seeded_stats_parsed_schema: Option, + existing_data: impl IntoIterator>> + 'static, _existing_predicate: Option, ) -> DeltaResult>>> { // TODO(#966): validate that the current predicate is compatible with the hint predicate. @@ -824,37 +921,37 @@ impl Scan { ))); } - // in order to be processed by our log replay, we must re-shape the existing scan metadata - // back into shape as we read it from the log. Since it is already reconciled data, - // we treat it as if it originated from a checkpoint. - let transform = engine.evaluation_handler().new_expression_evaluator( - scan_row_schema(), - get_scan_metadata_transform_expr(), - restored_add_schema().clone().into(), - )?; - let apply_transform = move |data: Box| { - Ok(ActionsBatch::new(transform.evaluate(data.as_ref())?, false)) - }; + // // in order to be processed by our log replay, we must re-shape the existing scan metadata + // // back into shape as we read it from the log. Since it is already reconciled data, + // // we treat it as if it originated from a checkpoint. + // let transform = engine.evaluation_handler().new_expression_evaluator( + // scan_row_schema(), + // get_scan_metadata_transform_expr(), + // restored_add_schema().clone().into(), + // )?; + // let apply_transform = move |data: Box| { + // Ok(ActionsBatch::new(transform.evaluate(data.as_ref())?, false)) + // }; let log_segment = self.snapshot.log_segment(); - // If the snapshot version corresponds to the hint version, we process the existing data - // to apply file skipping and provide the required transformations. - // Since we're only processing existing data (no checkpoint), we use the base schema - // and no stats_parsed optimization. - if existing_version == self.snapshot.version() { - let actions_with_checkpoint_info = ActionsWithCheckpointInfo { - actions: existing_data.into_iter().map(apply_transform), - checkpoint_info: CheckpointReadInfo { - has_stats_parsed: false, - has_partition_values_parsed: false, - checkpoint_read_schema: restored_add_schema().clone(), - }, - }; - return Ok(Box::new( - self.scan_metadata_inner(engine, actions_with_checkpoint_info)?, - )); - } + // // If the snapshot version corresponds to the hint version, we process the existing data + // // to apply file skipping and provide the required transformations. + // // Since we're only processing existing data (no checkpoint), we use the base schema + // // and no stats_parsed optimization. + // if existing_version == self.snapshot.version() { + // let actions_with_checkpoint_info = ActionsWithCheckpointInfo { + // actions: existing_data.into_iter().map(apply_transform), + // checkpoint_info: CheckpointReadInfo { + // has_stats_parsed: false, + // has_partition_values_parsed: false, + // checkpoint_read_schema: restored_add_schema().clone(), + // }, + // }; + // return Ok(Box::new( + // self.scan_metadata_inner(engine, actions_with_checkpoint_info)?, + // )); + // } // If the current log segment contains a checkpoint newer than the hint version // we disregard the existing data hint, and perform a full scan. The current log segment @@ -865,6 +962,36 @@ impl Scan { return Ok(Box::new(self.scan_metadata(engine)?)); } + // Existing scan metadata is already reconciled, so reshape it into checkpoint-style Add + // actions with one evaluator built from the declared seed schema. The same evaluator and + // checkpoint contract serve same-version and incremental seeded replay. + let (transform, checkpoint_info) = seed_replay( + engine, + &self.state_info, + self.stats_options(), + seeded_stats_parsed_schema, + )?; + let seed_actions = existing_data.into_iter().map(move |data| { + let data = data?; + Ok(ActionsBatch::new( + transform.evaluate(data.as_ref())?, + false, + )) + }); + + // If the snapshot version corresponds to the hint version, process only the existing data + // to apply file skipping and provide the required transformations. + if existing_version == self.snapshot.version() { + let actions_with_checkpoint_info = ActionsWithCheckpointInfo { + actions: seed_actions, + checkpoint_info, + }; + return Ok(Box::new(self.scan_metadata_inner( + engine, + actions_with_checkpoint_info, + )?)); + } + // create a new log segment containing only the commits added after the version hint. let mut ascending_commit_files = log_segment.listed.ascending_commit_files.clone(); ascending_commit_files.retain(|f| f.version > existing_version); @@ -882,30 +1009,23 @@ impl Scan { // For incremental reads, new_log_segment has no checkpoint but we use the // checkpoint schema returned by the function for consistency. - let (checkpoint_schema, meta_predicate) = if self.skip_stats() { - (CHECKPOINT_READ_SCHEMA_NO_STATS.clone(), None) - } else { - ( - CHECKPOINT_READ_SCHEMA.clone(), - self.build_actions_meta_predicate(), - ) - }; - let result = new_log_segment.read_actions_with_projected_checkpoint_actions( + let (checkpoint_schema, meta_predicate, physical_stats_schema) = + self.checkpoint_read_options(); + + let result = new_log_segment.read_actions_with_projected_checkpoint_actions_for_scan( engine, - COMMIT_READ_SCHEMA.clone(), + commit_read_schema(self.skip_stats()), checkpoint_schema, meta_predicate, - self.state_info - .physical_stats_schema - .as_ref() - .map(|s| s.as_ref()), + physical_stats_schema, None, + &self.stats_options(), )?; let actions_with_checkpoint_info = ActionsWithCheckpointInfo { - actions: result - .actions - .chain(existing_data.into_iter().map(apply_transform)), - checkpoint_info: result.checkpoint_info, + // JSON-log batches retain their source transform. Only cached batches pass through the + // declared seed transform before both streams enter log replay. + actions: result.actions.chain(seed_actions), + checkpoint_info, }; Ok(Box::new(self.scan_metadata_inner( @@ -965,29 +1085,22 @@ impl Scan { ) -> DeltaResult< ActionsWithCheckpointInfo> + Send>, > { - let (checkpoint_schema, meta_predicate) = if self.skip_stats() { - (CHECKPOINT_READ_SCHEMA_NO_STATS.clone(), None) - } else { - ( - CHECKPOINT_READ_SCHEMA.clone(), - self.build_actions_meta_predicate(), - ) - }; + let (checkpoint_schema, meta_predicate, physical_stats_schema) = + self.checkpoint_read_options(); + self.snapshot .log_segment() - .read_actions_with_projected_checkpoint_actions( + .read_actions_with_projected_checkpoint_actions_for_scan( engine, - COMMIT_READ_SCHEMA.clone(), + commit_read_schema(self.skip_stats()), checkpoint_schema, meta_predicate, - self.state_info - .physical_stats_schema - .as_ref() - .map(|s| s.as_ref()), + physical_stats_schema, self.state_info .physical_partition_schema .as_ref() .map(|s| s.as_ref()), + &self.stats_options(), ) } @@ -1096,7 +1209,7 @@ impl Scan { // since SequentialPhase reads checkpoints via CheckpointManifestReader which doesn't // currently support stats_parsed optimization. let checkpoint_read_schema = if self.skip_stats() { - CHECKPOINT_READ_SCHEMA_NO_STATS.clone() + CHECKPOINT_READ_SCHEMA_NO_JSON_STATS.clone() } else { CHECKPOINT_READ_SCHEMA.clone() }; @@ -1242,6 +1355,51 @@ impl Scan { } } +fn seed_replay( + engine: &dyn Engine, + state_info: &StateInfo, + stats_options: log_replay::ScanStatsOptions, + seeded_stats_parsed_schema: Option, +) -> DeltaResult<(Arc, CheckpointReadInfo)> { + // let available_stats_schema = seeded_input.stats_parsed_schema; + let effective_replay_stats_schema = if stats_options.skip_stats { + None + } else { + state_info.physical_stats_schema.clone() + }; + let input_schema = scan_row_schema_with_parsed_columns(seeded_stats_parsed_schema.clone(), None)?; + + let (transform_expr, checkpoint_read_schema, has_stats_parsed) = match ( + seeded_stats_parsed_schema.as_ref(), + effective_replay_stats_schema.as_ref(), + ) { + (Some(available), Some(effective)) => ( + get_scan_metadata_transform_expr_with_parsed_columns(available, effective), + restored_add_schema_with_parsed_columns(effective)?, + true, + ), + _ => ( + get_scan_metadata_transform_expr(), + restored_add_schema().clone(), + false, + ), + }; + let transform = engine.evaluation_handler().new_expression_evaluator( + input_schema, + transform_expr, + checkpoint_read_schema.as_ref().clone().into(), + )?; + + Ok(( + transform, + CheckpointReadInfo { + has_stats_parsed, + has_partition_values_parsed: false, + checkpoint_read_schema, + }, + )) +} + /// Get the base schema that scan rows (from [`Scan::scan_metadata`]) will be returned with. /// /// This is the base shape; engines may add trailing `*_parsed` columns by opting in via diff --git a/kernel/src/scan/state.rs b/kernel/src/scan/state.rs index e8eda6a6c3..8a6a96cda3 100644 --- a/kernel/src/scan/state.rs +++ b/kernel/src/scan/state.rs @@ -7,9 +7,10 @@ use roaring::RoaringTreemap; use serde::Deserialize; use tracing::warn; -use super::log_replay::SCAN_ROW_SCHEMA; +use super::log_replay::{SCAN_ROW_SCHEMA, STATS_PARSED_NAME}; use super::ScanMetadata; use crate::actions::deletion_vector::{deletion_treemap_to_bools, DeletionVectorDescriptor}; +use crate::actions::NUM_RECORDS; use crate::actions::visitors::visit_deletion_vector_at; use crate::engine_data::{FilteredRowVisitor, GetData, RowIndexIterator, TypedGetData}; use crate::scan::get_transform_for_row; @@ -161,6 +162,7 @@ impl ScanMetadata { callback, transforms: &self.scan_file_transforms, context, + has_typed_num_records: self.has_typed_num_records, }; visitor.visit_rows_of(&self.scan_files)?; Ok(visitor.context) @@ -171,23 +173,40 @@ struct ScanFileVisitor<'a, T> { callback: ScanCallback, transforms: &'a [Option], context: T, + has_typed_num_records: bool, } + impl FilteredRowVisitor for ScanFileVisitor<'_, T> { fn selected_column_names_and_types(&self) -> (&'static [ColumnName], &'static [DataType]) { static NAMES_AND_TYPES: LazyLock = LazyLock::new(|| SCAN_ROW_SCHEMA.leaves(None)); - NAMES_AND_TYPES.as_ref() + static WIDENED_NAMES_AND_TYPES: LazyLock = LazyLock::new(|| { + let (names, types) = NAMES_AND_TYPES.as_ref(); + let mut names = names.to_vec(); + let mut types = types.to_vec(); + names.push(ColumnName::new([STATS_PARSED_NAME, NUM_RECORDS])); + types.push(DataType::LONG); + (names, types).into() + }); + if self.has_typed_num_records { + WIDENED_NAMES_AND_TYPES.as_ref() + } else { + NAMES_AND_TYPES.as_ref() + } } fn visit_filtered<'a>( &mut self, getters: &[&'a dyn GetData<'a>], rows: RowIndexIterator<'_>, ) -> DeltaResult<()> { + let expected_len = self.selected_column_names_and_types().0.len(); + require!( - getters.len() == 14, + getters.len() == expected_len, Error::InternalError(format!( - "Wrong number of ScanFileVisitor getters: {}", - getters.len() + "Wrong number of ScanFileVisitor getters: {} (expected {})", + getters.len(), + expected_len, )) ); for row_index in rows { @@ -195,15 +214,28 @@ impl FilteredRowVisitor for ScanFileVisitor<'_, T> { if let Some(path) = getters[0].get_opt(row_index, "scanFile.path")? { let size = getters[1].get(row_index, "scanFile.size")?; let modification_time: i64 = getters[2].get(row_index, "add.modificationTime")?; - let stats: Option = getters[3].get_opt(row_index, "scanFile.stats")?; - let stats: Option = - stats.and_then(|json| match serde_json::from_str(json.as_str()) { + let stats_json: Option = + getters[3].get_opt(row_index, "scanFile.stats")?; + let mut stats: Option = stats_json.as_deref().and_then(|json| { + match serde_json::from_str(json) { Ok(stats) => Some(stats), Err(e) => { warn!("Invalid stats string in Add file {json}: {}", e); None } - }); + } + }); + if stats_json.is_none() && self.has_typed_num_records { + let num_records: Option = getters + .last() + .expect("typed numRecords getter must be present") + .get_opt(row_index, "scanFile.stats_parsed.numRecords")?; + if let Some(num_records) = num_records.filter(|value| *value >= 0) { + stats = Some(Stats { + num_records: num_records as u64, + }); + } + } let dv_index = SCAN_ROW_SCHEMA .index_of("deletionVector") diff --git a/kernel/src/scan/state_info.rs b/kernel/src/scan/state_info.rs index 6de5df87c0..4222f58ab0 100644 --- a/kernel/src/scan/state_info.rs +++ b/kernel/src/scan/state_info.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use tracing::{debug, enabled, warn, Level}; -use crate::actions::NULL_COUNT; +use crate::actions::{NULL_COUNT, NUM_RECORDS, TIGHT_BOUNDS}; use crate::expressions::ColumnName; use crate::scan::field_classifiers::TransformFieldClassifier; use crate::scan::transform_spec::{FieldTransformSpec, TransformSpec}; @@ -31,6 +31,11 @@ pub(crate) struct StateInfo { pub(crate) column_mapping_mode: ColumnMappingMode, /// Physical stats schema for reading/parsing stats from checkpoint files. /// Used to construct checkpoint read schema with stats_parsed. + /// Parsed statistics requested by the engine for scan metadata output. + pub(crate) requested_output_stats_schema: Option, + /// Effective physical statistics schema used while replaying the log. This is the union of + /// requested output fields and stats-eligible predicate fields, and can be wider than the + /// emitted schema. pub(crate) physical_stats_schema: Option, /// Physical partition schema with native types for `partitionValues_parsed`. Fields use /// physical column names (for column mapping) and are always nullable. Present when the @@ -121,112 +126,102 @@ fn validate_metadata_columns<'a>( Ok(metadata_info) } -/// Build data-skipping schemas based on `StructStats` and `PhysicalPredicate`. +/// Build the requested output and effective replay schemas used for data skipping. /// -/// Returns `(physical_stats_schema, physical_partition_schema)`, where: -/// - `physical_stats_schema` contains data-column stats for `stats_parsed`. -/// - `physical_partition_schema` contains typed partition values for `partitionValues_parsed`. -/// -/// All three arms route through `TableConfiguration::build_expected_stats_schemas`: the -/// `All` arm with no `requested_physical_columns` filter, and the two scoped arms with -/// the union of requested + predicate-referenced columns. That path applies the same -/// `BaseStatsTransform` -> `MinMaxStatsTransform` pipeline writers use, so the read-side -/// stats schema's shape matches the write-side exactly. +/// Returns `(requested_output_stats_schema, physical_stats_schema)`. fn build_data_skipping_schemas( struct_stats: &StructStats, physical_predicate: &PhysicalPredicate, predicate_column_names_logical: &[ColumnName], table_configuration: &TableConfiguration, - table_partition_schema: Option, ) -> DeltaResult<(Option, Option)> { - // Filter partition schema to only predicate-referenced columns. The DataSkippingFilter - // only needs partition columns that appear in the predicate, and the transform output - // should not include unused partition columns. - let predicate_partition_schema = match (&table_partition_schema, physical_predicate) { - (Some(tps), PhysicalPredicate::Some(_, ref_schema)) => { - // Partition values extracted from the string map via MapToStruct are always - // nullable (map lookup can return null), so we force all partition fields nullable. - ref_schema - .with_fields_filtered_nonempty(|f| tps.field(f.name()).is_some())? - .map(|partition_schema| { - let nullable_fields = partition_schema - .fields() - .map(|f| StructField::nullable(f.name(), f.data_type().clone())); - Arc::new(StructType::new_unchecked(nullable_fields)) - }) - } - _ => None, + let logical_schema = table_configuration.logical_schema(); + let physical_schema = table_configuration.physical_schema(); + let column_mapping_mode = table_configuration.column_mapping_mode(); + let resolve_logical_path = |column: &ColumnName| { + get_any_level_column_physical_name(&logical_schema, column, column_mapping_mode) + .inspect_err(|error| { + warn!("Failed to resolve physical name for column {column}: {error}") + }) + .ok() }; - - // `DataSkippingFilter` needs stats for every column its predicate references. Refs - // without stats fold to NULL and pruning collapses to "keep every file", even when - // the caller separately requested stats for some other set of columns via - // `StructStats::Columns`. Union the two so the schema serves both. Unresolvable - // refs (e.g. a predicate typo) are dropped here. - let union_to_physical = |requested_logical: &[ColumnName]| -> Vec { - let mut union_logical: Vec = requested_logical.to_vec(); - let existing: HashSet<&ColumnName> = requested_logical.iter().collect(); - for col in predicate_column_names_logical { - if !existing.contains(col) { - union_logical.push(col.clone()); - } - } - let logical_schema = table_configuration.logical_schema(); - let column_mapping_mode = table_configuration.column_mapping_mode(); - union_logical + let logical_to_physical = |columns: &[ColumnName]| -> Vec { + columns.iter().filter_map(&resolve_logical_path).collect() + }; + let requested_to_physical = |columns: &[ColumnName]| -> Vec { + columns .iter() - .filter_map(|col| { - get_any_level_column_physical_name(&logical_schema, col, column_mapping_mode) - .inspect_err(|e| warn!("Failed to resolve physical name for column {col}: {e}")) - .ok() + .filter_map(|column| { + if physical_schema.field_at(column).is_ok() { + Some(column.clone()) + } else { + resolve_logical_path(column) + } }) .collect() }; - - // A stats schema with only `numRecords` and `tightBounds` (the bookkeeping fields - // `build_expected_stats_schemas` always emits) has nothing to prune by. Return `None` - // in that case so the caller skips building a `DataSkippingFilter`. `nullCount` is the - // per-column stats wrapper, so its presence is the signal that at least one data - // column survived. The Delta protocol allows `minValues` / `maxValues` without - // `nullCount`, but `build_expected_stats_schemas` always emits `nullCount` whenever it - // emits min/max; this check relies on that implementation property. - let with_data_cols = |stats_schema: SchemaRef| -> Option { - stats_schema - .field(NULL_COUNT) - .is_some() - .then_some(stats_schema) + let build_selected_stats_schema = |columns: &[ColumnName]| -> DeltaResult { + if columns.is_empty() { + return Ok(Arc::new(StructType::new_unchecked([ + StructField::nullable(NUM_RECORDS, DataType::LONG), + StructField::nullable(TIGHT_BOUNDS, DataType::BOOLEAN), + ]))); + } + Ok(table_configuration + .build_expected_stats_schemas(None, Some(columns))? + .physical) }; - let stats_schema = match (struct_stats, physical_predicate) { - // Full table stats schema for stats_parsed. - (StructStats::All, _) => with_data_cols( + // Requested paths may already be physical. Otherwise, resolve them as logical paths. Predicate + // paths are always logical and are physicalized exactly once. + let requested_physical_columns = match struct_stats { + StructStats::Columns(columns) => Some(requested_to_physical(columns)), + _ => None, + }; + let requested_output_stats_schema = match (struct_stats, &requested_physical_columns) { + (StructStats::None, _) => None, + (StructStats::All, _) => Some( table_configuration .build_expected_stats_schemas(None, None)? .physical, ), - // Explicit requested columns. Union in predicate refs so the stats schema covers - // both sources. - (StructStats::Columns(requested_columns), _) if !requested_columns.is_empty() => { - let requested_physical = union_to_physical(requested_columns); - with_data_cols( - table_configuration - .build_expected_stats_schemas(None, Some(&requested_physical))? - .physical, - ) + (StructStats::Columns(_), Some(columns)) => { + Some(build_selected_stats_schema(columns)?) + } + (StructStats::Columns(_), None) => unreachable!(), + }; + + let predicate_physical_columns = + if matches!(physical_predicate, PhysicalPredicate::Some(_, _)) { + logical_to_physical(predicate_column_names_logical) + } else { + Vec::new() + }; + + let physical_stats_schema = match (struct_stats, requested_physical_columns) { + (StructStats::All, _) => requested_output_stats_schema.clone(), + (StructStats::Columns(_), Some(mut replay_physical_columns)) => { + for predicate_column in predicate_physical_columns { + if !replay_physical_columns.contains(&predicate_column) { + replay_physical_columns.push(predicate_column); + } + } + Some(build_selected_stats_schema(&replay_physical_columns)?) } - // No explicit requested columns, but a predicate is present. Use just the predicate - // refs so the stats schema is trimmed to what the rewritten predicate needs. - (_, PhysicalPredicate::Some(_, _)) => { - let predicate_refs_physical = union_to_physical(&[]); - with_data_cols( - table_configuration - .build_expected_stats_schemas(None, Some(&predicate_refs_physical))? - .physical, - ) + (StructStats::Columns(_), None) => unreachable!(), + (StructStats::None, _) if predicate_physical_columns.is_empty() => None, + (StructStats::None, _) => { + let stats_schema = table_configuration + .build_expected_stats_schemas(None, Some(&predicate_physical_columns))? + .physical; + stats_schema + .field(NULL_COUNT) + .is_some() + .then_some(stats_schema) } - (_, _) => None, }; - Ok((stats_schema, predicate_partition_schema)) + + Ok((requested_output_stats_schema, physical_stats_schema)) } impl StateInfo { @@ -384,6 +379,14 @@ impl StateInfo { } } + let (requested_output_stats_schema, physical_stats_schema) = + build_data_skipping_schemas( + &stats.struct_stats, + &physical_predicate, + &predicate_column_names, + table_configuration, + )?; + // Build partition schema with physical names, used for partition pruning in data // skipping and for the engine-facing `partitionValues_parsed` output column. Needed // when partition columns exist and either a predicate is present or the engine @@ -414,13 +417,24 @@ impl StateInfo { None }; - let (physical_stats_schema, predicate_partition_schema) = build_data_skipping_schemas( - &stats.struct_stats, - &physical_predicate, - &predicate_column_names, - table_configuration, - table_partition_schema.clone(), - )?; + // Filter partition schema to only predicate-referenced columns. The DataSkippingFilter + // only needs partition columns that appear in the predicate, and the transform output + // should not include unused partition columns. + let predicate_partition_schema = match (&table_partition_schema, &physical_predicate) { + (Some(tps), PhysicalPredicate::Some(_, ref_schema)) => { + // Partition values extracted from the string map via MapToStruct are always + // nullable (map lookup can return null), so we force all partition fields nullable. + ref_schema + .with_fields_filtered_nonempty(|f| tps.field(f.name()).is_some())? + .map(|partition_schema| { + let nullable_fields = partition_schema + .fields() + .map(|f| StructField::nullable(f.name(), f.data_type().clone())); + Arc::new(StructType::new_unchecked(nullable_fields)) + }) + } + _ => None, + }; // When the engine requested the typed struct, emit all partition columns rather than // the predicate-narrowed subset. The data skipping filter only references the columns @@ -452,6 +466,7 @@ impl StateInfo { physical_predicate, transform_spec, column_mapping_mode, + requested_output_stats_schema, physical_stats_schema, physical_partition_schema, physical_stats_columns, @@ -1079,6 +1094,7 @@ pub(crate) mod tests { StatsOptions { synthesize_json: true, struct_stats: StructStats::Columns(vec![column_name!("value")]), + ..Default::default() }, ) .unwrap(); @@ -1125,6 +1141,7 @@ pub(crate) mod tests { StatsOptions { synthesize_json: true, struct_stats: StructStats::Columns(vec![column_name!("value")]), + ..Default::default() }, ) .unwrap(); @@ -1281,6 +1298,7 @@ pub(crate) mod tests { StatsOptions { synthesize_json: true, struct_stats: StructStats::Columns(vec![column_name!("col_a")]), + ..Default::default() }, ) .unwrap(); diff --git a/kernel/src/scan/test_utils.rs b/kernel/src/scan/test_utils.rs index aaeaed5752..693ae05c05 100644 --- a/kernel/src/scan/test_utils.rs +++ b/kernel/src/scan/test_utils.rs @@ -164,6 +164,7 @@ pub(crate) fn run_with_validate_callback( physical_predicate: PhysicalPredicate::None, transform_spec, column_mapping_mode: ColumnMappingMode::None, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), diff --git a/kernel/src/scan/tests.rs b/kernel/src/scan/tests.rs index 951d64d657..28105c4591 100644 --- a/kernel/src/scan/tests.rs +++ b/kernel/src/scan/tests.rs @@ -1,11 +1,14 @@ +use std::fs; use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use ::test_utils::{get_column, load_test_data}; use bytes::Bytes; use rstest::rstest; use super::*; + +mod stats_read_policy; use crate::actions::{MAX_VALUES, MIN_VALUES, NULL_COUNT, NUM_RECORDS}; use crate::arrow::array::{Array, BooleanArray, Int64Array, StringArray, StructArray}; use crate::arrow::compute::filter_record_batch; @@ -662,7 +665,13 @@ fn test_scan_metadata_from_same_version() { .try_collect() .unwrap(); let new_files: Vec<_> = scan - .scan_metadata_from(engine.as_ref(), version, files, None) + .scan_metadata_from( + engine.as_ref(), + version, + None, + files.into_iter().map(Ok), + None, + ) .unwrap() .try_collect() .unwrap(); @@ -708,7 +717,13 @@ fn test_scan_metadata_from_with_update() { .unwrap(); let scan = snapshot.scan_builder().build().unwrap(); let new_files: Vec<_> = scan - .scan_metadata_from(engine.as_ref(), 0, files, None) + .scan_metadata_from( + engine.as_ref(), + 0, + None, + files.into_iter().map(Ok), + None, + ) .unwrap() .map_ok(|ScanMetadata { scan_files, .. }| { let (underlying_data, selection_vector) = scan_files.into_parts(); @@ -1396,6 +1411,221 @@ fn test_checkpoint_row_group_skipping( } } +#[derive(Debug)] +struct RecordedParquetRead { + files: Vec, + physical_schema: schema::SchemaRef, + predicate: Option, +} + +struct RecordingParquetHandler { + inner: Arc, + reads: Mutex>, +} + +impl RecordingParquetHandler { + fn new(inner: Arc) -> Self { + Self { + inner, + reads: Mutex::new(Vec::new()), + } + } + + fn take_reads(&self) -> Vec { + std::mem::take(&mut *self.reads.lock().unwrap()) + } +} + +impl ParquetHandler for RecordingParquetHandler { + fn read_parquet_files( + &self, + files: &[FileMeta], + physical_schema: schema::SchemaRef, + predicate: Option, + ) -> DeltaResult { + self.reads.lock().unwrap().push(RecordedParquetRead { + files: files.iter().map(|file| file.location.to_string()).collect(), + physical_schema: physical_schema.clone(), + predicate: predicate.clone(), + }); + self.inner + .read_parquet_files(files, physical_schema, predicate) + } + + fn read_parquet_footer(&self, file: &FileMeta) -> DeltaResult { + self.inner.read_parquet_footer(file) + } + + fn write_parquet_file( + &self, + location: url::Url, + data: DeltaResultIteratorStatic>, + ) -> DeltaResult<()> { + self.inner.write_parquet_file(location, data) + } +} + +struct RecordingParquetEngine { + inner: Arc, + parquet: Arc, +} + +impl RecordingParquetEngine { + fn new(inner: Arc) -> Self { + Self { + parquet: Arc::new(RecordingParquetHandler { + inner: inner.parquet_handler(), + reads: Mutex::new(Vec::new()), + }), + inner, + } + } + + fn take_reads(&self) -> Vec { + std::mem::take(&mut *self.parquet.reads.lock().unwrap()) + } +} + +impl Engine for RecordingParquetEngine { + fn evaluation_handler(&self) -> Arc { + self.inner.evaluation_handler() + } + + fn json_handler(&self) -> Arc { + self.inner.json_handler() + } + + fn parquet_handler(&self) -> Arc { + self.parquet.clone() + } + + fn storage_handler(&self) -> Arc { + self.inner.storage_handler() + } +} + +#[rstest] +#[case::all_struct(StatsOptions::all_struct(), false, false)] +#[case::all(StatsOptions::all(), true, false)] +#[case::none_with_predicate(StatsOptions::none(), false, true)] +fn test_checkpoint_stats_projection_matches_requested_output( + #[values( + "v1-single-part-struct-stats-only", + "v2-parquet-sidecars-struct-stats-only", + "v2-checkpoints-parquet-with-sidecars" + )] + table: &str, + #[case] stats: StatsOptions, + #[case] request_json_stats: bool, + #[case] skip_stats: bool, +) { + let extracted = load_test_data("tests/data", table).ok(); + let path = extracted + .as_ref() + .map(|dir| dir.path().join(table)) + .unwrap_or_else(|| { + fs::canonicalize(PathBuf::from(format!("./tests/data/{table}/"))).unwrap() + }); + let url = Url::from_directory_path(path).unwrap(); + let engine = RecordingParquetEngine::new(Arc::new(SyncEngine::new())); + let snapshot = Snapshot::builder_for(url).build(&engine).unwrap(); + engine.take_reads(); + + let predicate: Option = skip_stats + .then(|| Arc::new(Pred::gt(column_expr!("id"), Expr::literal(0i64))) as PredicateRef); + let scan = snapshot + .scan_builder() + .with_predicate(predicate) + .with_stats(stats) + .build() + .unwrap(); + for action in scan.replay_for_scan_metadata(&engine).unwrap().actions { + action.unwrap(); + } + + let reads = engine.take_reads(); + let compatible_structured_stats = table != "v2-checkpoints-parquet-with-sidecars"; + let expect_parsed_stats = !skip_stats && compatible_structured_stats; + let expect_json_stats = !skip_stats && (request_json_stats || !expect_parsed_stats); + let expected_file_fragment = if table.starts_with("v2-") { + "_sidecars/" + } else { + ".checkpoint." + }; + let action_reads: Vec<_> = reads + .iter() + .filter(|read| { + read.files + .iter() + .any(|file| file.contains(expected_file_fragment)) + && read.physical_schema.field("add").is_some() + }) + .collect(); + assert!(!action_reads.is_empty(), "expected checkpoint Add reads"); + for read in action_reads { + let add_field = read + .physical_schema + .field("add") + .expect("checkpoint read schema must contain add"); + let DataType::Struct(add) = add_field.data_type() else { + panic!("checkpoint add field must be a struct"); + }; + assert_eq!( + add.field("stats").is_some(), + expect_json_stats, + "JSON checkpoint stats projection must match the requested output" + ); + assert_eq!( + add.field("stats_parsed").is_some(), + expect_parsed_stats, + "structured checkpoint stats projection must match the requested output" + ); + } +} + +#[test] +fn test_all_struct_parses_json_commit_stats() { + let path = fs::canonicalize(PathBuf::from( + "./tests/data/v1-single-part-struct-stats-only/", + )) + .unwrap(); + let url = Url::from_directory_path(path).unwrap(); + let engine = Arc::new(SyncEngine::new()); + // The table's checkpoint is at version 5, so version 4 replays only JSON commits. + let snapshot = Snapshot::builder_for(url) + .at_version(4) + .build(engine.as_ref()) + .unwrap(); + let scan = snapshot + .scan_builder() + .with_stats(StatsOptions::all_struct()) + .build() + .unwrap(); + + let mut file_count = 0; + for scan_metadata in scan + .scan_metadata(engine.as_ref()) + .unwrap() + .collect::, _>>() + .unwrap() + { + let (underlying_data, selection_vector) = scan_metadata.scan_files.into_parts(); + let batch: RecordBatch = ArrowEngineData::try_from_engine_data(underlying_data) + .unwrap() + .into(); + let filtered = filter_record_batch(&batch, &BooleanArray::from(selection_vector)).unwrap(); + let stats_parsed = get_column!(filtered, "stats_parsed", StructArray); + let num_records = get_column!(stats_parsed, NUM_RECORDS, Int64Array); + + for row in 0..filtered.num_rows() { + assert!(!stats_parsed.is_null(row)); + assert_eq!(num_records.value(row), 1); + file_count += 1; + } + } + assert_eq!(file_count, 4); +} + #[test] fn test_skip_stats_disables_data_skipping() { let path = std::fs::canonicalize(PathBuf::from("./tests/data/parsed-stats/")).unwrap(); @@ -1508,6 +1738,7 @@ fn test_default_stats_options_no_struct_output() { #[case::with_json(StatsOptions { synthesize_json: true, struct_stats: StructStats::Columns(vec![column_name!("id")]), + ..Default::default() })] #[case::struct_columns_ctor(StatsOptions::struct_columns(vec![column_name!("id")]))] fn test_scan_metadata_with_specific_stats_columns(#[case] stats: StatsOptions) { @@ -1562,6 +1793,7 @@ fn test_scan_metadata_with_multiple_stats_columns() { .with_stats(StatsOptions { synthesize_json: true, struct_stats: StructStats::Columns(vec![column_name!("id"), column_name!("name")]), + ..Default::default() }) .build() .unwrap(); @@ -1634,6 +1866,7 @@ fn test_scan_metadata_with_nonexistent_stats_columns() { .with_stats(StatsOptions { synthesize_json: true, struct_stats: StructStats::Columns(vec![column_name!("nonexistent_column")]), + ..Default::default() }) .build() .unwrap(); diff --git a/kernel/src/scan/tests/stats_read_policy.rs b/kernel/src/scan/tests/stats_read_policy.rs new file mode 100644 index 0000000000..a29e76319a --- /dev/null +++ b/kernel/src/scan/tests/stats_read_policy.rs @@ -0,0 +1,758 @@ +use std::path::PathBuf; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; + +use super::super::{ + ScanMetadata, StatsOptions, StructStats, COMMIT_READ_SCHEMA, + COMMIT_READ_SCHEMA_NO_JSON_STATS, +}; +use crate::actions::{ADD_NAME, REMOVE_NAME}; +use crate::arrow::array::{Array, Int64Array, StringArray, StructArray}; +use crate::arrow::record_batch::RecordBatch; +use crate::engine::arrow_data::ArrowEngineData; +use crate::engine::sync::SyncEngine; +use crate::expressions::{ColumnName, Expression, Predicate}; +use crate::schema::{DataType, SchemaRef}; +use crate::{ + DeltaResult, DeltaResultIterator, DeltaResultIteratorStatic, Engine, EngineData, + EvaluationHandler, FileDataReadResultIterator, FileMeta, FilteredEngineData, JsonHandler, + ParquetFooter, ParquetHandler, PredicateRef, Snapshot, StorageHandler, +}; + +const RAW_CHECKPOINT_PATH: &str = + "part-00000-a190be9e-e3df-439e-b366-06a863f51e99-c000.snappy.parquet"; +const COMMIT_4_PATH: &str = + "part-00000-40525115-50e1-4475-aae1-c8edc59274e6-c000.snappy.parquet"; +const COMMIT_5_PATH: &str = + "part-00000-c0cbdedc-d11b-4e4c-b6f2-5f40c55ef515-c000.snappy.parquet"; +const COMMIT_4_STATS: &str = r#"{"numRecords":100,"minValues":{"id":401,"name":"name_401","age":20,"salary":90100,"ts_col":"1970-01-01T00:00:09.000Z"},"maxValues":{"id":500,"name":"name_500","age":69,"salary":100000,"ts_col":"1970-01-01T00:00:10.000Z"},"nullCount":{"id":0,"name":0,"age":0,"salary":0,"ts_col":0}}"#; +const COMMIT_5_STATS: &str = r#"{"numRecords":100,"minValues":{"id":501,"name":"name_501","age":20,"salary":100100,"ts_col":"1970-01-01T00:00:11.000Z"},"maxValues":{"id":600,"name":"name_600","age":69,"salary":110000,"ts_col":"1970-01-01T00:00:12.000Z"},"nullCount":{"id":0,"name":0,"age":0,"salary":0,"ts_col":0}}"#; +const PARSED_STATS_PATHS: [&str; 6] = [ + "part-00000-065eae2b-b4ea-4708-bb30-0888f35cabdd-c000.snappy.parquet", + "part-00000-06d85a38-b141-479b-a315-4157335e9a11-c000.snappy.parquet", + "part-00000-2d9663e0-37c0-425e-98df-2e7141f9b5fb-c000.snappy.parquet", + COMMIT_4_PATH, + "part-00000-a4c1def5-742e-4248-8c58-fc9f4018e43d-c000.snappy.parquet", + COMMIT_5_PATH, +]; + +#[derive(Clone, Copy, Default)] +struct ForbiddenStats { + raw: bool, + parsed: bool, +} + +#[derive(Clone)] +struct RequestedSchema { + files: Vec, + schema: SchemaRef, +} + +fn action_has_field(schema: &SchemaRef, action_name: &str, field_name: &str) -> bool { + schema + .field(action_name) + .and_then(|field| match field.data_type() { + DataType::Struct(action) => action.field(field_name), + _ => None, + }) + .is_some() +} + +fn check_schema(schema: &SchemaRef, forbidden: ForbiddenStats, handler: &str) -> DeltaResult<()> { + if forbidden.raw + && [ADD_NAME, REMOVE_NAME] + .iter() + .any(|action| action_has_field(schema, action, "stats")) + { + return Err(crate::Error::generic(format!( + "{handler} handler received forbidden raw stats schema" + ))); + } + if forbidden.parsed && action_has_field(schema, ADD_NAME, "stats_parsed") { + return Err(crate::Error::generic(format!( + "{handler} handler received forbidden parsed stats schema" + ))); + } + Ok(()) +} + +fn request(files: &[FileMeta], schema: SchemaRef) -> RequestedSchema { + RequestedSchema { + files: files + .iter() + .map(|file| file.location.path().to_string()) + .collect(), + schema, + } +} + +struct SchemaCheckingJsonHandler { + inner: Arc, + calls: Arc, + requests: Arc>>, + forbidden: ForbiddenStats, +} + +impl JsonHandler for SchemaCheckingJsonHandler { + fn parse_json( + &self, + json_strings: Box, + output_schema: SchemaRef, + ) -> DeltaResult> { + self.inner.parse_json(json_strings, output_schema) + } + + fn read_json_files( + &self, + files: &[FileMeta], + physical_schema: SchemaRef, + predicate: Option, + ) -> DeltaResult { + if !files.is_empty() { + self.calls.fetch_add(1, Ordering::Relaxed); + self.requests + .lock() + .unwrap() + .push(request(files, physical_schema.clone())); + } + check_schema(&physical_schema, self.forbidden, "JSON")?; + self.inner + .read_json_files(files, physical_schema, predicate) + } + + fn write_json_file( + &self, + path: &url::Url, + data: DeltaResultIterator<'_, FilteredEngineData>, + overwrite: bool, + ) -> DeltaResult<()> { + self.inner.write_json_file(path, data, overwrite) + } +} + +struct SchemaCheckingParquetHandler { + inner: Arc, + calls: Arc, + requests: Arc>>, + forbidden: ForbiddenStats, +} + +impl ParquetHandler for SchemaCheckingParquetHandler { + fn read_parquet_files( + &self, + files: &[FileMeta], + physical_schema: SchemaRef, + predicate: Option, + ) -> DeltaResult { + if !files.is_empty() { + self.calls.fetch_add(1, Ordering::Relaxed); + self.requests + .lock() + .unwrap() + .push(request(files, physical_schema.clone())); + } + check_schema(&physical_schema, self.forbidden, "Parquet")?; + self.inner + .read_parquet_files(files, physical_schema, predicate) + } + + fn write_parquet_file( + &self, + location: url::Url, + data: DeltaResultIteratorStatic>, + ) -> DeltaResult<()> { + self.inner.write_parquet_file(location, data) + } + + fn read_parquet_footer(&self, file: &FileMeta) -> DeltaResult { + self.inner.read_parquet_footer(file) + } +} + +struct SchemaCheckingEngine { + inner: Arc, + json_handler: Arc, + parquet_handler: Arc, +} + +impl SchemaCheckingEngine { + fn new(json_forbidden: ForbiddenStats, parquet_forbidden: ForbiddenStats) -> Self { + let inner = Arc::new(SyncEngine::new()); + let json_handler = Arc::new(SchemaCheckingJsonHandler { + inner: inner.json_handler(), + calls: Arc::new(AtomicUsize::new(0)), + requests: Arc::new(Mutex::new(vec![])), + forbidden: json_forbidden, + }); + let parquet_handler = Arc::new(SchemaCheckingParquetHandler { + inner: inner.parquet_handler(), + calls: Arc::new(AtomicUsize::new(0)), + requests: Arc::new(Mutex::new(vec![])), + forbidden: parquet_forbidden, + }); + Self { + inner, + json_handler, + parquet_handler, + } + } + + fn reset_requests(&self) { + self.json_handler.calls.store(0, Ordering::Relaxed); + self.parquet_handler.calls.store(0, Ordering::Relaxed); + self.json_handler.requests.lock().unwrap().clear(); + self.parquet_handler.requests.lock().unwrap().clear(); + } + + fn json_calls(&self) -> usize { + self.json_handler.calls.load(Ordering::Relaxed) + } + + fn parquet_calls(&self) -> usize { + self.parquet_handler.calls.load(Ordering::Relaxed) + } + + fn json_requests(&self) -> Vec { + self.json_handler.requests.lock().unwrap().clone() + } + + fn parquet_requests(&self) -> Vec { + self.parquet_handler.requests.lock().unwrap().clone() + } +} + +impl Engine for SchemaCheckingEngine { + fn evaluation_handler(&self) -> Arc { + self.inner.evaluation_handler() + } + + fn storage_handler(&self) -> Arc { + self.inner.storage_handler() + } + + fn json_handler(&self) -> Arc { + self.json_handler.clone() + } + + fn parquet_handler(&self) -> Arc { + self.parquet_handler.clone() + } +} + +fn snapshot_at(path: &str, engine: &dyn Engine) -> Arc { + let path = std::fs::canonicalize(PathBuf::from(path)).unwrap(); + let url = url::Url::from_directory_path(path).unwrap(); + Snapshot::builder_for(url).build(engine).unwrap() +} + +fn snapshot_at_version(path: &str, version: u64, engine: &dyn Engine) -> Arc { + let path = std::fs::canonicalize(PathBuf::from(path)).unwrap(); + let url = url::Url::from_directory_path(path).unwrap(); + Snapshot::builder_for(url) + .at_version(version) + .build(engine) + .unwrap() +} + +fn assert_requests_stats( + handler: &str, + requests: &[RequestedSchema], + raw: bool, + parsed: bool, +) { + assert!(!requests.is_empty(), "{handler} handler was not called"); + for request in requests { + assert!(!request.files.is_empty(), "{handler} request had no files"); + assert_eq!( + action_has_field(&request.schema, ADD_NAME, "stats"), + raw, + "unexpected add.stats projection for {handler} request {:?}", + request.files + ); + assert_eq!( + action_has_field(&request.schema, ADD_NAME, "stats_parsed"), + parsed, + "unexpected add.stats_parsed projection for {handler} request {:?}", + request.files + ); + } +} + +fn assert_request_extension(handler: &str, requests: &[RequestedSchema], extension: &str) { + assert!(!requests.is_empty(), "{handler} handler was not called"); + assert!( + requests + .iter() + .flat_map(|request| &request.files) + .all(|file| file.ends_with(extension)), + "{handler} request did not use only {extension} files" + ); +} + +#[derive(Debug)] +struct MetadataRow { + path: String, + selected: bool, + stats: Option, + stats_parsed_is_null: Option, + num_records: Option, + min_id: Option, +} + +fn metadata_rows(batches: Vec) -> Vec { + let mut rows = Vec::new(); + for scan_metadata in batches { + let (data, selection_vector) = scan_metadata.scan_files.into_parts(); + let batch: RecordBatch = ArrowEngineData::try_from_engine_data(data).unwrap().into(); + let paths = batch + .column_by_name("path") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let stats = batch + .column_by_name("stats") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let stats_parsed = batch + .column_by_name("stats_parsed") + .map(|array| array.as_any().downcast_ref::().unwrap()); + let num_records = stats_parsed.and_then(|parsed| { + parsed + .column_by_name("numRecords") + .map(|array| array.as_any().downcast_ref::().unwrap()) + }); + let min_id = stats_parsed + .and_then(|parsed| parsed.column_by_name("minValues")) + .map(|array| array.as_any().downcast_ref::().unwrap()) + .and_then(|min_values| min_values.column_by_name("id")) + .map(|array| array.as_any().downcast_ref::().unwrap()); + + for row in 0..batch.num_rows() { + if paths.is_null(row) { + continue; + } + rows.push(MetadataRow { + path: paths.value(row).to_string(), + selected: selection_vector.get(row).copied().unwrap_or(true), + stats: (!stats.is_null(row)).then(|| stats.value(row).to_string()), + stats_parsed_is_null: stats_parsed.map(|parsed| parsed.is_null(row)), + num_records: num_records + .filter(|values| !values.is_null(row)) + .map(|values| values.value(row)), + min_id: min_id + .filter(|values| !values.is_null(row)) + .map(|values| values.value(row)), + }); + } + } + rows +} + +fn collect_metadata(scan: &super::super::Scan, engine: &dyn Engine) -> Vec { + scan.scan_metadata(engine) + .unwrap() + .collect::, _>>() + .unwrap() +} + +fn selected_paths(rows: &[MetadataRow]) -> Vec { + let mut paths: Vec<_> = rows + .iter() + .filter(|row| row.selected) + .map(|row| row.path.clone()) + .collect(); + paths.sort(); + paths +} + +fn expected_parsed_stats_paths() -> Vec { + let mut paths: Vec<_> = PARSED_STATS_PATHS + .iter() + .map(|path| path.to_string()) + .collect(); + paths.sort(); + paths +} + +#[test] +fn existing_stats_options_enable_checkpoint_json_fallback() { + let options = [ + StatsOptions::default(), + StatsOptions::json_only(), + StatsOptions::all_struct(), + StatsOptions::struct_columns(vec![ColumnName::new(["id"])]), + StatsOptions::all(), + StatsOptions::none(), + ]; + + assert!(options + .iter() + .all(|option| option.checkpoint_stats_json_fallback)); +} + +#[test] +fn builder_disables_checkpoint_json_fallback_without_changing_projection() { + let option = StatsOptions::struct_columns(vec![ColumnName::new(["id"])]) + .with_checkpoint_stats_json_fallback(false); + + assert!(!option.checkpoint_stats_json_fallback); + assert!(!option.synthesize_json); + assert!(matches!(option.struct_stats, StructStats::Columns(_))); +} + +#[test] +fn stats_options_missing_fallback_field_defaults_true() { + let mut serialized = serde_json::to_value(StatsOptions::all_struct()).unwrap(); + serialized + .as_object_mut() + .unwrap() + .remove("checkpoint_stats_json_fallback"); + + let deserialized: StatsOptions = serde_json::from_value(serialized).unwrap(); + assert!(deserialized.checkpoint_stats_json_fallback); +} + +#[test] +fn no_stats_commit_schema_preserves_actions_but_omits_nested_stats() { + for action_name in [ADD_NAME, REMOVE_NAME] { + let expected: Vec<_> = match COMMIT_READ_SCHEMA.field(action_name).unwrap().data_type() { + DataType::Struct(action) => action + .fields() + .filter(|field| field.name() != "stats") + .map(|field| field.name().to_string()) + .collect(), + _ => panic!("{action_name} must be a struct"), + }; + let actual: Vec<_> = match COMMIT_READ_SCHEMA_NO_JSON_STATS + .field(action_name) + .unwrap() + .data_type() + { + DataType::Struct(action) => action + .fields() + .map(|field| field.name().to_string()) + .collect(), + _ => panic!("{action_name} must be a struct"), + }; + + assert_eq!(actual, expected); + assert!(!action_has_field( + &COMMIT_READ_SCHEMA_NO_JSON_STATS, + action_name, + "stats" + )); + } +} + +#[test] +fn existing_typed_options_request_raw_only_checkpoint_stats_and_parse_them() { + for (name, options) in [ + ("all_struct", StatsOptions::all_struct()), + ( + "struct_columns", + StatsOptions::struct_columns(vec![ColumnName::new(["int"])]), + ), + ] { + let engine = SchemaCheckingEngine::new( + ForbiddenStats { + raw: false, + parsed: true, + }, + ForbiddenStats { + raw: false, + parsed: true, + }, + ); + let checkpoint_snapshot = snapshot_at_version( + "./tests/data/with_checkpoint_no_last_checkpoint/", + 2, + &engine, + ); + let latest_snapshot = snapshot_at( + "./tests/data/with_checkpoint_no_last_checkpoint/", + &engine, + ); + engine.reset_requests(); + + let checkpoint_scan = checkpoint_snapshot + .scan_builder() + .with_stats(options.clone()) + .build() + .unwrap(); + let rows = metadata_rows(collect_metadata(&checkpoint_scan, &engine)); + let latest_scan = latest_snapshot + .scan_builder() + .with_stats(options) + .build() + .unwrap(); + let _ = collect_metadata(&latest_scan, &engine); + + assert!(engine.json_calls() > 0, "{name}: commit handler was not called"); + assert!( + engine.parquet_calls() > 0, + "{name}: checkpoint handler was not called" + ); + let json_requests = engine.json_requests(); + let parquet_requests = engine.parquet_requests(); + assert_requests_stats("JSON", &json_requests, true, false); + assert_requests_stats("Parquet", &parquet_requests, true, false); + assert_request_extension("JSON", &json_requests, ".json"); + assert_request_extension("Parquet", &parquet_requests, ".parquet"); + + assert_eq!(selected_paths(&rows), vec![RAW_CHECKPOINT_PATH]); + let row = rows.iter().find(|row| row.path == RAW_CHECKPOINT_PATH).unwrap(); + assert_eq!(row.stats_parsed_is_null, Some(false), "{name}"); + assert_eq!(row.num_records, Some(5), "{name}"); + } +} + +#[test] +fn fallback_disabled_raw_only_checkpoint_omits_stats_and_keeps_file_with_typed_null() { + for (name, options) in [ + ( + "all_struct", + StatsOptions::all_struct().with_checkpoint_stats_json_fallback(false), + ), + ( + "struct_columns", + StatsOptions::struct_columns(vec![ColumnName::new(["int"])]) + .with_checkpoint_stats_json_fallback(false), + ), + ] { + let engine = SchemaCheckingEngine::new( + ForbiddenStats::default(), + ForbiddenStats { + raw: true, + parsed: true, + }, + ); + let snapshot = snapshot_at_version( + "./tests/data/with_checkpoint_no_last_checkpoint/", + 2, + &engine, + ); + let latest_snapshot = snapshot_at( + "./tests/data/with_checkpoint_no_last_checkpoint/", + &engine, + ); + engine.reset_requests(); + let predicate = Arc::new(Predicate::gt( + Expression::column(["int"]), + Expression::literal(10_000i64), + )); + + let scan = snapshot + .scan_builder() + .with_predicate(predicate) + .with_stats(options.clone()) + .build() + .unwrap(); + let rows = metadata_rows(collect_metadata(&scan, &engine)); + let latest_scan = latest_snapshot + .scan_builder() + .with_stats(options) + .build() + .unwrap(); + let _ = collect_metadata(&latest_scan, &engine); + + assert!(engine.json_calls() > 0, "{name}: commit handler was not called"); + assert!( + engine.parquet_calls() > 0, + "{name}: checkpoint handler was not called" + ); + let json_requests = engine.json_requests(); + let parquet_requests = engine.parquet_requests(); + assert_requests_stats("JSON", &json_requests, true, false); + assert_requests_stats("Parquet", &parquet_requests, false, false); + assert_request_extension("JSON", &json_requests, ".json"); + assert_request_extension("Parquet", &parquet_requests, ".parquet"); + + assert_eq!( + selected_paths(&rows), + vec![RAW_CHECKPOINT_PATH], + "{name}: missing stats must conservatively keep the checkpoint file" + ); + let row = rows.iter().find(|row| row.path == RAW_CHECKPOINT_PATH).unwrap(); + assert_eq!(row.stats, None, "{name}"); + assert_eq!(row.stats_parsed_is_null, Some(true), "{name}"); + assert_eq!(row.num_records, None, "{name}"); + } +} + +#[test] +fn compatible_checkpoint_and_newer_commits_use_their_own_stats_sources() { + let engine = SchemaCheckingEngine::new( + ForbiddenStats { + raw: false, + parsed: true, + }, + ForbiddenStats { + raw: true, + parsed: false, + }, + ); + let snapshot = snapshot_at("./tests/data/parsed-stats/", &engine); + engine.reset_requests(); + + let scan = snapshot + .scan_builder() + .with_stats( + StatsOptions::struct_columns(vec![ColumnName::new(["id"])]) + .with_checkpoint_stats_json_fallback(false), + ) + .build() + .unwrap(); + let rows = metadata_rows(collect_metadata(&scan, &engine)); + + assert!(engine.json_calls() > 0, "commit handler was not called"); + assert!( + engine.parquet_calls() > 0, + "checkpoint handler was not called" + ); + let json_requests = engine.json_requests(); + let parquet_requests = engine.parquet_requests(); + assert_requests_stats("JSON", &json_requests, true, false); + assert_requests_stats("Parquet", &parquet_requests, false, true); + assert_request_extension("JSON", &json_requests, ".json"); + assert_request_extension("Parquet", &parquet_requests, ".parquet"); + + assert_eq!(selected_paths(&rows), expected_parsed_stats_paths()); + for (path, expected_json, expected_min_id) in [ + (COMMIT_4_PATH, COMMIT_4_STATS, 401), + (COMMIT_5_PATH, COMMIT_5_STATS, 501), + ] { + let row = rows.iter().find(|row| row.path == path).unwrap(); + assert!(row.selected); + assert_eq!(row.stats.as_deref(), Some(expected_json)); + assert_eq!(row.stats_parsed_is_null, Some(false)); + assert_eq!(row.num_records, Some(100)); + assert_eq!(row.min_id, Some(expected_min_id)); + } + + for row in rows + .iter() + .filter(|row| row.path != COMMIT_4_PATH && row.path != COMMIT_5_PATH) + { + assert!(row.selected); + assert_eq!(row.stats, None, "checkpoint raw JSON must not be projected"); + assert_eq!(row.stats_parsed_is_null, Some(false)); + assert_eq!(row.num_records, Some(100)); + assert!(row.min_id.is_some()); + } +} + +#[test] +fn none_requests_no_stats_and_preserves_the_complete_active_file_set() { + let forbidden = ForbiddenStats { + raw: true, + parsed: true, + }; + let engine = SchemaCheckingEngine::new(forbidden, forbidden); + let snapshot = snapshot_at("./tests/data/parsed-stats/", &engine); + engine.reset_requests(); + + let scan = snapshot + .scan_builder() + .with_stats(StatsOptions::none()) + .build() + .unwrap(); + let rows = metadata_rows(collect_metadata(&scan, &engine)); + + assert!(engine.json_calls() > 0, "commit handler was not called"); + assert!( + engine.parquet_calls() > 0, + "checkpoint handler was not called" + ); + let json_requests = engine.json_requests(); + let parquet_requests = engine.parquet_requests(); + assert_requests_stats("JSON", &json_requests, false, false); + assert_requests_stats("Parquet", &parquet_requests, false, false); + assert_request_extension("JSON", &json_requests, ".json"); + assert_request_extension("Parquet", &parquet_requests, ".parquet"); + + assert_eq!(selected_paths(&rows), expected_parsed_stats_paths()); + assert_eq!(rows.iter().filter(|row| row.selected).count(), 6); + assert!(rows + .iter() + .filter(|row| row.selected) + .all(|row| row.stats.is_none() && row.stats_parsed_is_null.is_none())); +} + +#[test] +fn seeded_typed_stats_and_newer_json_commits_preserve_both_sources() { + let seed_engine = SyncEngine::new(); + let seed_snapshot = snapshot_at_version("./tests/data/parsed-stats/", 3, &seed_engine); + let seed_scan = seed_snapshot + .scan_builder() + .with_stats( + StatsOptions::all_struct().with_checkpoint_stats_json_fallback(false), + ) + .build() + .unwrap(); + let seed_stats_schema = seed_scan.state_info.requested_output_stats_schema.clone(); + let seed_data: Vec<_> = seed_scan + .scan_metadata(&seed_engine) + .unwrap() + .map(|result| result.and_then(|metadata| metadata.scan_files.apply_selection_vector())) + .collect::, _>>() + .unwrap(); + + let engine = SchemaCheckingEngine::new( + ForbiddenStats { + raw: false, + parsed: true, + }, + ForbiddenStats::default(), + ); + let snapshot = snapshot_at("./tests/data/parsed-stats/", &engine); + let scan = snapshot + .scan_builder() + .with_stats( + StatsOptions::all_struct().with_checkpoint_stats_json_fallback(false), + ) + .build() + .unwrap(); + engine.reset_requests(); + + let batches = scan + .scan_metadata_from( + &engine, + 3, + seed_stats_schema, + seed_data.into_iter().map(Ok), + None, + ) + .unwrap() + .collect::, _>>() + .unwrap(); + let rows = metadata_rows(batches); + + assert!(engine.json_calls() > 0, "new commit handler was not called"); + assert_eq!( + engine.parquet_calls(), + 0, + "seeded replay must not reread the checkpoint" + ); + let json_requests = engine.json_requests(); + assert_requests_stats("JSON", &json_requests, true, false); + assert_request_extension("JSON", &json_requests, ".json"); + assert_eq!(selected_paths(&rows), expected_parsed_stats_paths()); + + for (path, expected_json, expected_min_id) in [ + (COMMIT_4_PATH, COMMIT_4_STATS, 401), + (COMMIT_5_PATH, COMMIT_5_STATS, 501), + ] { + let row = rows.iter().find(|row| row.path == path).unwrap(); + assert_eq!(row.stats.as_deref(), Some(expected_json)); + assert_eq!(row.stats_parsed_is_null, Some(false)); + assert_eq!(row.num_records, Some(100)); + assert_eq!(row.min_id, Some(expected_min_id)); + } + assert!(rows + .iter() + .filter(|row| row.path != COMMIT_4_PATH && row.path != COMMIT_5_PATH) + .all(|row| row.stats_parsed_is_null == Some(false) && row.num_records == Some(100))); +} diff --git a/kernel/src/snapshot/mod.rs b/kernel/src/snapshot/mod.rs index 63d912eefc..9277abf73d 100644 --- a/kernel/src/snapshot/mod.rs +++ b/kernel/src/snapshot/mod.rs @@ -22,6 +22,7 @@ use crate::crc::{ use crate::expressions::ColumnName; use crate::incremental_scan::IncrementalScanBuilder; use crate::log_segment::{DomainMetadataMap, LogSegment}; +use crate::log_segment_files::LogSegmentFiles; use crate::metrics::events::{DOMAIN_METADATA_LOADED_SPAN, SET_TRANSACTION_LOADED_SPAN}; use crate::metrics::SnapshotLoadMetricContext; use crate::path::ParsedLogPath; @@ -67,13 +68,13 @@ pub enum CheckpointWriteResult { /// have a defined schema (which may change over time for any given table), specific version, and /// frozen log segment. pub struct Snapshot { - span: tracing::Span, - log_segment: LogSegment, - table_configuration: TableConfiguration, + pub span: tracing::Span, + pub log_segment: LogSegment, + pub table_configuration: TableConfiguration, /// CRC at this snapshot's version, eagerly resolved at construction time. `Some(crc)` /// means `crc.version == self.version()` and the CRC can be queried at zero I/O. `None` /// means no CRC was loadable (no CRC on disk at this version, or the read failed). - crc: Option>, + pub crc: Option>, } impl PartialEq for Snapshot { @@ -1596,7 +1597,6 @@ mod tests { } #[test] - #[ignore = "log compaction disabled (#2337)"] fn test_log_compaction_writer() { let path = std::fs::canonicalize(PathBuf::from("./tests/data/table-with-dv-small/")).unwrap(); @@ -1622,20 +1622,6 @@ mod tests { assert_result_error_with_message(result, "Invalid version range"); } - // TODO(#2337): remove this test when log compaction is re-enabled. - #[test] - fn test_log_compaction_writer_unsupported() { - let path = - std::fs::canonicalize(PathBuf::from("./tests/data/table-with-dv-small/")).unwrap(); - let url = url::Url::from_directory_path(path).unwrap(); - - let engine = SyncEngine::new(); - let snapshot = Snapshot::builder_for(url).build(&engine).unwrap(); - - let result = snapshot.log_compaction_writer(0, 1); - assert_result_error_with_message(result, "not currently supported"); - } - #[tokio::test] async fn test_timestamp_with_ict_disabled() -> Result<(), Box> { let store = Arc::new(InMemory::new()); diff --git a/kernel/src/table_changes/physical_to_logical.rs b/kernel/src/table_changes/physical_to_logical.rs index c7a5694ec5..a8cf6070be 100644 --- a/kernel/src/table_changes/physical_to_logical.rs +++ b/kernel/src/table_changes/physical_to_logical.rs @@ -186,6 +186,7 @@ mod tests { physical_predicate: PhysicalPredicate::None, transform_spec: Some(Arc::new(transform_spec)), column_mapping_mode: ColumnMappingMode::None, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), @@ -410,6 +411,7 @@ mod tests { physical_predicate: PhysicalPredicate::None, transform_spec: Some(Arc::new(transform_spec)), column_mapping_mode: ColumnMappingMode::None, + requested_output_stats_schema: None, physical_stats_schema: None, physical_partition_schema: None, physical_stats_columns: HashSet::new(), diff --git a/kernel/src/table_configuration.rs b/kernel/src/table_configuration.rs index 1602f562ec..b99c833d78 100644 --- a/kernel/src/table_configuration.rs +++ b/kernel/src/table_configuration.rs @@ -102,20 +102,20 @@ fn validate_partition_columns(metadata: &Metadata, logical_schema: &StructType) #[internal_api] #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct TableConfiguration { - metadata: Metadata, - protocol: Protocol, + pub metadata: Metadata, + pub protocol: Protocol, /// Logical schema: field names are the user-facing (logical) column names. - logical_schema: SchemaRef, + pub logical_schema: SchemaRef, /// The subset of the logical schema that remains after excluding partition columns. - logical_schema_without_partition_columns: SchemaRef, + pub logical_schema_without_partition_columns: SchemaRef, /// Physical schema for all columns (field names respect column mapping mode). - physical_schema: SchemaRef, + pub physical_schema: SchemaRef, /// The subset of the physical schema that remains after excluding partition columns. - physical_data_schema_without_partition_columns: SchemaRef, - table_properties: TableProperties, - column_mapping_mode: ColumnMappingMode, - table_root: Url, - version: Version, + pub physical_data_schema_without_partition_columns: SchemaRef, + pub table_properties: TableProperties, + pub column_mapping_mode: ColumnMappingMode, + pub table_root: Url, + pub version: Version, } impl TableConfiguration { @@ -165,7 +165,7 @@ impl TableConfiguration { ) } - fn try_new_inner( + pub fn try_new_inner( metadata: Metadata, protocol: Protocol, table_root: Url, diff --git a/kernel/tests/integration/log/log_compaction.rs b/kernel/tests/integration/log/log_compaction.rs index 3f7da96de3..001425723f 100644 --- a/kernel/tests/integration/log/log_compaction.rs +++ b/kernel/tests/integration/log/log_compaction.rs @@ -21,7 +21,6 @@ fn url_to_object_store_path(url: &Url) -> Result Result<(), Box> { let _ = tracing_subscriber::fmt::try_init(); @@ -192,7 +191,6 @@ async fn action_reconciliation_round_trip() -> Result<(), Box Result<(), Box> { let _ = tracing_subscriber::fmt::try_init(); diff --git a/kernel/tests/integration/read.rs b/kernel/tests/integration/read.rs index b64c9ee5bc..0ce8fccae6 100644 --- a/kernel/tests/integration/read.rs +++ b/kernel/tests/integration/read.rs @@ -2182,13 +2182,13 @@ fn checkpoint_stats_skipping( // writeStatsAsStruct=true, writeStatsAsJson=false (no JSON stats in checkpoint), // schema (id: long, value: string), 5 files with 1 row each, checkpoint at v5. // Cross-product covers all five checkpoint variants against four stats option -// shapes: ScanFile.stats should be populated via the COALESCE/ToJson fallback -// when both `json=true` and `struct_stats=All` are set; otherwise null on these -// struct-stats-only checkpoints. +// shapes: JSON output synthesizes stats from typed checkpoint values, while +// struct-only output uses the typed numRecords fallback. With no requested +// statistics, ScanFile.stats remains null. #[rstest::rstest] -#[case::default_json_only(StatsOptions::default(), false)] +#[case::default_json_only(StatsOptions::default(), true)] #[case::all_both(StatsOptions::all(), true)] -#[case::all_struct_only(StatsOptions::all_struct(), false)] +#[case::all_struct_only(StatsOptions::all_struct(), true)] #[case::none(StatsOptions::none(), false)] fn struct_stats_surfaced_in_scan_file( #[values( @@ -2312,9 +2312,10 @@ fn struct_stats_only_preserves_data_skipping( "data skipping via stats_parsed should leave only 2 files (id=4, id=5)" ); for scan_file in &scan_files { - assert!( - scan_file.stats.is_none(), - "ScanFile.stats must remain null when synthesis is skipped, path: {}", + assert_eq!( + scan_file.stats.as_ref().map(|stats| stats.num_records), + Some(1), + "ScanFile.stats must contain num_records=1 from typed stats_parsed fallback, not JSON synthesis, path: {}", scan_file.path ); }