[data] support new pyiceberg version (#51744)

<!-- Thank you for your contribution! Please review
https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before
opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR.
If you don't have the access to it, we will shortly find a reviewer and
assign them to your PR. -->

## Why are these changes needed?

<!-- Please give a short summary of the change and the problem this
solves. -->

## Related issue number

<!-- For example: "Closes #1234" -->

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for
https://docs.ray.io/en/master/.
- [ ] I've added any new APIs to the API Reference. For example, if I
added a
method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a
few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: jukejian <jukejian@bytedance.com>
Signed-off-by: Richard Liaw <rliaw@berkeley.edu>
Co-authored-by: Richard Liaw <rliaw@berkeley.edu>

Author: 2 people

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

@@ -11,6 +11,7 @@
 from ray.data._internal.execution.interfaces import TaskContext
 from ray.data.datasource.datasink import WriteResult
 import uuid
+from packaging import version
 
 if TYPE_CHECKING:
     from pyiceberg.catalog import Catalog
@@ -85,7 +86,15 @@ def _get_catalog(self) -> "Catalog":
 
     def on_write_start(self) -> None:
         """Prepare for the transaction"""
-        from pyiceberg.table import PropertyUtil, TableProperties
+        import pyiceberg
+        from pyiceberg.table import TableProperties
+
+        if version.parse(pyiceberg.__version__) >= version.parse("0.9.0"):
+            from pyiceberg.utils.properties import property_as_bool
+        else:
+            from pyiceberg.table import PropertyUtil
+
+            property_as_bool = PropertyUtil.property_as_bool
 
         catalog = self._get_catalog()
         table = catalog.load_table(self.table_identifier)
@@ -103,7 +112,7 @@ def on_write_start(self) -> None:
                 f"Not all partition types are supported for writes. Following partitions cannot be written using pyarrow: {unsupported_partitions}."
             )
 
