feat: setTransactionRetentionDuration (delta-io#1013)

feat: add SetTransaction retention duration filtering

## What changes are proposed in this pull request?

This PR implements the `delta.setTransactionRetentionDuration` table
property in delta-kernel-rs, which allows filtering out expired
transaction identifiers based on their `lastUpdated` timestamp. This
feature helps prevent unbounded growth of transaction logs and reduces
checkpoint file sizes.

### Key changes:
1. **Transaction filtering during log replay**: Added retention
timestamp filtering to `SetTransactionVisitor` to skip transactions
older than the retention duration
2. **Checkpoint writing**: Updated `CheckpointVisitor` to exclude
expired transactions when creating checkpoints
3. **Backward compatibility**: Transactions without `lastUpdated` field
are preserved forever (never expire)
4. **Consistent behavior**: The same retention logic is applied in both
checkpoint writing and batch transaction reading operations

### Implementation details:
- Modified `SetTransactionVisitor` to accept an optional retention
timestamp
- Updated `CheckpointLogReplayProcessor` to calculate and propagate
transaction retention timestamps
- Enhanced `CheckpointVisitor` to filter transactions based on retention
during checkpoint creation
- Single transaction lookups (`get_app_id_version`) intentionally do
filter to
- Batch operations (future `get_all_app_id_versions`, checkpointing) do
filter expired transactions

## How was this change tested?

Added comprehensive unit tests covering:

1. **Checkpoint writing tests** (`checkpoint::log_replay::tests`):
- `test_checkpoint_visitor_txn_retention`: Verifies individual
transaction filtering with various timestamp scenarios
- `test_checkpoint_actions_iter_with_txn_retention`: Tests transaction
filtering across multiple batches during checkpoint creation

2. **Transaction reading tests** (`actions::set_transaction::tests`):
- `test_visitor_retention_with_null_last_updated`: Ensures transactions
without `lastUpdated` are never filtered
- Modified existing tests to support the new retention timestamp
parameter

3. **Test scenarios covered**:
- Transactions with `lastUpdated <= retention_timestamp` are filtered
out
- Transactions with `lastUpdated = None` are always kept (backward
compatibility)
- Boundary condition: transactions exactly at retention timestamp are
filtered
   - Cross-batch behavior in checkpoint processing
- Verification that filtered transactions are excluded from checkpoint
files

All tests pass with `cargo t --all-features --all-targets`.

---------

Co-authored-by: Ryan Johnson <scovich@users.noreply.github.com>
Co-authored-by: Zach Schuermann <zachary.zvs@gmail.com>

Author: 3 people

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);
                     }