Merge branch 'main' into manndp/remove_domain_metadata

Author: mann-db

ffi/src/transaction/mod.rs

@@ -157,10 +157,18 @@ mod tests {
 
     use std::sync::Arc;
 
+    const ZERO_UUID: &str = "00000000-0000-0000-0000-000000000000";
+
     use super::*;
 
     use tempfile::tempdir;
 
+    fn check_txn_id_exists(commit_info: &serde_json::Value) {
+        commit_info["txnId"]
+            .as_str()
+            .expect("txnId should be present in commitInfo");
+    }
+
     fn create_arrow_ffi_from_json(
         schema: ArrowSchema,
         json_string: &str,
@@ -341,9 +349,12 @@ mod tests {
                 .into_iter::<serde_json::Value>()
                 .try_collect()?;
 
-            // set timestamps to 0 and paths to known string values for comparison
-            // (otherwise timestamps are non-deterministic and paths are random UUIDs)
+            check_txn_id_exists(&parsed_commits[0]["commitInfo"]);
+
+            // set timestamps to 0, paths and txn_id to known string values for comparison
+            // (otherwise timestamps are non-deterministic, paths and txn_id are random UUIDs)
             set_json_value(&mut parsed_commits[0], "commitInfo.timestamp", json!(0))?;
+            set_json_value(&mut parsed_commits[0], "commitInfo.txnId", json!(ZERO_UUID))?;
             set_json_value(&mut parsed_commits[1], "add.modificationTime", json!(0))?;
             set_json_value(&mut parsed_commits[1], "add.size", json!(0))?;
 
@@ -355,6 +366,7 @@ mod tests {
                         "operation": "UNKNOWN",
                         "kernelVersion": format!("v{}", env!("CARGO_PKG_VERSION")),
                         "operationParameters": {},
+                        "txnId": ZERO_UUID
                     }
                 }),
                 json!({

kernel/src/action_reconciliation/mod.rs

@@ -0,0 +1,290 @@
+//! # Action Reconciliation
+//! This module implements APIs related to action reconciliation.
+//! Please see the [Delta Lake Protocol](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#action-reconciliation)
+//! for more details about action reconciliation.
+//!
+//! ## Retention and Cleanup
+//!
+//! This module provides utilities for calculating retention timestamps used during action reconciliation:
+//!
+//! - **Deleted File Retention**: Determines when `remove` actions can be excluded from checkpoints
+//! - **Transaction Retention**: Calculates when expired app ids can be cleaned up
+use std::time::Duration;
+
+use crate::table_properties::TableProperties;
+use crate::{DeltaResult, Error};
+
+const SECONDS_PER_MINUTE: u64 = 60;
+const MINUTES_PER_HOUR: u64 = 60;
+const HOURS_PER_DAY: u64 = 24;
+
+/// The default retention period for deleted files in seconds.
+/// This is set to 7 days, which is the default in delta-spark.
+pub(crate) const DEFAULT_RETENTION_SECS: u64 =
+    7 * HOURS_PER_DAY * MINUTES_PER_HOUR * SECONDS_PER_MINUTE;
+
+/// Provides common functionality for calculating file retention timestamps
+/// and transaction expiration timestamps.
+pub(crate) trait RetentionCalculator {
+    /// Get the table properties for accessing retention durations
+    fn table_properties(&self) -> &TableProperties;
+
+    /// Determines the minimum timestamp before which deleted files
+    /// are eligible for permanent removal during VACUUM operations. It is used
+    /// during checkpointing to decide whether to include `remove` actions.
+    ///
+    /// If a deleted file's timestamp is older than this threshold (based on the
+    /// table's `deleted_file_retention_duration`), the corresponding `remove` action
+    /// is included in the checkpoint, allowing VACUUM operations to later identify
+    /// and clean up those files.
+    ///
+    /// # Returns:
+    /// The cutoff timestamp in milliseconds since epoch, matching the remove action's
+    /// `deletion_timestamp` field format for comparison.
+    ///
+    /// Note: The default retention period is 7 days, matching delta-spark's behavior.
+    fn deleted_file_retention_timestamp(&self) -> DeltaResult<i64> {
+        let retention_duration = self.table_properties().deleted_file_retention_duration;
+
+        deleted_file_retention_timestamp_with_time(
+            retention_duration,
+            crate::utils::current_time_duration()?,
+        )
+    }
+
+    /// Calculate the transaction expiration timestamp
+    ///
+    /// Calculates the timestamp threshold for transaction expiration based on
+    /// the table's `set_transaction_retention_duration` property. Transactions that expired
+    /// before this timestamp can be cleaned up.
+    ///
+    /// # Returns
+    /// The timestamp in milliseconds since epoch before which transactions are considered expired,
+    /// or `None` if transaction retention is not configured.
+    ///
+    /// # Errors
+    /// Returns an error if the current system time cannot be obtained or if the retention
+    /// duration exceeds the maximum representable value for i64.
+    fn get_transaction_expiration_timestamp(&self) -> DeltaResult<Option<i64>> {
+        calculate_transaction_expiration_timestamp(self.table_properties())
+    }
+}
+
+/// Calculates the timestamp threshold for deleted file retention based on the provided duration.
+/// This is factored out to allow testing with an injectable time and duration parameter.
+///
+/// # Parameters
+/// - `retention_duration`: The duration to retain deleted files. The table property
+///   `deleted_file_retention_duration` is passed here. If `None`, defaults to 7 days.
+/// - `now_duration`: The current time as a [`Duration`]. This allows for testing with
+///   a specific time instead of using `SystemTime::now()`.
+///
+/// # Returns: The timestamp in milliseconds since epoch
+pub(crate) fn deleted_file_retention_timestamp_with_time(
+    retention_duration: Option<Duration>,
+    now_duration: Duration,
+) -> DeltaResult<i64> {
+    // Use provided retention duration or default (7 days)
+    let retention_duration =
+        retention_duration.unwrap_or_else(|| Duration::from_secs(DEFAULT_RETENTION_SECS));
+
+    // Convert to milliseconds for remove action deletion_timestamp comparison
+    let now_ms = i64::try_from(now_duration.as_millis())
+        .map_err(|_| Error::checkpoint_write("Current timestamp exceeds i64 millisecond range"))?;
+
+    let retention_ms = i64::try_from(retention_duration.as_millis())
+        .map_err(|_| Error::checkpoint_write("Retention duration exceeds i64 millisecond range"))?;
+
+    // Simple subtraction - will produce negative values if retention > now
+    Ok(now_ms - retention_ms)
+}
+
+/// Calculates the transaction expiration timestamp based on table properties.
+/// Returns None if set_transaction_retention_duration is not set.
+pub(crate) fn calculate_transaction_expiration_timestamp(
+    table_properties: &TableProperties,
+) -> DeltaResult<Option<i64>> {
+    table_properties
+        .set_transaction_retention_duration
+        .map(|duration| -> DeltaResult<i64> {
+            let now_ms = crate::utils::current_time_ms()?;
+
+            let expiration_ms = i64::try_from(duration.as_millis())
+                .map_err(|_| Error::generic("Retention duration exceeds i64 millisecond range"))?;
+
+            Ok(now_ms - expiration_ms)
+        })
+        .transpose()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    #[test]
+    fn test_deleted_file_retention_timestamp_with_time() -> DeltaResult<()> {
+        // Test with default retention (7 days)
+        let reference_time = Duration::from_secs(1_000_000_000);
+        let result = deleted_file_retention_timestamp_with_time(None, reference_time)?;
+        let expected = 1_000_000_000_000 - (7 * 24 * 60 * 60 * 1000);
+        assert_eq!(result, expected);
+
+        // Test with custom retention (1 day)
+        let retention = Duration::from_secs(24 * 60 * 60); // 1 day
+        let result = deleted_file_retention_timestamp_with_time(Some(retention), reference_time)?;
+        let expected = 1_000_000_000_000 - (24 * 60 * 60 * 1000); // 1 day in milliseconds
+        assert_eq!(result, expected);
+
+        // Test with zero retention
+        let retention = Duration::from_secs(0);
+        let result = deleted_file_retention_timestamp_with_time(Some(retention), reference_time)?;
+        let expected = 1_000_000_000_000; // Same as reference time
+        assert_eq!(result, expected);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_deleted_file_retention_timestamp_edge_cases() {
+        // Test with very large retention duration
+        let reference_time = Duration::from_secs(1_000_000_000);
+        let large_retention = Duration::from_secs(u64::MAX);
+        let result =
+            deleted_file_retention_timestamp_with_time(Some(large_retention), reference_time);
+        assert!(result.is_err());
+        assert!(result
+            .unwrap_err()
+            .to_string()
+            .contains("Retention duration exceeds i64 millisecond range"));
+    }
+
+    #[test]
+    fn test_deleted_file_retention_timestamp_with_large_now_time() {
+        // Test with very large current time that would overflow i64 milliseconds
+        let reference_time = Duration::from_secs(u64::MAX);
+        let retention = Duration::from_secs(1);
+        let result = deleted_file_retention_timestamp_with_time(Some(retention), reference_time);
+        assert!(result.is_err());
+        assert!(result
+            .unwrap_err()
+            .to_string()
+            .contains("Current timestamp exceeds i64 millisecond range"));
+    }
+
+    #[test]
+    fn test_calculate_transaction_expiration_timestamp() -> DeltaResult<()> {
+        // No set_transaction_retention_duration
+        let properties = TableProperties::default();
+        let result = calculate_transaction_expiration_timestamp(&properties)?;
+        assert_eq!(result, None);
+
+        // Test with set_transaction_retention_duration
+        let properties = TableProperties {
+            set_transaction_retention_duration: Some(Duration::from_secs(3600)), // 1 hour
+            ..Default::default()
+        };
+        let result = calculate_transaction_expiration_timestamp(&properties)?;
+        assert!(result.is_some());
+
+        // The result should be current time minus 1 hour (approximately)
+        // We can't test exact value due to timing, but we can verify it's reasonable
+        let timestamp = result.unwrap();
+        let now_ms = crate::utils::current_time_ms().unwrap();
+        let one_hour_ms = 3600 * 1000;
+
+        // Should be within a reasonable range (allowing for test execution time)
+        assert!(timestamp < now_ms);
+        assert!(timestamp > now_ms - one_hour_ms - 1000); // Allow 1 second buffer
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_calculate_transaction_expiration_timestamp_edge_cases() {
+        // Test with very large retention duration that would overflow
+        let properties = TableProperties {
+            set_transaction_retention_duration: Some(Duration::from_secs(u64::MAX)),
+            ..Default::default()
+        };
+        let result = calculate_transaction_expiration_timestamp(&properties);
+        assert!(result.is_err());
+        assert!(result
+            .unwrap_err()
+            .to_string()
+            .contains("Retention duration exceeds i64 millisecond range"));
+    }
+
+    // Mock implementation of RetentionCalculator for testing trait methods
+    struct MockRetentionCalculator {
+        properties: TableProperties,
+    }
+
+    impl MockRetentionCalculator {
+        fn new(properties: TableProperties) -> Self {
+            Self { properties }
+        }
+    }
+
+    impl RetentionCalculator for MockRetentionCalculator {
+        fn table_properties(&self) -> &TableProperties {
+            &self.properties
+        }
+    }
+
+    #[test]
+    fn test_retention_calculator_trait_deleted_file_retention_timestamp() -> DeltaResult<()> {
+        // Test with default retention
+        let properties = TableProperties::default();
+        let calculator = MockRetentionCalculator::new(properties);
+        let result = calculator.deleted_file_retention_timestamp()?;
+
+        // Should be current time minus 7 days (approximately)
+        let now_ms = crate::utils::current_time_ms().unwrap();
+        let seven_days_ms = 7 * 24 * 60 * 60 * 1000;
+
+        assert!(result < now_ms);
+        assert!(result > now_ms - seven_days_ms - 1000); // Allow a small 1 second buffer
+
+        // Test with custom retention
+        let properties = TableProperties {
+            deleted_file_retention_duration: Some(Duration::from_secs(1800)), // 30 minutes
+            ..Default::default()
+        };
+        let calculator = MockRetentionCalculator::new(properties);
+        let result = calculator.deleted_file_retention_timestamp()?;
+
+        let thirty_minutes_ms = 30 * 60 * 1000;
+        assert!(result < now_ms);
+        assert!(result > now_ms - thirty_minutes_ms - 1000); // Allow 1 second buffer
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_retention_calculator_trait_get_transaction_expiration_timestamp() -> DeltaResult<()> {
+        // Test with no transaction retention
+        let properties = TableProperties::default();
+        let calculator = MockRetentionCalculator::new(properties);
+        let result = calculator.get_transaction_expiration_timestamp()?;
+        assert_eq!(result, None);
+
+        // Test with transaction retention
+        let properties = TableProperties {
+            set_transaction_retention_duration: Some(Duration::from_secs(7200)), // 2 hours
+            ..Default::default()
+        };
+        let calculator = MockRetentionCalculator::new(properties);
+        let result = calculator.get_transaction_expiration_timestamp()?;
+        assert!(result.is_some());
+
+        let timestamp = result.unwrap();
+        let now_ms = crate::utils::current_time_ms().unwrap();
+        let two_hours_ms = 2 * 60 * 60 * 1000;
+
+        assert!(timestamp < now_ms);
+        assert!(timestamp > now_ms - two_hours_ms - 1000); // Allow 1 second buffer
+
+        Ok(())
+    }
+}

kernel/src/actions/mod.rs

@@ -584,8 +584,7 @@ pub(crate) struct CommitInfo {
     pub(crate) kernel_version: Option<String>,
     /// A place for the engine to store additional metadata associated with this commit
     pub(crate) engine_info: Option<String>,
-    /// A unique transaction identified for this commit. When the `catalogManaged` table feature is
-    /// enabled (not yet implemented), this field will be required. Otherwise, it is optional.
+    /// A unique transaction identifier for this commit.
     pub(crate) txn_id: Option<String>,
 }
 
@@ -602,7 +601,7 @@ impl CommitInfo {
             operation_parameters: None,
             kernel_version: Some(format!("v{KERNEL_VERSION}")),
             engine_info,
-            txn_id: None,
+            txn_id: Some(uuid::Uuid::new_v4().to_string()),
         }
     }
 }
@@ -1405,6 +1404,7 @@ mod tests {
         let engine = ExprEngine::new();
 
         let commit_info = CommitInfo::new(0, None, None);
+        let commit_info_txn_id = commit_info.txn_id.clone();
 
         let engine_data = commit_info.into_engine_data(CommitInfo::to_schema().into(), &engine);
 
@@ -1428,7 +1428,7 @@ mod tests {
                 operation_parameters,
                 Arc::new(StringArray::from(vec![Some(format!("v{KERNEL_VERSION}"))])),
                 Arc::new(StringArray::from(vec![None::<String>])),
-                Arc::new(StringArray::from(vec![None::<String>])),
+                Arc::new(StringArray::from(vec![commit_info_txn_id])),
             ],
         )
         .unwrap();