zeroshade commented on code in PR #1959:
URL: https://github.com/apache/iceberg-go/pull/1959#discussion_r3960397500
##########
catalog/rest/scan_planning.go:
##########
@@ -293,35 +296,153 @@ func (r *Catalog) planIOBaseProps(req
table.ScanPlanningRequest) iceberg.Propert
return props
}
-// collectScanTasks expands plan-task handles into their task envelopes,
walking
-// the fanout: a fetchScanTasks response can itself return more plan-tasks. A
-// handle is fetched at most once; a server that re-issues one would otherwise
-// loop forever. The envelope boundaries are retained because delete-file
-// references are local to each response.
-func (r *Catalog) collectScanTasks(ctx context.Context, ident
table.Identifier, tasks ScanTasks) ([]ScanTasks, error) {
+// collectScanTasksWithConcurrency expands plan-task handles into their task
+// envelopes, walking the fanout: a response can itself return more
plan-tasks. Each
+// frontier is fetched concurrently, but its responses are appended in handle
+// order so completion timing cannot change the result order. A handle is
+// fetched at most once; a server that re-issues one would otherwise loop
+// forever. The envelope boundaries are retained because delete-file references
+// are local to each response.
+//
+// maxConcurrency bounds the number of concurrent fetches. A non-positive limit
+// uses runtime.GOMAXPROCS. On a fetch error, at most the configured number of
+// requests can already be in flight when cancellation reaches their contexts.
+func (r *Catalog) collectScanTasksWithConcurrency(
+ ctx context.Context,
+ ident table.Identifier,
+ tasks ScanTasks,
+ maxConcurrency int,
+) ([]ScanTasks, error) {
+ if maxConcurrency <= 0 {
+ maxConcurrency = runtime.GOMAXPROCS(0)
+ }
+
envelopes := []ScanTasks{tasks}
- queue := append([]string(nil), tasks.PlanTasks...)
- seen := make(map[string]bool, len(queue))
- for len(queue) > 0 {
- handle := queue[0]
- queue = queue[1:]
- if seen[handle] {
- continue
+ frontier := append([]string(nil), tasks.PlanTasks...)
+ seen := make(map[string]bool, len(frontier))
+ for len(frontier) > 0 {
+ handles := make([]string, 0, len(frontier))
+ for _, handle := range frontier {
+ if seen[handle] {
+ continue
+ }
+ seen[handle] = true
+ handles = append(handles, handle)
+ }
+ if len(handles) == 0 {
+ break
}
- seen[handle] = true
- resp, err := r.FetchScanTasks(ctx, ident,
FetchScanTasksRequest{PlanTask: handle})
+ responses, err := r.fetchScanTaskFrontier(ctx, ident, handles,
maxConcurrency)
if err != nil {
return nil, err
}
- envelopes = append(envelopes, resp.ScanTasks)
- queue = append(queue, resp.PlanTasks...)
+
+ var nextFrontier []string
+ for _, response := range responses {
+ envelopes = append(envelopes, response.ScanTasks)
+ nextFrontier = append(nextFrontier,
response.PlanTasks...)
+ }
+ frontier = nextFrontier
}
return envelopes, nil
}
+// fetchScanTaskFrontier fetches one breadth-first frontier concurrently while
+// placing responses back into handle order. When a handle fails, unfinished
+// later handles are canceled, but earlier handles are allowed to finish so the
+// first error in the old serial handle order remains deterministic.
+func (r *Catalog) fetchScanTaskFrontier(
+ ctx context.Context,
+ ident table.Identifier,
+ handles []string,
+ maxConcurrency int,
+) ([]FetchScanTasksResponse, error) {
+ responses := make([]FetchScanTasksResponse, len(handles))
+ errs := make([]error, len(handles))
+ if len(handles) == 0 {
+ return responses, nil
+ }
+ if maxConcurrency <= 0 {
+ maxConcurrency = runtime.GOMAXPROCS(0)
Review Comment:
**nit** — Unreachable maxConcurrency normalization in fetchScanTaskFrontier
fetchScanTaskFrontier re-applies the 'if maxConcurrency <= 0 {
maxConcurrency = runtime.GOMAXPROCS(0) }' fallback, but its only caller
(collectScanTasksWithConcurrency, scan_planning.go:337) already normalized the
value at :316. The branch is dead for every current call path. Harmless
defensiveness in an unexported helper; drop it or keep normalization in exactly
one place.
--
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]
