singhpk234 commented on code in PR #3724:
URL: https://github.com/apache/iceberg-python/pull/3724#discussion_r3692512523
##########
mkdocs/docs/configuration.md:
##########
@@ -386,6 +386,10 @@ 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. Async plans (`status=submitted`) are polled via `GET
.../plan/{plan-id}` until completion. |
Review Comment:
this is something loadTable return too, in case the server wants to ask
client for server side scan planning ...
##########
pyiceberg/catalog/rest/__init__.py:
##########
@@ -600,6 +690,81 @@ def plan_scan(self, identifier: str | Identifier, request:
PlanTableScanRequest)
return tasks
+ def _plan_scan_result(self, identifier: str | Identifier, request:
PlanTableScanRequest) -> PlannedScanResult:
+ """Plan a table scan and return tasks with optional plan storage
credentials.
+
+ Handles the full scan planning lifecycle including async polling and
pagination.
+ """
+ response = self._plan_table_scan(identifier, request)
+
+ if isinstance(response, PlanFailed):
+ error_msg = response.error.message if response.error else "unknown
error"
+ raise RuntimeError(f"Received status: failed: {error_msg}")
+
+ if isinstance(response, PlanCancelled):
+ raise RuntimeError("Received status: cancelled")
+
+ if isinstance(response, PlanSubmitted):
+ if not response.plan_id:
+ raise ValueError("Async scan planning submitted without
plan-id")
+ response = self._poll_until_completed(identifier, response.plan_id)
+
+ if not isinstance(response, PlanCompleted):
+ raise RuntimeError(f"Invalid planStatus for response:
{type(response).__name__}")
+
+ tasks = self._expand_plan_tasks(identifier, response)
+ return PlannedScanResult(
+ tasks=tasks,
+ storage_credentials=list(response.storage_credentials or []),
+ plan_id=response.plan_id,
+ )
+
+ def plan_scan(self, identifier: str | Identifier, request:
PlanTableScanRequest) -> list[FileScanTask]:
+ """Plan a table scan and return FileScanTasks.
+
+ Handles the full scan planning lifecycle including async polling and
pagination.
+
+ Args:
+ identifier: Table identifier.
+ request: The scan plan request parameters.
+
+ Returns:
+ List of FileScanTask objects ready for execution.
+
+ Raises:
+ RuntimeError: If planning fails, is cancelled, or returns
unexpected response.
+ RemotePlanTimeoutError: If async planning does not complete in
time.
+ ValueError: If a submitted plan is missing plan-id.
+ """
+ return self._plan_scan_result(identifier, request).tasks
+
+ def _file_io_from_plan(
+ self,
+ existing_properties: Properties,
+ storage_credentials: list[StorageCredential],
+ location: str | None = None,
+ ) -> FileIO | None:
+ """Build a scan-scoped FileIO from plan storage credentials.
Review Comment:
in this scenario can there be more than one plan active ?
##########
pyiceberg/catalog/rest/__init__.py:
##########
@@ -551,45 +571,115 @@ def _fetch_scan_tasks(self, identifier: str |
Identifier, plan_task: str) -> Sca
return ScanTasks.model_validate_json(response.text)
- def plan_scan(self, identifier: str | Identifier, request:
PlanTableScanRequest) -> list[FileScanTask]:
- """Plan a table scan and return FileScanTasks.
-
- Handles the full scan planning lifecycle including pagination.
+ @retry(**_RETRY_ARGS)
+ def _fetch_planning_result(self, identifier: str | Identifier, plan_id:
str) -> PlanningResponse:
+ """Fetch the result of an async scan plan by plan-id.
Args:
identifier: Table identifier.
- request: The scan plan request parameters.
+ plan_id: Plan id returned from a submitted planTableScan response.
Returns:
- List of FileScanTask objects ready for execution.
+ PlanningResponse with the current plan status.
Raises:
- RuntimeError: If planning fails, is cancelled, or returns
unexpected response.
- NotImplementedError: If async planning is required but not yet
supported.
+ NoSuchPlanIdError: If the plan-id does not exist.
+ NoSuchTableError: If the table does not exist.
"""
- response = self._plan_table_scan(identifier, request)
+ self._check_endpoint(Capability.V1_FETCH_TABLE_SCAN_PLAN)
+ response = self._session.get(
+ self.url(
+ Endpoints.fetch_planning_result,
+ prefixed=True,
+ plan_id=quote(plan_id, safe=""),
+ **self._split_identifier_for_path(identifier),
+ ),
+ )
+ try:
+ response.raise_for_status()
+ except HTTPError as exc:
+ _handle_non_200_response(exc, {404: NoSuchPlanIdError})
- if isinstance(response, PlanFailed):
- error_msg = response.error.message if response.error else "unknown
error"
- raise RuntimeError(f"Received status: failed: {error_msg}")
+ return _PLANNING_RESPONSE_ADAPTER.validate_json(response.text)
- if isinstance(response, PlanCancelled):
- raise RuntimeError("Received status: cancelled")
+ def _cancel_planning(self, identifier: str | Identifier, plan_id: str) ->
bool:
+ """Best-effort cancel of an async scan plan.
- if isinstance(response, PlanSubmitted):
- # TODO: implement polling for async planning
- raise NotImplementedError(f"Async scan planning not yet supported
for planId: {response.plan_id}")
+ Returns:
+ True if the cancel request was accepted, False otherwise.
+ """
+ if Capability.V1_CANCEL_TABLE_SCAN_PLAN not in
self._supported_endpoints:
+ return False
- if not isinstance(response, PlanCompleted):
- raise RuntimeError(f"Invalid planStatus for response:
{type(response).__name__}")
+ try:
+ response = self._session.delete(
+ self.url(
+ Endpoints.cancel_planning,
+ prefixed=True,
+ plan_id=quote(plan_id, safe=""),
+ **self._split_identifier_for_path(identifier),
+ ),
+ )
+ response.raise_for_status()
+ return True
+ except Exception:
+ # Plan may have already completed, failed, or been cancelled.
+ return False
+
+ def _poll_until_completed(self, identifier: str | Identifier, plan_id:
str) -> PlanCompleted:
+ """Poll fetchPlanningResult until the plan completes or times out.
+
+ Uses exponential backoff matching Java RESTTableScan defaults.
+ """
+ max_wait_ms = property_as_int(
+ self.properties,
+ REST_SCAN_PLANNING_POLL_TIMEOUT_MS,
+ REST_SCAN_PLANNING_POLL_TIMEOUT_MS_DEFAULT,
+ )
+ if max_wait_ms is None or max_wait_ms <= 0:
+ raise ValueError(f"Invalid value for
{REST_SCAN_PLANNING_POLL_TIMEOUT_MS}: {max_wait_ms} (must be positive)")
+
+ sleep_ms = float(REST_SCAN_PLANNING_POLL_MIN_SLEEP_MS)
+ start = time.monotonic()
+ retries = 0
+
+ while True:
+ response = self._fetch_planning_result(identifier, plan_id)
+
+ if isinstance(response, PlanCompleted):
+ return response
+
+ if isinstance(response, PlanFailed):
+ error_msg = response.error.message if response.error else
"unknown error"
+ self._cancel_planning(identifier, plan_id)
+ raise RuntimeError(f"Remote scan planning failed for planId:
{plan_id}: {error_msg}")
+
+ if isinstance(response, PlanCancelled):
+ raise RuntimeError(f"Remote scan planning cancelled for
planId: {plan_id}")
+
+ if not isinstance(response, PlanSubmitted):
+ self._cancel_planning(identifier, plan_id)
+ raise RuntimeError(f"Invalid planStatus for planId: {plan_id}:
{type(response).__name__}")
+
+ elapsed_ms = (time.monotonic() - start) * 1000
+ if retries >= REST_SCAN_PLANNING_POLL_MAX_RETRIES or elapsed_ms >=
max_wait_ms:
+ self._cancel_planning(identifier, plan_id)
+ raise RemotePlanTimeoutError(
+ f"Remote scan planning for planId: {plan_id} did not
complete within configured limits "
+ f"(timeout={max_wait_ms} ms,
maxRetries={REST_SCAN_PLANNING_POLL_MAX_RETRIES})"
+ )
+ time.sleep(sleep_ms / 1000.0)
+ sleep_ms = min(sleep_ms * REST_SCAN_PLANNING_POLL_SCALE_FACTOR,
REST_SCAN_PLANNING_POLL_MAX_SLEEP_MS)
+ retries += 1
+
+ def _expand_plan_tasks(self, identifier: str | Identifier, response:
PlanCompleted) -> list[FileScanTask]:
+ """Expand a completed plan response into FileScanTask objects,
including pagination."""
tasks: list[FileScanTask] = []
- # Collect tasks from initial response
for task in response.file_scan_tasks:
tasks.append(FileScanTask.from_rest_response(task,
response.delete_files))
- # Fetch and collect from additional batches
Review Comment:
should we keep the comments still ?
##########
mkdocs/docs/api.md:
##########
@@ -1695,6 +1695,8 @@ scan = table.scan(
[task.file.file_path for task in scan.plan_files()]
```
+When the REST catalog is configured with `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; see [REST Catalog configuration](configuration.md#rest-catalog).
Review Comment:
```suggestion
When the REST catalog returns with `scan-planning-mode=server` and
advertises the plan endpoint, `plan_files()` / `to_arrow()` for clients to use
server-side scan planning. Catalogs that return async plans
(`status=submitted`) are polled automatically till they reach terminal state;
see [REST Catalog configuration](configuration.md#rest-catalog).
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
