[Converter] Commit high watermark of converter processed up to as part of snapshot properties (#589)

* [Converter] Commit high watermark of converter processed up to as part of snapshot properties

* Remove print statement

---------

Co-authored-by: Miranda <yiqin121@gmail.com>

Author: Zyiqin-Miranda

deltacat/__init__.py

@@ -126,7 +126,7 @@
 
 deltacat.logs.configure_deltacat_logger(logging.getLogger(__name__))
 
-__version__ = "2.0.0.post6"
+__version__ = "2.0.0.post7"
 
 
 __all__ = [

deltacat/compute/converter/converter_session.py

@@ -80,6 +80,8 @@ def converter_session(
             - file_system: File system instance
             - location_provider_prefix_override: Optional prefix override for file locations
             - position_delete_for_multiple_data_files: Whether to generate position deletes for multiple data files
+            - start_snapshot_id: Optional starting snapshot ID for filtering files (files from this snapshot onwards will be processed)
+            - start_sequence_number: Optional starting sequence number for filtering files (used in conjunction with start_snapshot_id)
         **kwargs: Additional keyword arguments (currently unused)
 
     Returns:
@@ -123,14 +125,26 @@ def converter_session(
     position_delete_for_multiple_data_files = (
         params.position_delete_for_multiple_data_files
     )
+    start_snapshot_id = params.start_snapshot_id
+    start_sequence_number = params.start_sequence_number
+    logger.info(
+        f"Converter session parameters - start_snapshot_id: {start_snapshot_id}, start_sequence_number: {start_sequence_number}"
+    )
 
     logger.info(f"Fetching all bucket files for table {table_identifier}...")
-    data_file_dict, equality_delete_dict, pos_delete_dict = fetch_all_bucket_files(
-        table=iceberg_table
-    )
+    (
+        data_file_dict,
+        equality_delete_dict,
+        pos_delete_dict,
+        latest_snapshot_id,
+        largest_sequence_number,
+    ) = fetch_all_bucket_files(table=iceberg_table, start_snapshot_id=start_snapshot_id)
     logger.info(
         f"Fetched files - data: {len(data_file_dict)}, equality_delete: {len(equality_delete_dict)}, pos_delete: {len(pos_delete_dict)}"
     )
+    logger.info(
+        f"Latest snapshot ID: {latest_snapshot_id}, Largest sequence number: {largest_sequence_number}"
+    )
 
     convert_input_files_for_all_buckets = group_all_files_to_each_bucket(
         data_file_dict=data_file_dict,
@@ -350,20 +364,26 @@ def convert_input_provider(index: int, item: Any) -> Dict[str, ConvertInput]:
             converter_snapshot_id = commit_append_snapshot(
                 iceberg_table=iceberg_table,
                 new_position_delete_files=to_be_added_files_list,
+                latest_snapshot_id=latest_snapshot_id,
+                largest_sequence_number=largest_sequence_number,
             )
         elif snapshot_type == SnapshotType.REPLACE:
             logger.info(f"Committing replace snapshot for {table_identifier}.")
             converter_snapshot_id = commit_replace_snapshot(
                 iceberg_table=iceberg_table,
                 to_be_deleted_files=to_be_deleted_files_list,
                 new_position_delete_files=to_be_added_files_list,
+                latest_snapshot_id=latest_snapshot_id,
+                largest_sequence_number=largest_sequence_number,
             )
         elif snapshot_type == SnapshotType.DELETE:
             logger.info(f"Committing delete snapshot for {table_identifier}.")
             converter_snapshot_id = commit_replace_snapshot(
                 iceberg_table=iceberg_table,
                 to_be_deleted_files=to_be_deleted_files_list,
                 new_position_delete_files=[],  # No new files to add
+                latest_snapshot_id=latest_snapshot_id,
+                largest_sequence_number=largest_sequence_number,
             )
         else:
             logger.warning(f"Unexpected snapshot type: {snapshot_type}")

deltacat/compute/converter/model/converter_session_params.py

@@ -49,6 +49,8 @@ def of(params: Optional[Dict[str, Any]]) -> ConverterSessionParams:
         )
         result.s3_prefix_override = params.get("s3_prefix_override", None)
         result.fileio_override = params.get("fileio_override", None)
+        result.start_snapshot_id = params.get("start_snapshot_id", None)
+        result.start_sequence_number = params.get("start_sequence_number", None)
 
         return result
 
@@ -155,3 +157,19 @@ def fileio_override(self) -> Optional[FileIO]:
     @fileio_override.setter
     def fileio_override(self, fileio_override: Optional[FileIO]) -> None:
         self["fileio_override"] = fileio_override
+
+    @property
+    def start_snapshot_id(self) -> Optional[int]:
+        return self["start_snapshot_id"]
+
+    @start_snapshot_id.setter
+    def start_snapshot_id(self, start_snapshot_id: Optional[int]) -> None:
+        self["start_snapshot_id"] = start_snapshot_id
+
+    @property
+    def start_sequence_number(self) -> Optional[int]:
+        return self["start_sequence_number"]
+
+    @start_sequence_number.setter
+    def start_sequence_number(self, start_sequence_number: Optional[int]) -> None:
+        self["start_sequence_number"] = start_sequence_number

deltacat/compute/converter/pyiceberg/overrides.py

@@ -203,7 +203,13 @@ def parquet_files_dict_to_iceberg_data_files(
 def fetch_all_bucket_files(
     table: Table,
     start_snapshot_id: Optional[int] = None,
-) -> Tuple[Dict[Any, DataFileList], Dict[Any, DataFileList], Dict[Any, DataFileList]]:
+) -> Tuple[
+    Dict[Any, DataFileList],
+    Dict[Any, DataFileList],
+    Dict[Any, DataFileList],
+    Optional[int],
+    Optional[int],
+]:
     # step 1: filter manifests using partition summaries
     # the filter depends on the partition spec used to write the manifest file, so create a cache of filters for each spec id
 
@@ -213,7 +219,7 @@ def fetch_all_bucket_files(
     current_snapshot = data_scan.snapshot()
 
     if not current_snapshot:
-        return {}, {}, {}
+        return {}, {}, {}, None, None
 
     snapshots = list(table.metadata.snapshots)
     expected_start_sequence_number = -1
@@ -350,11 +356,33 @@ def fetch_all_bucket_files(
     for partition_value, files_dict in positional_delete_entries_registry.items():
         positional_delete_entries[partition_value] = list(files_dict.values())
 
+    # Calculate latest snapshot ID and largest sequence number from snapshots
+    latest_snapshot_id = None
+    largest_sequence_number = None
+
+    if snapshots:
+        # Get the latest snapshot ID (current snapshot)
+        latest_snapshot_id = current_snapshot.snapshot_id
+
+        # Find the largest sequence number across all snapshots
+        largest_sequence_number = max(
+            snapshot.sequence_number for snapshot in snapshots
+        )
+
     logger.info(
         f"Fetched {sum(len(files) for files in data_entries.values())} data files from table, "
         f"{sum(len(files) for files in equality_data_entries.values())} equality delete files, "
         f"{sum(len(files) for files in positional_delete_entries.values())} position delete files"
     )
+    logger.info(
+        f"Latest snapshot ID: {latest_snapshot_id}, Largest sequence number: {largest_sequence_number}"
+    )
     for k, v in data_entries.items():
         logger.info(f"{len(v)} files for partition value :{k}")
-    return data_entries, equality_data_entries, positional_delete_entries
+    return (
+        data_entries,
+        equality_data_entries,
+        positional_delete_entries,
+        latest_snapshot_id,
+        largest_sequence_number,
+    )

deltacat/compute/converter/pyiceberg/update_snapshot_overrides.py

@@ -167,7 +167,10 @@ def _get_entries(manifest: ManifestFile) -> List[ManifestEntry]:
 
 
 def commit_append_snapshot(
-    iceberg_table: Table, new_position_delete_files: List[DataFile]
+    iceberg_table: Table,
+    new_position_delete_files: List[DataFile],
+    latest_snapshot_id: int = None,
+    largest_sequence_number: int = None,
 ) -> str:
     tx = iceberg_table.transaction()
     try:
@@ -177,7 +180,21 @@ def commit_append_snapshot(
                     "schema.name-mapping.default": tx.table_metadata.schema().name_mapping.model_dump_json()
                 }
             )
-        with append_delete_files_override(tx.update_snapshot()) as append_snapshot:
+
+        # Prepare snapshot properties
+        snapshot_properties = {}
+        if latest_snapshot_id is not None:
+            snapshot_properties["sourceSnapshotId"] = str(latest_snapshot_id)
+        if largest_sequence_number is not None:
+            snapshot_properties["sourceSequenceNumber"] = str(largest_sequence_number)
+
+        logger.info(
+            f"Committing append snapshot with properties: {snapshot_properties}"
+        )
+
+        with append_delete_files_override(
+            tx.update_snapshot(snapshot_properties=snapshot_properties)
+        ) as append_snapshot:
             if new_position_delete_files:
                 for data_file in new_position_delete_files:
                     append_snapshot.append_data_file(data_file)
@@ -207,6 +224,7 @@ def commit_snapshot_properties_change(iceberg_table: Table):
     except Exception as e:
         raise e
     else:
+        logger.info(f"Commit only table properties changes: {current_snapshot_id}")
         metadata = tx.commit_transaction().metadata
         logger.info(
             f"Successfully committed only table properties change with ray.converter.snapshot_id:{current_snapshot_id}"
@@ -305,6 +323,8 @@ def commit_replace_snapshot(
     iceberg_table: Table,
     new_position_delete_files: List[DataFile],
     to_be_deleted_files: List[DataFile],
+    latest_snapshot_id: int = None,
+    largest_sequence_number: int = None,
 ) -> str:
     tx = iceberg_table.transaction()
     try:
@@ -314,8 +334,20 @@ def commit_replace_snapshot(
                     "schema.name-mapping.default": tx.table_metadata.schema().name_mapping.model_dump_json()
                 }
             )
+
+        # Prepare snapshot properties
+        snapshot_properties = {}
+        if latest_snapshot_id is not None:
+            snapshot_properties["sourceSnapshotId"] = str(latest_snapshot_id)
+        if largest_sequence_number is not None:
+            snapshot_properties["sourceSequenceNumber"] = str(largest_sequence_number)
+
+        logger.info(
+            f"Committing replace snapshot with properties: {snapshot_properties}"
+        )
+
         with replace_delete_files_override(
-            tx.update_snapshot()
+            tx.update_snapshot(snapshot_properties=snapshot_properties)
         ) as replace_delete_snapshot:
             if new_position_delete_files:
                 for data_file in new_position_delete_files:

deltacat/tests/compute/converter/integration/test_convert_session.py

@@ -490,7 +490,13 @@ def test_converter(
 
     # Get files and create convert input
     tbl = session_catalog.load_table(identifier)
-    data_file_dict, equality_delete_dict, pos_delete_dict = fetch_all_bucket_files(tbl)
+    (
+        data_file_dict,
+        equality_delete_dict,
+        pos_delete_dict,
+        _,
+        _,
+    ) = fetch_all_bucket_files(tbl)
 
     # Handle equality delete if present
     if "equality_delete_data" in test_case:
@@ -606,7 +612,13 @@ def test_converter_session_duplicate_position_deletes_spark_compatibility(
 
     # Load table and run converter first time
     tbl = session_catalog.load_table(identifier)
-    data_file_dict, equality_delete_dict, pos_delete_dict = fetch_all_bucket_files(tbl)
+    (
+        data_file_dict,
+        equality_delete_dict,
+        pos_delete_dict,
+        _,
+        _,
+    ) = fetch_all_bucket_files(tbl)
 
     convert_input_files_for_all_buckets = group_all_files_to_each_bucket(
         data_file_dict=data_file_dict,
@@ -675,7 +687,13 @@ def test_converter_session_duplicate_position_deletes_spark_compatibility(
     # This could happen in scenarios where the converter is run multiple times on the same data
 
     # Get files again (now includes position delete files from first run)
-    data_file_dict, equality_delete_dict, pos_delete_dict = fetch_all_bucket_files(tbl)
+    (
+        data_file_dict,
+        equality_delete_dict,
+        pos_delete_dict,
+        _,
+        _,
+    ) = fetch_all_bucket_files(tbl)
 
     convert_input_files_for_all_buckets = group_all_files_to_each_bucket(
         data_file_dict=data_file_dict,
@@ -850,9 +868,13 @@ def test_converter_session_no_input_files(
             )
 
         # Verify table is empty (no data files)
-        data_file_dict, equality_delete_dict, pos_delete_dict = fetch_all_bucket_files(
-            tbl
-        )
+        (
+            data_file_dict,
+            equality_delete_dict,
+            pos_delete_dict,
+            _,
+            _,
+        ) = fetch_all_bucket_files(tbl)
 
         assert (
             len(data_file_dict) == 0

deltacat/tests/compute/converter/integration/test_converter_commit_conflict_resolution.py

@@ -414,7 +414,13 @@ def test_converter_commit_conflict_resolution(
     # Get files and create convert input
     tbl = session_catalog.load_table(identifier)
 
-    data_file_dict, equality_delete_dict, pos_delete_dict = fetch_all_bucket_files(tbl)
+    (
+        data_file_dict,
+        equality_delete_dict,
+        pos_delete_dict,
+        _,
+        _,
+    ) = fetch_all_bucket_files(tbl)
 
     # Handle equality delete if present
     if "equality_delete_data" in test_case: