From 17cf5e3e72a874b31529b8d480f5d9fd9380c63c Mon Sep 17 00:00:00 2001 From: Tanmay Rauth Date: Wed, 1 Jul 2026 14:38:40 -0700 Subject: [PATCH] [Data] Support incremental reads for Iceberg tables Extend ray.data.read_iceberg(...) with start_snapshot_id / end_snapshot_id to read only the rows appended between two Iceberg snapshots, built on PyIceberg's Table.incremental_append_scan(...) (apache/iceberg-python#3512). - IcebergDatasource: incremental-scan branch reusing the existing plan_files()/projection() read path; validates mutual exclusivity with snapshot_id; guards on the PyIceberg capability with a clear error. - read_iceberg: additive params, docstring, and example. - Tests: validation, capability guard (runs on all PyIceberg versions), and functional incremental reads (skipped until the pin ships #3512). Related to #64464 Signed-off-by: Tanmay Rauth --- .../datasource/iceberg_datasource.py | 67 +++++++- python/ray/data/read_api.py | 22 ++- .../ray/data/tests/datasource/test_iceberg.py | 150 ++++++++++++++++++ 3 files changed, 235 insertions(+), 4 deletions(-) diff --git a/python/ray/data/_internal/datasource/iceberg_datasource.py b/python/ray/data/_internal/datasource/iceberg_datasource.py index 41f8afe6d031..caeab334902e 100644 --- a/python/ray/data/_internal/datasource/iceberg_datasource.py +++ b/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,7 +390,7 @@ 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() @@ -375,6 +398,9 @@ def _get_data_scan(self) -> "DataScan": 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 diff --git a/python/ray/data/read_api.py b/python/ray/data/read_api.py index 79b62718e3d7..0aaa407791a3 100644 --- a/python/ray/data/read_api.py +++ b/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, ) diff --git a/python/ray/data/tests/datasource/test_iceberg.py b/python/ray/data/tests/datasource/test_iceberg.py index a7473c71262b..34eef5d3a894 100644 --- a/python/ray/data/tests/datasource/test_iceberg.py +++ b/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