[Data] Support incremental reads for Iceberg tables

python/ray/data/_internal/datasource/iceberg_datasource.py

@@ -77,7 +77,7 @@
     from pyiceberg.io import FileIO
     from pyiceberg.manifest import DataFile
     from pyiceberg.schema import Schema
-    from pyiceberg.table import DataScan, FileScanTask, Table
+    from pyiceberg.table import DataScan, FileScanTask, IncrementalAppendScan, Table
     from pyiceberg.table.metadata import TableMetadata
 
     from ray.data.context import DataContext
@@ -272,6 +272,8 @@ def __init__(
         row_filter: Union[str, "BooleanExpression"] = None,
         selected_fields: Tuple[str, ...] = ("*",),
         snapshot_id: Optional[int] = None,
+        start_snapshot_id: Optional[int] = None,
+        end_snapshot_id: Optional[int] = None,
         scan_kwargs: Optional[Dict[str, Any]] = None,
         catalog_kwargs: Optional[Dict[str, Any]] = None,
     ):
@@ -285,7 +287,17 @@ def __init__(
                  to reading
             selected_fields: Which columns from the data to read, passed directly to
                 PyIceberg's load functions
-            snapshot_id: Optional snapshot ID for the Iceberg table
+            snapshot_id: Optional snapshot ID for the Iceberg table. Reads a single
+                snapshot as of the given ID. Mutually exclusive with
+                ``start_snapshot_id`` / ``end_snapshot_id``.
+            start_snapshot_id: Optional snapshot ID to start an incremental append scan
+                from, *exclusively* (maps to PyIceberg's
+                ``from_snapshot_id_exclusive``). When set, only rows appended after this
+                snapshot are read.
+            end_snapshot_id: Optional snapshot ID to end an incremental append scan at,
+                *inclusively* (maps to PyIceberg's ``to_snapshot_id_inclusive``).
+                Defaults to the table's current snapshot when only
+                ``start_snapshot_id`` is given.
             scan_kwargs: Optional arguments to pass to PyIceberg's Table.scan()
                 function
             catalog_kwargs: Optional arguments to use when setting up the Iceberg
@@ -297,6 +309,17 @@ def __init__(
         _check_import(self, module="pyiceberg", package="pyiceberg")
         from pyiceberg.expressions import AlwaysTrue
 
+        # An incremental (append) scan is requested when either snapshot bound is set.
+        self._incremental = start_snapshot_id is not None or end_snapshot_id is not None
+        if snapshot_id is not None and self._incremental:
+            raise ValueError(
+                "`snapshot_id` is mutually exclusive with `start_snapshot_id` / "
+                "`end_snapshot_id`. Pass `snapshot_id` to read a single snapshot, or "
+                "the start/end pair to read data appended between two snapshots."
+            )
+        self._start_snapshot_id = start_snapshot_id
+        self._end_snapshot_id = end_snapshot_id
+
         self._scan_kwargs = scan_kwargs if scan_kwargs is not None else {}
         self._catalog_kwargs = catalog_kwargs if catalog_kwargs is not None else {}
 
@@ -367,14 +390,17 @@ def _get_combined_filter(self) -> "BooleanExpression":
 
         return combined_filter
 
-    def _get_data_scan(self) -> "DataScan":
+    def _get_data_scan(self) -> Union["DataScan", "IncrementalAppendScan"]:
         # Get the combined filter
         combined_filter = self._get_combined_filter()
 
         # Convert back to tuple for PyIceberg API (None -> ("*",))
         data_columns = self._get_data_columns()
         selected_fields = ("*",) if data_columns is None else tuple(data_columns)
 
+        if self._incremental:
+            return self._get_incremental_scan(combined_filter, selected_fields)
+
         data_scan = self.table.scan(
             row_filter=combined_filter,
             selected_fields=selected_fields,
@@ -383,6 +409,41 @@ def _get_data_scan(self) -> "DataScan":
 
         return data_scan
 
+    def _get_incremental_scan(
+        self,
+        combined_filter: "BooleanExpression",
+        selected_fields: Tuple[str, ...],
+    ) -> "IncrementalAppendScan":
+        """Build a PyIceberg incremental append scan between the requested snapshots.
+
+        Reads only the rows appended between ``start_snapshot_id`` (exclusive) and
+        ``end_snapshot_id`` (inclusive). The resulting scan exposes the same
+        ``plan_files()`` / ``projection()`` interface as a regular ``DataScan``, so the
+        rest of the read path is unchanged.
+        """
+        table = self.table
+        # Incremental append scans landed in PyIceberg after 0.11.0
+        # (apache/iceberg-python#3512). Guard on the capability rather than a version
+        # string so backports are picked up automatically.
+        if not hasattr(table, "incremental_append_scan"):
+            import pyiceberg
+
+            raise ValueError(
+                "Incremental Iceberg reads (`start_snapshot_id` / `end_snapshot_id`) "
+                "require a PyIceberg version that provides "
+                "`Table.incremental_append_scan(...)` (see apache/iceberg-python#3512). "
+                f"The installed PyIceberg version ({pyiceberg.__version__}) does not "
+                "expose it; please upgrade PyIceberg."
+            )
+
+        return table.incremental_append_scan(
+            from_snapshot_id_exclusive=self._start_snapshot_id,
+            to_snapshot_id_inclusive=self._end_snapshot_id,
+            row_filter=combined_filter,
+            selected_fields=selected_fields,
+            **self._scan_kwargs,
+        )
+
     def estimate_inmemory_data_size(self) -> Optional[int]:
         # Approximate the size by using the plan files - this will not
         # incorporate the deletes, but that's a reasonable approximation

python/ray/data/read_api.py

@@ -4390,6 +4390,8 @@ def read_iceberg(
     parallelism: int = -1,
     selected_fields: Tuple[str, ...] = ("*",),
     snapshot_id: Optional[int] = None,
+    start_snapshot_id: Optional[int] = None,
+    end_snapshot_id: Optional[int] = None,
     scan_kwargs: Optional[Dict[str, str]] = None,
     catalog_kwargs: Optional[Dict[str, str]] = None,
     catalog: Optional["Catalog"] = None,
@@ -4424,6 +4426,13 @@ def read_iceberg(
         ... ).filter(col("column_name") == "literal_value")
         >>> # Select specific columns
         >>> ds = ds.select_columns(["col1", "col2"])  #doctest: +SKIP
+        >>> # Incrementally read only rows appended between two snapshots
+        >>> ds = ray.data.read_iceberg( #doctest: +SKIP
+        ...     table_identifier="db_name.table_name",
+        ...     start_snapshot_id=1234567890,  # exclusive
+        ...     end_snapshot_id=9876543210,  # inclusive
+        ...     catalog_kwargs={"name": "default", "type": "glue"},
+        ... )
 
     Args:
         table_identifier: Fully qualified table identifier (``db_name.table_name``)
@@ -4435,7 +4444,16 @@ def read_iceberg(
             Which columns from the data to read, passed directly to
             PyIceberg's load functions. Should be an tuple of string column names.
         snapshot_id: Optional snapshot ID for the Iceberg table, by default the latest
-            snapshot is used
+            snapshot is used. Mutually exclusive with ``start_snapshot_id`` /
+            ``end_snapshot_id``.
+        start_snapshot_id: Optional snapshot ID to start an incremental read from,
+            *exclusively*. When set, only rows appended after this snapshot are read
+            (via PyIceberg's ``Table.incremental_append_scan``). Requires a PyIceberg
+            version that supports incremental append scans. Mutually exclusive with
+            ``snapshot_id``.
+        end_snapshot_id: Optional snapshot ID to end an incremental read at,
+            *inclusively*. Defaults to the table's current snapshot when only
+            ``start_snapshot_id`` is provided. Mutually exclusive with ``snapshot_id``.
         scan_kwargs: Optional arguments to pass to PyIceberg's Table.scan() function
              (e.g., case_sensitive, limit, etc.)
         catalog_kwargs: Optional arguments to pass to PyIceberg's catalog.load_catalog()
@@ -4506,6 +4524,8 @@ def read_iceberg(
         row_filter=row_filter,
         selected_fields=selected_fields,
         snapshot_id=snapshot_id,
+        start_snapshot_id=start_snapshot_id,
+        end_snapshot_id=end_snapshot_id,
         scan_kwargs=scan_kwargs,
         catalog_kwargs=catalog_kwargs,
     )

python/ray/data/tests/datasource/test_iceberg.py

@@ -2027,6 +2027,156 @@ def flaky_dataframe_to_data_files(*args, **kwargs):
     assert len(result.data_files) > 0, "Expected data files in result"
 
 
+def _supports_incremental_scan() -> bool:
+    """Whether the installed PyIceberg exposes incremental append scans."""
+    return hasattr(Table, "incremental_append_scan")
+
+
+_INCREMENTAL_UNSUPPORTED_REASON = (
+    "Installed PyIceberg does not support Table.incremental_append_scan "
+    "(apache/iceberg-python#3512)"
+)
+
+
+def _append_batch(col_a_values: List[int]) -> int:
+    """Append a batch of rows to the test table and return the new snapshot ID."""
+    sql_catalog = pyi_catalog.load_catalog(**_CATALOG_KWARGS)
+    table = sql_catalog.load_table(f"{_DB_NAME}.{_TABLE_NAME}")
+    batch = pa.Table.from_pydict(
+        {
+            "col_a": col_a_values,
+            "col_b": ["x"] * len(col_a_values),
+            "col_c": [1] * len(col_a_values),
+        },
+        schema=_SCHEMA,
+    )
+    table.append(batch)
+    return table.current_snapshot().snapshot_id
+
+
+@pytest.mark.skipif(
+    get_pyarrow_version() < parse_version("14.0.0"),
+    reason="PyIceberg 0.7.0 fails on pyarrow <= 14.0.0",
+)
+def test_incremental_snapshot_id_mutually_exclusive():
+    """`snapshot_id` cannot be combined with the incremental snapshot bounds."""
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        IcebergDatasource(
+            table_identifier=f"{_DB_NAME}.{_TABLE_NAME}",
+            snapshot_id=123,
+            start_snapshot_id=456,
+            catalog_kwargs=_CATALOG_KWARGS.copy(),
+        )
+
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        IcebergDatasource(
+            table_identifier=f"{_DB_NAME}.{_TABLE_NAME}",
+            snapshot_id=123,
+            end_snapshot_id=789,
+            catalog_kwargs=_CATALOG_KWARGS.copy(),
+        )
+
+
+@pytest.mark.skipif(
+    get_pyarrow_version() < parse_version("14.0.0"),
+    reason="PyIceberg 0.7.0 fails on pyarrow <= 14.0.0",
+)
+def test_incremental_read_requires_pyiceberg_support(monkeypatch):
+    """A clear error is raised when PyIceberg lacks incremental_append_scan.
+
+    This runs on all PyIceberg versions: on versions without the capability the
+    attribute is simply absent; on versions with it we remove it to exercise the
+    guard.
+    """
+    start_snapshot_id = _append_batch([200, 201, 202])
+
+    iceberg_ds = IcebergDatasource(
+        table_identifier=f"{_DB_NAME}.{_TABLE_NAME}",
+        start_snapshot_id=start_snapshot_id,
+        catalog_kwargs=_CATALOG_KWARGS.copy(),
+    )
+    # Force the capability check to fail regardless of the installed version.
+    monkeypatch.delattr(Table, "incremental_append_scan", raising=False)
+
+    with pytest.raises(ValueError, match="incremental_append_scan"):
+        # Triggers _get_data_scan via the plan_files property.
+        _ = iceberg_ds.plan_files
+
+
+@pytest.mark.skipif(
+    get_pyarrow_version() < parse_version("14.0.0"),
+    reason="PyIceberg 0.7.0 fails on pyarrow <= 14.0.0",
+)
+@pytest.mark.skipif(
+    not _supports_incremental_scan(), reason=_INCREMENTAL_UNSUPPORTED_REASON
+)
+def test_incremental_read_between_snapshots():
+    """Reading between two snapshots returns only the rows appended in that range."""
+    # The autouse fixture appends 120 rows then deletes some; append two more
+    # batches so we have distinct append snapshots to read between.
+    start_snapshot_id = _append_batch([200, 201, 202])  # start bound (exclusive)
+    _append_batch([300, 301])  # appended within the range
+    end_snapshot_id = _append_batch([400, 401, 402])  # end bound (inclusive)
+
+    ds = read_iceberg(
+        table_identifier=f"{_DB_NAME}.{_TABLE_NAME}",
+        start_snapshot_id=start_snapshot_id,
+        end_snapshot_id=end_snapshot_id,
+        catalog_kwargs=_CATALOG_KWARGS.copy(),
+    )
+
+    result = ds.to_pandas().sort_values("col_a").reset_index(drop=True)
+    # Start is exclusive, so the 200/201/202 batch is not included; end is
+    # inclusive, so the 400/401/402 batch is.
+    assert result["col_a"].tolist() == [300, 301, 400, 401, 402]
+
+
+@pytest.mark.skipif(
+    get_pyarrow_version() < parse_version("14.0.0"),
+    reason="PyIceberg 0.7.0 fails on pyarrow <= 14.0.0",
+)
+@pytest.mark.skipif(
+    not _supports_incremental_scan(), reason=_INCREMENTAL_UNSUPPORTED_REASON
+)
+def test_incremental_read_defaults_end_to_current_snapshot():
+    """Omitting end_snapshot_id reads everything appended after the start snapshot."""
+    start_snapshot_id = _append_batch([200, 201])
+    _append_batch([300, 301])
+    _append_batch([400])
+
+    ds = read_iceberg(
+        table_identifier=f"{_DB_NAME}.{_TABLE_NAME}",
+        start_snapshot_id=start_snapshot_id,
+        catalog_kwargs=_CATALOG_KWARGS.copy(),
+    )
+
+    result = ds.to_pandas().sort_values("col_a").reset_index(drop=True)
+    assert result["col_a"].tolist() == [300, 301, 400]
+
+
+@pytest.mark.skipif(
+    get_pyarrow_version() < parse_version("14.0.0"),
+    reason="PyIceberg 0.7.0 fails on pyarrow <= 14.0.0",
+)
+@pytest.mark.skipif(
+    not _supports_incremental_scan(), reason=_INCREMENTAL_UNSUPPORTED_REASON
+)
+def test_incremental_read_with_predicate_pushdown():
+    """Predicate pushdown composes with incremental reads."""
+    start_snapshot_id = _append_batch([200, 201, 202])
+    end_snapshot_id = _append_batch([300, 301, 302])
+
+    ds = read_iceberg(
+        table_identifier=f"{_DB_NAME}.{_TABLE_NAME}",
+        start_snapshot_id=start_snapshot_id,
+        end_snapshot_id=end_snapshot_id,
+        catalog_kwargs=_CATALOG_KWARGS.copy(),
+    ).filter(expr=col("col_a") >= 301)
+
+    result = ds.to_pandas().sort_values("col_a").reset_index(drop=True)
+    assert result["col_a"].tolist() == [301, 302]
+
+
 if __name__ == "__main__":
     import sys