zeroshade commented on code in PR #1959:
URL: https://github.com/apache/iceberg-go/pull/1959#discussion_r3937963726
##########
catalog/rest/scan_planning.go:
##########
@@ -247,7 +249,8 @@ func (r *Catalog) PlanFiles(ctx context.Context, req
table.ScanPlanningRequest)
"%w: unexpected plan status %q from planTableScan",
ErrRESTError, resp.Status)
}
- envelopes, err := r.collectScanTasks(ctx, req.Identifier,
completed.ScanTasks)
+ envelopes, err := r.collectScanTasksWithConcurrency(
+ ctx, req.Identifier, completed.ScanTasks, req.MaxConcurrency)
if err != nil {
Review Comment:
**minor** — PR description still advertises a fixed 8 workers; the code
derives concurrency from the scan
The summary and benchmark table describe 'up to 8 concurrent REST requests'
and an '8 workers' row, but concurrency now comes from req.MaxConcurrency
(populated from the scan's WithMaxConcurrency) and falls back to
runtime.GOMAXPROCS(0), not 8. Since the tunability request was addressed in
code, please update the description and re-label the benchmark row, otherwise
the recorded numbers cannot be reproduced from the stated configuration.
##########
catalog/rest/scan_planning.go:
##########
@@ -294,34 +297,156 @@ func (r *Catalog) planIOBaseProps(req
table.ScanPlanningRequest) iceberg.Propert
}
// 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.
+// the fanout: a fetchScanTasks 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.
func (r *Catalog) collectScanTasks(ctx context.Context, ident
table.Identifier, tasks ScanTasks) ([]ScanTasks, error) {
+ return r.collectScanTasksWithConcurrency(ctx, ident, tasks,
runtime.GOMAXPROCS(0))
+}
Review Comment:
**nit** — collectScanTasks is now reachable only from tests
PlanFiles calls collectScanTasksWithConcurrency directly, so the
collectScanTasks wrapper survives solely as a test entry point with a
GOMAXPROCS default. That is harmless, but a reader will assume it is the
production path. Either drop it and have the four tests pass an explicit limit,
or note in its doc comment that it is the default-concurrency convenience
wrapper.
##########
schema.go:
##########
@@ -352,6 +352,8 @@ func (s *Schema) MarshalJSON() ([]byte, error) {
type Alias Schema
+ // Keep this literal in sync with every JSON-marshaled Schema field.
Copying
+ // the whole Schema would also copy its lazy atomic caches after they
are used.
aliasCopy := Alias{ID: s.ID, IdentifierFieldIDs: ids}
Review Comment:
**minor** — Hand-maintained alias literal can silently drop a future Schema
field
Replacing *(*Alias)(s) with Alias{ID: ..., IdentifierFieldIDs: ...} is the
right fix for copying the lazy atomic caches, but correctness is now maintained
by comment alone: adding a json-tagged field to Schema would drop it from the
serialized output with no compile error and no failing test. A small
reflection-based test asserting the marshaled key set matches Schema's
json-tagged exported fields would make the invariant self-enforcing.
--
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]
