delta table replays in snapshot mode (#9312)

GitOrigin-RevId: de0553ffc030bda6aac15fef1f3e4c9063c36b78

Author: zxqfd555-pw

CHANGELOG.md

@@ -5,10 +5,12 @@ All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ## [Unreleased]
 
+### Changed
+- `pw.io.deltalake.read` now accepts the `start_from_timestamp_ms` parameter for non-append-only tables. In this case, the connector will replay the history of changes in the table version by version starting from the state of the table at the given timestamp. The differences between versions will be applied atomically.
+
 ## [0.26.3] - 2025-10-03
 
 ### Added
-
 - New parser `pathway.xpacks.llm.parsers.PaddleOCRParser` supporting parsing of PDF, PPTX and images.
 
 ## [0.26.2] - 2025-10-01

python/pathway/io/deltalake/__init__.py

@@ -356,8 +356,9 @@ def read(
             custom endpoint is left blank, the authorized user's credentials for S3 will
             be used.
         start_from_timestamp_ms: If defined, only changes that occurred after the specified
-            timestamp will be read. This parameter can only be used for tables with
-            append-only behavior.
+            timestamp are read. When used with **non-append-only tables**, the state
+            of the table at the given timestamp is loaded first, and then all updates are read
+            incrementally.
         autocommit_duration_ms: The maximum time between two commits. Every
             ``autocommit_duration_ms`` milliseconds, the updates received by the connector are
             committed and pushed into Pathway's computation graph.

python/pathway/tests/test_io.py

@@ -5174,3 +5174,107 @@ def run_test(n_expected_entries: int, n_different_times: int):
         ]
         write_lines(input_path, input_contents)
         run_test(10, 1)
+
+
+def test_delta_snapshot_mode_rewind(tmp_path):
+    input_path = tmp_path / "input.jsonl"
+    delta_table_path = tmp_path / "delta"
+    output_path = tmp_path / "output.jsonl"
+    pstorage_path = tmp_path / "PStorage"
+
+    row_1 = {
+        "key": 1,
+        "value": "one",
+    }
+    row_2 = {
+        "key": 2,
+        "value": "two",
+    }
+    row_3 = {
+        "key": 3,
+        "value": "three",
+    }
+
+    class InputSchema(pw.Schema):
+        key: int = pw.column_definition(primary_key=True)
+        value: str
+
+    def update_delta_table_snapshot(rows: list[dict]):
+        G.clear()
+        with open(input_path, "w") as f:
+            for row in rows:
+                f.write(json.dumps(row))
+                f.write("\n")
+        table = pw.io.jsonlines.read(input_path, schema=InputSchema, mode="static")
+        pw.io.deltalake.write(table, delta_table_path, output_table_type="snapshot")
+        run_all(
+            persistence_config=pw.persistence.Config(
+                backend=pw.persistence.Backend.filesystem(pstorage_path)
+            )
+        )
+
+    def from_delta_table_to_file(start_from_timestamp_ms: int | None) -> list[dict]:
+        G.clear()
+        table = pw.io.deltalake.read(
+            delta_table_path,
+            schema=InputSchema,
+            mode="static",
+            start_from_timestamp_ms=start_from_timestamp_ms,
+        )
+        pw.io.jsonlines.write(table, output_path)
+        run_all()
+        result = []
+        with open(output_path, "r") as f:
+            for row in f:
+                parsed_row = json.loads(row)
+                if start_from_timestamp_ms is None:
+                    assert parsed_row["diff"] == 1
+                result.append(parsed_row)
+        return result
+
+    time_start_1 = int(time.time() * 1000)
+    time.sleep(0.05)
+    update_delta_table_snapshot([row_1, row_2, row_3])
+    assert len(from_delta_table_to_file(None)) == 3
+
+    time_start_2 = int(time.time() * 1000)
+    time.sleep(0.05)
+    update_delta_table_snapshot([row_1, row_2])
+    assert len(from_delta_table_to_file(None)) == 2
+
+    time_start_3 = int(time.time() * 1000)
+    time.sleep(0.05)
+    update_delta_table_snapshot([row_1, row_3])
+    assert len(from_delta_table_to_file(None)) == 2
+
+    time_start_4 = int(time.time() * 1000)
+    time.sleep(0.05)
+    update_delta_table_snapshot([row_2])
+    assert len(from_delta_table_to_file(None)) == 1
+    time.sleep(0.05)
+    time_start_5 = int(time.time() * 1000)
+
+    # We start with an empty set.
+    # Then, we move to [1, 2, 3]. It's 3 actions.
+    # Then, we move to [1, 2] via just one action: -1. It's 4 actions in total.
+    # Then, we move to [1, 3], which takes -2, +3. It's 6 actions in total.
+    # Then, we move to [2], which takes -1, -3, +2. It's 9 actions.
+    assert len(from_delta_table_to_file(time_start_1)) == 9
+
+    # We the state at `time_start_2` corresponds to the snapshot with 3 elements.
+    # Then we apply all diffs, so the size of the log is the same as in the previous
+    # case.
+    assert len(from_delta_table_to_file(time_start_2)) == 9
+
+    # We start with [1, 2]. It's 2 events.
+    # Then, we do -2, +3 to advance to [1, 3]. It's 4 events.
+    # Then, we do -1, -3, +2 to advance to [2]. It's 7 events.
+    assert len(from_delta_table_to_file(time_start_3)) == 7
+
+    # We start with [1, 3]. It's 2 events.
+    # Then, we do -1, -3, +2 to advance to [2]. It's 5 events.
+    assert len(from_delta_table_to_file(time_start_4)) == 5
+
+    # There are no events following after `time_start_5`, so we take the snapshot
+    # that is actual at this point of time. It's just one action: [2]
+    assert len(from_delta_table_to_file(time_start_5)) == 1

src/connectors/data_lake/delta.rs

@@ -51,7 +51,9 @@ use crate::connectors::data_format::{
 };
 use crate::connectors::data_lake::buffering::PayloadType;
 use crate::connectors::data_lake::ArrowDataType;
-use crate::connectors::data_storage::{ConnectorMode, ConversionError, ValuesMap};
+use crate::connectors::data_storage::{
+    CommitPossibility, ConnectorMode, ConversionError, ValuesMap,
+};
 use crate::connectors::metadata::ParquetMetadata;
 use crate::connectors::scanner::S3Scanner;
 use crate::connectors::{
@@ -701,48 +703,26 @@ impl DeltaTableReader {
         }
         let mut current_version = table.version();
 
-        let mut parquet_files_queue = {
-            let history = runtime.block_on(async {
-                Ok::<Vec<DeltaTableCommitInfo>, ReadError>(table.history(None).await?)
-            })?;
-            Self::get_reader_actions(&table, path, history, &column_types)?
-        };
+        let mut parquet_files_queue = VecDeque::new();
         let mut backfilling_entries_queue = VecDeque::new();
+        let mut snapshot_loading_needed = backfilling_thresholds.is_empty();
 
         if let Some(start_from_timestamp_ms) = start_from_timestamp_ms {
             assert!(backfilling_thresholds.is_empty()); // Checked upstream in python_api.rs
-            let current_timestamp = current_unix_timestamp_ms();
-            if start_from_timestamp_ms > current_timestamp.try_into().unwrap() {
-                warn!("The timestamp {start_from_timestamp_ms} is greater than the current timestamp {current_timestamp}. All new entries will be read.");
-            }
-            let (earliest_version, latest_version) = runtime.block_on(async {
-                Ok::<(i64, i64), ReadError>((
-                    table.get_earliest_version().await?,
-                    table.get_latest_version().await?,
-                ))
-            })?;
-            let snapshot = table.snapshot()?;
-
-            let mut last_version_below_threshold = None;
-            for version in earliest_version..=latest_version {
-                let Some(timestamp) = snapshot.version_timestamp(version) else {
-                    continue;
-                };
-                if timestamp < start_from_timestamp_ms {
-                    last_version_below_threshold = Some(version);
-                } else {
-                    break;
-                }
-            }
-            if let Some(last_version_below_threshold) = last_version_below_threshold {
-                runtime
-                    .block_on(async { table.load_version(last_version_below_threshold).await })?;
-                current_version = last_version_below_threshold;
-            } else {
-                current_version = earliest_version;
-                warn!("All available versions are newer than the specified timestamp {start_from_timestamp_ms}. The read will start from the beginning.");
-            }
-            parquet_files_queue.clear();
+            Self::handle_start_from_timestamp_ms(
+                &runtime,
+                &mut table,
+                start_from_timestamp_ms,
+                is_append_only,
+                &mut current_version,
+                &mut snapshot_loading_needed,
+            )?;
+        } else {
+            snapshot_loading_needed = true;
+        }
+        if snapshot_loading_needed {
+            parquet_files_queue =
+                Self::get_reader_actions_for_table(&runtime, &table, path, &column_types)?;
         }
 
         if !backfilling_thresholds.is_empty() {
@@ -771,6 +751,64 @@ impl DeltaTableReader {
         })
     }
 
+    fn handle_start_from_timestamp_ms(
+        runtime: &TokioRuntime,
+        table: &mut DeltaTable,
+        start_from_timestamp_ms: i64,
+        is_append_only: bool,
+        current_version: &mut i64,
+        snapshot_loading_needed: &mut bool,
+    ) -> Result<(), ReadError> {
+        let current_timestamp = current_unix_timestamp_ms();
+        if start_from_timestamp_ms > current_timestamp.try_into().unwrap() {
+            warn!("The timestamp {start_from_timestamp_ms} is greater than the current timestamp {current_timestamp}. All new entries will be read.");
+        }
+        let (earliest_version, latest_version) = runtime.block_on(async {
+            Ok::<(i64, i64), ReadError>((
+                table.get_earliest_version().await?,
+                table.get_latest_version().await?,
+            ))
+        })?;
+        let snapshot = table.snapshot()?;
+
+        let mut last_version_below_threshold = None;
+        let mut version_at_threshold = None;
+        for version in earliest_version..=latest_version {
+            let Some(timestamp) = snapshot.version_timestamp(version) else {
+                continue;
+            };
+            if timestamp < start_from_timestamp_ms {
+                last_version_below_threshold = Some(version);
+            } else {
+                if timestamp == start_from_timestamp_ms {
+                    version_at_threshold = Some(version);
+                }
+                break;
+            }
+        }
+
+        if !is_append_only && version_at_threshold.is_some() {
+            *current_version = version_at_threshold.unwrap();
+        } else if let Some(last_version_below_threshold) = last_version_below_threshold {
+            *current_version = last_version_below_threshold;
+        } else {
+            *current_version = earliest_version;
+            warn!(
+                    "All available versions are newer than the specified timestamp {start_from_timestamp_ms}. The read will start from the beginning, version {current_version}."
+                );
+            // NB: All versions are newer than the requested one, meaning that we need to read the
+            // full state at the `earliest_version` and then continue incrementally.
+        }
+
+        if is_append_only && last_version_below_threshold.is_some() {
+            // We've found the threshold version, we read only diffs from this version onwards.
+            *snapshot_loading_needed = false;
+        }
+
+        runtime.block_on(async { table.load_version(*current_version).await })?;
+        Ok(())
+    }
+
     fn record_batch_has_pathway_fields(batch: &ArrowRecordBatch) -> bool {
         for (field, _) in SPECIAL_OUTPUT_FIELDS {
             if let Some(time_column) = batch.column_by_name(field) {
@@ -891,7 +929,8 @@ impl DeltaTableReader {
                 if is_new_block {
                     backfilling_entries_queue.push_back(BackfillingEntry::SourceEvent(
                         ReadResult::FinishedSource {
-                            commit_allowed: true,
+                            // Applicable only for append-only tables, hence no need to avoid squashing diff = +1 with diff = -1
+                            commit_possibility: CommitPossibility::Possible,
                         },
                     ));
                     backfilling_entries_queue.push_back(BackfillingEntry::SourceEvent(
@@ -909,7 +948,9 @@ impl DeltaTableReader {
             }
             backfilling_entries_queue.push_back(BackfillingEntry::SourceEvent(
                 ReadResult::FinishedSource {
-                    commit_allowed: true,
+                    // Same as above, we don't force commits, since the situation with losing/collapsing +1 and -1 events
+                    // is not possible here
+                    commit_possibility: CommitPossibility::Possible,
                 },
             ));
             if pathway_meta_column_added {
@@ -943,6 +984,18 @@ impl DeltaTableReader {
         }
     }
 
+    fn get_reader_actions_for_table(
+        runtime: &TokioRuntime,
+        table: &DeltaTable,
+        base_path: &str,
+        column_types: &HashMap<String, Type>,
+    ) -> Result<VecDeque<DeltaReaderAction>, ReadError> {
+        let history = runtime.block_on(async {
+            Ok::<Vec<DeltaTableCommitInfo>, ReadError>(table.history(None).await?)
+        })?;
+        Self::get_reader_actions(table, base_path, history, column_types)
+    }
+
     fn get_reader_actions(
         table: &DeltaTable,
         base_path: &str,
@@ -1219,12 +1272,21 @@ impl DeltaTableReader {
                 None => {
                     // The Pathway time advancement (e.g. commit) is only possible if it was the
                     // last Parquet block within a version.
+                    let is_last_in_version = self
+                        .current_action
+                        .as_ref()
+                        .expect("current action must be set if there's a reader")
+                        .is_last_in_version;
+
                     let source_event = ReadResult::FinishedSource {
-                        commit_allowed: self
-                            .current_action
-                            .as_ref()
-                            .expect("current action must be set if there's a reader")
-                            .is_last_in_version,
+                        commit_possibility: if is_last_in_version {
+                            // The versions are read on-line, force to avoid squashing same-key events
+                            // with the previous or the next versions.
+                            // Note that it can be less strict if the batch only has additions.
+                            CommitPossibility::Forced
+                        } else {
+                            CommitPossibility::Forbidden
+                        },
                     };
                     self.reader = None;
                     self.current_action = None;

src/connectors/data_lake/iceberg.rs

@@ -34,7 +34,7 @@ use crate::async_runtime::create_async_tokio_runtime;
 use crate::connectors::data_format::NDARRAY_SINGLE_ELEMENT_FIELD_NAME;
 use crate::connectors::data_lake::buffering::PayloadType;
 use crate::connectors::data_lake::MetadataPerColumn;
-use crate::connectors::data_storage::ConnectorMode;
+use crate::connectors::data_storage::{CommitPossibility, ConnectorMode};
 use crate::connectors::metadata::IcebergMetadata;
 use crate::connectors::{
     DataEventType, OffsetKey, OffsetValue, ReadError, ReadResult, Reader, ReaderContext,
@@ -475,7 +475,7 @@ impl IcebergReader {
                     self.diff_queue
                         .push_front(ReadResult::NewSource(new_source_metadata.into()));
                     self.diff_queue.push_back(ReadResult::FinishedSource {
-                        commit_allowed: true,
+                        commit_possibility: CommitPossibility::Possible,
                     });
                 }