impl pr feedback

Author: MannDP

kernel/src/transaction/mod.rs

@@ -285,7 +285,7 @@ impl Transaction {
         engine: &'a dyn Engine,
         domain_metadatas: &'a [DomainMetadata],
         read_snapshot: &Snapshot,
-        row_tracking_high_watermark: Option<i64>,
+        row_tracking_high_watermark: Option<RowTrackingDomainMetadata>,
     ) -> DeltaResult<impl Iterator<Item = DeltaResult<Box<dyn EngineData>>> + 'a> {
         // if there are domain metadata actions, the table must support it
         if !domain_metadatas.is_empty()
@@ -315,7 +315,7 @@ impl Transaction {
         }
 
         let system_domains = row_tracking_high_watermark
-            .map(|hwm| DomainMetadata::try_from(RowTrackingDomainMetadata::new(hwm)))
+            .map(|hwm| DomainMetadata::try_from(hwm))
             .transpose()?
             .into_iter();
 
@@ -370,7 +370,39 @@ impl Transaction {
         &'a self,
         engine: &dyn Engine,
         commit_version: u64,
-    ) -> DeltaResult<(EngineDataResultIterator<'a>, Option<i64>)> {
+    ) -> DeltaResult<(
+        EngineDataResultIterator<'a>,
+        Option<RowTrackingDomainMetadata>,
+    )> {
+        fn build_add_actions<'a, I, T>(
+            engine: &dyn Engine,
+            add_files_metadata: I,
+            input_schema: SchemaRef,
+            output_schema: SchemaRef,
+        ) -> impl Iterator<Item = DeltaResult<Box<dyn EngineData>>> + 'a
+        where
+            I: Iterator<Item = DeltaResult<T>> + Send + 'a,
+            T: Deref<Target = dyn EngineData> + Send + 'a,
+        {
+            let evaluation_handler = engine.evaluation_handler();
+
+            add_files_metadata.map(move |add_files_batch| {
+                // Convert stats to a JSON string and nest the add action in a top-level struct
+                let adds_expr = Expression::struct_from([Expression::transform(
+                    Transform::new_top_level().with_replaced_field(
+                        "stats",
+                        Expression::unary(ToJson, Expression::column(["stats"])).into(),
+                    ),
+                )]);
+                let adds_evaluator = evaluation_handler.new_expression_evaluator(
+                    input_schema.clone(),
+                    Arc::new(adds_expr),
+                    output_schema.clone().into(),
+                );
+                adds_evaluator.evaluate(add_files_batch?.deref())
+            })
+        }
+
         if self.add_files_metadata.is_empty() {
             return Ok((Box::new(iter::empty()), None));
         }
@@ -411,7 +443,7 @@ impl Transaction {
                 },
             );
 
-            let add_actions = self.create_add_actions(
+            let add_actions = build_add_actions(
                 engine,
                 extended_add_files,
                 with_row_tracking_cols(add_files_schema()),
@@ -420,13 +452,13 @@ impl Transaction {
                 ))),
             );
 
-            Ok((
-                Box::new(add_actions),
-                Some(row_tracking_visitor.row_id_high_water_mark),
-            ))
+            let row_tracking_domain_metadata =
+                RowTrackingDomainMetadata::new(row_tracking_visitor.row_id_high_water_mark);
+
+            Ok((Box::new(add_actions), Some(row_tracking_domain_metadata)))
         } else {
             // Simple case without row tracking
-            let add_actions = self.create_add_actions(
+            let add_actions = build_add_actions(
                 engine,
                 self.add_files_metadata.iter().map(|a| Ok(a.deref())),
                 add_files_schema().clone(),
@@ -436,37 +468,6 @@ impl Transaction {
             Ok((Box::new(add_actions), None))
         }
     }
-
-    /// Convert file metadata provided by the engine into protocol-compliant add actions.
-    fn create_add_actions<'a, I, T>(
-        &'a self,
-        engine: &dyn Engine,
-        add_files_metadata: I,
-        input_schema: SchemaRef,
-        output_schema: SchemaRef,
-    ) -> impl Iterator<Item = DeltaResult<Box<dyn EngineData>>> + 'a
-    where
-        I: Iterator<Item = DeltaResult<T>> + Send + 'a,
-        T: Deref<Target = dyn EngineData> + Send + 'a,
-    {
-        let evaluation_handler = engine.evaluation_handler();
-
-        add_files_metadata.map(move |add_files_batch| {
-            // Convert stats to a JSON string and nest the add action in a top-level struct
-            let adds_expr = Expression::struct_from([Expression::transform(
-                Transform::new_top_level().with_replaced_field(
-                    "stats",
-                    Expression::unary(ToJson, Expression::column(["stats"])).into(),
-                ),
-            )]);
-            let adds_evaluator = evaluation_handler.new_expression_evaluator(
-                input_schema.clone(),
-                Arc::new(adds_expr),
-                output_schema.clone().into(),
-            );
-            adds_evaluator.evaluate(add_files_batch?.deref())
-        })
-    }
 }
 
 /// WriteContext is data derived from a [`Transaction`] that can be provided to writers in order to

