lukeFalsina commented on code in PR #3724:
URL: https://github.com/apache/iceberg-python/pull/3724#discussion_r3704204655
##########
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:
I got some help from the LLM to verify this: Yes on the server side you can
have more than one active plan; but still only one plan per client call.
- Server: plans are keyed by `plan-id`; many can be in flight (Java’s
`InMemoryPlanningState` is a map of plan IDs).
- This client: each `_plan_scan_result` owns one `plan_id`, blocks in
`_poll_until_completed`, then `_file_io_from_plan` builds IO from that plan’s
storage-credentials. The `list[StorageCredential]` is prefix-scoped creds
within one plan, not multiple plans.
- Concurrent table.scan() calls (threads/processes) can each have their own
plan and id; RestCatalog does not keep a shared active-plan registry.
So I think `_file_io_from_plan` does not need multi-plan awareness (so the
interface should stay as is).
What do you think?
--
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]
