feat: Honor loadTable scan-planning-mode overrides per table

Prefer scan-planning-mode from LoadTableResponse.config over the
catalog-level property (matching Java), so REST catalogs can enable
server-side planning only for selected tables. Invalid catalog values
are ignored with a warning and no longer block a valid table override.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: lukeFalsina

mkdocs/docs/api.md

@@ -1695,7 +1695,7 @@ scan = table.scan(
 [task.file.file_path for task in scan.plan_files()]
 ```
 
-When the REST catalog returns `scan-planning-mode=server` and advertises the plan endpoint, `plan_files()` / `to_arrow()` use server-side scan planning. Catalogs that return async plans (`status=submitted`) are polled automatically until they reach a terminal state; see [REST Catalog configuration](configuration.md#rest-catalog).
+When the REST catalog returns `scan-planning-mode=server` and advertises the plan endpoint, `plan_files()` / `to_arrow()` use server-side scan planning. The mode can also be returned per table in the `loadTable` response `config`, which takes precedence over the catalog-level setting, so a server can require server-side planning for some tables while others keep client-side planning. Catalogs that return async plans (`status=submitted`) are polled automatically until they reach a terminal state; see [REST Catalog configuration](configuration.md#rest-catalog).
 
 The low level API `plan_files` methods returns a set of tasks that provide the files that might contain matching rows:
 

mkdocs/docs/configuration.md

@@ -386,7 +386,7 @@ catalog:
 | snapshot-loading-mode | refs                           | The snapshots to return in the body of the metadata. Setting the value to `all` would return the full set of snapshots currently valid for the table. Setting the value to `refs` would load all snapshots referenced by branches or tags. |
 | `header.X-Iceberg-Access-Delegation` | `vended-credentials` | Signal to the server that the client supports delegated access via a comma-separated list of access mechanisms. The server may choose to supply access via any or none of the requested mechanisms. When using `vended-credentials`, the server provides temporary credentials to the client. When using `remote-signing`, the server signs requests on behalf of the client. (default: `vended-credentials`) |
 | view-endpoints-supported | false                           | For backwards compatibility with older REST servers. Set to `true` if the server supports view endpoints but doesn't send the `endpoints` field in the ConfigResponse. |
-| scan-planning-mode | client | When set to `server`, and the catalog advertises the plan-table-scan endpoint, `table.scan()` uses REST server-side scan planning. May be set by the client or returned by the catalog via `GET /v1/config` defaults/overrides when the server wants clients to use server-side planning. Async plans (`status=submitted`) are polled via `GET .../plan/{plan-id}` until completion. |
+| scan-planning-mode | client | When set to `server`, and the catalog advertises the plan-table-scan endpoint, `table.scan()` uses REST server-side scan planning. May be set by the client, returned by the catalog via `GET /v1/config` defaults/overrides, or returned per table in the `config` of the `loadTable` response. The `loadTable` value takes precedence over the catalog-level value, which lets a server request server-side planning only for specific tables. Async plans (`status=submitted`) are polled via `GET .../plan/{plan-id}` until completion. |
 | rest-scan-planning.poll-timeout-ms | 300000 | Maximum time to wait for an async scan plan to complete before failing (default: 5 minutes). |
 
 When server-side planning returns `storage-credentials` on a completed plan, PyIceberg applies them to the scan-scoped FileIO (layered on top of the existing table/load-time IO properties) so planned data and delete files can be read.

pyiceberg/catalog/__init__.py

@@ -804,8 +804,13 @@ def namespace_to_string(identifier: str | Identifier, err: type[ValueError] | ty
         return ".".join(segment.strip() for segment in tuple_identifier)
 
     @abstractmethod
-    def supports_server_side_planning(self) -> bool:
-        """Check if the catalog supports server-side scan planning."""
+    def supports_server_side_planning(self, table_config: Properties = EMPTY_DICT) -> bool:
+        """Check if server-side scan planning should be used.
+
+        Args:
+            table_config: Table configuration returned by the catalog when loading the table,
+                which may override the catalog-level scan planning mode for a single table.
+        """
 
     @staticmethod
     def identifier_to_database(
@@ -907,7 +912,7 @@ def __init__(self, name: str, **properties: str):
         super().__init__(name, **properties)
 
     @override
-    def supports_server_side_planning(self) -> bool:
+    def supports_server_side_planning(self, table_config: Properties = EMPTY_DICT) -> bool:
         return False
 
     @override

pyiceberg/catalog/noop.py

@@ -98,7 +98,7 @@ def drop_table(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 
     @override
-    def supports_server_side_planning(self) -> bool:
+    def supports_server_side_planning(self, table_config: Properties = EMPTY_DICT) -> bool:
         return False
 
     @override

pyiceberg/catalog/rest/__init__.py

@@ -16,6 +16,7 @@
 #  under the License.
 from __future__ import annotations
 
+import logging
 import time
 from collections import deque
 from enum import Enum
@@ -100,6 +101,8 @@
 if TYPE_CHECKING:
     import pyarrow as pa
 
+logger = logging.getLogger(__name__)
+
 
 class HttpMethod(str, Enum):
     GET = "GET"
@@ -295,6 +298,23 @@ class ScanPlanningMode(Enum):
 DEFAULT_NAMESPACE_SEPARATOR = b"\x1f".decode(UTF8)
 
 
+def _parse_scan_planning_mode(properties: Properties, *, strict: bool = True) -> ScanPlanningMode | None:
+    """Read the scan planning mode from a set of properties, returning None when it is not set.
+
+    When ``strict`` is False, an unrecognized value is logged and treated as unset so a higher-priority
+    source (for example a loadTable override) or the default mode can still decide.
+    """
+    if (mode := properties.get(SCAN_PLANNING_MODE)) is None:
+        return None
+    try:
+        return ScanPlanningMode(str(mode).strip().lower())
+    except ValueError as exc:
+        if strict:
+            raise ValueError(f"Invalid {SCAN_PLANNING_MODE}: {mode}") from exc
+        logger.warning("Ignoring invalid %s=%r", SCAN_PLANNING_MODE, mode)
+        return None
+
+
 def _retry_hook(retry_state: RetryCallState) -> None:
     rest_catalog: RestCatalog = retry_state.args[0]
     rest_catalog._refresh_token()  # pylint: disable=protected-access
@@ -512,11 +532,31 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: str | Non
             merged_properties[AUTH_MANAGER] = self._auth_manager
         return load_file_io(merged_properties, location)
 
+    def _effective_scan_planning_mode(self, table_config: Properties) -> ScanPlanningMode:
+        """Resolve the scan planning mode, where a loadTable override wins over the catalog property.
+
+        An invalid catalog-level value is ignored (with a warning) so it cannot block a valid
+        loadTable override or the default client-side mode. An invalid loadTable value still fails.
+        """
+        # Parse the table override first so a valid loadTable value is not blocked by a bad catalog property.
+        table_mode = _parse_scan_planning_mode(table_config)
+        catalog_mode = _parse_scan_planning_mode(self.properties, strict=False)
+
+        if catalog_mode is not None and table_mode is not None and catalog_mode != table_mode:
+            logger.warning(
+                "Scan planning mode mismatch: client config=%s, server config=%s. Server config takes precedence.",
+                catalog_mode.value,
+                table_mode.value,
+            )
+
+        return table_mode or catalog_mode or ScanPlanningMode(SCAN_PLANNING_MODE_DEFAULT)
+
     @override
-    def supports_server_side_planning(self) -> bool:
-        """Check if the catalog supports server-side scan planning."""
-        scan_planning_mode = ScanPlanningMode(self.properties.get(SCAN_PLANNING_MODE, SCAN_PLANNING_MODE_DEFAULT))
-        return Capability.V1_SUBMIT_TABLE_SCAN_PLAN in self._supported_endpoints and scan_planning_mode == ScanPlanningMode.SERVER
+    def supports_server_side_planning(self, table_config: Properties = EMPTY_DICT) -> bool:
+        """Check if server-side scan planning should be used, honoring a per-table loadTable override."""
+        if Capability.V1_SUBMIT_TABLE_SCAN_PLAN not in self._supported_endpoints:
+            return False
+        return self._effective_scan_planning_mode(table_config) == ScanPlanningMode.SERVER
 
     @retry(**_RETRY_ARGS)
     def _plan_table_scan(self, identifier: str | Identifier, request: PlanTableScanRequest) -> PlanningResponse:

pyiceberg/table/__init__.py

@@ -1267,6 +1267,7 @@ def scan(
             limit=limit,
             catalog=self.catalog,
             table_identifier=self._identifier,
+            table_config=self.config,
         )
 
     def incremental_append_scan(
@@ -1985,6 +1986,7 @@ class TableScan(BaseScan):
     snapshot_id: int | None
     catalog: Catalog | None
     table_identifier: Identifier | None
+    table_config: Properties
 
     def __init__(
         self,
@@ -1998,6 +2000,7 @@ def __init__(
         limit: int | None = None,
         catalog: Catalog | None = None,
         table_identifier: Identifier | None = None,
+        table_config: Properties = EMPTY_DICT,
     ):
         super().__init__(
             table_metadata=table_metadata,
@@ -2011,6 +2014,7 @@ def __init__(
         self.snapshot_id = snapshot_id
         self.catalog = catalog
         self.table_identifier = table_identifier
+        self.table_config = table_config
 
     def snapshot(self) -> Snapshot | None:
         if self.snapshot_id:
@@ -2255,7 +2259,7 @@ 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.supports_server_side_planning()
+        return self.catalog.supports_server_side_planning(self.table_config)
 
     def _plan_files_server_side(self) -> Iterable[FileScanTask]:
         """Plan files using REST server-side scan planning."""

tests/catalog/test_rest.py

@@ -2857,6 +2857,60 @@ def test_server_side_planning_enabled_from_server_config(self, rest_mock: Mocker
 
         assert catalog.supports_server_side_planning() is True
 
+    def test_server_side_planning_enabled_by_table_config(self, rest_mock: Mocker) -> None:
+        catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+        assert catalog.supports_server_side_planning() is False
+        assert catalog.supports_server_side_planning({"scan-planning-mode": ScanPlanningMode.SERVER.value}) is True
+
+    def test_server_side_planning_table_config_overrides_catalog_property(self, rest_mock: Mocker) -> None:
+        catalog = RestCatalog(
+            "rest",
+            uri=TEST_URI,
+            token=TEST_TOKEN,
+            **{"scan-planning-mode": ScanPlanningMode.SERVER.value},
+        )
+
+        assert catalog.supports_server_side_planning({"scan-planning-mode": ScanPlanningMode.CLIENT.value}) is False
+
+    def test_server_side_planning_table_config_ignored_when_endpoint_unsupported(self, requests_mock: Mocker) -> None:
+        requests_mock.get(
+            f"{TEST_URI}v1/config",
+            json={"defaults": {}, "overrides": {}},
+            status_code=200,
+        )
+        catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+        assert catalog.supports_server_side_planning({"scan-planning-mode": ScanPlanningMode.SERVER.value}) is False
+
+    def test_server_side_planning_invalid_mode(self, rest_mock: Mocker) -> None:
+        catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+        with pytest.raises(ValueError, match="Invalid scan-planning-mode: remote"):
+            catalog.supports_server_side_planning({"scan-planning-mode": "remote"})
+
+    def test_server_side_planning_invalid_catalog_mode_falls_back_to_default(self, rest_mock: Mocker) -> None:
+        catalog = RestCatalog(
+            "rest",
+            uri=TEST_URI,
+            token=TEST_TOKEN,
+            **{"scan-planning-mode": "servr"},
+        )
+
+        # Bad catalog config is ignored; default remains client-side planning.
+        assert catalog.supports_server_side_planning() is False
+
+    def test_server_side_planning_table_override_survives_invalid_catalog_mode(self, rest_mock: Mocker) -> None:
+        catalog = RestCatalog(
+            "rest",
+            uri=TEST_URI,
+            token=TEST_TOKEN,
+            **{"scan-planning-mode": "servr"},
+        )
+
+        assert catalog.supports_server_side_planning({"scan-planning-mode": ScanPlanningMode.SERVER.value}) is True
+        assert catalog.supports_server_side_planning({"scan-planning-mode": ScanPlanningMode.CLIENT.value}) is False
+
     def test_supported_endpoint(self, requests_mock: Mocker) -> None:
         requests_mock.get(
             f"{TEST_URI}v1/config",

tests/catalog/test_scan_planning_models.py

@@ -52,6 +52,7 @@ def rest_scan_catalog(requests_mock: Mocker) -> RestCatalog:
             "defaults": {"scan-planning-mode": "server"},
             "overrides": {},
             "endpoints": [
+                "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}",
                 "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan",
                 "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}",
                 "DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}",
@@ -68,6 +69,28 @@ def rest_scan_catalog(requests_mock: Mocker) -> RestCatalog:
     )
 
 
+@pytest.fixture
+def rest_client_planning_catalog(requests_mock: Mocker) -> RestCatalog:
+    """A catalog that advertises the plan endpoints but leaves the mode at the client-side default."""
+    requests_mock.get(
+        f"{TEST_URI}v1/config",
+        json={
+            "defaults": {},
+            "overrides": {},
+            "endpoints": [
+                "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}",
+                "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan",
+                "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}",
+                "DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}",
+                "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks",
+            ],
+        },
+        status_code=200,
+    )
+
+    return RestCatalog("test", uri=TEST_URI)
+
+
 def _rest_data_file(
     *,
     file_path: str = "s3://bucket/table/data/file.parquet",
@@ -694,3 +717,102 @@ def test_plan_scan_equality_deletes_not_supported(rest_scan_catalog: RestCatalog
     request = PlanTableScanRequest()
     with pytest.raises(NotImplementedError, match="PyIceberg does not yet support equality deletes"):
         rest_scan_catalog.plan_scan(("db", "tbl"), request)
+
+
+def _mock_load_table(requests_mock: Mocker, metadata: dict[str, Any], config: dict[str, str]) -> None:
+    requests_mock.get(
+        f"{TEST_URI}v1/namespaces/db/tables/tbl",
+        json={
+            "metadata-location": "s3://bucket/tbl/metadata/00000.metadata.json",
+            "metadata": metadata,
+            "config": config,
+        },
+        status_code=200,
+    )
+
+
+def test_scan_uses_server_side_planning_from_load_table_config(
+    rest_client_planning_catalog: RestCatalog, requests_mock: Mocker, example_table_metadata_v2: dict[str, Any]
+) -> None:
+    _mock_load_table(requests_mock, example_table_metadata_v2, {"scan-planning-mode": "server"})
+    plan_mock = requests_mock.post(
+        f"{TEST_URI}v1/namespaces/db/tables/tbl/plan",
+        json={
+            "status": "completed",
+            "plan-id": "plan-123",
+            "delete-files": [],
+            "file-scan-tasks": [{"data-file": _rest_data_file(file_path="s3://bucket/tbl/data/file1.parquet")}],
+            "plan-tasks": [],
+        },
+        status_code=200,
+    )
+
+    table = rest_client_planning_catalog.load_table(("db", "tbl"))
+    scan = table.scan()
+
+    assert scan._should_use_server_side_planning() is True
+
+    tasks = list(scan.plan_files())
+    assert [task.file.file_path for task in tasks] == ["s3://bucket/tbl/data/file1.parquet"]
+    assert plan_mock.call_count == 1
+
+
+def test_scan_keeps_client_side_planning_without_load_table_config(
+    rest_client_planning_catalog: RestCatalog, requests_mock: Mocker, example_table_metadata_v2: dict[str, Any]
+) -> None:
+    _mock_load_table(requests_mock, example_table_metadata_v2, {})
+
+    table = rest_client_planning_catalog.load_table(("db", "tbl"))
+
+    assert table.scan()._should_use_server_side_planning() is False
+
+
+def test_scan_load_table_config_overrides_catalog_property(
+    rest_scan_catalog: RestCatalog, requests_mock: Mocker, example_table_metadata_v2: dict[str, Any]
+) -> None:
+    _mock_load_table(requests_mock, example_table_metadata_v2, {"scan-planning-mode": "client"})
+
+    table = rest_scan_catalog.load_table(("db", "tbl"))
+
+    assert table.scan()._should_use_server_side_planning() is False
+
+
+def test_scan_load_table_override_survives_invalid_catalog_mode(
+    requests_mock: Mocker, example_table_metadata_v2: dict[str, Any]
+) -> None:
+    requests_mock.get(
+        f"{TEST_URI}v1/config",
+        json={
+            "defaults": {},
+            "overrides": {},
+            "endpoints": [
+                "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}",
+                "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan",
+                "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}",
+                "DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}",
+                "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks",
+            ],
+        },
+        status_code=200,
+    )
+    catalog = RestCatalog("test", uri=TEST_URI, **{"scan-planning-mode": "servr"})
+    _mock_load_table(requests_mock, example_table_metadata_v2, {"scan-planning-mode": "server"})
+    plan_mock = requests_mock.post(
+        f"{TEST_URI}v1/namespaces/db/tables/tbl/plan",
+        json={
+            "status": "completed",
+            "plan-id": "plan-123",
+            "delete-files": [],
+            "file-scan-tasks": [{"data-file": _rest_data_file(file_path="s3://bucket/tbl/data/file1.parquet")}],
+            "plan-tasks": [],
+        },
+        status_code=200,
+    )
+
+    assert catalog.supports_server_side_planning() is False
+
+    table = catalog.load_table(("db", "tbl"))
+    scan = table.scan()
+    assert scan._should_use_server_side_planning() is True
+    assert [task.file.file_path for task in scan.plan_files()] == ["s3://bucket/tbl/data/file1.parquet"]
+    assert plan_mock.call_count == 1