kernel/tests/write.rs

@@ -31,7 +31,10 @@ use tempfile::tempdir;
 
 use delta_kernel::schema::{DataType, SchemaRef, StructField, StructType};
 
-use test_utils::{create_table, engine_store_setup, setup_test_tables, test_read};
+use test_utils::{
+    assert_result_error_with_message, create_table, engine_store_setup, setup_test_tables,
+    test_read,
+};
 
 mod common;
 use url::Url;
@@ -1283,24 +1286,24 @@ async fn test_set_domain_metadata_errors() -> Result<(), Box<dyn std::error::Err
 
     // System domain rejection
     let txn = snapshot.clone().transaction()?;
-    let err = txn
+    let res = txn
         .with_domain_metadata("delta.system".to_string(), "config".to_string())
-        .commit(&engine)
-        .unwrap_err();
-    assert!(err
-        .to_string()
-        .contains("Users cannot modify system controlled metadata domains"));
+        .commit(&engine);
+    assert_result_error_with_message(
+        res,
+        "Users cannot modify system controlled metadata domains",
+    );
 
     // Duplicate domain rejection
     let txn2 = snapshot.clone().transaction()?;
-    let err = txn2
+    let res = txn2
         .with_domain_metadata("app.config".to_string(), "v1".to_string())
         .with_domain_metadata("app.config".to_string(), "v2".to_string())
-        .commit(&engine)
-        .unwrap_err();
-    assert!(err
-        .to_string()
-        .contains("Metadata for domain app.config already specified in this transaction"));
+        .commit(&engine);
+    assert_result_error_with_message(
+        res,
+        "Metadata for domain app.config already specified in this transaction",
+    );
 
     Ok(())
 }
@@ -1331,14 +1334,12 @@ async fn test_set_domain_metadata_unsupported_writer_feature(
     .await?;
 
     let snapshot = Arc::new(Snapshot::builder(table_url.clone()).build(&engine)?);
-    let result = snapshot
+    let res = snapshot
         .transaction()?
         .with_domain_metadata("app.config".to_string(), "test_config".to_string())
         .commit(&engine);
 
-    assert!(result.is_err());
-    let err = result.unwrap_err();
-    assert!(err.to_string().contains("Domain metadata operations require writer version 7 and the 'domainMetadata' writer feature"));
+    assert_result_error_with_message(res, "Domain metadata operations require writer version 7 and the 'domainMetadata' writer feature");
 
     Ok(())
 }

test-utils/src/lib.rs

@@ -426,3 +426,16 @@ pub fn set_json_value(
     *v = new_value;
     Ok(())
 }
+
+pub fn assert_result_error_with_message<T, E: ToString>(res: Result<T, E>, message: &str) {
+    match res {
+        Ok(_) => panic!("Expected error, but got Ok result"),
+        Err(error) => {
+            let error_str = error.to_string();
+            assert!(
+                error_str.contains(message),
+                "Error message does not contain the expected message.\nExpected message:\t{message}\nActual message:\t\t{error_str}"
+            );
+        }
+    }
+}