This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 1857b7b00 feat(rest): add scan-planning retries, telemetry, and
documentation (#2020)
1857b7b00 is described below
commit 1857b7b0097da80b9b3e4562484cc382f5b5436c
Author: Revanth <[email protected]>
AuthorDate: Tue Sep 22 14:41:31 2026 -0500
feat(rest): add scan-planning retries, telemetry, and documentation (#2020)
* feat(rest): add scan-planning retries, telemetry, and documentation
* fix(rest): address scan-planning review feedback
---
README.md | 108 +++++++++++-
catalog/rest/scan_planning.go | 78 ++++++---
catalog/rest/scan_planning_parity_test.go | 8 +-
catalog/rest/scan_planning_retry.go | 68 ++++++++
catalog/rest/scan_planning_retry_test.go | 239 +++++++++++++++++++++++++++
catalog/rest/scan_planning_telemetry_test.go | 207 +++++++++++++++++++++++
expr_json_test.go | 19 +++
go.mod | 4 +-
internal/scanmetrics/telemetry.go | 84 ++++++++++
internal/scanmetrics/telemetry_test.go | 96 +++++++++++
table/scan_splits_test.go | 19 +++
table/scanner.go | 8 +-
12 files changed, 909 insertions(+), 29 deletions(-)
diff --git a/README.md b/README.md
index afb0cf1fa..23888222e 100644
--- a/README.md
+++ b/README.md
@@ -117,8 +117,112 @@ Tests requiring Spark 4 (currently the variant and
unknown-type tests) automatic
| Get Partition Specs | X |
| Get Manifests | X |
| Create New Manifests | X |
-| Plan Scan | x |
-| Plan Scan for Snapshot | x |
+| Plan Scan | Local + Remote |
+| Plan Scan for Snapshot | Local + Remote |
+
+### REST Scan Planning
+
+REST catalogs can plan scans on the server, including asynchronous plans,
+batched task retrieval, and plan-scoped storage credentials. Enable this per
+scan with `table.WithScanPlanningMode`; existing scans continue to plan
locally.
+
+| Scan option | Behavior |
+| :---------- | :------- |
+| `table.ScanPlanningLocal` (default) | Read manifests locally; no planning
endpoints required. |
+| `table.ScanPlanningRemote` | Require the advertised plan endpoint. Async or
fanout responses also require their continuation endpoints; missing
capabilities return an error. |
+| `table.ScanPlanningAuto` | Use remote planning when submission, polling, and
task retrieval endpoints are advertised; otherwise plan locally. Errors after
choosing remote are returned, without falling back to local. |
+
+Capabilities come from `GET /v1/config`. `rest.Catalog.SupportsPlanTableScan()`
+checks submission support, while `SupportsFullRemoteScanPlanning()` checks
+submission, polling, and task retrieval. Cancellation is best-effort and is not
+required for automatic remote planning. In `auto` mode, catalogs
+without a planner retain local planning.
+
+The scanner does not honor the REST `scan-planning-mode` configuration key
+(`client`/`server`). A catalog requiring `server` mode may rely on planning to
+vend plan-scoped storage credentials. Default local planning, explicit `local`,
+or an `auto` fallback can instead read manifests with the table's storage
+credentials, which may use a different identity or lack the required access.
+For these deployments, explicitly select `table.ScanPlanningRemote` on every
+scan; do not rely on the server configuration or `auto` to enforce remote
+planning. Remote mode returns an error if a required capability is missing.
+
+```go
+// tbl is a table loaded from a rest.Catalog.
+ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+defer cancel()
+
+scan := tbl.Scan(
+ table.WithScanPlanningMode(table.ScanPlanningRemote),
+ table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("tenant_id"),
"acme")),
+ table.WithSelectedFields("id", "tenant_id"),
+)
+defer scan.Close()
+
+tasks, err := scan.PlanFiles(ctx)
+if err != nil {
+ return err
+}
+// Read using this scan so its plan-scoped credentials remain attached.
+_, records, err := scan.ReadTasks(ctx, tasks)
+if err != nil {
+ return err
+}
+for record, err := range records {
+ if err != nil {
+ return err
+ }
+ // Consume the Arrow record batch here.
+ record.Release()
+}
+```
+
+`WithSnapshotID`, `WithSnapshotAsOf`, and tag scans use the historical schema;
+branch scans use the current table schema. File ranges follow
+`read.split.target-size` for both local and remote planning. Remote incremental
+scans and `_last_updated_sequence_number` projection are not supported. For
+row-lineage scans requiring that column, `auto` chooses local planning and
+`remote` returns an error.
+
+Plan and task POSTs retry HTTP 408, 429, 500, 502, 503, and 504 up to three
times
+with jittered backoff, reusing the same UUIDv7 idempotency key. Other HTTP
errors,
+transport failures, and malformed successful responses are returned directly.
+`Retry-After` is honored up to the caller's deadline (or capped at five seconds
+without a deadline). A new explicit call generates a new key unless one is
+supplied by the caller. `WaitForPlanOptions` controls polling delays and retry
+limits; use a context deadline to bound the complete operation. Cancellation
+during polling and exhausted polling trigger best-effort server cleanup.
Recognized expired
+plans return `rest.ErrPlanExpired`; expired task handles return
+`rest.ErrNoSuchPlanTask`. Callers decide whether to submit a new plan.
+
+Close the scan after consuming its tasks, including when tasks are never read.
+Plan-scoped credentials remain alive for active readers; closing or replacing
+the plan releases them when those readers finish. They cannot be renewed from
+the table-credentials endpoint. Once `ReadTasks` returns an iterator, iterate
it
+(even if stopping early) to release its reader lease.
+
+OpenTelemetry uses the application's global tracer and meter providers, with
+scope `github.com/apache/iceberg-go/scan-planning`; no SDK is installed by the
+library. Logical endpoint calls emit `iceberg.scan.planning.<operation>` spans
+and the following instruments:
+
+| Instrument | Meaning |
+| :--------- | :------ |
+| `iceberg.scan.planning.requests` | Logical endpoint calls, labeled by
operation and success/error outcome. |
+| `iceberg.scan.planning.request.duration` | Endpoint call duration including
retries, in milliseconds. |
+| `iceberg.scan.planning.retries` | Additional attempts after transient HTTP
failures. |
+| `iceberg.scan.planning.expirations` | Recognized expired plan or task
responses. |
+| `iceberg.scan.planning.fallbacks` | Auto mode choosing local due to
capability or row-lineage constraints. |
+
+Operations are `plan`, `fetch-result`, `fetch-tasks`, and `cancel`. Attributes
+contain only operation, outcome, or fallback reason; they omit table names,
+filters, tokens, credentials, and raw error messages. These client metrics
+complement the existing scan/commit metrics reporter.
+
+The in-process local/remote parity test runs with `go test ./catalog/rest`.
+For Java interoperability, run `make integration-rest-scan-planning` (Docker
+required). Remote planning is opt-in, with no migration required for existing
+local scans.
### Catalog Support
diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go
index 93ddfb6d1..b8c974651 100644
--- a/catalog/rest/scan_planning.go
+++ b/catalog/rest/scan_planning.go
@@ -39,6 +39,7 @@ import (
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/internal/scanmetrics"
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
"github.com/google/uuid"
@@ -594,7 +595,17 @@ func marshalScanFilter(req table.ScanPlanningRequest)
([]byte, error) {
// Any other status — including the empty status of a 200 with no body, which
// bypasses the response UnmarshalJSON validation — returns an ErrRESTError so
a
// malformed response cannot masquerade as an empty completed plan.
-func (r *Catalog) PlanTableScan(ctx context.Context, ident table.Identifier,
req PlanTableScanRequest) (PlanTableScanResponse, error) {
+//
+// Transient HTTP responses (408, 429, 500, 502, 503, 504) are retried up to
+// three times with jittered backoff and the same idempotency key. Retry-After
+// is honored, capped at five seconds when ctx has no deadline. Transport
+// failures and malformed successful responses are not retried.
+func (r *Catalog) PlanTableScan(ctx context.Context, ident table.Identifier,
req PlanTableScanRequest) (result PlanTableScanResponse, err error) {
+ ctx, finish := scanmetrics.Start(ctx, "plan")
+ defer func() {
+ finish(err)
+ }()
+
if err := r.endpoints.check(endpointPlanTableScan); err != nil {
return PlanTableScanResponse{}, err
}
@@ -609,10 +620,12 @@ func (r *Catalog) PlanTableScan(ctx context.Context,
ident table.Identifier, req
return PlanTableScanResponse{}, err
}
- resp, err := doPost[PlanTableScanRequest, PlanTableScanResponse](
- ctx, r.baseURI, path, req, r.cl,
- map[int]error{http.StatusNotFound: ErrRESTError},
- withHeaders(headers),
withErrorTypeOverride(planTableScanErrorTypes))
+ resp, err := retryScanPlanning(ctx, "plan", func()
(PlanTableScanResponse, error) {
+ return doPost[PlanTableScanRequest, PlanTableScanResponse](
+ ctx, r.baseURI, path, req, r.cl,
+ map[int]error{http.StatusNotFound: ErrRESTError},
+ withHeaders(headers),
withErrorTypeOverride(planTableScanErrorTypes))
+ })
if err != nil {
return PlanTableScanResponse{}, err
}
@@ -641,7 +654,15 @@ func (r *Catalog) PlanTableScan(ctx context.Context, ident
table.Identifier, req
// namespace is catalog.ErrNoSuchTable / catalog.ErrNoSuchNamespace, so the
poller
// can tell retry-with-a-new-plan from abort. A bare or unrecognized 404 stays
an
// ambiguous ErrRESTError rather than being guessed as an expiry.
-func (r *Catalog) FetchPlanningResult(ctx context.Context, ident
table.Identifier, planID string, opts FetchPlanningResultOptions)
(FetchPlanningResultResponse, error) {
+func (r *Catalog) FetchPlanningResult(ctx context.Context, ident
table.Identifier, planID string, opts FetchPlanningResultOptions) (result
FetchPlanningResultResponse, err error) {
+ ctx, finish := scanmetrics.Start(ctx, "fetch-result")
+ defer func() {
+ if errors.Is(err, ErrPlanExpired) {
+ scanmetrics.Expired(ctx, "fetch-result")
+ }
+ finish(err)
+ }()
+
if err := r.endpoints.check(endpointFetchPlanResult); err != nil {
return FetchPlanningResultResponse{}, err
}
@@ -680,7 +701,9 @@ func (r *Catalog) FetchPlanningResult(ctx context.Context,
ident table.Identifie
// the session-default access-delegation header (cancel vends no credentials).
A
// 404 (already-expired or unknown plan) is not special-cased: cancel is
// best-effort, so the generic REST error is acceptable.
-func (r *Catalog) CancelPlanning(ctx context.Context, ident table.Identifier,
planID string) error {
+func (r *Catalog) CancelPlanning(ctx context.Context, ident table.Identifier,
planID string) (err error) {
+ ctx, finish := scanmetrics.Start(ctx, "cancel")
+ defer func() { finish(err) }()
if err := r.endpoints.check(endpointCancelPlanning); err != nil {
return err
}
@@ -705,7 +728,17 @@ func (r *Catalog) CancelPlanning(ctx context.Context,
ident table.Identifier, pl
// vanishing. A bare or unrecognized 404 stays an ambiguous ErrRESTError. An
// empty 200 body is rejected (requireBody) so a truncated response is not read
// as a successfully completed empty task set.
-func (r *Catalog) FetchScanTasks(ctx context.Context, ident table.Identifier,
req FetchScanTasksRequest) (FetchScanTasksResponse, error) {
+// Transient HTTP responses use the same retry policy as
[Catalog.PlanTableScan],
+// retaining the idempotency key and plan-task handle across attempts.
+func (r *Catalog) FetchScanTasks(ctx context.Context, ident table.Identifier,
req FetchScanTasksRequest) (result FetchScanTasksResponse, err error) {
+ ctx, finish := scanmetrics.Start(ctx, "fetch-tasks")
+ defer func() {
+ if errors.Is(err, ErrNoSuchPlanTask) {
+ scanmetrics.Expired(ctx, "fetch-tasks")
+ }
+ finish(err)
+ }()
+
if err := r.endpoints.check(endpointFetchScanTasks); err != nil {
return FetchScanTasksResponse{}, err
}
@@ -720,11 +753,13 @@ func (r *Catalog) FetchScanTasks(ctx context.Context,
ident table.Identifier, re
return FetchScanTasksResponse{}, err
}
- return doPost[FetchScanTasksRequest, FetchScanTasksResponse](
- ctx, r.baseURI, path, req, r.cl,
- map[int]error{http.StatusNotFound: ErrRESTError},
- withHeaders(headers),
withSuppressedHeaders(headerIcebergAccessDelegation),
- withErrorTypeOverride(fetchScanTasksErrorTypes), requireBody())
+ return retryScanPlanning(ctx, "fetch-tasks", func()
(FetchScanTasksResponse, error) {
+ return doPost[FetchScanTasksRequest, FetchScanTasksResponse](
+ ctx, r.baseURI, path, req, r.cl,
+ map[int]error{http.StatusNotFound: ErrRESTError},
+ withHeaders(headers),
withSuppressedHeaders(headerIcebergAccessDelegation),
+ withErrorTypeOverride(fetchScanTasksErrorTypes),
requireBody())
+ })
}
// WaitForPlan polls a submitted plan to completion using jittered backoff. It
@@ -845,6 +880,9 @@ func (r *Catalog) WaitForPlan(ctx context.Context, ident
table.Identifier, planI
return CompletedPlanningResult{}, fmt.Errorf("waiting
for plan %q on %v: %w", planID, ident, ctx.Err())
case <-timer.C:
+ if err != nil {
+ scanmetrics.Retry(ctx, "fetch-result")
+ }
}
}
}
@@ -1056,8 +1094,8 @@ func idempotencyHeaderValue(idempotencyKey *string)
(string, error) {
// Plan and task POSTs always send an idempotency key. The spec
pins it
// to a UUIDv7 string (RFC 9562) so a server can key its dedup
window off
// the embedded timestamp, matching Java's
UUIDUtil.generateUuidV7(). A
- // v4 key would be treated as undefined by such servers. There
is no
- // transport retry here, so nil means a fresh key for this call.
+ // v4 key would be treated as undefined by such servers. All
automatic
+ // retries within this call reuse the generated key.
key, err := uuid.NewV7()
if err != nil {
return "", fmt.Errorf("generating idempotency key: %w",
err)
@@ -1145,9 +1183,8 @@ type CompletedPlanningResult struct {
type PlanTableScanRequest struct {
// IdempotencyKey is sent as the Idempotency-Key header, not in the
JSON body.
// If set, it must be a UUIDv7 string (RFC 9562). If nil, a fresh
UUIDv7 is
- // generated per call and not returned, so passing nil and retrying on a
- // transient error sends a different key each attempt and defeats
server-side
- // dedup: pass an explicit key to get idempotency across retries.
+ // generated per call and reused for automatic retries. Supply an
explicit
+ // key to preserve idempotency across separate calls.
IdempotencyKey *string `json:"-"`
// AccessDelegation is sent as the X-Iceberg-Access-Delegation header,
not
// in the JSON body. Nil uses the catalog default.
@@ -1286,9 +1323,8 @@ func decodePlanningError(raw json.RawMessage)
*PlanningError {
type FetchScanTasksRequest struct {
// IdempotencyKey is sent as the Idempotency-Key header, not in the
JSON body.
// If set, it must be a UUIDv7 string (RFC 9562). If nil, a fresh
UUIDv7 is
- // generated per call and not returned, so passing nil and retrying on a
- // transient error sends a different key each attempt and defeats
server-side
- // dedup: pass an explicit key to get idempotency across retries.
+ // generated per call and reused for automatic retries. Supply an
explicit
+ // key to preserve idempotency across separate calls.
IdempotencyKey *string `json:"-"`
PlanTask string `json:"plan-task"`
diff --git a/catalog/rest/scan_planning_parity_test.go
b/catalog/rest/scan_planning_parity_test.go
index 859e349d7..526f9ff52 100644
--- a/catalog/rest/scan_planning_parity_test.go
+++ b/catalog/rest/scan_planning_parity_test.go
@@ -219,8 +219,12 @@ func newPlanningParityTable(t *testing.T) (table.Metadata,
*iceio.MemFS) {
}
snapshotID, seq := int64((i+1)*10), int64(i+1)
path := fmt.Sprintf("mem://parity/list-%d.avro", snapshotID)
+ var parentSnapshotID *int64
+ if i == 1 {
+ parentSnapshotID = parityPtr(int64(10))
+ }
var list bytes.Buffer
- require.NoError(t, iceberg.WriteManifestList(2, &list,
snapshotID, nil, &seq, 0, manifests))
+ require.NoError(t, iceberg.WriteManifestList(2, &list,
snapshotID, parentSnapshotID, &seq, 0, manifests))
require.NoError(t, fs.WriteFile(path, list.Bytes()))
if i == 0 {
committed, err :=
iceberg.ReadManifestList(bytes.NewReader(list.Bytes()))
@@ -228,7 +232,7 @@ func newPlanningParityTable(t *testing.T) (table.Metadata,
*iceio.MemFS) {
mf = committed[0]
}
require.NoError(t, builder.AddSnapshot(&table.Snapshot{
- SnapshotID: snapshotID, SequenceNumber: seq,
+ SnapshotID: snapshotID, ParentSnapshotID:
parentSnapshotID, SequenceNumber: seq,
TimestampMs: time.Now().UnixMilli() + int64(i+1),
ManifestList: path, SchemaID: parityPtr(0),
}))
}
diff --git a/catalog/rest/scan_planning_retry.go
b/catalog/rest/scan_planning_retry.go
new file mode 100644
index 000000000..79ace6471
--- /dev/null
+++ b/catalog/rest/scan_planning_retry.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"
+ "errors"
+ "time"
+
+ "github.com/apache/iceberg-go/internal/scanmetrics"
+)
+
+// retryScanPlanning retries only explicit transient HTTP responses. In
+// particular, it never replays an ambiguous transport failure or a malformed
+// successful response. POST callers create their idempotency header before
+// entering this helper and reuse it (and the request body) for every attempt.
+func retryScanPlanning[T any](ctx context.Context, operation string, request
func() (T, error)) (T, error) {
+ const (
+ maxRetries = 3
+ minDelay = 100 * time.Millisecond
+ maxDelay = 5 * time.Second
+ )
+ _, hasDeadline := ctx.Deadline()
+ delay := minDelay
+ for attempt := 0; ; attempt++ {
+ if err := ctx.Err(); err != nil {
+ var zero T
+
+ return zero, err
+ }
+ result, err := request()
+ var responseError errorResponse
+ if !errors.As(err, &responseError) {
+ return result, err
+ }
+ retryAfter, retryable := scanPlanPollRetry(err)
+ if !retryable || attempt == maxRetries {
+ return result, err
+ }
+ delay = nextScanPlanBackoff(delay, minDelay, maxDelay)
+ wait := applyRetryAfter(delay, retryAfter, minDelay, maxDelay,
hasDeadline)
+ timer := time.NewTimer(wait)
+ select {
+ case <-ctx.Done():
+ timer.Stop()
+ var zero T
+
+ return zero, ctx.Err()
+ case <-timer.C:
+ scanmetrics.Retry(ctx, operation)
+ }
+ }
+}
diff --git a/catalog/rest/scan_planning_retry_test.go
b/catalog/rest/scan_planning_retry_test.go
new file mode 100644
index 000000000..68f9e7fdb
--- /dev/null
+++ b/catalog/rest/scan_planning_retry_test.go
@@ -0,0 +1,239 @@
+// 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"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "strconv"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/apache/iceberg-go/table"
+ "github.com/google/uuid"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestScanPlanningPOSTRetriesPreserveIdempotency(t *testing.T) {
+ t.Parallel()
+ for _, operation := range []string{"plan", "tasks"} {
+ for _, status := range []int{408, 429, 500, 502, 503, 504} {
+ t.Run(fmt.Sprintf("%s/%d", operation, status), func(t
*testing.T) {
+ t.Parallel()
+ var attempts atomic.Int32
+ var firstKey, firstBody string
+ var mu sync.Mutex
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointPlanTableScan, endpointFetchScanTasks}, func(mux
*http.ServeMux) {
+
mux.HandleFunc("/v1/namespaces/db/tables/tbl/"+operation, func(w
http.ResponseWriter, req *http.Request) {
+ mu.Lock()
+ defer mu.Unlock()
+ key :=
req.Header.Get(headerIdempotencyKey)
+ body, err :=
io.ReadAll(req.Body)
+ require.NoError(t, err)
+ parsed, err := uuid.Parse(key)
+ require.NoError(t, err)
+ assert.Equal(t,
uuid.Version(7), parsed.Version())
+ if attempts.Add(1) == 1 {
+ firstKey, firstBody =
key, string(body)
+ w.WriteHeader(status)
+
+ return
+ }
+ assert.Equal(t, firstKey, key)
+ assert.Equal(t, firstBody,
string(body))
+ if operation == "plan" {
+ _, _ =
w.Write([]byte(`{"status":"completed","plan-id":"p"}`))
+ } else {
+ _, _ =
w.Write([]byte(`{}`))
+ }
+ })
+ })
+ var err error
+ if operation == "plan" {
+ _, err = cat.PlanTableScan(t.Context(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{})
+ } else {
+ _, err =
cat.FetchScanTasks(t.Context(), table.Identifier{"db", "tbl"},
FetchScanTasksRequest{PlanTask: "opaque"})
+ }
+ require.NoError(t, err)
+ assert.Equal(t, int32(2), attempts.Load())
+ })
+ }
+ }
+}
+
+func TestScanPlanningPOSTDoesNotRetryTerminalResponses(t *testing.T) {
+ t.Parallel()
+ for _, status := range []int{400, 401, 403, 404, 501} {
+ t.Run(strconv.Itoa(status), func(t *testing.T) {
+ t.Parallel()
+ var attempts atomic.Int32
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointPlanTableScan}, func(mux *http.ServeMux) {
+
mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w http.ResponseWriter,
req *http.Request) {
+ attempts.Add(1)
+ w.WriteHeader(status)
+ _, _ = w.Write([]byte(`invalid json`))
+ })
+ })
+ _, err := cat.PlanTableScan(t.Context(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{})
+ require.Error(t, err)
+ assert.Equal(t, int32(1), attempts.Load())
+ })
+ }
+}
+
+func TestScanPlanningPOSTDoesNotRetryMalformedSuccess(t *testing.T) {
+ t.Parallel()
+ for _, operation := range []string{"plan", "tasks"} {
+ t.Run(operation, func(t *testing.T) {
+ t.Parallel()
+ var attempts atomic.Int32
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointPlanTableScan, endpointFetchScanTasks}, func(mux
*http.ServeMux) {
+ mux.HandleFunc("POST
/v1/namespaces/db/tables/tbl/"+operation, func(w http.ResponseWriter, _
*http.Request) {
+ attempts.Add(1)
+ _, _ = w.Write([]byte(`invalid json`))
+ })
+ })
+ var err error
+ if operation == "plan" {
+ _, err = cat.PlanTableScan(t.Context(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{})
+ } else {
+ _, err = cat.FetchScanTasks(t.Context(),
table.Identifier{"db", "tbl"}, FetchScanTasksRequest{PlanTask: "opaque"})
+ }
+ require.ErrorIs(t, err, ErrRESTError)
+ require.ErrorContains(t, err, "error decoding json
payload")
+ var responseError errorResponse
+ assert.False(t, errors.As(err, &responseError))
+ assert.Equal(t, int32(1), attempts.Load())
+ })
+ }
+}
+
+func TestScanPlanningRetryBoundsAndCancellation(t *testing.T) {
+ t.Parallel()
+ t.Run("bounded", func(t *testing.T) {
+ t.Parallel()
+ attempts := 0
+ _, err := retryScanPlanning(t.Context(), "plan", func() (int,
error) {
+ attempts++
+
+ return 0, errorResponse{statusCode:
http.StatusServiceUnavailable, wrapping: ErrServiceUnavailable}
+ })
+ require.ErrorIs(t, err, ErrServiceUnavailable)
+ assert.Equal(t, 4, attempts)
+ })
+ t.Run("transport failure", func(t *testing.T) {
+ attempts := 0
+ want := errors.New("connection reset after submission")
+ _, err := retryScanPlanning(t.Context(), "plan", func() (int,
error) {
+ attempts++
+
+ return 0, want
+ })
+ require.ErrorIs(t, err, want)
+ assert.Equal(t, 1, attempts)
+ })
+ t.Run("transport sentinel without HTTP response", func(t *testing.T) {
+ attempts := 0
+ _, err := retryScanPlanning(t.Context(), "plan", func() (int,
error) {
+ attempts++
+
+ return 0, ErrServiceUnavailable
+ })
+ require.ErrorIs(t, err, ErrServiceUnavailable)
+ assert.Equal(t, 1, attempts)
+ })
+ t.Run("cancel during retry after", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ var attempts atomic.Int32
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointPlanTableScan}, func(mux *http.ServeMux) {
+ mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan",
func(w http.ResponseWriter, req *http.Request) {
+ attempts.Add(1)
+ w.Header().Set("Retry-After", "3600")
+ w.WriteHeader(http.StatusServiceUnavailable)
+ cancel()
+ })
+ })
+ started := time.Now()
+ _, err := cat.PlanTableScan(ctx, table.Identifier{"db", "tbl"},
PlanTableScanRequest{})
+ require.ErrorIs(t, err, context.Canceled)
+ assert.Less(t, time.Since(started), time.Second)
+ assert.Equal(t, int32(1), attempts.Load())
+ })
+ t.Run("already cancelled", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ _, err := retryScanPlanning(ctx, "plan", func() (int, error) {
+ t.Fatal("request after cancellation")
+
+ return 0, nil
+ })
+ require.ErrorIs(t, err, context.Canceled)
+ })
+}
+
+func TestScanPlanningRetryAfterTiming(t *testing.T) {
+ t.Parallel()
+ for _, tc := range []struct {
+ name string
+ deadline time.Duration
+ wait time.Duration
+ }{
+ {name: "capped without deadline", wait: 5 * time.Second},
+ {name: "honored with deadline", deadline: 10 * time.Second,
wait: 8 * time.Second},
+ {name: "deadline interrupts wait", deadline: 2 * time.Second,
wait: 2 * time.Second},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ ctx := t.Context()
+ if tc.deadline != 0 {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithTimeout(ctx,
tc.deadline)
+ defer cancel()
+ }
+ started := time.Now()
+ attempts := 0
+ result, err := retryScanPlanning(ctx, "plan",
func() (int, error) {
+ attempts++
+ if attempts == 1 {
+ return 0,
errorResponse{statusCode: http.StatusTooManyRequests, retryAfter: "8"}
+ }
+
+ return 42, nil
+ })
+ assert.Equal(t, tc.wait, time.Since(started))
+ if tc.deadline != 0 && tc.deadline <
8*time.Second {
+ require.ErrorIs(t, err,
context.DeadlineExceeded)
+ assert.Equal(t, 1, attempts)
+ assert.Zero(t, result)
+ } else {
+ require.NoError(t, err)
+ assert.Equal(t, 2, attempts)
+ assert.Equal(t, 42, result)
+ }
+ })
+ })
+ }
+}
diff --git a/catalog/rest/scan_planning_telemetry_test.go
b/catalog/rest/scan_planning_telemetry_test.go
new file mode 100644
index 000000000..835b8210c
--- /dev/null
+++ b/catalog/rest/scan_planning_telemetry_test.go
@@ -0,0 +1,207 @@
+// 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"
+ "net/http"
+ "sync/atomic"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/attribute"
+ "go.opentelemetry.io/otel/codes"
+ sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+ "go.opentelemetry.io/otel/sdk/metric/metricdata"
+ sdktrace "go.opentelemetry.io/otel/sdk/trace"
+ "go.opentelemetry.io/otel/sdk/trace/tracetest"
+ "go.opentelemetry.io/otel/trace"
+)
+
+// Keep this test serial: it temporarily replaces the global OTel providers.
+func TestScanPlanningTelemetryWiring(t *testing.T) {
+ // Deterministic sampling also makes accidental t.Parallel calls fail
before
+ // the test can replace process-wide providers.
+ t.Setenv("OTEL_TRACES_SAMPLER", "always_on")
+ reader := sdkmetric.NewManualReader()
+ meter := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader),
+ sdkmetric.WithExemplarFilter(func(ctx context.Context) bool {
+ span := trace.SpanFromContext(ctx)
+ assert.True(t, span.IsRecording(), "metrics must be
recorded before the operation span ends")
+
+ return span.SpanContext().IsSampled()
+ }))
+ recorder := tracetest.NewSpanRecorder()
+ tracer :=
sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder))
+ oldMeter, oldTracer := otel.GetMeterProvider(), otel.GetTracerProvider()
+ otel.SetMeterProvider(meter)
+ otel.SetTracerProvider(tracer)
+ t.Cleanup(func() {
+ otel.SetMeterProvider(oldMeter)
+ otel.SetTracerProvider(oldTracer)
+ require.NoError(t, meter.Shutdown(context.Background()))
+ require.NoError(t, tracer.Shutdown(context.Background()))
+ })
+
+ var plans, polls, tasks atomic.Int32
+ cat := newScanPlanningTestCatalog(t, []endpoint{endpointPlanTableScan,
endpointFetchPlanResult, endpointFetchScanTasks, endpointCancelPlanning},
func(mux *http.ServeMux) {
+ mux.HandleFunc("POST /v1/namespaces/db/tables/tbl/plan", func(w
http.ResponseWriter, _ *http.Request) {
+ if plans.Add(1) == 1 {
+ w.WriteHeader(http.StatusServiceUnavailable)
+
+ return
+ }
+ _, _ =
w.Write([]byte(`{"status":"submitted","plan-id":"sensitive-plan"}`))
+ })
+ mux.HandleFunc("GET
/v1/namespaces/db/tables/tbl/plan/sensitive-plan", func(w http.ResponseWriter,
_ *http.Request) {
+ switch polls.Add(1) {
+ case 1:
+ w.WriteHeader(http.StatusServiceUnavailable)
+ case 2:
+ _, _ = w.Write([]byte(`{"status":"submitted"}`))
+ default:
+ _, _ = w.Write([]byte(`{"status":"completed"}`))
+ }
+ })
+ mux.HandleFunc("GET /v1/namespaces/db/tables/tbl/plan/expired",
func(w http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusNotFound)
+ _, _ =
w.Write([]byte(`{"error":{"type":"NoSuchPlanIdException","message":"sensitive-plan","code":404}}`))
+ })
+ mux.HandleFunc("POST /v1/namespaces/db/tables/tbl/tasks",
func(w http.ResponseWriter, _ *http.Request) {
+ if tasks.Add(1) == 1 {
+ w.WriteHeader(http.StatusServiceUnavailable)
+
+ return
+ }
+ w.WriteHeader(http.StatusNotFound)
+ _, _ =
w.Write([]byte(`{"error":{"type":"NoSuchPlanTaskException","message":"sensitive-task","code":404}}`))
+ })
+ mux.HandleFunc("DELETE
/v1/namespaces/db/tables/tbl/plan/sensitive-plan", func(w http.ResponseWriter,
_ *http.Request) {
+ w.WriteHeader(http.StatusNoContent)
+ })
+ })
+ ctx, parent := tracer.Tracer("test").Start(t.Context(), "caller")
+ defer parent.End()
+ ident := table.Identifier{"db", "tbl"}
+ _, err := cat.PlanTableScan(ctx, ident, PlanTableScanRequest{})
+ require.NoError(t, err)
+ _, err = cat.WaitForPlan(ctx, ident, "sensitive-plan", fastWaitOpts)
+ require.NoError(t, err)
+ _, err = cat.FetchPlanningResult(ctx, ident, "expired",
FetchPlanningResultOptions{})
+ require.ErrorIs(t, err, ErrPlanExpired)
+ _, err = cat.FetchScanTasks(ctx, ident, FetchScanTasksRequest{PlanTask:
"sensitive-task"})
+ require.ErrorIs(t, err, ErrNoSuchPlanTask)
+ require.NoError(t, cat.CancelPlanning(ctx, ident, "sensitive-plan"))
+
+ // Exercise both scanner fallback paths and confirm explicit local scans
+ // don't count as fallbacks. The full-capability catalog must not be
called.
+ meta, fs := newPlanningParityTable(t)
+ local := table.New(ident, meta, "metadata.json", func(context.Context)
(iceio.IO, error) { return fs, nil }, nil)
+ capable := table.New(ident, meta, "metadata.json",
func(context.Context) (iceio.IO, error) { return fs, nil }, cat)
+ for _, scan := range []*table.Scan{
+ local.Scan(),
+ local.Scan(table.WithScanPlanningMode(table.ScanPlanningAuto)),
+
capable.Scan(table.WithScanPlanningMode(table.ScanPlanningAuto),
table.WithSelectedFields(iceberg.LastUpdatedSequenceNumberColumnName)),
+ } {
+ _, err := scan.PlanFiles(ctx)
+ require.NoError(t, err)
+ require.NoError(t, scan.Close())
+ }
+ assert.Equal(t, int32(2), plans.Load())
+ assert.Equal(t, int32(3), polls.Load())
+ assert.Equal(t, int32(2), tasks.Load())
+
+ spans := recorder.Ended()
+ require.Len(t, spans, 7, "POST retries belong to one logical span; each
poll has its own")
+ outcomes := make(map[string]int)
+ for _, span := range spans {
+ assert.Equal(t, parent.SpanContext().SpanID(),
span.Parent().SpanID())
+ assert.Equal(t, trace.SpanKindClient, span.SpanKind())
+ assert.Equal(t, "github.com/apache/iceberg-go/scan-planning",
span.InstrumentationScope().Name)
+ require.Len(t, span.Attributes(), 1)
+ assert.Equal(t,
attribute.Key("iceberg.scan.planning.operation"), span.Attributes()[0].Key)
+ assert.Empty(t, span.Events())
+ outcome := "success"
+ if span.Status().Code == codes.Error {
+ outcome = "error"
+ assert.Equal(t, "scan planning failed",
span.Status().Description)
+ }
+ outcomes[span.Attributes()[0].Value.AsString()+"/"+outcome]++
+ }
+ assert.Equal(t, map[string]int{"plan/success": 1,
"fetch-result/success": 2, "fetch-result/error": 2, "fetch-tasks/error": 1,
"cancel/success": 1}, outcomes)
+
+ var resource metricdata.ResourceMetrics
+ require.NoError(t, reader.Collect(ctx, &resource))
+ counts := make(map[string]map[string]int64)
+ var durations uint64
+ for _, scope := range resource.ScopeMetrics {
+ for _, m := range scope.Metrics {
+ switch data := m.Data.(type) {
+ case metricdata.Sum[int64]:
+ counts[m.Name] = make(map[string]int64)
+ for _, point := range data.DataPoints {
+ op, _ :=
point.Attributes.Value(attribute.Key("iceberg.scan.planning.operation"))
+ outcome, _ :=
point.Attributes.Value(attribute.Key("iceberg.scan.planning.outcome"))
+ reason, _ :=
point.Attributes.Value(attribute.Key("iceberg.scan.planning.reason"))
+ key := op.AsString()
+ if m.Name ==
"iceberg.scan.planning.requests" {
+ assert.Equal(t, 2,
point.Attributes.Len())
+ key += "/" + outcome.AsString()
+ } else {
+ assert.Equal(t, 1,
point.Attributes.Len())
+ }
+ if m.Name ==
"iceberg.scan.planning.fallbacks" {
+ key = reason.AsString()
+ }
+ if m.Name ==
"iceberg.scan.planning.expirations" {
+ require.Len(t, point.Exemplars,
1)
+ exemplar := point.Exemplars[0]
+ var matched bool
+ for _, span := range spans {
+ if span.Name() ==
"iceberg.scan.planning."+op.AsString() && span.Status().Code == codes.Error {
+ spanID, traceID
:= span.SpanContext().SpanID(), span.SpanContext().TraceID()
+ if
string(exemplar.SpanID) == string(spanID[:]) && string(exemplar.TraceID) ==
string(traceID[:]) {
+ matched
= true
+ }
+ }
+ }
+ assert.True(t, matched,
"expiration exemplar must identify its failed operation span")
+ }
+ counts[m.Name][key] += point.Value
+ }
+ case metricdata.Histogram[float64]:
+ assert.Equal(t, "ms", m.Unit)
+ for _, point := range data.DataPoints {
+ durations += point.Count
+ }
+ }
+ }
+ }
+ assert.Equal(t, map[string]map[string]int64{
+ "iceberg.scan.planning.requests": {"plan/success": 1,
"fetch-result/success": 2, "fetch-result/error": 2, "fetch-tasks/error": 1,
"cancel/success": 1},
+ "iceberg.scan.planning.retries": {"plan": 1,
"fetch-result": 1, "fetch-tasks": 1},
+ "iceberg.scan.planning.expirations": {"fetch-result": 1,
"fetch-tasks": 1},
+ "iceberg.scan.planning.fallbacks": {"capability": 1,
"row-lineage": 1},
+ }, counts)
+ assert.Equal(t, uint64(7), durations)
+}
diff --git a/expr_json_test.go b/expr_json_test.go
index 64a4519a1..69a4c0fed 100644
--- a/expr_json_test.go
+++ b/expr_json_test.go
@@ -668,3 +668,22 @@ func TestMarshalBoundNestedReferencePreservesPath(t
*testing.T) {
})
}
}
+
+func TestUnboundTransformRejectsWrappedReference(t *testing.T) {
+ t.Parallel()
+ // External wrappers can embed Reference, but binding
+ // rejects them before they can reach BoundTransform.MarshalJSON.
+ type wrappedReference struct {
+ iceberg.Reference
+ }
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "Nested", Type:
&iceberg.StructType{FieldList: []iceberg.NestedField{
+ {ID: 2, Name: "Value", Type:
iceberg.PrimitiveTypes.Int32},
+ }}},
+ )
+ term := iceberg.NewUnboundTransform(iceberg.BucketTransform{NumBuckets:
16},
+ wrappedReference{iceberg.Reference("Nested.Value")})
+ _, err := term.Bind(schema, true)
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ require.ErrorContains(t, err, "transform terms must wrap a direct
reference")
+}
diff --git a/go.mod b/go.mod
index 3169b1a21..09addd984 100644
--- a/go.mod
+++ b/go.mod
@@ -62,7 +62,9 @@ require (
github.com/uptrace/bun/extra/bundebug v1.2.18
go.opentelemetry.io/otel v1.46.0
go.opentelemetry.io/otel/metric v1.46.0
+ go.opentelemetry.io/otel/sdk v1.46.0
go.opentelemetry.io/otel/sdk/metric v1.46.0
+ go.opentelemetry.io/otel/trace v1.46.0
gocloud.dev v0.46.0
golang.org/x/exp v0.0.0-20260603202125-055de637280b
golang.org/x/oauth2 v0.36.0
@@ -257,8 +259,6 @@ require (
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.45.0
// indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0
// indirect
- go.opentelemetry.io/otel/sdk v1.46.0 // indirect
- go.opentelemetry.io/otel/trace v1.46.0 // indirect
go.opentelemetry.io/proto/otlp v1.11.0 // indirect
go.yaml.in/yaml/v3 v3.0.5 // indirect
go.yaml.in/yaml/v4 v4.0.0-rc.6 // indirect
diff --git a/internal/scanmetrics/telemetry.go
b/internal/scanmetrics/telemetry.go
new file mode 100644
index 000000000..8dd63ac97
--- /dev/null
+++ b/internal/scanmetrics/telemetry.go
@@ -0,0 +1,84 @@
+// 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 scanmetrics instruments client-side scan planning. Attribute values
+// are bounded operation/outcome names; filters, credentials, table names, plan
+// tokens, paths, and server error messages are never exported.
+package scanmetrics
+
+import (
+ "context"
+ "time"
+
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/attribute"
+ "go.opentelemetry.io/otel/codes"
+ "go.opentelemetry.io/otel/metric"
+ "go.opentelemetry.io/otel/trace"
+)
+
+const scope = "github.com/apache/iceberg-go/scan-planning"
+
+// Start traces a logical operation, including its retries. Instruments come
+// from the application's global provider and are no-ops without an SDK.
+func Start(ctx context.Context, operation string) (context.Context,
func(error)) {
+ ctx, span := otel.Tracer(scope).Start(ctx,
"iceberg.scan.planning."+operation,
+ trace.WithSpanKind(trace.SpanKindClient),
+
trace.WithAttributes(attribute.String("iceberg.scan.planning.operation",
operation)))
+ start := time.Now()
+
+ return ctx, func(err error) {
+ outcome := "success"
+ if err != nil {
+ outcome = "error"
+ // Error strings can include filters, paths, or opaque
plan tokens.
+ span.SetStatus(codes.Error, "scan planning failed")
+ }
+ attrs := metric.WithAttributes(
+ attribute.String("iceberg.scan.planning.operation",
operation),
+ attribute.String("iceberg.scan.planning.outcome",
outcome))
+ meter := otel.Meter(scope)
+ if count, e :=
meter.Int64Counter("iceberg.scan.planning.requests"); e == nil {
+ count.Add(ctx, 1, attrs)
+ }
+ if duration, e :=
meter.Float64Histogram("iceberg.scan.planning.request.duration",
metric.WithUnit("ms")); e == nil {
+ duration.Record(ctx,
float64(time.Since(start))/float64(time.Millisecond), attrs)
+ }
+ span.End()
+ }
+}
+
+// Retry records an additional attempt after a transient HTTP failure.
+func Retry(ctx context.Context, operation string) {
+ count(ctx, "retries", "operation", operation)
+}
+
+// Expired records a recognized expired plan or plan-task response.
+func Expired(ctx context.Context, operation string) {
+ count(ctx, "expirations", "operation", operation)
+}
+
+// Fallback records auto mode choosing local planning before submitting a plan.
+func Fallback(ctx context.Context, reason string) {
+ count(ctx, "fallbacks", "reason", reason)
+}
+
+func count(ctx context.Context, name, key, value string) {
+ if counter, err :=
otel.Meter(scope).Int64Counter("iceberg.scan.planning." + name); err == nil {
+ counter.Add(ctx, 1,
metric.WithAttributes(attribute.String("iceberg.scan.planning."+key, value)))
+ }
+}
diff --git a/internal/scanmetrics/telemetry_test.go
b/internal/scanmetrics/telemetry_test.go
new file mode 100644
index 000000000..35513a494
--- /dev/null
+++ b/internal/scanmetrics/telemetry_test.go
@@ -0,0 +1,96 @@
+// 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 scanmetrics
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/codes"
+ sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+ "go.opentelemetry.io/otel/sdk/metric/metricdata"
+ sdktrace "go.opentelemetry.io/otel/sdk/trace"
+ "go.opentelemetry.io/otel/sdk/trace/tracetest"
+)
+
+func TestPlanningTelemetry(t *testing.T) {
+ // Provider replacement is process-wide; Setenv also prevents
t.Parallel.
+ t.Setenv("OTEL_TRACES_SAMPLER", "always_on")
+ reader := sdkmetric.NewManualReader()
+ meter := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+ recorder := tracetest.NewSpanRecorder()
+ tracer :=
sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder))
+ oldMeter, oldTracer := otel.GetMeterProvider(), otel.GetTracerProvider()
+ otel.SetMeterProvider(meter)
+ otel.SetTracerProvider(tracer)
+ t.Cleanup(func() {
+ otel.SetMeterProvider(oldMeter)
+ otel.SetTracerProvider(oldTracer)
+ require.NoError(t, meter.Shutdown(context.Background()))
+ require.NoError(t, tracer.Shutdown(context.Background()))
+ })
+
+ ctx, finish := Start(t.Context(), "plan")
+ Retry(ctx, "plan")
+ finish(nil)
+ ctx, finish = Start(t.Context(), "fetch-result")
+ Expired(ctx, "fetch-result")
+ finish(errors.New("secret token and sensitive filter value"))
+ Fallback(t.Context(), "capability")
+
+ spans := recorder.Ended()
+ require.Len(t, spans, 2)
+ assert.Equal(t, "iceberg.scan.planning.plan", spans[0].Name())
+ assert.Equal(t, codes.Unset, spans[0].Status().Code)
+ assert.Equal(t, codes.Error, spans[1].Status().Code)
+ assert.Equal(t, "scan planning failed", spans[1].Status().Description)
+ assert.Empty(t, spans[1].Events(), "raw errors must not be exported")
+ require.Len(t, spans[1].Attributes(), 1)
+ assert.Equal(t, "fetch-result",
spans[1].Attributes()[0].Value.AsString())
+
+ var resource metricdata.ResourceMetrics
+ require.NoError(t, reader.Collect(t.Context(), &resource))
+ counts := make(map[string]int64)
+ durationCount := uint64(0)
+ for _, scope := range resource.ScopeMetrics {
+ for _, m := range scope.Metrics {
+ switch data := m.Data.(type) {
+ case metricdata.Sum[int64]:
+ for _, point := range data.DataPoints {
+ counts[m.Name] += point.Value
+ }
+ case metricdata.Histogram[float64]:
+ assert.Equal(t, "ms", m.Unit)
+ for _, point := range data.DataPoints {
+ durationCount += point.Count
+ }
+ }
+ }
+ }
+ assert.Equal(t, map[string]int64{
+ "iceberg.scan.planning.requests": 2,
+ "iceberg.scan.planning.retries": 1,
+ "iceberg.scan.planning.expirations": 1,
+ "iceberg.scan.planning.fallbacks": 1,
+ }, counts)
+ assert.Equal(t, uint64(2), durationCount)
+}
diff --git a/table/scan_splits_test.go b/table/scan_splits_test.go
index 6d3e08024..b7d9937a9 100644
--- a/table/scan_splits_test.go
+++ b/table/scan_splits_test.go
@@ -237,6 +237,25 @@ func TestSplitParquetScanTaskKeepsUnsafeTasksIntact(t
*testing.T) {
}
}
+func TestSplitRemoteScanTasksPreservesPartialTasks(t *testing.T) {
+ t.Parallel()
+ file := splitTestDataFile(t, iceberg.ParquetFile, 100, []int64{8, 40,
70})
+ partial := []FileScanTask{
+ {File: file, Start: 0, Length: 40, Residual:
iceberg.AlwaysTrue{}},
+ {File: file, Start: 40, Length: 60, Residual:
iceberg.AlwaysFalse{}},
+ }
+ got := splitRemoteScanTasks(partial, 20)
+ assert.Equal(t, partial, got)
+ assert.Same(t, &partial[0], &got[0], "pass-through should retain the
original slice")
+
+ // Partial tasks must also survive when another file causes a new slice.
+ mixed := []FileScanTask{partial[0], {File: file, Start: 0, Length:
100}, partial[1]}
+ got = splitRemoteScanTasks(mixed, 20)
+ require.Len(t, got, 5)
+ assert.Equal(t, partial[0], got[0])
+ assert.Equal(t, partial[1], got[len(got)-1])
+}
+
func TestAdjustParquetTaskRangeToPhysicalFileSize(t *testing.T) {
file := splitTestDataFile(t, iceberg.ParquetFile, 100, []int64{8, 40,
70})
tests := []struct {
diff --git a/table/scanner.go b/table/scanner.go
index 713867b5e..e86c811b4 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -34,6 +34,7 @@ import (
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/iceberg-go"
iceberginternal "github.com/apache/iceberg-go/internal"
+ "github.com/apache/iceberg-go/internal/scanmetrics"
"github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/metrics"
"golang.org/x/sync/errgroup"
@@ -1630,8 +1631,11 @@ func (scan *Scan) planFiles(ctx context.Context,
projectScanColumns bool) ([]Fil
case ScanPlanningRemote:
return scan.planFilesRemote(ctx)
case ScanPlanningAuto:
- if supportsAutomaticRemotePlanning(scan.planner) &&
- !scan.requiresLastUpdatedSequenceNumber() {
+ if !supportsAutomaticRemotePlanning(scan.planner) {
+ scanmetrics.Fallback(ctx, "capability")
+ } else if scan.requiresLastUpdatedSequenceNumber() {
+ scanmetrics.Fallback(ctx, "row-lineage")
+ } else {
return scan.planFilesRemote(ctx)
}
case ScanPlanningLocal:
