Merge branch 'delta-io:main' into feat/public-ctor

Author: gotocoding-DB

kernel/src/actions/set_transaction.rs

@@ -20,9 +20,14 @@ impl SetTransactionScanner {
         log_segment: &LogSegment,
         application_id: &str,
         engine: &dyn Engine,
+        expiration_timestamp: Option<i64>,
     ) -> DeltaResult<Option<SetTransaction>> {
-        let mut transactions =
-            scan_application_transactions(log_segment, Some(application_id), engine)?;
+        let mut transactions = scan_application_transactions(
+            log_segment,
+            Some(application_id),
+            engine,
+            expiration_timestamp,
+        )?;
         Ok(transactions.remove(application_id))
     }
 
@@ -34,8 +39,9 @@ impl SetTransactionScanner {
     pub(crate) fn get_all(
         log_segment: &LogSegment,
         engine: &dyn Engine,
+        expiration_timestamp: Option<i64>,
     ) -> DeltaResult<SetTransactionMap> {
-        scan_application_transactions(log_segment, None, engine)
+        scan_application_transactions(log_segment, None, engine, expiration_timestamp)
     }
 }
 
@@ -46,8 +52,10 @@ fn scan_application_transactions(
     log_segment: &LogSegment,
     application_id: Option<&str>,
     engine: &dyn Engine,
+    expiration_timestamp: Option<i64>,
 ) -> DeltaResult<SetTransactionMap> {
-    let mut visitor = SetTransactionVisitor::new(application_id.map(|s| s.to_owned()));
+    let mut visitor =
+        SetTransactionVisitor::new(application_id.map(|s| s.to_owned()), expiration_timestamp);
     // If a specific id is requested then we can terminate log replay early as soon as it was
     // found. If all ids are requested then we are forced to replay the entire log.
     for maybe_data in replay_for_app_ids(log_segment, engine)? {
@@ -91,8 +99,10 @@ mod tests {
 
     use super::*;
     use crate::engine::sync::SyncEngine;
+    use crate::utils::test_utils::parse_json_batch;
     use crate::Table;
 
+    use crate::arrow::array::StringArray;
     use itertools::Itertools;
 
     fn get_latest_transactions(
@@ -108,8 +118,8 @@ mod tests {
         let log_segment = snapshot.log_segment();
 
         (
-            SetTransactionScanner::get_all(log_segment, &engine).unwrap(),
-            SetTransactionScanner::get_one(log_segment, app_id, &engine).unwrap(),
+            SetTransactionScanner::get_all(log_segment, &engine, None).unwrap(),
+            SetTransactionScanner::get_one(log_segment, app_id, &engine, None).unwrap(),
         )
     }
 
@@ -165,4 +175,45 @@ mod tests {
             .unwrap();
         assert_eq!(data.len(), 2);
     }
+
+    #[test]
+    fn test_txn_retention_filtering() {
+        let path = std::fs::canonicalize(PathBuf::from("./tests/data/app-txn-with-last-updated/"));
+        let url = url::Url::from_directory_path(path.unwrap()).unwrap();
+        let engine = SyncEngine::new();
+
+        let table = Table::new(url);
+        let snapshot = table.snapshot(&engine, None).unwrap();
+        let log_segment = snapshot.log_segment();
+
+        // Test with no retention (should get all transactions)
+        let all_txns = SetTransactionScanner::get_all(log_segment, &engine, None).unwrap();
+        assert_eq!(all_txns.len(), 4);
+
+        // Test with retention that filters out old transactions
+        let expiration_timestamp = Some(100); // Very old timestamp
+        let filtered_txns =
+            SetTransactionScanner::get_all(log_segment, &engine, expiration_timestamp).unwrap();
+
+        // Exact count depends on test data
+        assert!(filtered_txns.len() <= all_txns.len());
+    }
+
+    #[test]
+    fn test_visitor_retention_with_null_last_updated() {
+        let json_strings: StringArray = vec![
+            r#"{"txn":{"appId":"app_with_time","version":1,"lastUpdated":100}}"#,
+            r#"{"txn":{"appId":"app_without_time","version":2}}"#,
+        ]
+        .into();
+        let batch = parse_json_batch(json_strings);
+
+        let mut visitor = SetTransactionVisitor::new(None, Some(1000));
+        visitor.visit_rows_of(batch.as_ref()).unwrap();
+
+        // app_with_last_updated should be filtered out (100 < 1000)
+        // app_without_last_updated should be kept
+        assert_eq!(visitor.set_transactions.len(), 1);
+        assert!(visitor.set_transactions.contains_key("app_without_time"));
+    }
 }

kernel/src/actions/visitors.rs

@@ -301,14 +301,18 @@ pub(crate) type SetTransactionMap = HashMap<String, SetTransaction>;
 pub(crate) struct SetTransactionVisitor {
     pub(crate) set_transactions: SetTransactionMap,
     pub(crate) application_id: Option<String>,
+    /// Minimum timestamp for transaction retention. Transactions with last_updated
+    /// older than or equal to this timestamp will be filtered out. None means no filtering.
+    expiration_timestamp: Option<i64>,
 }
 
 impl SetTransactionVisitor {
     /// Create a new visitor. When application_id is set then bookkeeping is only for that id only
-    pub(crate) fn new(application_id: Option<String>) -> Self {
+    pub(crate) fn new(application_id: Option<String>, expiration_timestamp: Option<i64>) -> Self {
         SetTransactionVisitor {
             set_transactions: HashMap::default(),
             application_id,
+            expiration_timestamp,
         }
     }
 
@@ -353,6 +357,14 @@ impl RowVisitor for SetTransactionVisitor {
                     .is_none_or(|requested| requested.eq(&app_id))
                 {
                     let txn = SetTransactionVisitor::visit_txn(i, app_id, getters)?;
+                    // Check retention: filter out transactions that are old
+                    // If last_updated is None, the transaction never expires
+                    match self.expiration_timestamp.zip(txn.last_updated) {
+                        Some((expiration_ts, last_updated)) if last_updated <= expiration_ts => {
+                            continue
+                        }
+                        _ => (),
+                    }
                     if !self.set_transactions.contains_key(&txn.app_id) {
                         self.set_transactions.insert(txn.app_id.clone(), txn);
                     }