lukeFalsina commented on code in PR #3724:
URL: https://github.com/apache/iceberg-python/pull/3724#discussion_r3703945687
##########
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:
Sure, let me put those back.
--
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]
