From d279eaaf870ddce23941386989e9c0e79dea553b Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Fri, 28 Aug 2026 23:49:34 +0200
Subject: [PATCH 01/12] perf(rest): fetch remote scan plan tasks concurrently

Signed-off-by: Minh Vu <vuhoangminh97@gmail.com>
---
 catalog/rest/scan_planning.go            |  74 +++++++++---
 catalog/rest/scan_planning_bench_test.go |  68 +++++++++++
 catalog/rest/scan_planning_test.go       | 147 +++++++++++++++++++++++
 3 files changed, 274 insertions(+), 15 deletions(-)
 create mode 100644 catalog/rest/scan_planning_bench_test.go

diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go
index b44399221..8500d9a21 100644
--- a/catalog/rest/scan_planning.go
+++ b/catalog/rest/scan_planning.go
@@ -41,6 +41,7 @@ import (
 	iceio "github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
 	"github.com/google/uuid"
+	"golang.org/x/sync/errgroup"
 	"golang.org/x/sync/semaphore"
 )
 
@@ -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
+
 // ErrPlanExpired is returned when polling a plan that the server no longer
 // knows about: a fetchPlanningResult 404 whose error.type is exactly
 // NoSuchPlanIdException. It is distinct from a table/namespace-gone 404
@@ -294,34 +300,72 @@ 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))
+	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 {
+				return err
+			}
+
+			responses[i] = response
+
+			return nil
+		})
+	}
+
+	if err := group.Wait(); err != nil {
+		return nil, err
+	}
+
+	return responses, nil
+}
+
 // remoteScanTasks decodes each server task envelope into domain FileScanTasks.
 // The decoder must run before envelopes are combined because delete-file
 // references are indexes into the envelope-local delete-files array.
