chore!: rename log_schema to commit_schema (delta-io#1419)

Rename schema's for more accurately describe schemas. This is follow-up
from delta-io#1407

Author: emkornfield

kernel/examples/inspect-table/src/main.rs

@@ -4,7 +4,7 @@ use delta_kernel::actions::visitors::{
     SetTransactionVisitor,
 };
 use delta_kernel::actions::{
-    get_log_schema, ADD_NAME, CDC_NAME, METADATA_NAME, PROTOCOL_NAME, REMOVE_NAME,
+    get_commit_schema, ADD_NAME, CDC_NAME, METADATA_NAME, PROTOCOL_NAME, REMOVE_NAME,
     SET_TRANSACTION_NAME,
 };
 use delta_kernel::engine_data::{GetData, RowVisitor, TypedGetData as _};
@@ -70,7 +70,7 @@ enum Action {
 }
 
 static NAMES_AND_TYPES: LazyLock<ColumnNamesAndTypes> =
-    LazyLock::new(|| get_log_schema().leaves(None));
+    LazyLock::new(|| get_commit_schema().leaves(None));
 
 struct LogVisitor {
     actions: Vec<(Action, usize)>,
@@ -205,10 +205,11 @@ fn try_main() -> DeltaResult<()> {
             }
         }
         Commands::Actions { oldest_first } => {
-            let log_schema = get_log_schema();
-            let actions = snapshot
-                .log_segment()
-                .read_actions(&engine, log_schema.clone(), None)?;
+            let commit_schema = get_commit_schema();
+            let actions =
+                snapshot
+                    .log_segment()
+                    .read_actions(&engine, commit_schema.clone(), None)?;
 
             let mut visitor = LogVisitor::new();
             for action in actions {

kernel/src/actions/mod.rs

@@ -67,7 +67,7 @@ pub(crate) const DOMAIN_METADATA_NAME: &str = "domainMetadata";
 
 pub(crate) const INTERNAL_DOMAIN_PREFIX: &str = "delta.";
 
-static LOG_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
+static COMMIT_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
     Arc::new(StructType::new_unchecked([
         StructField::nullable(ADD_NAME, Add::to_schema()),
         StructField::nullable(REMOVE_NAME, Remove::to_schema()),
@@ -76,17 +76,16 @@ static LOG_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
         StructField::nullable(SET_TRANSACTION_NAME, SetTransaction::to_schema()),
         StructField::nullable(COMMIT_INFO_NAME, CommitInfo::to_schema()),
         StructField::nullable(CDC_NAME, Cdc::to_schema()),
-        StructField::nullable(CHECKPOINT_METADATA_NAME, CheckpointMetadata::to_schema()),
         StructField::nullable(DOMAIN_METADATA_NAME, DomainMetadata::to_schema()),
     ]))
 });
 
 static ALL_ACTIONS_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
     Arc::new(StructType::new_unchecked(
-        get_log_schema()
-            .fields()
-            .cloned()
-            .chain([StructField::nullable(SIDECAR_NAME, Sidecar::to_schema())]),
+        get_commit_schema().fields().cloned().chain([
+            StructField::nullable(CHECKPOINT_METADATA_NAME, CheckpointMetadata::to_schema()),
+            StructField::nullable(SIDECAR_NAME, Sidecar::to_schema()),
+        ]),
     ))
 });
 
@@ -121,8 +120,8 @@ static LOG_DOMAIN_METADATA_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
 #[internal_api]
 /// Gets the schema for all actions that can appear in commits
 /// logs.  This excludes actions that can only appear in checkpoints.
-pub(crate) fn get_log_schema() -> &'static SchemaRef {
-    &LOG_SCHEMA
+pub(crate) fn get_commit_schema() -> &'static SchemaRef {
+    &COMMIT_SCHEMA
 }
 
 #[internal_api]
@@ -1047,7 +1046,7 @@ mod tests {
 
     #[test]
     fn test_metadata_schema() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&[METADATA_NAME])
             .expect("Couldn't get metaData field");
 
@@ -1081,7 +1080,7 @@ mod tests {
 
     #[test]
     fn test_add_schema() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&[ADD_NAME])
             .expect("Couldn't get add field");
 
