laskoviymishka commented on code in PR #1959:
URL: https://github.com/apache/iceberg-go/pull/1959#discussion_r3903429389
##########
schema.go:
##########
@@ -346,8 +346,7 @@ func (s *Schema) MarshalJSON() ([]byte, error) {
type Alias Schema
- aliasCopy := *(*Alias)(s)
- aliasCopy.IdentifierFieldIDs = ids
+ aliasCopy := Alias{ID: s.ID, IdentifierFieldIDs: ids}
Review Comment:
agreed with this fix, and I like that it landed as its own `fix(schema):`
commit so it's cherry-pickable.
One small thing on the literal itself: the old `*(*Alias)(s)` picked up
every exported field automatically, and this now hard-codes the two we have.
Correct today, but a new json-tagged field on `Schema` would be silently
dropped here with no compile error and no test failure (the JSONEq golden only
catches unexpected additions, not omissions). A short comment above the literal
saying it must list every marshalled field, and why it can't just copy the
struct, would keep the next person from missing it. Non-blocking, wdyt?
##########
catalog/rest/scan_planning.go:
##########
@@ -51,6 +52,11 @@ var _ table.ScanPlanner = (*Catalog)(nil)
// extension used by table.Scan's auto planning mode.
var _ table.FullRemoteScanPlanner = (*Catalog)(nil)
+// remoteScanTaskFetchConcurrency bounds in-flight fetchScanTasks requests for
+// each frontier. Keeping frontiers separate preserves breadth-first response
+// ordering while allowing independent plan-task handles to fetch concurrently.
+const remoteScanTaskFetchConcurrency = 8
Review Comment:
not a blocker, but I wonder if this wants to be tunable rather than a fixed
8. Java derives its fetch parallelism from `max(2, availableProcessors())` via
the worker pool size, so it scales with the box and can be tuned for server
rate limits. A catalog option or table property (something like
`rest.fetch-scan-tasks.concurrency`) would match that. Fine as a follow-up,
just flagging.
##########
catalog/rest/scan_planning.go:
##########
@@ -294,34 +300,84 @@ 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) {
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)
}
- seen[handle] = true
- resp, err := r.FetchScanTasks(ctx, ident,
FetchScanTasksRequest{PlanTask: handle})
+ responses, err := r.fetchScanTaskFrontier(ctx, ident, handles)
Review Comment:
one perf thought, definitely not blocking: because we fetch a whole frontier
and only advance once `fetchScanTaskFrontier` returns, each BFS level waits on
its slowest handle before the next level's handles start. Java's
`ScanTaskIterable` pushes child handles onto a shared queue as each response
lands, so idle workers pick up the next level without waiting for their peers.
For a deep, uneven fanout (1 to 8 to 64) that tier barrier can leave workers
idle. The current shape is simpler and keeps ordering clean, so fine to keep,
but maybe worth a follow-up if fanout depth ever gets expensive.
##########
catalog/rest/scan_planning.go:
##########
@@ -294,34 +300,84 @@ 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) {
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)
}
- seen[handle] = true
- resp, err := r.FetchScanTasks(ctx, ident,
FetchScanTasksRequest{PlanTask: handle})
+ responses, err := r.fetchScanTaskFrontier(ctx, ident, handles)
if err != nil {
return nil, err
}
- envelopes = append(envelopes, resp.ScanTasks)
- queue = append(queue, resp.PlanTasks...)
+
+ nextFrontier := make([]string, 0)
Review Comment:
tiny thing: this is `make([]string, 0)` while the `handles` allocation a few
lines up uses a capacity hint. `var nextFrontier []string` reads a touch
cleaner. Purely cosmetic, I checked and staticcheck doesn't flag either form so
it won't affect CI.
##########
catalog/rest/scan_planning.go:
##########
@@ -294,34 +300,84 @@ 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) {
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)
}
- seen[handle] = true
- resp, err := r.FetchScanTasks(ctx, ident,
FetchScanTasksRequest{PlanTask: handle})
+ responses, err := r.fetchScanTaskFrontier(ctx, ident, handles)
if err != nil {
return nil, err
}
- envelopes = append(envelopes, resp.ScanTasks)
- queue = append(queue, resp.PlanTasks...)
+
+ nextFrontier := make([]string, 0)
+ for _, response := range responses {
+ envelopes = append(envelopes, response.ScanTasks)
+ nextFrontier = append(nextFrontier,
response.PlanTasks...)
+ }
+ frontier = nextFrontier
}
return envelopes, nil
}
+func (r *Catalog) fetchScanTaskFrontier(
+ ctx context.Context,
+ ident table.Identifier,
+ handles []string,
+) ([]FetchScanTasksResponse, error) {
+ responses := make([]FetchScanTasksResponse, len(handles))
+ errs := make([]error, len(handles))
+ group, groupCtx := errgroup.WithContext(ctx)
+ group.SetLimit(remoteScanTaskFetchConcurrency)
+
+ for i, handle := range handles {
+ group.Go(func() error {
+ response, err := r.FetchScanTasks(groupCtx, ident,
FetchScanTasksRequest{PlanTask: handle})
+ if err != nil {
+ errs[i] = err
Review Comment:
nice fix on the ordering here, forcing the first handle-order error to win
instead of whichever request loses the race is exactly right, and the new test
pins it well.
Small follow-on while you're in here: the error still goes back bare, so a
failure inside an 8-way frontier doesn't say which handle expired, and the rest
of the file wraps (`WaitForPlan`, `remoteScanTasks`). Recording it wrapped,
e.g. `errs[i] = fmt.Errorf("fetching scan tasks for handle %q: %w", handle,
err)`, keeps the `errors.Is` chain (and the handle-order selection below)
intact.
##########
schema_test.go:
##########
@@ -2293,3 +2294,48 @@ func
TestVisitGeoSchemaWithSchemaVisitorPerPrimitiveType(t *testing.T) {
assert.Equal(t, 1, v.geometryCalls)
assert.Equal(t, 1, v.geographyCalls)
}
+
+func TestSchemaMarshalJSONConcurrentLazyLookups(t *testing.T) {
Review Comment:
this is a nice race reproducer, but it only proves anything under `go test
-race`. There's no assertion that fails if the race comes back, so without the
detector it passes against the old code too. A one-line comment noting it's
meant to run under `-race` would stop the next person assuming it guards the
fix on its own.
--
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]