diff --git a/catalog/rest/scan_planning_bench_test.go b/catalog/rest/scan_planning_bench_test.go
new file mode 100644
index 000000000..8dc639b04
--- /dev/null
+++ b/catalog/rest/scan_planning_bench_test.go
@@ -0,0 +1,68 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package rest
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/apache/iceberg-go/table"
+)
+
+func BenchmarkCollectScanTasks64Handles(b *testing.B) {
+	const (
+		handleCount = 64
+		latency     = 10 * time.Millisecond
+	)
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		time.Sleep(latency)
+		_, _ = w.Write([]byte(`{"file-scan-tasks":[]}`))
+	}))
+	b.Cleanup(server.Close)
+
+	serverURL, err := url.Parse(server.URL)
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	catalog := &Catalog{
+		baseURI:   serverURL.JoinPath("v1"),
+		cl:        server.Client(),
+		endpoints: newEndpointSet([]endpoint{endpointFetchScanTasks}),
+	}
+	handles := make([]string, handleCount)
+	for i := range handles {
+		handles[i] = fmt.Sprintf("task-%d", i)
+	}
+	tasks := ScanTasks{PlanTasks: handles}
+	ident := table.Identifier{"db", "tbl"}
+
+	b.ResetTimer()
+	for range b.N {
+		_, err := catalog.collectScanTasks(context.Background(), ident, tasks)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index a45972f6a..edadad71a 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -27,6 +27,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -1245,6 +1246,152 @@ func TestPlanFilesExpandsPlanTasks(t *testing.T) {
 	assert.Equal(t, []string{"h1", "h2"}, fetched)
 }
 
+func TestCollectScanTasksFetchesFrontierConcurrentlyInOrder(t *testing.T) {
+	t.Parallel()
+
+	h1Started := make(chan struct{})
+	h2Started := make(chan struct{})
+	releaseH1 := make(chan struct{})
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(releaseH1) }) }
+
+	cat := newScanPlanningTestCatalog(t, []endpoint{endpointFetchScanTasks}, func(mux *http.ServeMux) {
+		mux.HandleFunc("/v1/namespaces/db/tables/tbl/tasks", func(w http.ResponseWriter, req *http.Request) {
+			var body FetchScanTasksRequest
+			if err := json.NewDecoder(req.Body).Decode(&body); err != nil {
+				http.Error(w, err.Error(), http.StatusBadRequest)
+
+				return
+			}
+
+			switch body.PlanTask {
+			case "h1":
+				close(h1Started)
+				<-releaseH1
+				_, _ = w.Write([]byte(`{"file-scan-tasks":[{"data-file":{"file-path":"h1"}}]}`))
+			case "h2":
+				close(h2Started)
+				_, _ = w.Write([]byte(`{"file-scan-tasks":[{"data-file":{"file-path":"h2"}}]}`))
+			default:
+				http.Error(w, "unexpected plan task", http.StatusBadRequest)
+			}
+		})
+	})
+	t.Cleanup(release)
+
+	type outcome struct {
+		envelopes []ScanTasks
+		err       error
+	}
+	done := make(chan outcome, 1)
+	go func() {
+		envelopes, err := cat.collectScanTasks(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
+			PlanTasks: []string{"h1", "h2"},
+		})
+		done <- outcome{envelopes: envelopes, err: err}
+	}()
+
+	for name, started := range map[string]<-chan struct{}{
+		"h1": h1Started,
+		"h2": h2Started,
+	} {
+		select {
+		case <-started:
+		case <-time.After(time.Second):
+			t.Fatalf("timed out waiting for %s to start", name)
+		}
+	}
+
+	release()
+	select {
+	case result := <-done:
+		require.NoError(t, result.err)
+		require.Len(t, result.envelopes, 3)
+		require.Len(t, result.envelopes[1].FileScanTasks, 1)
+		require.Len(t, result.envelopes[2].FileScanTasks, 1)
+		require.NotNil(t, result.envelopes[1].FileScanTasks[0].DataFile)
+		require.NotNil(t, result.envelopes[2].FileScanTasks[0].DataFile)
+		assert.Equal(t, "h1", result.envelopes[1].FileScanTasks[0].DataFile.FilePath)
+		assert.Equal(t, "h2", result.envelopes[2].FileScanTasks[0].DataFile.FilePath)
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for frontier fetches")
+	}
+}
+
+func TestCollectScanTasksBoundsFrontierConcurrency(t *testing.T) {
+	t.Parallel()
+
+	const handleCount = remoteScanTaskFetchConcurrency + 1
+	started := make(chan string, handleCount)
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	finish := func() { releaseOnce.Do(func() { close(release) }) }
+
+	cat := newScanPlanningTestCatalog(t, []endpoint{endpointFetchScanTasks}, func(mux *http.ServeMux) {
+		mux.HandleFunc("/v1/namespaces/db/tables/tbl/tasks", func(w http.ResponseWriter, req *http.Request) {
+			var body FetchScanTasksRequest
+			if err := json.NewDecoder(req.Body).Decode(&body); err != nil {
+				http.Error(w, err.Error(), http.StatusBadRequest)
+
+				return
+			}
+
+			started <- body.PlanTask
+			<-release
+			_, _ = w.Write([]byte(`{"file-scan-tasks":[]}`))
+		})
+	})
+	t.Cleanup(finish)
+
+	handles := make([]string, handleCount)
+	for i := range handles {
+		handles[i] = fmt.Sprintf("h%d", i)
+	}
+	done := make(chan error, 1)
+	go func() {
+		_, err := cat.collectScanTasks(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
+			PlanTasks: handles,
+		})
+		done <- err
+	}()
+
+	for range remoteScanTaskFetchConcurrency {
+		select {
+		case <-started:
+		case <-time.After(time.Second):
+			t.Fatal("timed out waiting for bounded frontier workers")
+		}
+	}
+
+	select {
+	case handle := <-started:
+		t.Fatalf("frontier exceeded concurrency limit with %s", handle)
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	finishOne := func() {
+		select {
+		case release <- struct{}{}:
+		case <-time.After(time.Second):
+			t.Fatal("timed out releasing a frontier worker")
+		}
+	}
+	finishOne()
+	select {
+	case <-started:
+	case <-time.After(time.Second):
+		t.Fatal("timed out starting the queued frontier handle")
+	}
+	finish()
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for bounded frontier fetches")
+	}
+}
+
 // TestPlanFilesFanoutCycleTerminates guards the seen-set: a server that re-issues
 // a handle it already returned must not loop forever.
 func TestPlanFilesFanoutCycleTerminates(t *testing.T) {

From 94314ec82628f34306845750fee5cd259e71aea7 Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Sun, 30 Aug 2026 23:01:20 +0200
Subject: [PATCH 02/12] test(rest): cover fanout deduplication and cancellation

Signed-off-by: Minh Vu <vuhoangminh97@gmail.com>
---
 catalog/rest/scan_planning_test.go | 91 ++++++++++++++++++++++++++++++
 1 file changed, 91 insertions(+)

diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index edadad71a..82565b733 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -1832,3 +1832,94 @@ func TestPlanFilesPropagatesFailure(t *testing.T) {
 	_, err := cat.PlanFiles(context.Background(), planFilesReq())
 	require.ErrorIs(t, err, ErrPlanFailed)
 }
+
+func TestCollectScanTasksDeduplicatesAcrossFrontiers(t *testing.T) {
+	t.Parallel()
+
+	children := map[string][]string{
+		"a": {"c", "d", "a"},
+		"b": {"d", "e"},
+		"c": {"b"},
+	}
+	var mu sync.Mutex
+	calls := make(map[string]int)
+	cat := newScanPlanningTestCatalog(t, []endpoint{endpointFetchScanTasks}, func(mux *http.ServeMux) {
+		mux.HandleFunc("/v1/namespaces/db/tables/tbl/tasks", func(w http.ResponseWriter, req *http.Request) {
+			var body FetchScanTasksRequest
+			if err := json.NewDecoder(req.Body).Decode(&body); err != nil {
+				http.Error(w, err.Error(), http.StatusBadRequest)
+
+				return
+			}
+			mu.Lock()
+			calls[body.PlanTask]++
+			mu.Unlock()
+			response := ScanTasks{
+				PlanTasks: children[body.PlanTask],
+				FileScanTasks: []RESTFileScanTask{{DataFile: &RESTDataFile{
+					RESTContentFile: RESTContentFile{FilePath: body.PlanTask},
+				}}},
+			}
+			_ = json.NewEncoder(w).Encode(response)
+		})
+	})
+
+	envelopes, err := cat.collectScanTasks(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
+		PlanTasks: []string{"a", "a", "b"},
+	})
+	require.NoError(t, err)
+	require.Len(t, envelopes, 6)
+	for i, name := range []string{"a", "b", "c", "d", "e"} {
+		require.Len(t, envelopes[i+1].FileScanTasks, 1)
+		assert.Equal(t, name, envelopes[i+1].FileScanTasks[0].DataFile.FilePath)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, map[string]int{"a": 1, "b": 1, "c": 1, "d": 1, "e": 1}, calls)
+}
+
+func TestCollectScanTasksCancelsSiblingRequestsOnFailure(t *testing.T) {
+	t.Parallel()
+
+	started := make(chan struct{})
+	cancelled := make(chan struct{})
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	cat := newScanPlanningTestCatalog(t, []endpoint{endpointFetchScanTasks}, func(mux *http.ServeMux) {
+		mux.HandleFunc("/v1/namespaces/db/tables/tbl/tasks", func(w http.ResponseWriter, req *http.Request) {
+			var body FetchScanTasksRequest
+			if err := json.NewDecoder(req.Body).Decode(&body); err != nil {
+				http.Error(w, err.Error(), http.StatusBadRequest)
+
+				return
+			}
+			switch body.PlanTask {
+			case "slow":
+				close(started)
+				<-req.Context().Done()
+				close(cancelled)
+			case "failed":
+				select {
+				case <-started:
+					w.Header().Set("Content-Type", "application/json")
+					w.WriteHeader(http.StatusBadRequest)
+					_, _ = w.Write([]byte(`{"error":{"message":"invalid handle","type":"BadRequestException","code":400}}`))
+				case <-req.Context().Done():
+				}
+			default:
+				http.Error(w, "unexpected handle", http.StatusBadRequest)
+			}
+		})
+	})
+
+	envelopes, err := cat.collectScanTasks(ctx, table.Identifier{"db", "tbl"}, ScanTasks{
+		PlanTasks: []string{"slow", "failed"},
+	})
+	require.ErrorIs(t, err, ErrBadRequest)
+	assert.Nil(t, envelopes)
+	select {
+	case <-cancelled:
+	case <-ctx.Done():
+		t.Fatal("sibling request was not cancelled after a fetch failure")
+	}
+}

