feat: Add scan planning endpoints and configs

Author: geruh

pyiceberg/catalog/rest/__init__.py

@@ -96,6 +96,8 @@ class Endpoints:
     list_views: str = "namespaces/{namespace}/views"
     drop_view: str = "namespaces/{namespace}/views/{view}"
     view_exists: str = "namespaces/{namespace}/views/{view}"
+    plan_table_scan: str = "namespaces/{namespace}/tables/{table}/plan"
+    fetch_scan_tasks: str = "namespaces/{namespace}/tables/{table}/tasks"
 
 
 class IdentifierKind(Enum):
@@ -130,6 +132,7 @@ class IdentifierKind(Enum):
 SNAPSHOT_LOADING_MODE = "snapshot-loading-mode"
 AUTH = "auth"
 CUSTOM = "custom"
+REST_SCAN_PLANNING_ENABLED = "rest-scan-planning-enabled"
 
 NAMESPACE_SEPARATOR = b"\x1f".decode(UTF8)
 
@@ -269,6 +272,14 @@ def _create_session(self) -> Session:
 
         return session
 
+    def is_scan_planning_enabled(self) -> bool:
+        """Check if server-side scan planning is enabled.
+
+        Returns:
+            True if enabled, False otherwise.
+        """
+        return property_as_bool(self.properties, REST_SCAN_PLANNING_ENABLED, False)
+
     def _create_legacy_oauth2_auth_manager(self, session: Session) -> AuthManager:
         """Create the LegacyOAuth2AuthManager by fetching required properties.
 

tests/catalog/test_rest.py

@@ -1993,3 +1993,53 @@ 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_scan_planning_disabled_by_default(self, rest_mock: Mocker) -> None:
+        rest_mock.get(
+            f"{TEST_URI}v1/config",
+            json={"defaults": {}, "overrides": {}},
+            status_code=200,
+        )
+        catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+        assert catalog.is_scan_planning_enabled() is False
+
+    def test_scan_planning_enabled_by_property(self, rest_mock: Mocker) -> None:
+        rest_mock.get(
+            f"{TEST_URI}v1/config",
+            json={"defaults": {}, "overrides": {}},
+            status_code=200,
+        )
+        catalog = RestCatalog(
+            "rest",
+            uri=TEST_URI,
+            token=TEST_TOKEN,
+            **{"rest-scan-planning-enabled": "true"},
+        )
+
+        assert catalog.is_scan_planning_enabled() is True
+
+    def test_scan_planning_explicitly_disabled(self, rest_mock: Mocker) -> None:
+        rest_mock.get(
+            f"{TEST_URI}v1/config",
+            json={"defaults": {}, "overrides": {}},
+            status_code=200,
+        )
+        catalog = RestCatalog(
+            "rest",
+            uri=TEST_URI,
+            token=TEST_TOKEN,
+            **{"rest.scan-planning.enabled": "false"},
+        )
+
+        assert catalog.is_scan_planning_enabled() is False
+
+    def test_scan_planning_enabled_from_server_config(self, rest_mock: Mocker) -> None:
+        rest_mock.get(
+            f"{TEST_URI}v1/config",
+            json={"defaults": {"rest-scan-planning-enabled": "true"}, "overrides": {}},
+            status_code=200,
+        )
+        catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+        assert catalog.is_scan_planning_enabled() is True