@@ -1139,7 +1138,7 @@ mod tests {
 
     #[test]
     fn test_remove_schema() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&[REMOVE_NAME])
             .expect("Couldn't get remove field");
         let expected = Arc::new(StructType::new_unchecked([StructField::nullable(
@@ -1162,7 +1161,7 @@ mod tests {
 
     #[test]
     fn test_cdc_schema() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&[CDC_NAME])
             .expect("Couldn't get cdc field");
         let expected = Arc::new(StructType::new_unchecked([StructField::nullable(
@@ -1195,7 +1194,7 @@ mod tests {
 
     #[test]
     fn test_checkpoint_metadata_schema() {
-        let schema = get_log_schema()
+        let schema = get_all_actions_schema()
             .project(&[CHECKPOINT_METADATA_NAME])
             .expect("Couldn't get checkpointMetadata field");
         let expected = Arc::new(StructType::new_unchecked([StructField::nullable(
@@ -1210,7 +1209,7 @@ mod tests {
 
     #[test]
     fn test_transaction_schema() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&["txn"])
             .expect("Couldn't get transaction field");
 
@@ -1227,7 +1226,7 @@ mod tests {
 
     #[test]
     fn test_commit_info_schema() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&["commitInfo"])
             .expect("Couldn't get commitInfo field");
 
@@ -1251,7 +1250,7 @@ mod tests {
 
     #[test]
     fn test_domain_metadata_schema() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&[DOMAIN_METADATA_NAME])
             .expect("Couldn't get domainMetadata field");
         let expected = Arc::new(StructType::new_unchecked([StructField::nullable(
@@ -1789,9 +1788,9 @@ mod tests {
         let metadata_id = metadata.id.clone();
 
         // test with the full log schema that wraps metadata in a "metaData" field
-        let log_schema = get_log_schema().project(&[METADATA_NAME]).unwrap();
+        let commit_schema = get_commit_schema().project(&[METADATA_NAME]).unwrap();
         let actual: RecordBatch = metadata
-            .into_engine_data(log_schema, &engine)
+            .into_engine_data(commit_schema, &engine)
             .unwrap()
             .into_any()
             .downcast::<ArrowEngineData>()
@@ -1886,8 +1885,8 @@ mod tests {
         assert_eq!(record_batch, expected);
 
         // test with the full log schema that wraps protocol in a "protocol" field
-        let log_schema = get_log_schema().project(&[PROTOCOL_NAME]).unwrap();
-        let engine_data = protocol.into_engine_data(log_schema, &engine);
+        let commit_schema = get_commit_schema().project(&[PROTOCOL_NAME]).unwrap();
+        let engine_data = protocol.into_engine_data(commit_schema, &engine);
 
         let schema = Arc::new(Schema::new(vec![Field::new(
             "protocol",
@@ -1996,7 +1995,7 @@ mod tests {
 
     #[test]
     fn test_schema_contains_file_actions_with_add() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&[ADD_NAME, PROTOCOL_NAME])
             .unwrap();
         assert!(schema_contains_file_actions(&schema));
@@ -2007,7 +2006,7 @@ mod tests {
 
     #[test]
     fn test_schema_contains_file_actions_with_remove() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&[REMOVE_NAME, METADATA_NAME])
             .unwrap();
         assert!(schema_contains_file_actions(&schema));
@@ -2018,13 +2017,15 @@ mod tests {
 
     #[test]
     fn test_schema_contains_file_actions_with_both() {
-        let schema = get_log_schema().project(&[ADD_NAME, REMOVE_NAME]).unwrap();
+        let schema = get_commit_schema()
+            .project(&[ADD_NAME, REMOVE_NAME])
+            .unwrap();
         assert!(schema_contains_file_actions(&schema));
     }
 
     #[test]
     fn test_schema_contains_file_actions_with_neither() {
-        let schema = get_log_schema()
+        let schema = get_commit_schema()
             .project(&[PROTOCOL_NAME, METADATA_NAME])
             .unwrap();
         assert!(!schema_contains_file_actions(&schema));

kernel/src/actions/visitors.rs

@@ -1087,7 +1087,7 @@ mod tests {
         engine
             .evaluation_handler()
             .new_expression_evaluator(
-                get_log_schema().clone(),
+                get_commit_schema().clone(),
                 expression.into(),
                 InCommitTimestampVisitor::schema().into(),
             )

kernel/src/engine/arrow_data.rs

@@ -343,7 +343,7 @@ impl ArrowEngineData {
 mod tests {
     use std::sync::Arc;
 
-    use crate::actions::{get_log_schema, Metadata, Protocol};
+    use crate::actions::{get_commit_schema, Metadata, Protocol};
     use crate::arrow::array::types::Int32Type;
     use crate::arrow::array::{Array, AsArray, Int32Array, RecordBatch, StringArray};
     use crate::arrow::datatypes::{
@@ -366,7 +366,7 @@ mod tests {
             r#"{"metaData":{"id":"aff5cb91-8cd9-4195-aef9-446908507302","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"c1\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"c2\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"c3\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["c1","c2"],"configuration":{},"createdTime":1670892997849}}"#,
         ]
         .into();
-        let output_schema = get_log_schema().clone();
+        let output_schema = get_commit_schema().clone();
         let parsed = handler
             .parse_json(string_array_to_engine_data(json_strings), output_schema)
             .unwrap();
@@ -385,7 +385,7 @@ mod tests {
             r#"{"protocol": {"minReaderVersion": 3, "minWriterVersion": 7, "readerFeatures": ["rw1"], "writerFeatures": ["rw1", "w2"]}}"#,
         ]
         .into();
-        let output_schema = get_log_schema().project(&["protocol"])?;
+        let output_schema = get_commit_schema().project(&["protocol"])?;
         let parsed = handler
             .parse_json(string_array_to_engine_data(json_strings), output_schema)
             .unwrap();

kernel/src/engine/default/json.rs

@@ -252,7 +252,7 @@ mod tests {
     use std::sync::{mpsc, Arc, Mutex};
     use std::task::Waker;
 
-    use crate::actions::get_log_schema;
+    use crate::actions::get_commit_schema;
     use crate::arrow::array::{AsArray, Int32Array, RecordBatch, StringArray};
     use crate::arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
     use crate::engine::arrow_data::ArrowEngineData;
@@ -488,7 +488,7 @@ mod tests {
             r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["deletionVectors"],"writerFeatures":["deletionVectors"]}}"#,
             r#"{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"value\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.enableDeletionVectors":"true","delta.columnMapping.mode":"none"},"createdTime":1677811175819}}"#,
         ]);
-        let output_schema = get_log_schema().clone();
+        let output_schema = get_commit_schema().clone();
 
         let batch = handler
             .parse_json(string_array_to_engine_data(json_strings), output_schema)
@@ -503,7 +503,7 @@ mod tests {
         let json_strings = StringArray::from(vec![
             r#"{"add":{"path":"part-00000-fae5310a-a37d-4e51-827b-c3d5516560ca-c000.snappy.parquet","partitionValues":{},"size":635,"modificationTime":1677811178336,"dataChange":true,"stats":"{\"numRecords\":10,\"minValues\":{\"value\":0},\"maxValues\":{\"value\":9},\"nullCount\":{\"value\":0},\"tightBounds\":false}","tags":{"INSERTION_TIME":"1677811178336000","MIN_INSERTION_TIME":"1677811178336000","MAX_INSERTION_TIME":"1677811178336000","OPTIMIZE_TARGET_SIZE":"268435456"},"deletionVector":{"storageType":"u","pathOrInlineDv":"vBn[lx{q8@P<9BNH/isA","offset":1,"sizeInBytes":36,"cardinality":2, "maxRowId": 3}}}"#,
         ]);
-        let output_schema = get_log_schema().clone();
+        let output_schema = get_commit_schema().clone();
 
         let batch: RecordBatch = handler
             .parse_json(string_array_to_engine_data(json_strings), output_schema)
@@ -542,7 +542,7 @@ mod tests {
 
         let handler = DefaultJsonHandler::new(store, Arc::new(TokioBackgroundExecutor::new()));
         let data: Vec<RecordBatch> = handler
-            .read_json_files(files, get_log_schema().clone(), None)
+            .read_json_files(files, get_commit_schema().clone(), None)
             .unwrap()
             .map_ok(into_record_batch)
             .try_collect()
@@ -554,7 +554,7 @@ mod tests {
         // limit batch size
         let handler = handler.with_batch_size(2);
         let data: Vec<RecordBatch> = handler
-            .read_json_files(files, get_log_schema().clone(), None)
+            .read_json_files(files, get_commit_schema().clone(), None)
             .unwrap()
             .map_ok(into_record_batch)
             .try_collect()

kernel/src/log_segment.rs

@@ -5,7 +5,7 @@ use std::sync::{Arc, LazyLock};
 
 use crate::actions::visitors::SidecarVisitor;
 use crate::actions::{
-    get_log_schema, schema_contains_file_actions, Metadata, Protocol, Sidecar, METADATA_NAME,
+    get_commit_schema, schema_contains_file_actions, Metadata, Protocol, Sidecar, METADATA_NAME,
     PROTOCOL_NAME, SIDECAR_NAME,
 };
 use crate::last_checkpoint_hint::LastCheckpointHint;
@@ -556,7 +556,7 @@ impl LogSegment {
         &self,
         engine: &dyn Engine,
     ) -> DeltaResult<impl Iterator<Item = DeltaResult<ActionsBatch>> + Send> {
-        let schema = get_log_schema().project(&[PROTOCOL_NAME, METADATA_NAME])?;
+        let schema = get_commit_schema().project(&[PROTOCOL_NAME, METADATA_NAME])?;
         // filter out log files that do not contain metadata or protocol information
         static META_PREDICATE: LazyLock<Option<PredicateRef>> = LazyLock::new(|| {
             Some(Arc::new(Predicate::or(

kernel/src/log_segment/tests.rs

@@ -9,7 +9,7 @@ use url::Url;
 
 use crate::actions::visitors::AddVisitor;
 use crate::actions::{
-    get_all_actions_schema, get_log_schema, Add, Sidecar, ADD_NAME, METADATA_NAME, REMOVE_NAME,
+    get_all_actions_schema, get_commit_schema, Add, Sidecar, ADD_NAME, METADATA_NAME, REMOVE_NAME,
     SIDECAR_NAME,
 };
 use crate::engine::arrow_data::ArrowEngineData;
@@ -1070,15 +1070,15 @@ fn test_create_checkpoint_stream_returns_checkpoint_batches_as_is_if_schema_has_
     add_checkpoint_to_store(
         &store,
         // Create a checkpoint batch with sidecar actions to verify that the sidecar actions are not read.
-        sidecar_batch_with_given_paths(vec!["sidecar1.parquet"], get_log_schema().clone()),
+        sidecar_batch_with_given_paths(vec!["sidecar1.parquet"], get_commit_schema().clone()),
         "00000000000000000001.checkpoint.parquet",
     )?;
 
     let checkpoint_one_file = log_root
         .join("00000000000000000001.checkpoint.parquet")?
         .to_string();
 
-    let v2_checkpoint_read_schema = get_log_schema().project(&[METADATA_NAME])?;
+    let v2_checkpoint_read_schema = get_commit_schema().project(&[METADATA_NAME])?;
 
     let log_segment = LogSegment::try_new(
         ListedLogFiles::try_new(
@@ -1138,7 +1138,7 @@ fn test_create_checkpoint_stream_returns_checkpoint_batches_if_checkpoint_is_mul
     let checkpoint_one_file = log_root.join(checkpoint_part_1)?.to_string();
     let checkpoint_two_file = log_root.join(checkpoint_part_2)?.to_string();
 
-    let v2_checkpoint_read_schema = get_log_schema().project(&[ADD_NAME])?;
+    let v2_checkpoint_read_schema = get_commit_schema().project(&[ADD_NAME])?;
 
     let log_segment = LogSegment::try_new(
         ListedLogFiles::try_new(
@@ -1185,7 +1185,7 @@ fn test_create_checkpoint_stream_reads_parquet_checkpoint_batch_without_sidecars
 
     add_checkpoint_to_store(
         &store,
-        add_batch_simple(get_log_schema().clone()),
+        add_batch_simple(get_commit_schema().clone()),
         "00000000000000000001.checkpoint.parquet",
     )?;
 
@@ -1295,12 +1295,12 @@ fn test_create_checkpoint_stream_reads_checkpoint_file_and_returns_sidecar_batch
 
     add_sidecar_to_store(
         &store,
-        add_batch_simple(get_log_schema().project(&[ADD_NAME, REMOVE_NAME])?),
+        add_batch_simple(get_commit_schema().project(&[ADD_NAME, REMOVE_NAME])?),
         "sidecarfile1.parquet",
     )?;
     add_sidecar_to_store(
         &store,
-        add_batch_with_remove(get_log_schema().project(&[ADD_NAME, REMOVE_NAME])?),
+        add_batch_with_remove(get_commit_schema().project(&[ADD_NAME, REMOVE_NAME])?),
         "sidecarfile2.parquet",
     )?;