Address comments from fokko

Author: geruh

pyiceberg/catalog/__init__.py

@@ -722,6 +722,10 @@ def namespace_to_string(identifier: str | Identifier, err: type[ValueError] | ty
 
         return ".".join(segment.strip() for segment in tuple_identifier)
 
+    def supports_server_side_planning(self) -> bool:
+        """Check if the catalog supports server-side scan planning."""
+        return False
+
     @staticmethod
     def identifier_to_database(
         identifier: str | Identifier, err: type[ValueError] | type[NoSuchNamespaceError] = ValueError

pyiceberg/catalog/rest/__init__.py

@@ -390,12 +390,8 @@ def _create_session(self) -> Session:
 
         return session
 
-    def is_rest_scan_planning_enabled(self) -> bool:
-        """Check if rest server-side scan planning is enabled.
-
-        Returns:
-            True if enabled, False otherwise.
-        """
+    def supports_server_side_planning(self) -> bool:
+        """Check if the catalog supports server-side scan planning."""
         return Capability.V1_SUBMIT_TABLE_SCAN_PLAN in self._supported_endpoints and property_as_bool(
             self.properties, REST_SCAN_PLANNING_ENABLED, REST_SCAN_PLANNING_ENABLED_DEFAULT
         )
@@ -410,6 +406,7 @@ def _plan_table_scan(self, identifier: str | Identifier, request: PlanTableScanR
 
         Returns:
             PlanningResponse the result of the scan plan request representing the status
+
         Raises:
             NoSuchTableError: If a table with the given identifier does not exist.
         """

pyiceberg/catalog/rest/scan_planning.py

@@ -25,16 +25,9 @@
 
 from pyiceberg.catalog.rest.response import ErrorResponseMessage
 from pyiceberg.expressions import BooleanExpression, SerializableBooleanExpression
-from pyiceberg.manifest import DataFileContent, FileFormat
+from pyiceberg.manifest import FileFormat
 from pyiceberg.typedef import IcebergBaseModel
 
-# REST content-type to DataFileContent
-CONTENT_TYPE_MAP: dict[str, DataFileContent] = {
-    "data": DataFileContent.DATA,
-    "position-deletes": DataFileContent.POSITION_DELETES,
-    "equality-deletes": DataFileContent.EQUALITY_DELETES,
-}
-
 # Primitive types that can appear in partition values and bounds
 PrimitiveTypeValue: TypeAlias = bool | int | float | str | Decimal | UUID | date | time | datetime | bytes
 

pyiceberg/manifest.py

@@ -69,6 +69,28 @@ def __repr__(self) -> str:
         """Return the string representation of the DataFileContent class."""
         return f"DataFileContent.{self.name}"
 
+    @staticmethod
+    def from_rest_type(content_type: str) -> DataFileContent:
+        """Convert REST API content type string to DataFileContent.
+
+        Args:
+            content_type: REST API content type.
+
+        Returns:
+            The corresponding DataFileContent enum value.
+
+        Raises:
+            ValueError: If the content type is unknown.
+        """
+        mapping = {
+            "data": DataFileContent.DATA,
+            "position-deletes": DataFileContent.POSITION_DELETES,
+            "equality-deletes": DataFileContent.EQUALITY_DELETES,
+        }
+        if content_type not in mapping:
+            raise ValueError(f"Invalid file content value: {content_type}")
+        return mapping[content_type]
+
 
 class ManifestContent(int, Enum):
     DATA = 0

pyiceberg/table/__init__.py

@@ -1849,7 +1849,7 @@ def from_rest_response(
 
 def _rest_file_to_data_file(rest_file: RESTContentFile) -> DataFile:
     """Convert a REST content file to a manifest DataFile."""
-    from pyiceberg.catalog.rest.scan_planning import CONTENT_TYPE_MAP, RESTDataFile
+    from pyiceberg.catalog.rest.scan_planning import RESTDataFile
 
     if isinstance(rest_file, RESTDataFile):
         column_sizes = rest_file.column_sizes.to_dict() if rest_file.column_sizes else None
@@ -1863,7 +1863,7 @@ def _rest_file_to_data_file(rest_file: RESTContentFile) -> DataFile:
         nan_value_counts = None
 
     data_file = DataFile.from_args(
-        content=CONTENT_TYPE_MAP[rest_file.content],
+        content=DataFileContent.from_rest_type(rest_file.content),
         file_path=rest_file.file_path,
         file_format=rest_file.file_format,
         partition=Record(*rest_file.partition) if rest_file.partition else Record(),
@@ -2051,15 +2051,13 @@ def scan_plan_helper(self) -> Iterator[list[ManifestEntry]]:
             ],
         )
 
-    def _should_use_rest_planning(self) -> bool:
-        """Check if REST scan planning should be used for this scan."""
-        from pyiceberg.catalog.rest import RestCatalog
-
-        if not isinstance(self.catalog, RestCatalog):
+    def _should_use_server_side_planning(self) -> bool:
+        """Check if server-side scan planning should be used for this scan."""
+        if not self.catalog:
             return False
-        return self.catalog.is_rest_scan_planning_enabled()
+        return self.catalog.supports_server_side_planning()
 
-    def _plan_files_rest(self) -> Iterable[FileScanTask]:
+    def _plan_files_server_side(self) -> Iterable[FileScanTask]:
         """Plan files using REST server-side scan planning."""
         from pyiceberg.catalog.rest import RestCatalog
         from pyiceberg.catalog.rest.scan_planning import PlanTableScanRequest
@@ -2120,8 +2118,8 @@ def plan_files(self) -> Iterable[FileScanTask]:
         Returns:
             List of FileScanTasks that contain both data and delete files.
         """
-        if self._should_use_rest_planning():
-            return self._plan_files_rest()
+        if self._should_use_server_side_planning():
+            return self._plan_files_server_side()
         return self._plan_files_local()
 
     def to_arrow(self) -> pa.Table:

tests/catalog/test_rest.py

@@ -2016,22 +2016,22 @@ def test_rest_catalog_context_manager_with_exception_sigv4(self, rest_mock: Mock
         assert catalog is not None and hasattr(catalog, "_session")
         assert len(catalog._session.adapters) == self.EXPECTED_ADAPTERS_SIGV4
 
-    def test_rest_scan_planning_disabled_by_default(self, rest_mock: Mocker) -> None:
+    def test_server_side_planning_disabled_by_default(self, rest_mock: Mocker) -> None:
         catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
 
-        assert catalog.is_rest_scan_planning_enabled() is False
+        assert catalog.supports_server_side_planning() is False
 
-    def test_rest_scan_planning_enabled_by_property(self, rest_mock: Mocker) -> None:
+    def test_server_side_planning_enabled_by_property(self, rest_mock: Mocker) -> None:
         catalog = RestCatalog(
             "rest",
             uri=TEST_URI,
             token=TEST_TOKEN,
             **{"rest-scan-planning-enabled": "true"},
         )
 
-        assert catalog.is_rest_scan_planning_enabled() is True
+        assert catalog.supports_server_side_planning() is True
 
-    def test_rest_scan_planning_disabled_when_endpoint_unsupported(self, requests_mock: Mocker) -> None:
+    def test_server_side_planning_disabled_when_endpoint_unsupported(self, requests_mock: Mocker) -> None:
         # config endpoint does not populate endpoint falling back to default
         requests_mock.get(
             f"{TEST_URI}v1/config",
@@ -2045,19 +2045,19 @@ def test_rest_scan_planning_disabled_when_endpoint_unsupported(self, requests_mo
             **{"rest-scan-planning-enabled": "true"},
         )
 
-        assert catalog.is_rest_scan_planning_enabled() is False
+        assert catalog.supports_server_side_planning() is False
 
-    def test_rest_scan_planning_explicitly_disabled(self, rest_mock: Mocker) -> None:
+    def test_server_side_planning_explicitly_disabled(self, rest_mock: Mocker) -> None:
         catalog = RestCatalog(
             "rest",
             uri=TEST_URI,
             token=TEST_TOKEN,
             **{"rest-scan-planning-enabled": "false"},
         )
 
-        assert catalog.is_rest_scan_planning_enabled() is False
+        assert catalog.supports_server_side_planning() is False
 
-    def test_rest_scan_planning_enabled_from_server_config(self, rest_mock: Mocker) -> None:
+    def test_server_side_planning_enabled_from_server_config(self, rest_mock: Mocker) -> None:
         rest_mock.get(
             f"{TEST_URI}v1/config",
             json={
@@ -2069,7 +2069,7 @@ def test_rest_scan_planning_enabled_from_server_config(self, rest_mock: Mocker)
         )
         catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
 
-        assert catalog.is_rest_scan_planning_enabled() is True
+        assert catalog.supports_server_side_planning() is True
 
     def test_supported_endpoint(self, requests_mock: Mocker) -> None:
         requests_mock.get(