-        self._manifest_merge_enabled = PropertyUtil.property_as_bool(
+        self._manifest_merge_enabled = property_as_bool(
             self._table_metadata.properties,
             TableProperties.MANIFEST_MERGE_ENABLED,
             TableProperties.MANIFEST_MERGE_ENABLED_DEFAULT,

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

@@ -13,6 +13,9 @@
 from ray.data.datasource.datasource import Datasource, ReadTask
 from ray.util.annotations import DeveloperAPI
 
+from packaging import version
+import pyarrow as pa
+
 if TYPE_CHECKING:
     from pyiceberg.catalog import Catalog
     from pyiceberg.expressions import BooleanExpression
@@ -34,21 +37,47 @@ def _get_read_task(
     limit: Optional[int],
     schema: "Schema",
 ) -> Iterable[Block]:
-    from pyiceberg.io import pyarrow as pyi_pa_io
-
-    # Use the PyIceberg API to read only a single task (specifically, a
-    # FileScanTask) - note that this is not as simple as reading a single
-    # parquet file, as there might be delete files, etc. associated, so we
-    # must use the PyIceberg API for the projection.
-    yield pyi_pa_io.project_table(
-        tasks=tasks,
-        table_metadata=table_metadata,
-        io=table_io,
-        row_filter=row_filter,
-        projected_schema=schema,
-        case_sensitive=case_sensitive,
-        limit=limit,
-    )
+    # Determine the PyIceberg version to handle backward compatibility
+    import pyiceberg
+
+    if version.parse(pyiceberg.__version__) >= version.parse("0.9.0"):
+        # Modern implementation using ArrowScan (PyIceberg 0.9.0+)
+        from pyiceberg.io.pyarrow import ArrowScan
+
+        # Initialize scanner with Iceberg metadata and query parameters
+        scanner = ArrowScan(
+            table_metadata=table_metadata,
+            io=table_io,
+            row_filter=row_filter,
+            projected_schema=schema,
+            case_sensitive=case_sensitive,
+            limit=limit,
+        )
+
+        # Convert scanned data to Arrow Table format
+        result_table = scanner.to_table(tasks=tasks)
+
+        # Stream results as RecordBatches for memory efficiency
+        for batch in result_table.to_batches():
+            yield pa.Table.from_batches([batch])
+
+    else:
+        # Legacy implementation using project_table (PyIceberg <0.9.0)
+        from pyiceberg.io import pyarrow as pyi_pa_io
+
+        # Use the PyIceberg API to read only a single task (specifically, a
+        # FileScanTask) - note that this is not as simple as reading a single
+        # parquet file, as there might be delete files, etc. associated, so we
+        # must use the PyIceberg API for the projection.
+        yield pyi_pa_io.project_table(
+            tasks=tasks,
+            table_metadata=table_metadata,
+            io=table_io,
+            row_filter=row_filter,
+            projected_schema=schema,
+            case_sensitive=case_sensitive,
+            limit=limit,
+        )
 
 
 @DeveloperAPI

python/requirements/ml/data-test-requirements.txt

@@ -19,6 +19,6 @@ deltalake
 pytest-mock
 decord
 snowflake-connector-python
-pyiceberg[sql-sqlite]==0.7.0
+pyiceberg[sql-sqlite]==0.9.0
 clickhouse-connect
 hudi==0.2.0

python/requirements_compiled.txt

@@ -235,10 +235,11 @@ braceexpand==0.1.7
     # via webdataset
 bracex==2.4
     # via wcmatch
-cachetools==5.3.2
+cachetools==5.5.2
     # via
     #   aim
     #   google-auth
+    #   pyiceberg
 certifi==2025.1.31
     # via
     #   -r python/requirements/cloud-requirements.txt
@@ -1649,7 +1650,7 @@ pygments==2.18.0
     #   nbconvert
     #   rich
     #   sphinx
-pyiceberg==0.7.0
+pyiceberg==0.9.0
     # via -r python/requirements/ml/data-test-requirements.txt
 pyjwt==2.8.0
     # via

python/requirements_compiled_rayllm_py311_cpu.txt

@@ -241,9 +241,9 @@ blake3==1.0.4 \
     # via
     #   -c python/requirements_compiled_rayllm_test_py311_cpu.txt
     #   vllm
-cachetools==5.3.2 \
-    --hash=sha256:086ee420196f7b2ab9ca2db2520aca326318b68fe5ba8bc4d49cca91add450f2 \
-    --hash=sha256:861f35a13a451f94e301ce2bec7cac63e881232ccce7ed67fab9b5df4d3beaa1
+cachetools==5.5.2 \
+    --hash=sha256:1a661caa9175d26759571b2e19580f9d6393969e5dfca11fdb1f947a23e640d4 \
+    --hash=sha256:d26a22bcc62eb95c3beabd9f1ee5e820d3d2704fe2967cbe350e20c8ffcd3f0a
     # via
     #   -c python/requirements_compiled_rayllm_test_py311_cpu.txt
     #   google-auth

python/requirements_compiled_rayllm_py311_cu121.txt

@@ -241,9 +241,9 @@ blake3==1.0.4 \
     # via
     #   -c python/requirements_compiled_rayllm_test_py311_cu121.txt
     #   vllm
-cachetools==5.3.2 \
-    --hash=sha256:086ee420196f7b2ab9ca2db2520aca326318b68fe5ba8bc4d49cca91add450f2 \
-    --hash=sha256:861f35a13a451f94e301ce2bec7cac63e881232ccce7ed67fab9b5df4d3beaa1
+cachetools==5.5.2 \
+    --hash=sha256:1a661caa9175d26759571b2e19580f9d6393969e5dfca11fdb1f947a23e640d4 \
+    --hash=sha256:d26a22bcc62eb95c3beabd9f1ee5e820d3d2704fe2967cbe350e20c8ffcd3f0a
     # via
     #   -c python/requirements_compiled_rayllm_test_py311_cu121.txt
     #   google-auth

python/requirements_compiled_rayllm_py311_cu124.txt

@@ -241,9 +241,9 @@ blake3==1.0.4 \
     # via
     #   -c python/requirements_compiled_rayllm_test_py311_cu124.txt
     #   vllm
-cachetools==5.3.2 \
-    --hash=sha256:086ee420196f7b2ab9ca2db2520aca326318b68fe5ba8bc4d49cca91add450f2 \
-    --hash=sha256:861f35a13a451f94e301ce2bec7cac63e881232ccce7ed67fab9b5df4d3beaa1
+cachetools==5.5.2 \
+    --hash=sha256:1a661caa9175d26759571b2e19580f9d6393969e5dfca11fdb1f947a23e640d4 \
+    --hash=sha256:d26a22bcc62eb95c3beabd9f1ee5e820d3d2704fe2967cbe350e20c8ffcd3f0a
     # via
     #   -c python/requirements_compiled_rayllm_test_py311_cu124.txt
     #   google-auth