From 568c69a30b561998e5f0eb74e43fd18f7c9101a6 Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Tue, 1 Sep 2026 07:24:58 +0200
Subject: [PATCH 03/12] fix(rest): preserve scan task fetch error order

---
 catalog/rest/scan_planning.go      | 16 ++++++++--
 catalog/rest/scan_planning_test.go | 48 ++++++++++++++++++++++++++++++
 2 files changed, 62 insertions(+), 2 deletions(-)

diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go
index 8500d9a21..fb5028264 100644
--- a/catalog/rest/scan_planning.go
+++ b/catalog/rest/scan_planning.go
@@ -343,6 +343,7 @@ func (r *Catalog) fetchScanTaskFrontier(
 	handles []string,
 ) ([]FetchScanTasksResponse, error) {
 	responses := make([]FetchScanTasksResponse, len(handles))
+	errs := make([]error, len(handles))
 	group, groupCtx := errgroup.WithContext(ctx)
 	group.SetLimit(remoteScanTaskFetchConcurrency)
 
@@ -350,6 +351,8 @@ func (r *Catalog) fetchScanTaskFrontier(
 		group.Go(func() error {
 			response, err := r.FetchScanTasks(groupCtx, ident, FetchScanTasksRequest{PlanTask: handle})
 			if err != nil {
+				errs[i] = err
+
 				return err
 			}
 
@@ -359,8 +362,17 @@ func (r *Catalog) fetchScanTaskFrontier(
 		})
 	}
 
-	if err := group.Wait(); err != nil {
-		return nil, err
+	if waitErr := group.Wait(); waitErr != nil {
+		// Preserve the serial fetch contract: when multiple handles fail, the
+		// first error in handle order wins. Sibling requests cancelled by the
+		// first failure are not meaningful candidates for the returned error.
+		for _, err := range errs {
+			if err != nil && !errors.Is(err, context.Canceled) {
+				return nil, err
+			}
+		}
+
+		return nil, waitErr
 	}
 
 	return responses, nil
diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index 82565b733..a54e11de4 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -1318,6 +1318,54 @@ func TestCollectScanTasksFetchesFrontierConcurrentlyInOrder(t *testing.T) {
 	}
 }
 
+type orderedScanTaskErrorTransport struct {
+	planTaskReturned chan struct{}
+}
+
+func (t *orderedScanTaskErrorTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	var body FetchScanTasksRequest
+	if err := json.NewDecoder(req.Body).Decode(&body); err != nil {
+		return nil, err
+	}
+
+	var errType string
+	switch body.PlanTask {
+	case "table":
+		<-t.planTaskReturned
+		errType = errTypeNoSuchTable
+	case "plan-task":
+		close(t.planTaskReturned)
+		errType = errTypeNoSuchPlanTask
+	default:
+		return nil, fmt.Errorf("unexpected plan task %q", body.PlanTask)
+	}
+
+	data := fmt.Sprintf(`{"error":{"message":%q,"type":%q,"code":404}}`, errType, errType)
+	return &http.Response{
+		StatusCode:    http.StatusNotFound,
+		Header:        http.Header{"Content-Type": {"application/json"}},
+		Body:          io.NopCloser(strings.NewReader(data)),
+		ContentLength: int64(len(data)),
+		Request:       req,
+	}, nil
+}
+
+func TestCollectScanTasksReturnsFirstErrorInHandleOrder(t *testing.T) {
+	t.Parallel()
+
+	cat := newScanPlanningTestCatalog(t, []endpoint{endpointFetchScanTasks}, nil)
+	cat.cl = &http.Client{Transport: &orderedScanTaskErrorTransport{
+		planTaskReturned: make(chan struct{}),
+	}}
+
+	envelopes, err := cat.collectScanTasks(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
+		PlanTasks: []string{"table", "plan-task"},
+	})
+	require.ErrorIs(t, err, catalog.ErrNoSuchTable)
+	assert.NotErrorIs(t, err, ErrNoSuchPlanTask)
+	assert.Nil(t, envelopes)
+}
+
 func TestCollectScanTasksBoundsFrontierConcurrency(t *testing.T) {
 	t.Parallel()
 

From 0aa0ae6d9c20ee649b44e8e5b0010b3cded169aa Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Tue, 1 Sep 2026 07:30:31 +0200
Subject: [PATCH 04/12] style(rest): satisfy return spacing lint

---
 catalog/rest/scan_planning_test.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index a54e11de4..db9fa9ca6 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -1341,6 +1341,7 @@ func (t *orderedScanTaskErrorTransport) RoundTrip(req *http.Request) (*http.Resp
 	}
 
 	data := fmt.Sprintf(`{"error":{"message":%q,"type":%q,"code":404}}`, errType, errType)
+
 	return &http.Response{
 		StatusCode:    http.StatusNotFound,
 		Header:        http.Header{"Content-Type": {"application/json"}},

From d93bd8cf781d57295913a2db576467f7e9a51b35 Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Tue, 1 Sep 2026 13:58:19 +0200
Subject: [PATCH 05/12] fix(rest): add handle context to fetch errors

---
 catalog/rest/scan_planning.go      | 4 ++--
 catalog/rest/scan_planning_test.go | 1 +
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go
index fb5028264..d0547a5d6 100644
--- a/catalog/rest/scan_planning.go
+++ b/catalog/rest/scan_planning.go
@@ -326,7 +326,7 @@ func (r *Catalog) collectScanTasks(ctx context.Context, ident table.Identifier,
 			return nil, err
 		}
 
-		nextFrontier := make([]string, 0)
+		var nextFrontier []string
 		for _, response := range responses {
 			envelopes = append(envelopes, response.ScanTasks)
 			nextFrontier = append(nextFrontier, response.PlanTasks...)
@@ -351,7 +351,7 @@ func (r *Catalog) fetchScanTaskFrontier(
 		group.Go(func() error {
 			response, err := r.FetchScanTasks(groupCtx, ident, FetchScanTasksRequest{PlanTask: handle})
 			if err != nil {
-				errs[i] = err
+				errs[i] = fmt.Errorf("fetching scan tasks for handle %q: %w", handle, err)
 
 				return err
 			}
diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index db9fa9ca6..4e163874c 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -1363,6 +1363,7 @@ func TestCollectScanTasksReturnsFirstErrorInHandleOrder(t *testing.T) {
 		PlanTasks: []string{"table", "plan-task"},
 	})
 	require.ErrorIs(t, err, catalog.ErrNoSuchTable)
+	assert.Contains(t, err.Error(), `handle "table"`)
 	assert.NotErrorIs(t, err, ErrNoSuchPlanTask)
 	assert.Nil(t, envelopes)
 }

From 81bb112809315bfee4cde16c8450907453ad81ff Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Tue, 1 Sep 2026 13:58:21 +0200
Subject: [PATCH 06/12] docs(schema): explain JSON marshal alias

---
 schema.go      | 2 ++
 schema_test.go | 2 ++
 2 files changed, 4 insertions(+)

diff --git a/schema.go b/schema.go
index b2ca8b181..6c04df66b 100644
--- a/schema.go
+++ b/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}
 
 	return json.Marshal(struct {
diff --git a/schema_test.go b/schema_test.go
index 5661453e6..1170205cc 100644
--- a/schema_test.go
+++ b/schema_test.go
@@ -2302,6 +2302,8 @@ func TestVisitGeoSchemaWithSchemaVisitorPerPrimitiveType(t *testing.T) {
 	assert.Equal(t, 1, v.geographyCalls)
 }
 
+// This test is intended to be run with -race; without the detector, the old
+// MarshalJSON implementation also passes these assertions.
 func TestSchemaMarshalJSONConcurrentLazyLookups(t *testing.T) {
 	for range 32 {
 		schema := iceberg.NewSchemaWithIdentifiers(17, nil,

From 96acb8d3b8e76450d0461650a3f5456a916f94fc Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Thu, 3 Sep 2026 10:30:20 +0200
Subject: [PATCH 07/12] fix(rest): honor scan concurrency and cancellation

---
 catalog/rest/scan_planning.go            | 66 ++++++++++++++++++++----
 catalog/rest/scan_planning_bench_test.go | 17 +++---
 catalog/rest/scan_planning_test.go       | 15 ++++--
 table/scan_planning.go                   |  3 ++
 table/scan_planning_test.go              |  2 +
 table/scanner.go                         |  1 +
 6 files changed, 82 insertions(+), 22 deletions(-)

diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go
index d0547a5d6..35ec0215d 100644
--- a/catalog/rest/scan_planning.go
+++ b/catalog/rest/scan_planning.go
@@ -30,6 +30,7 @@ import (
 	"math/rand/v2"
 	"net/http"
 	"net/url"
+	"runtime"
 	"slices"
 	"strconv"
 	"strings"
@@ -52,11 +53,6 @@ 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
-
 // ErrPlanExpired is returned when polling a plan that the server no longer
 // knows about: a fetchPlanningResult 404 whose error.type is exactly
 // NoSuchPlanIdException. It is distinct from a table/namespace-gone 404
@@ -253,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 {
 		cleanup()
 
@@ -307,6 +304,23 @@ func (r *Catalog) planIOBaseProps(req table.ScanPlanningRequest) iceberg.Propert
 // 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))
+}
+
+// collectScanTasksWithConcurrency expands plan-task handles using the
+// requested maximum 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}
 
 	frontier := append([]string(nil), tasks.PlanTasks...)
@@ -320,8 +334,11 @@ func (r *Catalog) collectScanTasks(ctx context.Context, ident table.Identifier,
 			seen[handle] = true
 			handles = append(handles, handle)
 		}
+		if len(handles) == 0 {
+			break
+		}
 
-		responses, err := r.fetchScanTaskFrontier(ctx, ident, handles)
+		responses, err := r.fetchScanTaskFrontier(ctx, ident, handles, maxConcurrency)
 		if err != nil {
 			return nil, err
 		}
@@ -337,21 +354,39 @@ func (r *Catalog) collectScanTasks(ctx context.Context, ident table.Identifier,
 	return envelopes, nil
 }
 
+// fetchScanTaskFrontier fetches one breadth-first frontier concurrently while
+// placing responses back into handle order. The explicit cancellation context
+// keeps sibling transport errors as context.Canceled instead of propagating an
+// errgroup cancellation cause into those errors.
 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))
-	group, groupCtx := errgroup.WithContext(ctx)
-	group.SetLimit(remoteScanTaskFetchConcurrency)
+	if len(handles) == 0 {
+		return responses, nil
+	}
+	if maxConcurrency <= 0 {
+		maxConcurrency = runtime.GOMAXPROCS(0)
+	}
+	maxConcurrency = min(maxConcurrency, len(handles))
+
+	fetchCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	var group errgroup.Group
+	group.SetLimit(maxConcurrency)
 
 	for i, handle := range handles {
+		i, handle := i, handle
 		group.Go(func() error {
-			response, err := r.FetchScanTasks(groupCtx, ident, FetchScanTasksRequest{PlanTask: handle})
+			response, err := r.FetchScanTasks(fetchCtx, ident, FetchScanTasksRequest{PlanTask: handle})
 			if err != nil {
 				errs[i] = fmt.Errorf("fetching scan tasks for handle %q: %w", handle, err)
+				cancel()
 
 				return err
 			}
@@ -367,7 +402,16 @@ func (r *Catalog) fetchScanTaskFrontier(
 		// first error in handle order wins. Sibling requests cancelled by the
 		// first failure are not meaningful candidates for the returned error.
 		for _, err := range errs {
-			if err != nil && !errors.Is(err, context.Canceled) {
+			if err != nil &&
+				!errors.Is(err, context.Canceled) &&
+				!errors.Is(err, context.DeadlineExceeded) {
+				return nil, err
+			}
+		}
+		// If cancellation is the only error, still return the handle-aware
+		// error recorded for the first handle rather than the bare group error.
+		for _, err := range errs {
+			if err != nil {
 				return nil, err
 			}
 		}
diff --git a/catalog/rest/scan_planning_bench_test.go b/catalog/rest/scan_planning_bench_test.go
index 8dc639b04..531bd2a12 100644
--- a/catalog/rest/scan_planning_bench_test.go
+++ b/catalog/rest/scan_planning_bench_test.go
@@ -58,11 +58,16 @@ func BenchmarkCollectScanTasks64Handles(b *testing.B) {
 	tasks := ScanTasks{PlanTasks: handles}
 	ident := table.Identifier{"db", "tbl"}
 
-	b.ResetTimer()
-	for range b.N {
-		_, err := catalog.collectScanTasks(context.Background(), ident, tasks)
-		if err != nil {
-			b.Fatal(err)
-		}
+	for _, maxConcurrency := range []int{1, 2, 4, 8, 16, 32, 64} {
+		b.Run(fmt.Sprintf("concurrency-%d", maxConcurrency), func(b *testing.B) {
+			b.ResetTimer()
+			for range b.N {
+				_, err := catalog.collectScanTasksWithConcurrency(
+					context.Background(), ident, tasks, maxConcurrency)
+				if err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
 	}
 }
diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index 4e163874c..1865d0fe7 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -1331,7 +1331,11 @@ func (t *orderedScanTaskErrorTransport) RoundTrip(req *http.Request) (*http.Resp
 	var errType string
 	switch body.PlanTask {
 	case "table":
-		<-t.planTaskReturned
+		select {
+		case <-t.planTaskReturned:
+		case <-req.Context().Done():
+			return nil, req.Context().Err()
+		}
 		errType = errTypeNoSuchTable
 	case "plan-task":
 		close(t.planTaskReturned)
@@ -1371,7 +1375,8 @@ func TestCollectScanTasksReturnsFirstErrorInHandleOrder(t *testing.T) {
 func TestCollectScanTasksBoundsFrontierConcurrency(t *testing.T) {
 	t.Parallel()
 
-	const handleCount = remoteScanTaskFetchConcurrency + 1
+	const maxConcurrency = 8
+	const handleCount = maxConcurrency + 1
 	started := make(chan string, handleCount)
 	release := make(chan struct{})
 	var releaseOnce sync.Once
@@ -1399,13 +1404,13 @@ func TestCollectScanTasksBoundsFrontierConcurrency(t *testing.T) {
 	}
 	done := make(chan error, 1)
 	go func() {
-		_, err := cat.collectScanTasks(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
+		_, err := cat.collectScanTasksWithConcurrency(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
 			PlanTasks: handles,
-		})
+		}, maxConcurrency)
 		done <- err
 	}()
 
-	for range remoteScanTaskFetchConcurrency {
+	for range maxConcurrency {
 		select {
 		case <-started:
 		case <-time.After(time.Second):
diff --git a/table/scan_planning.go b/table/scan_planning.go
index 24ce690ae..c41fca5ad 100644
--- a/table/scan_planning.go
+++ b/table/scan_planning.go
@@ -105,6 +105,9 @@ type ScanPlanningRequest struct {
 	RowFilter        iceberg.BooleanExpression
 	MinRowsRequested *int64
 	StatsFields      []string
+	// MaxConcurrency is the maximum number of concurrent operations a planner
+	// may use for this scan. Zero means use the planner's default.
+	MaxConcurrency int
 	// CaseSensitive must carry the Scan's value (which defaults to true), not
 	// Go's false zero value, or the wire request would flip the spec default.
 	// Nil means use the scan default.
diff --git a/table/scan_planning_test.go b/table/scan_planning_test.go
index b9489bb65..1363abdb0 100644
--- a/table/scan_planning_test.go
+++ b/table/scan_planning_test.go
@@ -399,12 +399,14 @@ func TestScanPlanningRemoteResolvesDefaultProjectionAndSchema(t *testing.T) {
 	planner := &fakeScanPlanner{supports: true}
 	scan := (&Table{metadata: metadata}).Scan(
 		WithScanPlanningMode(ScanPlanningRemote),
+		WithMaxConcurrency(3),
 	)
 	scan.planner = planner
 
 	_, err = scan.PlanFiles(context.Background())
 	require.NoError(t, err)
 	assert.Equal(t, []string{"id"}, planner.receivedRequest.SelectedFields)
+	assert.Equal(t, 3, planner.receivedRequest.MaxConcurrency)
 	assert.True(t, planner.receivedRequest.Schema.Equals(metadata.CurrentSchema()))
 	require.NotNil(t, planner.receivedRequest.UseSnapshotSchema)
 	assert.False(t, *planner.receivedRequest.UseSnapshotSchema)
diff --git a/table/scanner.go b/table/scanner.go
index c30c76ed8..40b596224 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -1689,6 +1689,7 @@ func (scan *Scan) planFilesRemote(ctx context.Context) ([]FileScanTask, error) {
 		SelectedFields:    selectedFields,
 		RowFilter:         scan.rowFilter,
 		MinRowsRequested:  minRowsRequested,
+		MaxConcurrency:    scan.concurrency,
 		CaseSensitive:     &caseSensitive,
 		UseSnapshotSchema: &useSnapshotSchema,
 	})

From 3ee3bd1586676b1328add47c4fb05d1441a01228 Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Thu, 3 Sep 2026 17:45:43 +0200
Subject: [PATCH 08/12] fix(rest): preserve ordered scan task errors

---
 catalog/rest/scan_planning.go      | 76 ++++++++++++++++++++----------
 catalog/rest/scan_planning_test.go |  2 +-
 2 files changed, 52 insertions(+), 26 deletions(-)

diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go
index 35ec0215d..4368b902e 100644
--- a/catalog/rest/scan_planning.go
+++ b/catalog/rest/scan_planning.go
@@ -355,9 +355,9 @@ func (r *Catalog) collectScanTasksWithConcurrency(
 }
 
 // fetchScanTaskFrontier fetches one breadth-first frontier concurrently while
-// placing responses back into handle order. The explicit cancellation context
-// keeps sibling transport errors as context.Canceled instead of propagating an
-// errgroup cancellation cause into those errors.
+// 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,
@@ -374,8 +374,23 @@ func (r *Catalog) fetchScanTaskFrontier(
 	}
 	maxConcurrency = min(maxConcurrency, len(handles))
 
-	fetchCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
+	requestCtxs := make([]context.Context, len(handles))
+	requestCancels := make([]context.CancelFunc, len(handles))
+	for i := range handles {
+		requestCtxs[i], requestCancels[i] = context.WithCancel(ctx)
+	}
+	defer func() {
+		for _, cancel := range requestCancels {
+			cancel()
+		}
+	}()
+
+	var (
+		mu               sync.Mutex
+		completed        = make([]bool, len(handles))
+		siblingCanceled  = make([]bool, len(handles))
+		lowestFailureIdx = -1
+	)
 
 	var group errgroup.Group
 	group.SetLimit(maxConcurrency)
@@ -383,35 +398,46 @@ func (r *Catalog) fetchScanTaskFrontier(
 	for i, handle := range handles {
 		i, handle := i, handle
 		group.Go(func() error {
-			response, err := r.FetchScanTasks(fetchCtx, ident, FetchScanTasksRequest{PlanTask: handle})
-			if err != nil {
-				errs[i] = fmt.Errorf("fetching scan tasks for handle %q: %w", handle, err)
-				cancel()
+			response, fetchErr := r.FetchScanTasks(requestCtxs[i], ident, FetchScanTasksRequest{PlanTask: handle})
+
+			mu.Lock()
+			defer mu.Unlock()
+			completed[i] = true
+			if fetchErr == nil {
+				responses[i] = response
+
+				return nil
+			}
 
-				return err
+			errs[i] = fmt.Errorf("fetching scan tasks for handle %q: %w", handle, fetchErr)
+			if siblingCanceled[i] || (lowestFailureIdx >= 0 && i > lowestFailureIdx) {
+				return fetchErr
 			}
 
-			responses[i] = response
+			if lowestFailureIdx < 0 || i < lowestFailureIdx {
+				lowestFailureIdx = i
+				for j := i + 1; j < len(handles); j++ {
+					if completed[j] || siblingCanceled[j] {
+						continue
+					}
+					siblingCanceled[j] = true
+					requestCancels[j]()
+				}
+			}
 
-			return nil
+			return fetchErr
 		})
 	}
 
 	if waitErr := group.Wait(); waitErr != nil {
-		// Preserve the serial fetch contract: when multiple handles fail, the
-		// first error in handle order wins. Sibling requests cancelled by the
-		// first failure are not meaningful candidates for the returned error.
-		for _, err := range errs {
-			if err != nil &&
-				!errors.Is(err, context.Canceled) &&
-				!errors.Is(err, context.DeadlineExceeded) {
-				return nil, err
-			}
+		if err := ctx.Err(); err != nil {
+			return nil, err
 		}
-		// If cancellation is the only error, still return the handle-aware
-		// error recorded for the first handle rather than the bare group error.
-		for _, err := range errs {
-			if err != nil {
+
+		mu.Lock()
+		defer mu.Unlock()
+		for i, err := range errs {
+			if err != nil && !siblingCanceled[i] {
 				return nil, err
 			}
 		}
diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index 1865d0fe7..190672f1b 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -1968,7 +1968,7 @@ func TestCollectScanTasksCancelsSiblingRequestsOnFailure(t *testing.T) {
 	})
 
 	envelopes, err := cat.collectScanTasks(ctx, table.Identifier{"db", "tbl"}, ScanTasks{
-		PlanTasks: []string{"slow", "failed"},
+		PlanTasks: []string{"failed", "slow"},
 	})
 	require.ErrorIs(t, err, ErrBadRequest)
 	assert.Nil(t, envelopes)

From f606e6ee2c9eb422cc0477d7f42d5150e02bcd55 Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Thu, 3 Sep 2026 18:17:11 +0200
Subject: [PATCH 09/12] fix(rest): tighten scan planning validation

---
 catalog/rest/fetch_scan_tasks_validation.go | 3 +--
 catalog/rest/scan_planning.go               | 1 -
 catalog/rest/scan_planning_test.go          | 6 ++++++
 3 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/catalog/rest/fetch_scan_tasks_validation.go b/catalog/rest/fetch_scan_tasks_validation.go
index 45803eb8e..cfe0858cd 100644
--- a/catalog/rest/fetch_scan_tasks_validation.go
+++ b/catalog/rest/fetch_scan_tasks_validation.go
@@ -35,8 +35,7 @@ func validatePlanningTaskEnvelope(data []byte, status PlanStatus, tasks ScanTask
 
 	if status != PlanStatusCompleted {
 		for _, name := range []string{"plan-tasks", "file-scan-tasks", "delete-files"} {
-			raw, ok := fields[name]
-			if ok && !isJSONNull(raw) {
+			if _, ok := fields[name]; ok {
 				return fmt.Errorf("%w: %s response includes %s for status %q", ErrRESTError, endpoint, name, status)
 			}
 		}
diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go
index 4368b902e..25b6c7e29 100644
--- a/catalog/rest/scan_planning.go
+++ b/catalog/rest/scan_planning.go
@@ -396,7 +396,6 @@ func (r *Catalog) fetchScanTaskFrontier(
 	group.SetLimit(maxConcurrency)
 
 	for i, handle := range handles {
-		i, handle := i, handle
 		group.Go(func() error {
 			response, fetchErr := r.FetchScanTasks(requestCtxs[i], ident, FetchScanTasksRequest{PlanTask: handle})
 
diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index 190672f1b..5447bf11b 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -214,8 +214,11 @@ func TestPlanTableScanResponseRejectsInvalidStatusEnvelope(t *testing.T) {
 
 	for i, payload := range []string{
 		`{"status":"submitted","plan-id":"abc","file-scan-tasks":[]}`,
+		`{"status":"submitted","plan-id":"abc","file-scan-tasks":null}`,
 		`{"status":"submitted","plan-id":"abc","delete-files":[]}`,
+		`{"status":"submitted","plan-id":"abc","delete-files":null}`,
 		`{"status":"failed","plan-tasks":[]}`,
+		`{"status":"failed","plan-tasks":null}`,
 		`{"status":"completed","plan-id":"abc","delete-files":[{}]}`,
 		`{"status":"failed","plan-id":"abc"}`,
 	} {
@@ -316,8 +319,11 @@ func TestFetchPlanningResultResponseValidation(t *testing.T) {
 
 		for i, payload := range []string{
 			`{"status":"submitted","plan-tasks":[]}`,
+			`{"status":"submitted","plan-tasks":null}`,
 			`{"status":"submitted","delete-files":[]}`,
+			`{"status":"submitted","delete-files":null}`,
 			`{"status":"cancelled","file-scan-tasks":[]}`,
+			`{"status":"cancelled","file-scan-tasks":null}`,
 			`{"status":"completed","delete-files":[{}]}`,
 		} {
 			t.Run(fmt.Sprintf("payload-%d", i), func(t *testing.T) {

From ca8224f4c6578d3bffe06b125b9b065147b16ebe Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Thu, 3 Sep 2026 18:27:07 +0200
Subject: [PATCH 10/12] fix(rest): reject null planning task fields

---
 catalog/rest/fetch_scan_tasks_validation.go | 15 ++++++++++-----
 catalog/rest/scan_planning_test.go          |  6 ++++++
 2 files changed, 16 insertions(+), 5 deletions(-)

diff --git a/catalog/rest/fetch_scan_tasks_validation.go b/catalog/rest/fetch_scan_tasks_validation.go
index cfe0858cd..ebb5ea185 100644
--- a/catalog/rest/fetch_scan_tasks_validation.go
+++ b/catalog/rest/fetch_scan_tasks_validation.go
@@ -33,11 +33,16 @@ func validatePlanningTaskEnvelope(data []byte, status PlanStatus, tasks ScanTask
 		return err
 	}
 
-	if status != PlanStatusCompleted {
-		for _, name := range []string{"plan-tasks", "file-scan-tasks", "delete-files"} {
-			if _, ok := fields[name]; ok {
-				return fmt.Errorf("%w: %s response includes %s for status %q", ErrRESTError, endpoint, name, status)
-			}
+	for _, name := range []string{"plan-tasks", "file-scan-tasks", "delete-files"} {
+		raw, ok := fields[name]
+		if !ok {
+			continue
+		}
+		if isJSONNull(raw) {
+			return fmt.Errorf("%w: %s response field %s must not be null", ErrRESTError, endpoint, name)
+		}
+		if status != PlanStatusCompleted {
+			return fmt.Errorf("%w: %s response includes %s for status %q", ErrRESTError, endpoint, name, status)
 		}
 	}
 
diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index 5447bf11b..80286e61b 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -220,6 +220,9 @@ func TestPlanTableScanResponseRejectsInvalidStatusEnvelope(t *testing.T) {
 		`{"status":"failed","plan-tasks":[]}`,
 		`{"status":"failed","plan-tasks":null}`,
 		`{"status":"completed","plan-id":"abc","delete-files":[{}]}`,
+		`{"status":"completed","plan-id":"abc","plan-tasks":null}`,
+		`{"status":"completed","plan-id":"abc","file-scan-tasks":null}`,
+		`{"status":"completed","plan-id":"abc","delete-files":null}`,
 		`{"status":"failed","plan-id":"abc"}`,
 	} {
 		t.Run(fmt.Sprintf("payload-%d", i), func(t *testing.T) {
@@ -325,6 +328,9 @@ func TestFetchPlanningResultResponseValidation(t *testing.T) {
 			`{"status":"cancelled","file-scan-tasks":[]}`,
 			`{"status":"cancelled","file-scan-tasks":null}`,
 			`{"status":"completed","delete-files":[{}]}`,
+			`{"status":"completed","plan-tasks":null}`,
+			`{"status":"completed","file-scan-tasks":null}`,
+			`{"status":"completed","delete-files":null}`,
 		} {
 			t.Run(fmt.Sprintf("payload-%d", i), func(t *testing.T) {
 				t.Parallel()

From 8656237a9dbcc7188878f8f100b4468fefda63bd Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Sat, 5 Sep 2026 00:01:07 +0200
Subject: [PATCH 11/12] test(schema): guard exported JSON fields

Signed-off-by: Minh Vu <vuhoangminh97@gmail.com>
---
 schema_test.go | 37 +++++++++++++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/schema_test.go b/schema_test.go
index 1170205cc..f0773a188 100644
--- a/schema_test.go
+++ b/schema_test.go
@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"reflect"
 	"runtime"
 	"strings"
 	"sync"
@@ -679,6 +680,42 @@ func TestSerializeSchema(t *testing.T) {
 	}`, string(data))
 }
 
+func TestMarshalSchemaIncludesExportedFields(t *testing.T) {
+	schema := iceberg.NewSchemaWithIdentifiers(17, []int{1},
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	data, err := json.Marshal(schema)
+	require.NoError(t, err)
+	var marshaled map[string]json.RawMessage
+	require.NoError(t, json.Unmarshal(data, &marshaled))
+
+	require.Contains(t, marshaled, "type")
+	require.Contains(t, marshaled, "fields")
+	wantKeys := 2
+	value := reflect.ValueOf(schema).Elem()
+	for i := range value.NumField() {
+		field := value.Type().Field(i)
+		tag, tagged := field.Tag.Lookup("json")
+		name, _, _ := strings.Cut(tag, ",")
+		if !field.IsExported() || !tagged || name == "-" {
+			continue
+		}
+		if name == "" {
+			name = field.Name
+		}
+		wantKeys++
+
+		// A nonzero fixture catches omitted assignments even when the alias
+		// still emits the field's key with a zero value. Extend it for new fields.
+		require.False(t, value.Field(i).IsZero(), "populate Schema.%s in the fixture", field.Name)
+		want, err := json.Marshal(value.Field(i).Interface())
+		require.NoError(t, err)
+		require.Contains(t, marshaled, name)
+		assert.JSONEq(t, string(want), string(marshaled[name]), "Schema.%s", field.Name)
+	}
+	assert.Len(t, marshaled, wantKeys)
+}
+
 func TestUnmarshalSchema(t *testing.T) {
 	var schema iceberg.Schema
 	require.NoError(t, json.Unmarshal([]byte(`{

From a954db0d75352241198c5fa57dc5be8226ea7086 Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Sat, 5 Sep 2026 00:01:07 +0200
Subject: [PATCH 12/12] refactor(rest): remove test-only scan task wrapper

Signed-off-by: Minh Vu <vuhoangminh97@gmail.com>
---
 catalog/rest/scan_planning.go      | 14 +++++---------
 catalog/rest/scan_planning_test.go | 21 +++++++++++++--------
 2 files changed, 18 insertions(+), 17 deletions(-)

diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go
index 25b6c7e29..93ddfb6d1 100644
--- a/catalog/rest/scan_planning.go
+++ b/catalog/rest/scan_planning.go
@@ -296,20 +296,16 @@ 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. Each
+// 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.
-func (r *Catalog) collectScanTasks(ctx context.Context, ident table.Identifier, tasks ScanTasks) ([]ScanTasks, error) {
-	return r.collectScanTasksWithConcurrency(ctx, ident, tasks, runtime.GOMAXPROCS(0))
-}
-
-// collectScanTasksWithConcurrency expands plan-task handles using the
-// requested maximum number of concurrent fetches. A non-positive limit uses
-// runtime.GOMAXPROCS. On a fetch error, at most the configured number of
+//
+// 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,
diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go
index 80286e61b..093eca2cd 100644
--- a/catalog/rest/scan_planning_test.go
+++ b/catalog/rest/scan_planning_test.go
@@ -1231,6 +1231,7 @@ func TestPlanFilesPollsSubmittedPlan(t *testing.T) {
 func TestPlanFilesExpandsPlanTasks(t *testing.T) {
 	t.Parallel()
 
+	var mu sync.Mutex
 	var fetched []string
 	cat := newScanPlanningTestCatalog(t, []endpoint{endpointPlanTableScan, endpointFetchScanTasks}, func(mux *http.ServeMux) {
 		mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w http.ResponseWriter, req *http.Request) {
@@ -1240,7 +1241,9 @@ func TestPlanFilesExpandsPlanTasks(t *testing.T) {
 		mux.HandleFunc("/v1/namespaces/db/tables/tbl/tasks", func(w http.ResponseWriter, req *http.Request) {
 			var body FetchScanTasksRequest
 			require.NoError(t, json.NewDecoder(req.Body).Decode(&body))
+			mu.Lock()
 			fetched = append(fetched, body.PlanTask)
+			mu.Unlock()
 			switch body.PlanTask {
 			case "h1":
 				_, err := w.Write([]byte(`{"plan-tasks":["h2"]}`))
@@ -1255,6 +1258,8 @@ func TestPlanFilesExpandsPlanTasks(t *testing.T) {
 	result, err := cat.PlanFiles(context.Background(), planFilesReq())
 	require.NoError(t, err)
 	assert.Empty(t, result.Tasks)
+	mu.Lock()
+	defer mu.Unlock()
 	assert.Equal(t, []string{"h1", "h2"}, fetched)
 }
 
@@ -1297,9 +1302,9 @@ func TestCollectScanTasksFetchesFrontierConcurrentlyInOrder(t *testing.T) {
 	}
 	done := make(chan outcome, 1)
 	go func() {
-		envelopes, err := cat.collectScanTasks(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
+		envelopes, err := cat.collectScanTasksWithConcurrency(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
 			PlanTasks: []string{"h1", "h2"},
-		})
+		}, 2)
 		done <- outcome{envelopes: envelopes, err: err}
 	}()
 
@@ -1375,9 +1380,9 @@ func TestCollectScanTasksReturnsFirstErrorInHandleOrder(t *testing.T) {
 		planTaskReturned: make(chan struct{}),
 	}}
 
-	envelopes, err := cat.collectScanTasks(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
+	envelopes, err := cat.collectScanTasksWithConcurrency(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
 		PlanTasks: []string{"table", "plan-task"},
-	})
+	}, 2)
 	require.ErrorIs(t, err, catalog.ErrNoSuchTable)
 	assert.Contains(t, err.Error(), `handle "table"`)
 	assert.NotErrorIs(t, err, ErrNoSuchPlanTask)
@@ -1931,9 +1936,9 @@ func TestCollectScanTasksDeduplicatesAcrossFrontiers(t *testing.T) {
 		})
 	})
 
-	envelopes, err := cat.collectScanTasks(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
+	envelopes, err := cat.collectScanTasksWithConcurrency(t.Context(), table.Identifier{"db", "tbl"}, ScanTasks{
 		PlanTasks: []string{"a", "a", "b"},
-	})
+	}, 2)
 	require.NoError(t, err)
 	require.Len(t, envelopes, 6)
 	for i, name := range []string{"a", "b", "c", "d", "e"} {
@@ -1979,9 +1984,9 @@ func TestCollectScanTasksCancelsSiblingRequestsOnFailure(t *testing.T) {
 		})
 	})
 
-	envelopes, err := cat.collectScanTasks(ctx, table.Identifier{"db", "tbl"}, ScanTasks{
+	envelopes, err := cat.collectScanTasksWithConcurrency(ctx, table.Identifier{"db", "tbl"}, ScanTasks{
 		PlanTasks: []string{"failed", "slow"},
-	})
+	}, 2)
 	require.ErrorIs(t, err, ErrBadRequest)
 	assert.Nil(t, envelopes)
 	select {