This is an automated email from the ASF dual-hosted git repository.
zeroshade 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 9dd8d308f feat(table,rest): enforce the catalog's scan-planning-mode
directive (#2162)
9dd8d308f is described below
commit 9dd8d308f6fa2119eccb4b7393b7dfc153a19b45
Author: Revanth <[email protected]>
AuthorDate: Wed Oct 21 12:26:55 2026 -0500
feat(table,rest): enforce the catalog's scan-planning-mode directive (#2162)
* fix(table): add ScanPlanningDirective() accessor on table
Previously, the table returned by a LoadTable() on a REST catalog did
not have any way of exposing what scan planning mode the catalog is
trying to enforce on this table. In cases where the catalog is enforcing
server-side scan planning, the caller would have to rely on errors
and/or missing files through the Scan() interface, to realize that
server-side scan planning is necessary.
This change exposes a ScanPlanningDirective() function on a table.Table
that exposes this value as returned from LoadTableResponse. If the catalog
is not imposing any particular scan planning mode,
ScanPlanningDirectiveUnknown is returned.
* fix(table): carry scan-planning directive state explicitly
Address review feedback on the ScanPlanningDirective accessor:
- WithScanPlanningDirective now takes (value, ok) so absence is explicit
and a comma-ok map read can be passed straight through.
- Add WithScanPlanningDirectiveFrom to copy a table's directive verbatim;
RefreshTableCredentials uses it instead of reconstructing the raw value
from SavedConfig, which could report the wrong directive for externally
built tables.
- Drop the positional loadConfig parameter from tableFromResponse; load
paths pass the directive as a caller option instead.
- Rename ScanPlanningDirectiveUnknown to ScanPlanningDirectiveNone and
document that it only means "no directive" with a nil error.
- Use a switch over the valid directives.
- Add coverage for Commit, StagedTable, Refresh, CreateTable,
RegisterTable, and LoadTable followed by a credential refresh.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* feat(table,rest): enforce the catalog's scan-planning-mode directive
* fix(table,rest): fail closed on UpdateTable and incremental scans under a
scan-planning directive
* fix(table,rest): tolerate a nil directive source and pin UpdateTable's
directive
---------
Co-authored-by: Bilal Akhtar <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
README.md | 36 ++-
catalog/rest/refresh_table_credentials_test.go | 111 ++++++++
catalog/rest/rest.go | 53 +++-
catalog/rest/rest_test.go | 370 +++++++++++++++++++++++++
catalog/rest/scan_planning_directive_test.go | 101 +++++++
table/incremental_append_scan.go | 13 +-
table/incremental_changelog_scan.go | 11 +-
table/scan_planning.go | 188 ++++++++++++-
table/scan_planning_directive_wiring_test.go | 160 +++++++++++
table/scan_planning_test.go | 223 +++++++++++++++
table/scanner.go | 18 +-
table/table.go | 30 +-
table/transaction.go | 9 +-
13 files changed, 1267 insertions(+), 56 deletions(-)
diff --git a/README.md b/README.md
index 01ec83776..7e3a034e0 100644
--- a/README.md
+++ b/README.md
@@ -138,14 +138,34 @@ 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.
+Scans follow the catalog's `scan-planning-mode` directive (`client`/`server`).
+The directive is resolved with the same precedence as the Java client: the
+table-load response's `config` block wins, and the catalog's own
+`scan-planning-mode` property (including `/v1/config` defaults and overrides)
+applies when the server sends none. A mismatch between the two is logged.
+Read it with `tbl.ScanPlanningDirective()`.
+
+| Directive | Scan option | Result |
+| :-------- | :---------- | :----- |
+| `server` | none, `auto`, or `remote` | Plan remotely. |
+| `server` | `local` | Error. |
+| `client` | none, `auto`, or `local` | Plan locally. |
+| `client` | `remote` | Error. |
+| none | any | As described above. |
+
+Some behaviors are specific to this client. The directive is checked when a
+scan is planned, not when the table is loaded, so a `server` table whose
+catalog does not advertise the plan endpoint can still be loaded and committed
+to; planning it returns an error. An unrecognized directive value likewise
+fails only at planning. An explicit scan option that conflicts with the
+directive is rejected rather than overridden. And incremental append and
+changelog scans, which only plan locally, return an error on a `server` table,
+where the Java client plans them locally.
+
+Scans inside a transaction plan staged metadata locally and do not apply the
+directive. Commit responses carry no table
+config, so a table returned by `UpdateTable` keeps only the catalog's own
+`scan-planning-mode` property; reload it to obtain a per-table directive.
```go
// tbl is a table loaded from a rest.Catalog.
diff --git a/catalog/rest/refresh_table_credentials_test.go
b/catalog/rest/refresh_table_credentials_test.go
index 5f46bc899..7dc37b9be 100644
--- a/catalog/rest/refresh_table_credentials_test.go
+++ b/catalog/rest/refresh_table_credentials_test.go
@@ -275,6 +275,117 @@ func
TestRefreshTableCredentialsPreservesMetricsReporter(t *testing.T) {
assert.Same(t, reporter, refreshed.MetricsReporter())
}
+// TestRefreshTableCredentialsPreservesScanPlanningDirective checks that the
+// load-time scan-planning directive survives a credential refresh, whose
+// response carries no table config to re-derive it from. The directive is
+// copied from the table, never reconstructed from its saved config.
+func TestRefreshTableCredentialsPreservesScanPlanningDirective(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ opts []table.Option
+ want table.ScanPlanningDirective
+ wantErr string
+ }{
+ {name: "absent", want: table.ScanPlanningDirectiveNone},
+ {
+ name: "server",
+ opts:
[]table.Option{table.WithScanPlanningDirective("server", true)},
+ want: table.ScanPlanningDirectiveServer,
+ },
+ {
+ name: "unrecognized",
+ opts:
[]table.Option{table.WithScanPlanningDirective("hybrid", true)},
+ wantErr: `"hybrid"`,
+ },
+ {
+ // A saved config built from metadata properties, as
the method's doc
+ // tells external callers to pass, must not leak into
the directive.
+ name: "unrecognized_with_saved_metadata_property",
+ opts: []table.Option{
+ table.WithScanPlanningDirective("hybrid", true),
+
table.WithSavedConfig(iceberg.Properties{table.ScanPlanningModeKey: "server"}),
+ },
+ wantErr: `"hybrid"`,
+ },
+ {
+ name: "absent_with_saved_metadata_property",
+ opts: []table.Option{
+
table.WithSavedConfig(iceberg.Properties{table.ScanPlanningModeKey: "server"}),
+ },
+ want: table.ScanPlanningDirectiveNone,
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ scheme := "restcreds-directive-" +
strings.ReplaceAll(tc.name, "_", "-")
+
+ cat := newCredsTestCatalog(t, credsCatalogOpts{
+ creds: func(w http.ResponseWriter, _
*http.Request) {
+ assert.NoError(t,
json.NewEncoder(w).Encode(storageCredentialsBody(
+
scheme+"://warehouse/database/table",
+
map[string]string{"s3.access-key-id": "vended-key"},
+ )))
+ },
+ })
+
+ tbl, _ := newExternalTable(t, cat, scheme, tc.opts...)
+
+ refreshed, err :=
cat.RefreshTableCredentials(context.Background(), tbl)
+ require.NoError(t, err)
+ require.NotSame(t, tbl, refreshed)
+
+ got, err := refreshed.ScanPlanningDirective()
+ if tc.wantErr != "" {
+ require.ErrorIs(t, err,
iceberg.ErrInvalidArgument)
+ assert.Contains(t, err.Error(), tc.wantErr)
+
+ return
+ }
+ require.NoError(t, err)
+ assert.Equal(t, tc.want, got)
+ })
+ }
+}
+
+// TestRefreshTableCredentialsAfterLoadTableKeepsScanPlanningDirective covers
+// the directive end to end: an unrecognized value from the load response's
+// config block must still be reported verbatim after a credential refresh.
+func TestRefreshTableCredentialsAfterLoadTableKeepsScanPlanningDirective(t
*testing.T) {
+ const scheme = "restcreds-loaded-directive"
+
+ metadataLoc := scheme +
"://warehouse/database/table/metadata/00000-a.metadata.json"
+ cat := newCredsTestCatalog(t, credsCatalogOpts{
+ loadTable: func(w http.ResponseWriter, req *http.Request) {
+ assert.Equal(t, http.MethodGet, req.Method)
+ assert.NoError(t,
json.NewEncoder(w).Encode(map[string]any{
+ "metadata-location": metadataLoc,
+ "metadata": json.RawMessage(strings.ReplaceAll(
+ exampleTableMetadataNoSnapshotV1,
"s3://", scheme+"://")),
+ "config":
map[string]string{table.ScanPlanningModeKey: "Hybrid"},
+ }))
+ },
+ creds: func(w http.ResponseWriter, _ *http.Request) {
+ assert.NoError(t,
json.NewEncoder(w).Encode(storageCredentialsBody(
+ scheme+"://warehouse/database/table",
+ map[string]string{iceio.S3AccessKeyID:
"vended-key"},
+ )))
+ },
+ })
+
+ tbl, err := cat.LoadTable(context.Background(),
catalog.ToIdentifier("db", "tbl"))
+ require.NoError(t, err)
+ _, err = tbl.ScanPlanningDirective()
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ require.Contains(t, err.Error(), `"Hybrid"`)
+
+ refreshed, err := cat.RefreshTableCredentials(context.Background(), tbl)
+ require.NoError(t, err)
+ require.NotSame(t, tbl, refreshed)
+
+ _, err = refreshed.ScanPlanningDirective()
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ assert.Contains(t, err.Error(), `"Hybrid"`)
+}
+
// TestRefreshTableCredentialsAfterLoadTable covers a table loaded through
// LoadTable: the per-table config block of the load response is in neither the
// catalog's props nor the table's metadata properties, so only the table's
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index 2d6db178b..fcd5d8aa7 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -34,6 +34,7 @@ import (
"strconv"
"strings"
"sync"
+ "sync/atomic"
"time"
"github.com/apache/iceberg-go"
@@ -935,6 +936,9 @@ type Catalog struct {
// from reporterProps) and the server advertises the endpoint. Owned
here and
// closed by Close.
metricsDispatcher *metricsDispatcher
+ // scanPlanningMismatchWarned limits the scan-planning-mode mismatch
warning
+ // to once per catalog; see scanPlanningDirective.
+ scanPlanningMismatchWarned atomic.Bool
}
func newCatalogFromProps(ctx context.Context, name string, uri string, p
iceberg.Properties) (*Catalog, error) {
@@ -1394,6 +1398,33 @@ func (r *Catalog) tableFromResponse(
), nil
}
+// scanPlanningDirective resolves a table's raw scan-planning-mode directive
and
+// whether one is present, in the shape table.WithScanPlanningDirective takes.
+// As in Java's RESTSessionCatalog, the server's table config takes precedence
+// and the catalog's own scan-planning-mode property applies when the server
+// sends none; table metadata is never consulted. r.props already folds in the
+// /v1/config defaults and overrides. A mismatch between the two is logged as a
+// warning once per catalog and at debug level after that, since Refresh
+// reloads the table. A nil loadConfig yields the catalog property alone.
+func (r *Catalog) scanPlanningDirective(ctx context.Context, identifier
[]string, loadConfig iceberg.Properties) (string, bool) {
+ server, hasServer := loadConfig[table.ScanPlanningModeKey]
+ client, hasClient := r.props[table.ScanPlanningModeKey]
+ if !hasServer {
+ return client, hasClient
+ }
+
+ if hasClient && !strings.EqualFold(client, server) {
+ level := slog.LevelDebug
+ if r.scanPlanningMismatchWarned.CompareAndSwap(false, true) {
+ level = slog.LevelWarn
+ }
+ slog.Log(ctx, level, "iceberg: scan-planning-mode mismatch
between catalog and server; using the server value",
+ "table", strings.Join(identifier, "."), "client",
client, "server", server)
+ }
+
+ return server, true
+}
+
func (r *Catalog) fetchTableCreds(ctx context.Context, ident []string,
location string) (iceberg.Properties, error) {
if err := r.endpoints.check(endpointTableCredentials); err != nil {
return nil, err
@@ -1445,9 +1476,14 @@ func (r *Catalog) RefreshTableCredentials(ctx
context.Context, tbl *table.Table)
scanCfg := maps.Clone(config)
mergeVendedCredentials(config, resp, metadataLoc)
+ opts := []table.Option{
+ table.WithSavedConfig(tbl.SavedConfig()),
+ // Credential refresh does not return table config, so carry the
+ // existing scan-planning directive over verbatim.
+ table.WithScanPlanningDirectiveFrom(tbl),
+ }
// Keep a reporter the caller set on the table rather than reverting it
to
// the catalog default, as Refresh does.
- opts := []table.Option{table.WithSavedConfig(tbl.SavedConfig())}
if reporter := tbl.MetricsReporter(); !metrics.IsNop(reporter) {
opts = append(opts, table.WithMetricsReporter(reporter))
}
@@ -1725,7 +1761,7 @@ func (r *Catalog) CreateTable(ctx context.Context,
identifier table.Identifier,
vended := resolveStorageCredentials(ret.StorageCredentials,
ret.MetadataLoc)
mergeVendedCredentials(config, vended, ret.MetadataLoc)
- return r.tableFromResponse(ctx, identifier, ret.Metadata,
ret.MetadataLoc, config, scanPlanningConfig, credsVended, vended, ret.Labels,
table.WithSavedConfig(saved))
+ return r.tableFromResponse(ctx, identifier, ret.Metadata,
ret.MetadataLoc, config, scanPlanningConfig, credsVended, vended, ret.Labels,
table.WithSavedConfig(saved),
table.WithScanPlanningDirective(r.scanPlanningDirective(ctx, identifier,
ret.Config)))
}
// commitStagedCreate performs the second phase of a staged table
@@ -1954,7 +1990,7 @@ func (r *Catalog) RegisterTable(ctx context.Context,
identifier table.Identifier
vended := resolveStorageCredentials(ret.StorageCredentials,
ret.MetadataLoc)
mergeVendedCredentials(config, vended, ret.MetadataLoc)
- return r.tableFromResponse(ctx, identifier, ret.Metadata,
ret.MetadataLoc, config, scanPlanningConfig, credsVended, vended, ret.Labels,
table.WithSavedConfig(saved))
+ return r.tableFromResponse(ctx, identifier, ret.Metadata,
ret.MetadataLoc, config, scanPlanningConfig, credsVended, vended, ret.Labels,
table.WithSavedConfig(saved),
table.WithScanPlanningDirective(r.scanPlanningDirective(ctx, identifier,
ret.Config)))
}
// LoadTable loads a table from the catalog. It implements [catalog.Catalog].
@@ -2017,7 +2053,7 @@ func (r *Catalog) loadTableWithMode(ctx context.Context,
identifier table.Identi
vended := resolveStorageCredentials(ret.StorageCredentials,
ret.MetadataLoc)
mergeVendedCredentials(config, vended, ret.MetadataLoc)
- return r.tableFromResponse(ctx, identifier, ret.Metadata,
ret.MetadataLoc, config, scanPlanningConfig, credsVended, vended, ret.Labels,
table.WithSavedConfig(saved))
+ return r.tableFromResponse(ctx, identifier, ret.Metadata,
ret.MetadataLoc, config, scanPlanningConfig, credsVended, vended, ret.Labels,
table.WithSavedConfig(saved),
table.WithScanPlanningDirective(r.scanPlanningDirective(ctx, identifier,
ret.Config)))
}
func (r *Catalog) UpdateTable(ctx context.Context, ident table.Identifier,
requirements []table.Requirement, updates []table.Update) (*table.Table, error)
{
@@ -2065,8 +2101,13 @@ func (r *Catalog) UpdateTable(ctx context.Context, ident
table.Identifier, requi
config := maps.Clone(r.props)
maps.Copy(config, metadata.Properties())
- // A commit response carries no labels (they are load-time enrichment).
- return r.tableFromResponse(ctx, ident, metadata, ret.MetadataLoc,
config, config, false, nil, nil, table.WithSavedConfig(metadata.Properties()))
+ // A commit response carries no labels or table config (they are
load-time
+ // enrichment). The catalog's own scan-planning-mode property still
applies,
+ // so the returned table enforces a catalog-wide directive; a per-table
+ // directive from the server's load config needs a reload.
+ return r.tableFromResponse(ctx, ident, metadata, ret.MetadataLoc,
config, config, false, nil, nil,
+ table.WithSavedConfig(metadata.Properties()),
+ table.WithScanPlanningDirective(r.scanPlanningDirective(ctx,
ident, nil)))
}
func (r *Catalog) DropTable(ctx context.Context, identifier table.Identifier)
error {
diff --git a/catalog/rest/rest_test.go b/catalog/rest/rest_test.go
index d04c958de..ee38c1a7c 100644
--- a/catalog/rest/rest_test.go
+++ b/catalog/rest/rest_test.go
@@ -1773,6 +1773,184 @@ func (r *RestCatalogSuite) TestLoadTableLabels() {
r.Nil(labels.Field(1))
// field present in schema but unlabeled
}
+func (r *RestCatalogSuite) TestLoadTableScanPlanningDirective() {
+ for _, tc := range []struct {
+ name string
+ config string
+ properties string
+ catOpts []rest.Option
+ want table.ScanPlanningDirective
+ wantErr bool
+ }{
+ {name: "absent", config: `{}`, want:
table.ScanPlanningDirectiveNone},
+ {name: "client", config: `{"scan-planning-mode": "client"}`,
want: table.ScanPlanningDirectiveClient},
+ {name: "server", config: `{"scan-planning-mode": "server"}`,
want: table.ScanPlanningDirectiveServer},
+ {name: "unrecognized", config: `{"scan-planning-mode":
"bogus"}`, want: table.ScanPlanningDirectiveNone, wantErr: true},
+ // Table metadata properties never carry the directive.
+ {name: "metadata_property_only", config: `{}`, properties:
`{"scan-planning-mode": "server"}`, want: table.ScanPlanningDirectiveNone},
+ // As in Java, the catalog's own property applies when the
server sends
+ // none, and the server's value wins when both are present.
+ {
+ name: "catalog_property_only",
+ config: `{}`,
+ catOpts:
[]rest.Option{rest.WithAdditionalProps(iceberg.Properties{"scan-planning-mode":
"client"})},
+ want: table.ScanPlanningDirectiveClient,
+ },
+ {
+ name: "config_wins_over_catalog_property",
+ config: `{"scan-planning-mode": "server"}`,
+ catOpts:
[]rest.Option{rest.WithAdditionalProps(iceberg.Properties{"scan-planning-mode":
"client"})},
+ want: table.ScanPlanningDirectiveServer,
+ },
+ {
+ name: "config_wins_over_metadata_property",
+ config: `{"scan-planning-mode": "client"}`,
+ properties: `{"scan-planning-mode": "server"}`,
+ want: table.ScanPlanningDirectiveClient,
+ },
+ } {
+ r.Run(tc.name, func() {
+ tableName := "table_" + tc.name
+ properties := tc.properties
+ if properties == "" {
+ properties = `{}`
+ }
+
r.mux.HandleFunc("/v1/namespaces/fokko/tables/"+tableName, func(w
http.ResponseWriter, req *http.Request) {
+ r.Require().Equal(http.MethodGet, req.Method)
+ w.Write([]byte(`{
+ "metadata-location":
"s3://warehouse/database/table/metadata/00001.metadata.json",
+ "metadata": {
+ "format-version": 1,
+ "table-uuid":
"b55d9dda-6561-423a-8bfc-787980ce421f",
+ "location":
"s3://warehouse/database/table",
+ "last-updated-ms":
1646787054459,
+ "last-column-id": 2,
+ "schema":
{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"id","required":false,"type":"int"},{"id":2,"name":"data","required":false,"type":"string"}]},
+ "current-schema-id": 0,
+ "schemas":
[{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"id","required":false,"type":"int"},{"id":2,"name":"data","required":false,"type":"string"}]}],
+ "partition-spec": [],
+ "default-spec-id": 0,
+ "partition-specs":
[{"spec-id":0,"fields":[]}],
+ "last-partition-id": 999,
+ "default-sort-order-id": 0,
+ "sort-orders":
[{"order-id":0,"fields":[]}],
+ "properties": ` + properties + `
+ },
+ "config": ` + tc.config + `
+ }`))
+ })
+
+ cat, err := rest.NewCatalog(context.Background(),
"rest", r.srv.URL,
+
append([]rest.Option{rest.WithOAuthToken(TestToken)}, tc.catOpts...)...)
+ r.Require().NoError(err)
+
+ tbl, err := cat.LoadTable(context.Background(),
catalog.ToIdentifier("fokko", tableName))
+ r.Require().NoError(err)
+ got, err := tbl.ScanPlanningDirective()
+ if tc.wantErr {
+ r.ErrorIs(err, iceberg.ErrInvalidArgument)
+ r.Equal(tc.want, got)
+
+ return
+ }
+ r.Require().NoError(err)
+ r.Equal(tc.want, got)
+ })
+ }
+}
+
+// TestCreateAndRegisterTableScanPlanningDirective pins that the other load
+// response paths read the directive from the config block too.
+func (r *RestCatalogSuite) TestCreateAndRegisterTableScanPlanningDirective() {
+ const metadataLoc =
"s3://warehouse/database/table/metadata/00001.metadata.json"
+
+ for _, op := range []struct {
+ name string
+ path string
+ call func(cat *rest.Catalog, ident table.Identifier)
(*table.Table, error)
+ }{
+ {
+ name: "create",
+ path: "/v1/namespaces/fokko/tables",
+ call: func(cat *rest.Catalog, ident table.Identifier)
(*table.Table, error) {
+ return cat.CreateTable(context.Background(),
ident, tableSchemaSimple)
+ },
+ },
+ {
+ name: "register",
+ path: "/v1/namespaces/fokko/register",
+ call: func(cat *rest.Catalog, ident table.Identifier)
(*table.Table, error) {
+ return cat.RegisterTable(context.Background(),
ident, metadataLoc)
+ },
+ },
+ } {
+ for _, tc := range []struct {
+ name string
+ config string
+ properties string
+ want table.ScanPlanningDirective
+ wantErr bool
+ }{
+ {name: "absent", config: `{}`, properties: `{}`, want:
table.ScanPlanningDirectiveNone},
+ {name: "server", config: `{"scan-planning-mode":
"server"}`, properties: `{}`, want: table.ScanPlanningDirectiveServer},
+ {name: "unrecognized", config: `{"scan-planning-mode":
"bogus"}`, properties: `{}`, wantErr: true},
+ {name: "metadata_property_only", config: `{}`,
properties: `{"scan-planning-mode": "server"}`, want:
table.ScanPlanningDirectiveNone},
+ } {
+ r.Run(op.name+"_"+tc.name, func() {
+ mux := http.NewServeMux()
+ srv := httptest.NewServer(mux)
+ defer srv.Close()
+
+ mux.HandleFunc("/v1/config", func(w
http.ResponseWriter, _ *http.Request) {
+
json.NewEncoder(w).Encode(map[string]any{
+ "defaults": map[string]any{},
+ "overrides": map[string]any{},
+ "endpoints":
rest.AllEndpointStrings,
+ })
+ })
+ mux.HandleFunc(op.path, func(w
http.ResponseWriter, req *http.Request) {
+ r.Require().Equal(http.MethodPost,
req.Method)
+ w.Write([]byte(`{
+ "metadata-location": "` +
metadataLoc + `",
+ "metadata": {
+ "format-version": 1,
+ "table-uuid":
"b55d9dda-6561-423a-8bfc-787980ce421f",
+ "location":
"s3://warehouse/database/table",
+ "last-updated-ms":
1646787054459,
+ "last-column-id": 2,
+ "schema":
{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"id","required":false,"type":"int"},{"id":2,"name":"data","required":false,"type":"string"}]},
+ "current-schema-id": 0,
+ "schemas":
[{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"id","required":false,"type":"int"},{"id":2,"name":"data","required":false,"type":"string"}]}],
+ "partition-spec": [],
+ "default-spec-id": 0,
+ "partition-specs":
[{"spec-id":0,"fields":[]}],
+ "last-partition-id":
999,
+
"default-sort-order-id": 0,
+ "sort-orders":
[{"order-id":0,"fields":[]}],
+ "properties": ` +
tc.properties + `
+ },
+ "config": ` + tc.config + `
+ }`))
+ })
+
+ cat, err :=
rest.NewCatalog(context.Background(), "rest", srv.URL,
rest.WithOAuthToken(TestToken))
+ r.Require().NoError(err)
+
+ tbl, err := op.call(cat,
catalog.ToIdentifier("fokko", "fokko2"))
+ r.Require().NoError(err)
+ got, err := tbl.ScanPlanningDirective()
+ if tc.wantErr {
+ r.ErrorIs(err,
iceberg.ErrInvalidArgument)
+
+ return
+ }
+ r.Require().NoError(err)
+ r.Equal(tc.want, got)
+ })
+ }
+ }
+}
+
func (r *RestCatalogSuite) TestCreateTableLabels() {
r.mux.HandleFunc("/v1/namespaces/fokko/tables", func(w
http.ResponseWriter, req *http.Request) {
if !r.Equal(http.MethodPost, req.Method) {
@@ -4489,3 +4667,195 @@ func TestEndpointNegotiation(t *testing.T) {
}
assert.False(t, tablesHit)
}
+
+// directiveTestMetadata is minimal table metadata for the scan-planning
+// directive tests.
+const directiveTestMetadata = `{
+ "format-version": 1,
+ "table-uuid": "b55d9dda-6561-423a-8bfc-787980ce421f",
+ "location": "s3://warehouse/database/table",
+ "last-updated-ms": 1646787054459,
+ "last-column-id": 2,
+ "schema":
{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"id","required":false,"type":"int"},{"id":2,"name":"data","required":false,"type":"string"}]},
+ "current-schema-id": 0,
+ "schemas":
[{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"id","required":false,"type":"int"},{"id":2,"name":"data","required":false,"type":"string"}]}],
+ "partition-spec": [],
+ "default-spec-id": 0,
+ "partition-specs": [{"spec-id":0,"fields":[]}],
+ "last-partition-id": 999,
+ "default-sort-order-id": 0,
+ "sort-orders": [{"order-id":0,"fields":[]}],
+ "properties": {}
+}`
+
+// A server directive without the plan endpoint must not block loading or
+// metadata-only use of the table; only planning a scan fails (#2102).
+func TestLoadTableServerDirectiveWithoutPlanEndpoint(t *testing.T) {
+ planning := map[string]bool{
+ "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan":
true,
+ "GET
/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}": true,
+ "DELETE
/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}": true,
+ "POST
/v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks": true,
+ }
+ var endpoints []string
+ for _, e := range rest.AllEndpointStrings {
+ if !planning[e] {
+ endpoints = append(endpoints, e)
+ }
+ }
+ require.Len(t, endpoints, len(rest.AllEndpointStrings)-len(planning),
"planning endpoint strings changed")
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/v1/config", func(w http.ResponseWriter, _
*http.Request) {
+ require.NoError(t, json.NewEncoder(w).Encode(map[string]any{
+ "defaults": map[string]any{}, "overrides":
map[string]any{}, "endpoints": endpoints,
+ }))
+ })
+ mux.HandleFunc("/v1/namespaces/fokko/tables/tbl", func(w
http.ResponseWriter, _ *http.Request) {
+ _, err := w.Write([]byte(`{
+ "metadata-location":
"s3://warehouse/database/table/metadata/00001.metadata.json",
+ "metadata": ` + directiveTestMetadata + `,
+ "config": {"scan-planning-mode": "server"}
+ }`))
+ require.NoError(t, err)
+ })
+ srv := httptest.NewServer(mux)
+ defer srv.Close()
+
+ cat, err := rest.NewCatalog(context.Background(), "rest", srv.URL,
rest.WithOAuthToken(TestToken))
+ require.NoError(t, err)
+ require.False(t, cat.SupportsPlanTableScan())
+
+ tbl, err := cat.LoadTable(context.Background(),
catalog.ToIdentifier("fokko", "tbl"))
+ require.NoError(t, err)
+ assert.Len(t, tbl.Schema().Fields(), 2)
+ directive, err := tbl.ScanPlanningDirective()
+ require.NoError(t, err)
+ assert.Equal(t, table.ScanPlanningDirectiveServer, directive)
+
+ _, err = tbl.Scan().PlanFiles(context.Background())
+ require.ErrorIs(t, err, table.ErrInvalidOperation)
+ assert.ErrorContains(t, err, "requires server-side scan planning")
+}
+
+// A catalog commonly advertises scan-planning-mode through the /v1/config
+// defaults block rather than client options. Tables from both LoadTable and
+// UpdateTable, whose commit response carries no table config, must enforce it.
+func TestCatalogConfigScanPlanningDirective(t *testing.T) {
+ const metadataLoc =
"s3://warehouse/database/table/metadata/00001.metadata.json"
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/v1/config", func(w http.ResponseWriter, _
*http.Request) {
+ require.NoError(t, json.NewEncoder(w).Encode(map[string]any{
+ "defaults": map[string]any{"scan-planning-mode":
"server"},
+ "overrides": map[string]any{},
+ }))
+ })
+ mux.HandleFunc("/v1/namespaces/fokko/tables/tbl", func(w
http.ResponseWriter, req *http.Request) {
+ body := `{"metadata-location": "` + metadataLoc + `",
"metadata": ` + directiveTestMetadata
+ if req.Method == http.MethodGet {
+ body += `, "config": {}`
+ }
+ _, err := w.Write([]byte(body + `}`))
+ require.NoError(t, err)
+ })
+ srv := httptest.NewServer(mux)
+ defer srv.Close()
+
+ cat, err := rest.NewCatalog(context.Background(), "rest", srv.URL,
rest.WithOAuthToken(TestToken))
+ require.NoError(t, err)
+ ident := catalog.ToIdentifier("fokko", "tbl")
+
+ loaded, err := cat.LoadTable(context.Background(), ident)
+ require.NoError(t, err)
+ updated, err := cat.UpdateTable(context.Background(), ident, nil, nil)
+ require.NoError(t, err)
+
+ for name, tbl := range map[string]*table.Table{"load": loaded,
"update": updated} {
+ t.Run(name, func(t *testing.T) {
+ directive, err := tbl.ScanPlanningDirective()
+ require.NoError(t, err)
+ assert.Equal(t, table.ScanPlanningDirectiveServer,
directive)
+
+ _, err = tbl.Scan().PlanFiles(context.Background())
+ require.ErrorIs(t, err, table.ErrInvalidOperation)
+ assert.ErrorContains(t, err, "requires server-side scan
planning")
+ })
+ }
+}
+
+// A commit response carries no table config, so UpdateTable keeps only the
+// catalog's own scan-planning-mode property and drops the per-table directive
+// from the load config. This pins both directions: a per-table server
directive
+// is dropped, and a per-table client directive under a catalog-wide server
+// directive comes back as server.
+func TestUpdateTableKeepsOnlyCatalogScanPlanningDirective(t *testing.T) {
+ const metadataLoc =
"s3://warehouse/database/table/metadata/00001.metadata.json"
+
+ for _, tc := range []struct {
+ name string
+ catalogMode string
+ tableConfig string
+ wantLoaded table.ScanPlanningDirective
+ wantUpdated table.ScanPlanningDirective
+ }{
+ {
+ name: "per-table server is dropped",
+ tableConfig: `{"scan-planning-mode": "server"}`,
+ wantLoaded: table.ScanPlanningDirectiveServer,
+ wantUpdated: table.ScanPlanningDirectiveNone,
+ },
+ {
+ name: "per-table client under catalog server
becomes server",
+ catalogMode: "server",
+ tableConfig: `{"scan-planning-mode": "client"}`,
+ wantLoaded: table.ScanPlanningDirectiveClient,
+ wantUpdated: table.ScanPlanningDirectiveServer,
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mux := http.NewServeMux()
+ mux.HandleFunc("/v1/config", func(w
http.ResponseWriter, _ *http.Request) {
+ require.NoError(t,
json.NewEncoder(w).Encode(map[string]any{
+ "defaults": map[string]any{},
"overrides": map[string]any{},
+ }))
+ })
+ mux.HandleFunc("/v1/namespaces/fokko/tables/tbl",
func(w http.ResponseWriter, req *http.Request) {
+ body := `{"metadata-location": "` + metadataLoc
+ `", "metadata": ` + directiveTestMetadata
+ if req.Method == http.MethodGet {
+ body += `, "config": ` + tc.tableConfig
+ }
+ _, err := w.Write([]byte(body + `}`))
+ require.NoError(t, err)
+ })
+ srv := httptest.NewServer(mux)
+ defer srv.Close()
+
+ opts := []rest.Option{rest.WithOAuthToken(TestToken)}
+ if tc.catalogMode != "" {
+ opts = append(opts,
rest.WithAdditionalProps(iceberg.Properties{"scan-planning-mode":
tc.catalogMode}))
+ }
+ cat, err := rest.NewCatalog(context.Background(),
"rest", srv.URL, opts...)
+ require.NoError(t, err)
+ ident := catalog.ToIdentifier("fokko", "tbl")
+
+ loaded, err := cat.LoadTable(context.Background(),
ident)
+ require.NoError(t, err)
+ got, err := loaded.ScanPlanningDirective()
+ require.NoError(t, err)
+ assert.Equal(t, tc.wantLoaded, got)
+
+ updated, err := cat.UpdateTable(context.Background(),
ident, nil, nil)
+ require.NoError(t, err)
+ got, err = updated.ScanPlanningDirective()
+ require.NoError(t, err)
+ assert.Equal(t, tc.wantUpdated, got)
+
+ if tc.wantUpdated == table.ScanPlanningDirectiveServer {
+ _, err =
updated.Scan(table.WithScanPlanningMode(table.ScanPlanningLocal)).PlanFiles(context.Background())
+ require.ErrorIs(t, err,
table.ErrInvalidOperation)
+ assert.ErrorContains(t, err, "requires
server-side scan planning")
+ }
+ })
+ }
+}
diff --git a/catalog/rest/scan_planning_directive_test.go
b/catalog/rest/scan_planning_directive_test.go
new file mode 100644
index 000000000..6ac174c27
--- /dev/null
+++ b/catalog/rest/scan_planning_directive_test.go
@@ -0,0 +1,101 @@
+// 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 (
+ "bytes"
+ "context"
+ "log/slog"
+ "strings"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+)
+
+// captureSlog replaces the default slog logger for the rest of the test. Tests
+// that use it must not be parallel.
+func captureSlog(t *testing.T) *bytes.Buffer {
+ t.Helper()
+
+ var logs bytes.Buffer
+ prev := slog.Default()
+ slog.SetDefault(slog.New(slog.NewTextHandler(&logs,
&slog.HandlerOptions{Level: slog.LevelDebug})))
+ t.Cleanup(func() { slog.SetDefault(prev) })
+
+ return &logs
+}
+
+func TestScanPlanningDirective(t *testing.T) {
+ logs := captureSlog(t)
+
+ key := table.ScanPlanningModeKey
+ for _, tc := range []struct {
+ name string
+ catalog iceberg.Properties
+ server iceberg.Properties
+ want string
+ wantPresent bool
+ wantWarn bool
+ }{
+ {name: "neither"},
+ {name: "server only", server: iceberg.Properties{key:
"server"}, want: "server", wantPresent: true},
+ {name: "catalog only", catalog: iceberg.Properties{key:
"client"}, want: "client", wantPresent: true},
+ {name: "empty server value", server: iceberg.Properties{key:
""}, want: "", wantPresent: true},
+ {name: "agree", catalog: iceberg.Properties{key: "SERVER"},
server: iceberg.Properties{key: "server"}, want: "server", wantPresent: true},
+ {name: "mismatch", catalog: iceberg.Properties{key: "client"},
server: iceberg.Properties{key: "server"}, want: "server", wantPresent: true,
wantWarn: true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ logs.Reset()
+ r := &Catalog{props: tc.catalog}
+
+ got, present :=
r.scanPlanningDirective(context.Background(), []string{"db", "tbl"}, tc.server)
+ assert.Equal(t, tc.want, got)
+ assert.Equal(t, tc.wantPresent, present)
+ if tc.wantWarn {
+ assert.Contains(t, logs.String(), "level=WARN")
+ assert.Contains(t, logs.String(),
"scan-planning-mode mismatch")
+ assert.Contains(t, logs.String(),
"table=db.tbl")
+ } else {
+ assert.Empty(t, logs.String())
+ }
+ })
+ }
+}
+
+// Refresh reloads the table, so a stable mismatch must not warn on every load.
+func TestScanPlanningDirectiveMismatchWarnsOncePerCatalog(t *testing.T) {
+ logs := captureSlog(t)
+
+ key := table.ScanPlanningModeKey
+ r := &Catalog{props: iceberg.Properties{key: "client"}}
+ for range 3 {
+ r.scanPlanningDirective(context.Background(), []string{"db",
"tbl"}, iceberg.Properties{key: "server"})
+ }
+
+ out := logs.String()
+ assert.Equal(t, 1, strings.Count(out, "level=WARN"), out)
+ assert.Equal(t, 2, strings.Count(out, "level=DEBUG"), out)
+
+ // A separate catalog warns again.
+ logs.Reset()
+ other := &Catalog{props: iceberg.Properties{key: "client"}}
+ other.scanPlanningDirective(context.Background(), []string{"db",
"tbl"}, iceberg.Properties{key: "server"})
+ assert.Contains(t, logs.String(), "level=WARN")
+}
diff --git a/table/incremental_append_scan.go b/table/incremental_append_scan.go
index ed77dbe1b..d682d4ca8 100644
--- a/table/incremental_append_scan.go
+++ b/table/incremental_append_scan.go
@@ -47,8 +47,9 @@ type IncrementalAppendScan struct {
// sensitivity, concurrency, planning mode, and reporting. Projection and row
// limits are not applied to the returned file tasks; callers can read planned
// tasks with a separately configured [Scan.ReadTasks]. Auto planning falls
back
-// to local planning. Remote planning returns ErrInvalidOperation until
-// incremental remote planning is implemented.
+// to local planning. Remote planning, including a catalog `server`
+// scan-planning-mode directive, returns ErrInvalidOperation until incremental
+// remote planning is implemented.
func (t Table) NewIncrementalAppendScan(opts ...ScanOption)
*IncrementalAppendScan {
return &IncrementalAppendScan{scan: t.Scan(opts...)}
}
@@ -86,12 +87,8 @@ func (s *IncrementalAppendScan) ToSnapshot(snapshotID int64)
*IncrementalAppendS
// through the configured reporter on successful planning. Delete files are not
// applied because appended files are not present before the append snapshot.
func (s *IncrementalAppendScan) PlanFiles(ctx context.Context)
([]FileScanTask, error) {
- switch s.scan.planningMode {
- case ScanPlanningLocal, ScanPlanningAuto:
- case ScanPlanningRemote:
- return nil, fmt.Errorf("%w: incremental append scans do not
support remote planning", ErrInvalidOperation)
- default:
- return nil, fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, s.scan.planningMode)
+ if err := s.scan.checkLocalOnlyPlanning("incremental append"); err !=
nil {
+ return nil, err
}
start := time.Now()
diff --git a/table/incremental_changelog_scan.go
b/table/incremental_changelog_scan.go
index 8b8006adc..f3b418102 100644
--- a/table/incremental_changelog_scan.go
+++ b/table/incremental_changelog_scan.go
@@ -48,7 +48,8 @@ type plannedChangelogTask struct {
// NewIncrementalChangelogScan creates an incremental changelog planner.
// Projection and row limits are not applied to returned tasks. Auto planning
-// falls back to local planning, while remote planning is not supported. Use
+// falls back to local planning, while remote planning, including a catalog
+// `server` scan-planning-mode directive, is not supported. Use
// ChangelogScanTask.ScanTask with Scan.ReadTasks to read the returned files.
// Row filters are attached to each task as residuals without
partition-specific
// simplification, matching the existing incremental append scan behavior.
@@ -102,12 +103,8 @@ func (s *IncrementalChangelogScan) PlanFiles(ctx
context.Context) ([]ChangelogSc
return nil, err
}
- switch s.scan.planningMode {
- case ScanPlanningLocal, ScanPlanningAuto:
- case ScanPlanningRemote:
- return nil, fmt.Errorf("%w: incremental changelog scans do not
support remote planning", ErrInvalidOperation)
- default:
- return nil, fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, s.scan.planningMode)
+ if err := s.scan.checkLocalOnlyPlanning("incremental changelog"); err
!= nil {
+ return nil, err
}
start := time.Now()
diff --git a/table/scan_planning.go b/table/scan_planning.go
index ddf31dfb2..f27c29195 100644
--- a/table/scan_planning.go
+++ b/table/scan_planning.go
@@ -23,27 +23,32 @@ package table
import (
"context"
+ "fmt"
"io"
+ "strings"
"github.com/apache/iceberg-go"
icebergio "github.com/apache/iceberg-go/io"
)
// ScanPlanningMode is the user-facing scan option: three values
-// (local/remote/auto) selecting how (*Scan).PlanFiles plans a scan. Local
-// planning remains the default; remote is opt-in via WithScanPlanningMode.
+// (local/remote/auto) selecting how (*Scan).PlanFiles plans a scan. Without a
+// catalog directive, local planning is the default and remote is opt-in via
+// WithScanPlanningMode.
//
-// This is deliberately distinct from the REST table-config key
-// `scan-planning-mode` (values `client`/`server`), which is a server directive
-// resolved separately (OQ4): a `client` table forces local planning, a
`server`
-// table forces remote planning, and explicit conflicting scan options fail
-// fast. There is intentionally no fourth `server` value here; the directive
-// lives in the table config, not the user option.
+// This is deliberately distinct from the catalog's `scan-planning-mode`
+// directive (values `client`/`server`, see ScanPlanningDirective), which
+// (*Scan).PlanFiles enforces: a `server` table plans remotely and a `client`
+// table plans locally, whatever the default or ScanPlanningAuto would choose.
+// A scan that explicitly selects the conflicting mode fails. Incremental scans
+// only plan locally, so they fail under a `server` directive. Scans inside a
+// transaction plan staged metadata locally and do not apply the directive.
type ScanPlanningMode string
const (
// ScanPlanningLocal always plans locally by reading manifests through
the
- // table's FileIO. This is the default and current behavior.
+ // table's FileIO. This is the default unless the catalog requires
+ // server-side planning.
ScanPlanningLocal ScanPlanningMode = "local"
// ScanPlanningRemote requires a planner that advertises remote
capability
// and fails loudly if remote planning is unavailable.
@@ -53,10 +58,169 @@ const (
ScanPlanningAuto ScanPlanningMode = "auto"
)
+// ScanPlanningModeKey is the REST table-config key a catalog uses to tell
+// clients which planning mode a table supports. Valid values are `client` and
+// `server`.
+const ScanPlanningModeKey = "scan-planning-mode"
+
+// ScanPlanningDirective is the catalog's `scan-planning-mode` table-config
+// value. It is distinct from ScanPlanningMode, the user-facing scan option.
+type ScanPlanningDirective string
+
+const (
+ // ScanPlanningDirectiveNone is the zero value. When returned with a nil
+ // error from Table.ScanPlanningDirective, it means the catalog
supplied no
+ // directive and does not constrain where scans are planned. It is also
+ // returned alongside a non-nil error for an unrecognized directive, in
+ // which case it carries no meaning.
+ ScanPlanningDirectiveNone ScanPlanningDirective = ""
+ // ScanPlanningDirectiveClient indicates the catalog requires the
client to
+ // plan scans locally.
+ ScanPlanningDirectiveClient ScanPlanningDirective = "client"
+ // ScanPlanningDirectiveServer indicates the catalog requires scans to
be
+ // planned remotely by the catalog server.
+ ScanPlanningDirectiveServer ScanPlanningDirective = "server"
+)
+
+// ScanPlanningDirective returns the catalog's `scan-planning-mode` directive
+// recorded when the table was loaded. For a REST catalog this is the value in
+// the `config` block of the table-load response, or, when the server sends
+// none, the catalog's own `scan-planning-mode` property. Table metadata
+// properties are not consulted. Values are matched case-insensitively.
+//
+// It returns ScanPlanningDirectiveNone and a nil error when the catalog
+// supplied no directive. Any value other than `client` or `server`, including
+// an empty value or a mode introduced by a future version of the REST spec,
+// returns an error wrapping iceberg.ErrInvalidArgument; callers should treat
+// that as a directive they cannot honor rather than as the absence of one.
+//
+// The directive is only known for tables built from a load response
+// (LoadTable, CreateTable, RegisterTable, and their refreshes); tables
+// returned by Transaction.Commit or StagedTable keep the directive of the
+// table they were derived from. Commit responses do not carry table config, so
+// a table returned by a REST catalog's UpdateTable reports only the catalog's
+// own `scan-planning-mode` property; reload the table to obtain a per-table
+// directive.
+//
+// (*Scan).PlanFiles enforces the directive; see ScanPlanningMode.
+func (t Table) ScanPlanningDirective() (ScanPlanningDirective, error) {
+ if !t.hasScanPlanningDirective {
+ return ScanPlanningDirectiveNone, nil
+ }
+
+ switch d :=
ScanPlanningDirective(strings.ToLower(t.scanPlanningDirective)); d {
+ case ScanPlanningDirectiveClient, ScanPlanningDirectiveServer:
+ return d, nil
+ default:
+ return ScanPlanningDirectiveNone, fmt.Errorf("%w: unrecognized
%s %q, expected %q or %q",
+ iceberg.ErrInvalidArgument, ScanPlanningModeKey,
t.scanPlanningDirective,
+ ScanPlanningDirectiveClient,
ScanPlanningDirectiveServer)
+ }
+}
+
+// WithScanPlanningDirective records the raw `scan-planning-mode` value from a
+// catalog's table-load response config. ok reports whether the key was
+// present, so callers can pass a comma-ok map read straight through:
+//
+// v, ok := cfg[table.ScanPlanningModeKey]
+// table.WithScanPlanningDirective(v, ok)
+//
+// When ok is false the table reports no directive. A present but empty value
+// is reported as unrecognized.
+func WithScanPlanningDirective(value string, ok bool) Option {
+ return func(t *Table) {
+ t.scanPlanningDirective = value
+ t.hasScanPlanningDirective = ok
+ }
+}
+
+// WithScanPlanningDirectiveFrom copies src's scan-planning directive verbatim,
+// including whether one was present and any unrecognized raw value. Catalogs
+// use it when rebuilding a table from a response that carries no table config,
+// such as a credential refresh. A nil src leaves the directive unchanged.
+func WithScanPlanningDirectiveFrom(src *Table) Option {
+ if src == nil {
+ return func(*Table) {}
+ }
+
+ return WithScanPlanningDirective(src.scanPlanningDirective,
src.hasScanPlanningDirective)
+}
+
// WithScanPlanningMode sets the scan-planning mode for a scan. The default is
-// ScanPlanningLocal.
+// ScanPlanningLocal, or the mode the catalog's directive requires. Explicitly
+// selecting a mode that conflicts with the directive makes planning fail.
func WithScanPlanningMode(mode ScanPlanningMode) ScanOption {
- return func(scan *Scan) { scan.planningMode = mode }
+ return func(scan *Scan) {
+ scan.planningMode = mode
+ scan.planningModeSet = true
+ }
+}
+
+// effectivePlanningMode applies the catalog's scan-planning-mode directive to
+// the scan's planning mode. A `server` directive selects remote planning and a
+// `client` directive selects local planning, unless the scan explicitly asked
+// for the other one, which is an error. The directive is checked here rather
+// than when the table is loaded, so metadata-only use of a table keeps working
+// whatever the catalog advertises.
+func (scan *Scan) effectivePlanningMode() (ScanPlanningMode, error) {
+ switch scan.planningMode {
+ case ScanPlanningLocal, ScanPlanningRemote, ScanPlanningAuto:
+ default:
+ return "", fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, scan.planningMode)
+ }
+
+ if scan.directiveErr != nil {
+ return "", scan.directiveErr
+ }
+
+ switch scan.directive {
+ case ScanPlanningDirectiveServer:
+ if scan.planningModeSet && scan.planningMode ==
ScanPlanningLocal {
+ return "", fmt.Errorf("%w: the catalog requires
server-side scan planning (%s=%s) but the scan selected %s planning",
+ ErrInvalidOperation, ScanPlanningModeKey,
ScanPlanningDirectiveServer, ScanPlanningLocal)
+ }
+ if scan.planner == nil ||
!scan.planner.SupportsRemoteScanPlanning() {
+ return "", fmt.Errorf("%w: the catalog requires
server-side scan planning (%s=%s) but remote scan planning is unavailable",
+ ErrInvalidOperation, ScanPlanningModeKey,
ScanPlanningDirectiveServer)
+ }
+
+ return ScanPlanningRemote, nil
+ case ScanPlanningDirectiveClient:
+ if scan.planningModeSet && scan.planningMode ==
ScanPlanningRemote {
+ return "", fmt.Errorf("%w: the catalog requires
client-side scan planning (%s=%s) but the scan selected %s planning",
+ ErrInvalidOperation, ScanPlanningModeKey,
ScanPlanningDirectiveClient, ScanPlanningRemote)
+ }
+
+ return ScanPlanningLocal, nil
+ default:
+ return scan.planningMode, nil
+ }
+}
+
+// checkLocalOnlyPlanning applies the catalog's scan-planning-mode directive to
+// a scan that can only plan locally, such as an incremental scan. A `server`
+// directive fails rather than reading manifests with the table's own
+// credentials, which the directive exists to prevent. kind names the scan in
+// errors.
+func (scan *Scan) checkLocalOnlyPlanning(kind string) error {
+ mode, err := scan.effectivePlanningMode()
+ if err != nil {
+ return err
+ }
+
+ switch mode {
+ case ScanPlanningLocal, ScanPlanningAuto:
+ return nil
+ case ScanPlanningRemote:
+ if scan.directive == ScanPlanningDirectiveServer {
+ return fmt.Errorf("%w: the catalog requires server-side
scan planning (%s=%s) but %s scans do not support remote planning",
+ ErrInvalidOperation, ScanPlanningModeKey,
ScanPlanningDirectiveServer, kind)
+ }
+
+ return fmt.Errorf("%w: %s scans do not support remote
planning", ErrInvalidOperation, kind)
+ default:
+ return fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, mode)
+ }
}
// ScanPlanningMetadata is the subset of table.Metadata a ScanPlanner needs:
@@ -171,7 +335,7 @@ type FullRemoteScanPlanner interface {
//
// type Scan struct {
// // ...existing fields...
-// planningMode ScanPlanningMode // set by WithScanPlanningMode;
default ScanPlanningLocal
+// planningMode ScanPlanningMode // set by WithScanPlanningMode;
default ScanPlanningLocal, overridden by the catalog directive
// planner ScanPlanner // non-nil only when the catalog
supplies one
// }
//
diff --git a/table/scan_planning_directive_wiring_test.go
b/table/scan_planning_directive_wiring_test.go
new file mode 100644
index 000000000..38319fa34
--- /dev/null
+++ b/table/scan_planning_directive_wiring_test.go
@@ -0,0 +1,160 @@
+// 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 table
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// directiveStubCatalog hands every LoadTable a table carrying the configured
+// scan-planning directive, so Refresh can be checked to re-hydrate it.
+type directiveStubCatalog struct {
+ directive string
+ present bool
+}
+
+func (c *directiveStubCatalog) LoadTable(_ context.Context, ident Identifier)
(*Table, error) {
+ return New(ident, nil, "",
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, c,
+ WithScanPlanningDirective(c.directive, c.present)), nil
+}
+
+func (c *directiveStubCatalog) CommitTable(context.Context, Identifier,
[]Requirement, []Update) (Metadata, string, error) {
+ return nil, "", nil
+}
+
+// TestRefreshRehydratesScanPlanningDirective pins that Refresh adopts the
+// directive from the reloaded table, including dropping one the reload no
+// longer carries.
+func TestRefreshRehydratesScanPlanningDirective(t *testing.T) {
+ cat := &directiveStubCatalog{directive: "server", present: true}
+ tbl := New(Identifier{"db", "directive_refresh"}, nil, "",
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, cat)
+
+ got, err := tbl.ScanPlanningDirective()
+ require.NoError(t, err)
+ require.Equal(t, ScanPlanningDirectiveNone, got)
+
+ require.NoError(t, tbl.Refresh(context.Background()))
+ got, err = tbl.ScanPlanningDirective()
+ require.NoError(t, err)
+ assert.Equal(t, ScanPlanningDirectiveServer, got)
+
+ cat.directive, cat.present = "", false
+ require.NoError(t, tbl.Refresh(context.Background()))
+ got, err = tbl.ScanPlanningDirective()
+ require.NoError(t, err)
+ assert.Equal(t, ScanPlanningDirectiveNone, got)
+}
+
+// TestCommitPreservesScanPlanningDirective pins the commit path: doCommit
+// rebuilds the table via New(...), so the returned table must keep the source
+// table's directive, including an unrecognized raw value.
+func TestCommitPreservesScanPlanningDirective(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ raw string
+ present bool
+ want ScanPlanningDirective
+ wantErr bool
+ }{
+ {name: "absent", want: ScanPlanningDirectiveNone},
+ {name: "server", raw: "server", present: true, want:
ScanPlanningDirectiveServer},
+ {name: "unrecognized", raw: "hybrid", present: true, wantErr:
true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ loc := filepath.ToSlash(t.TempDir())
+ meta, err := NewMetadata(schema,
iceberg.UnpartitionedSpec, UnsortedSortOrder, loc,
+ iceberg.Properties{PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+
+ cat := &committingLabelCatalog{metadata: meta}
+ tbl := New(Identifier{"db", "commit_directive"}, meta,
loc+"/metadata/v1.metadata.json",
+ func(context.Context) (iceio.IO, error) {
return iceio.LocalFS{}, nil }, cat,
+ WithScanPlanningDirective(tc.raw, tc.present))
+
+ txn := tbl.NewTransaction()
+ require.NoError(t,
txn.SetProperties(iceberg.Properties{"k": "v"}))
+ committed, err := txn.Commit(context.Background())
+ require.NoError(t, err)
+
+ got, err := committed.ScanPlanningDirective()
+ if tc.wantErr {
+ require.ErrorIs(t, err,
iceberg.ErrInvalidArgument)
+ assert.Contains(t, err.Error(), `"`+tc.raw+`"`)
+
+ return
+ }
+ require.NoError(t, err)
+ assert.Equal(t, tc.want, got)
+ })
+ }
+}
+
+// TestStagedTablePreservesScanPlanningDirective pins that
+// Transaction.StagedTable() forwards the transaction table's directive; it
+// rebuilds the table via New(...) too.
+func TestStagedTablePreservesScanPlanningDirective(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ raw string
+ present bool
+ want ScanPlanningDirective
+ wantErr bool
+ }{
+ {name: "absent", want: ScanPlanningDirectiveNone},
+ {name: "client", raw: "client", present: true, want:
ScanPlanningDirectiveClient},
+ {name: "unrecognized", raw: "hybrid", present: true, wantErr:
true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true,
+ })
+ meta, err := NewMetadata(schema,
iceberg.UnpartitionedSpec, UnsortedSortOrder,
+ "mem://default/staged",
iceberg.Properties{PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+
+ tbl := New(Identifier{"default", "staged"}, meta, "",
+ func(context.Context) (iceio.IO, error) {
return iceio.NewMemFS(), nil }, nil,
+ WithScanPlanningDirective(tc.raw, tc.present))
+
+ staged, err := tbl.NewTransaction().StagedTable()
+ require.NoError(t, err)
+
+ got, err := staged.ScanPlanningDirective()
+ if tc.wantErr {
+ require.ErrorIs(t, err,
iceberg.ErrInvalidArgument)
+ assert.Contains(t, err.Error(), `"`+tc.raw+`"`)
+
+ return
+ }
+ require.NoError(t, err)
+ assert.Equal(t, tc.want, got)
+ })
+ }
+}
diff --git a/table/scan_planning_test.go b/table/scan_planning_test.go
index c0a7cbc9e..f88343539 100644
--- a/table/scan_planning_test.go
+++ b/table/scan_planning_test.go
@@ -741,3 +741,226 @@ func mustPlanIOState(t *testing.T, planIO PlanIO)
*planIOState {
return state
}
+
+func TestTableScanPlanningDirective(t *testing.T) {
+ t.Parallel()
+
+ tests := []struct {
+ name string
+ opts []Option
+ want ScanPlanningDirective
+ wantErr bool
+ }{
+ {name: "absent", want: ScanPlanningDirectiveNone},
+ {
+ name: "io props ignored",
+ opts:
[]Option{WithScanPlanningIOProperties(iceberg.Properties{ScanPlanningModeKey:
"server"})},
+ want: ScanPlanningDirectiveNone,
+ },
+ {name: "not present", opts:
[]Option{WithScanPlanningDirective("server", false)}, want:
ScanPlanningDirectiveNone},
+ {name: "client", opts:
[]Option{WithScanPlanningDirective("client", true)}, want:
ScanPlanningDirectiveClient},
+ {name: "server", opts:
[]Option{WithScanPlanningDirective("server", true)}, want:
ScanPlanningDirectiveServer},
+ {name: "case insensitive", opts:
[]Option{WithScanPlanningDirective("SERVER", true)}, want:
ScanPlanningDirectiveServer},
+ {name: "unrecognized", opts:
[]Option{WithScanPlanningDirective("remote", true)}, want:
ScanPlanningDirectiveNone, wantErr: true},
+ {name: "empty value", opts:
[]Option{WithScanPlanningDirective("", true)}, want: ScanPlanningDirectiveNone,
wantErr: true},
+ {
+ name: "later option overrides",
+ opts: []Option{WithScanPlanningDirective("server",
true), WithScanPlanningDirective("", false)},
+ want: ScanPlanningDirectiveNone,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ t.Parallel()
+
+ tbl := New(Identifier{"db", "tbl"}, nil, "", nil, nil,
tt.opts...)
+ got, err := tbl.ScanPlanningDirective()
+ if tt.wantErr {
+ require.ErrorIs(t, err,
iceberg.ErrInvalidArgument)
+ assert.Equal(t, tt.want, got)
+
+ return
+ }
+ require.NoError(t, err)
+ assert.Equal(t, tt.want, got)
+ })
+ }
+}
+
+// TestWithScanPlanningDirectiveFrom pins that the directive is copied
+// verbatim, including its presence bit and an unrecognized raw value.
+func TestWithScanPlanningDirectiveFrom(t *testing.T) {
+ t.Parallel()
+
+ for _, tc := range []struct {
+ name string
+ opts []Option
+ }{
+ {name: "absent"},
+ {name: "server", opts:
[]Option{WithScanPlanningDirective("server", true)}},
+ {name: "unrecognized", opts:
[]Option{WithScanPlanningDirective("hybrid", true)}},
+ {name: "empty value", opts:
[]Option{WithScanPlanningDirective("", true)}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+
+ src := New(Identifier{"db", "tbl"}, nil, "", nil, nil,
tc.opts...)
+ dst := New(Identifier{"db", "tbl"}, nil, "", nil, nil,
+ WithScanPlanningDirective("client", true),
WithScanPlanningDirectiveFrom(src))
+
+ assert.Equal(t, src.scanPlanningDirective,
dst.scanPlanningDirective)
+ assert.Equal(t, src.hasScanPlanningDirective,
dst.hasScanPlanningDirective)
+ })
+ }
+
+ t.Run("nil source", func(t *testing.T) {
+ t.Parallel()
+
+ var dst *Table
+ require.NotPanics(t, func() {
+ dst = New(Identifier{"db", "tbl"}, nil, "", nil, nil,
+ WithScanPlanningDirective("client", true),
WithScanPlanningDirectiveFrom(nil))
+ })
+ assert.Equal(t, "client", dst.scanPlanningDirective)
+ assert.True(t, dst.hasScanPlanningDirective)
+ })
+}
+
+func TestScanEnforcesPlanningDirective(t *testing.T) {
+ t.Parallel()
+
+ remoteTasks := ScanPlanningResult{Tasks: []FileScanTask{{}}}
+ tests := []struct {
+ name string
+ directive *string
+ opts []ScanOption
+ supports bool
+ wantErr error
+ // wantRemote reports whether the planner was asked to plan.
+ wantRemote bool
+ }{
+ {name: "server default", directive: new("server"), supports:
true, wantRemote: true},
+ {name: "server auto", directive: new("server"), opts:
[]ScanOption{WithScanPlanningMode(ScanPlanningAuto)}, supports: true,
wantRemote: true},
+ {name: "server remote", directive: new("server"), opts:
[]ScanOption{WithScanPlanningMode(ScanPlanningRemote)}, supports: true,
wantRemote: true},
+ {name: "server case insensitive", directive: new("SERVER"),
supports: true, wantRemote: true},
+ {name: "server explicit local", directive: new("server"), opts:
[]ScanOption{WithScanPlanningMode(ScanPlanningLocal)}, supports: true, wantErr:
ErrInvalidOperation},
+ {name: "server without capable planner", directive:
new("server"), supports: false, wantErr: ErrInvalidOperation},
+ {name: "server unknown mode", directive: new("server"), opts:
[]ScanOption{WithScanPlanningMode("bogus")}, supports: true, wantErr:
iceberg.ErrInvalidArgument},
+ {name: "client default", directive: new("client"), supports:
true},
+ {name: "client auto with capable planner", directive:
new("client"), opts: []ScanOption{WithScanPlanningMode(ScanPlanningAuto)},
supports: true},
+ {name: "client explicit remote", directive: new("client"),
opts: []ScanOption{WithScanPlanningMode(ScanPlanningRemote)}, supports: true,
wantErr: ErrInvalidOperation},
+ {name: "unrecognized directive", directive: new("bogus"),
supports: true, wantErr: iceberg.ErrInvalidArgument},
+ {name: "no directive default", supports: true},
+ {name: "no directive auto", opts:
[]ScanOption{WithScanPlanningMode(ScanPlanningAuto)}, supports: true,
wantRemote: true},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ t.Parallel()
+
+ txn, _ := createTestTransactionWithMemIO(t,
*iceberg.UnpartitionedSpec)
+ tbl := txn.tbl
+ planner := &fakeScanPlanner{result: remoteTasks,
supports: tt.supports}
+ tbl.planner = planner
+ if tt.directive != nil {
+ WithScanPlanningDirective(*tt.directive,
true)(tbl)
+ }
+
+ // The directive never blocks building a scan, only
planning it.
+ scan := tbl.Scan(tt.opts...)
+ require.NotNil(t, scan)
+
+ tasks, err := scan.PlanFiles(context.Background())
+ if tt.wantErr != nil {
+ require.ErrorIs(t, err, tt.wantErr)
+ assert.False(t, planner.called, "planner must
not be called when planning fails")
+
+ return
+ }
+ require.NoError(t, err)
+ assert.Equal(t, tt.wantRemote, planner.called)
+ if tt.wantRemote {
+ assert.Len(t, tasks, 1)
+ }
+ })
+ }
+}
+
+func TestTransactionScanIgnoresPlanningDirective(t *testing.T) {
+ t.Parallel()
+
+ txn, _ := createTestTransactionWithMemIO(t, *iceberg.UnpartitionedSpec)
+ planner := &fakeScanPlanner{supports: true}
+ txn.tbl.planner = planner
+ WithScanPlanningDirective("server", true)(txn.tbl)
+
+ scan, err := txn.Scan()
+ require.NoError(t, err)
+ _, err = scan.PlanFiles(context.Background())
+ require.NoError(t, err)
+ assert.False(t, planner.called, "transaction scans plan staged metadata
locally")
+}
+
+// Incremental scans only plan locally, so a `server` directive must fail them
+// rather than let them read manifests with the table's own credentials.
+func TestIncrementalScansEnforcePlanningDirective(t *testing.T) {
+ t.Parallel()
+
+ scans := []struct {
+ name string
+ plan func(Table, ...ScanOption) error
+ }{
+ {name: "append", plan: func(tbl Table, opts ...ScanOption)
error {
+ _, err :=
tbl.NewIncrementalAppendScan(opts...).PlanFiles(context.Background())
+
+ return err
+ }},
+ {name: "changelog", plan: func(tbl Table, opts ...ScanOption)
error {
+ _, err :=
tbl.NewIncrementalChangelogScan(opts...).PlanFiles(context.Background())
+
+ return err
+ }},
+ }
+ tests := []struct {
+ name string
+ directive *string
+ opts []ScanOption
+ wantErr error
+ wantMsg string
+ }{
+ {name: "server default", directive: new("server"), wantErr:
ErrInvalidOperation, wantMsg: "requires server-side scan planning"},
+ {name: "server auto", directive: new("server"), opts:
[]ScanOption{WithScanPlanningMode(ScanPlanningAuto)}, wantErr:
ErrInvalidOperation, wantMsg: "requires server-side scan planning"},
+ {name: "server explicit local", directive: new("server"), opts:
[]ScanOption{WithScanPlanningMode(ScanPlanningLocal)}, wantErr:
ErrInvalidOperation, wantMsg: "selected local planning"},
+ {name: "client default", directive: new("client")},
+ {name: "client explicit remote", directive: new("client"),
opts: []ScanOption{WithScanPlanningMode(ScanPlanningRemote)}, wantErr:
ErrInvalidOperation, wantMsg: "requires client-side scan planning"},
+ {name: "unrecognized directive", directive: new("bogus"),
wantErr: iceberg.ErrInvalidArgument},
+ {name: "no directive auto", opts:
[]ScanOption{WithScanPlanningMode(ScanPlanningAuto)}},
+ }
+
+ for _, sc := range scans {
+ for _, tt := range tests {
+ t.Run(sc.name+"/"+tt.name, func(t *testing.T) {
+ t.Parallel()
+
+ txn, _ := createTestTransactionWithMemIO(t,
*iceberg.UnpartitionedSpec)
+ tbl := txn.tbl
+ planner := &fakeScanPlanner{supports: true}
+ tbl.planner = planner
+ if tt.directive != nil {
+
WithScanPlanningDirective(*tt.directive, true)(tbl)
+ }
+
+ err := sc.plan(*tbl, tt.opts...)
+ assert.False(t, planner.called, "incremental
scans never call the remote planner")
+ if tt.wantErr != nil {
+ require.ErrorIs(t, err, tt.wantErr)
+ assert.ErrorContains(t, err, tt.wantMsg)
+
+ return
+ }
+ require.NoError(t, err)
+ })
+ }
+ }
+}
diff --git a/table/scanner.go b/table/scanner.go
index 2d053e1e8..b3c98e041 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -552,6 +552,13 @@ type Scan struct {
planner ScanPlanner
scanPlanningIOProps iceberg.Properties
planningMode ScanPlanningMode
+ // planningModeSet records an explicit WithScanPlanningMode, so a mode
that
+ // conflicts with the catalog directive fails instead of being
overridden.
+ planningModeSet bool
+ // directive and directiveErr are the table's scan-planning-mode
directive
+ // (see ScanPlanningDirective). An unrecognized directive fails
planning only.
+ directive ScanPlanningDirective
+ directiveErr error
// planIO, when non-nil, is a plan-scoped FileIO loader set by remote
scan
// planning. ReadTasks leases it instead of falling back to ioF, and
replacing
// the plan retires it after all active readers finish. See PlanIO.
@@ -1637,7 +1644,12 @@ func (scan *Scan) planFiles(ctx context.Context,
projectScanColumns bool) ([]Fil
scan.asOfTimestamp = nil
}
- switch scan.planningMode {
+ mode, err := scan.effectivePlanningMode()
+ if err != nil {
+ return nil, err
+ }
+
+ switch mode {
case ScanPlanningRemote:
return scan.planFilesRemote(ctx)
case ScanPlanningAuto:
@@ -1650,7 +1662,9 @@ func (scan *Scan) planFiles(ctx context.Context,
projectScanColumns bool) ([]Fil
}
case ScanPlanningLocal:
default:
- return nil, fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, scan.planningMode)
+ // effectivePlanningMode validates the mode first; this keeps a
mode it
+ // accepts but this switch does not handle from planning
locally.
+ return nil, fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, mode)
}
start := time.Now()
diff --git a/table/table.go b/table/table.go
index ab06c9c35..b51424a7d 100644
--- a/table/table.go
+++ b/table/table.go
@@ -111,8 +111,13 @@ type Table struct {
// REST catalogs can return FileIO configuration in the response's
config
// block.
scanPlanningIOProps iceberg.Properties
- savedConfig iceberg.Properties
- reporter metrics.Reporter
+ // scanPlanningDirective is the raw `scan-planning-mode` value from the
+ // catalog load response's config block; hasScanPlanningDirective
records
+ // whether the key was present.
+ scanPlanningDirective string
+ hasScanPlanningDirective bool
+ savedConfig iceberg.Properties
+ reporter metrics.Reporter
// reporterSet records whether a caller injected a reporter via
// WithMetricsReporter. It distinguishes an explicit reporter
(including an
// explicit NopReporter opt-out) from the construction-time default, so
@@ -250,6 +255,8 @@ func (t *Table) Refresh(ctx context.Context) error {
t.manifestCache = newSnapshotManifestCacheForMetadata(fresh.metadata)
t.planner = fresh.planner
t.scanPlanningIOProps = maps.Clone(fresh.scanPlanningIOProps)
+ t.scanPlanningDirective = fresh.scanPlanningDirective
+ t.hasScanPlanningDirective = fresh.hasScanPlanningDirective
t.labels = fresh.labels
t.savedConfig = maps.Clone(fresh.savedConfig)
// Only inherit the catalog-derived reporter when the caller hasn't set
one
@@ -857,6 +864,7 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
t.cat,
withReporterState(t.reporter, t.reporterSet),
WithScanPlanningIOProperties(t.scanPlanningIOProps),
+ WithScanPlanningDirective(t.scanPlanningDirective,
t.hasScanPlanningDirective),
WithLabels(t.labels),
WithSavedConfig(t.savedConfig),
), nil
@@ -1364,6 +1372,7 @@ func WithArrowBatchSize(n int) ScanOption {
}
func (t Table) Scan(opts ...ScanOption) *Scan {
+ directive, directiveErr := t.ScanPlanningDirective()
s := &Scan{
identifier: slices.Clone(t.identifier),
metadata: t.metadata,
@@ -1372,14 +1381,15 @@ func (t Table) Scan(opts ...ScanOption) *Scan {
manifestCache: t.manifestCache,
planner: t.planner,
scanPlanningIOProps: maps.Clone(t.scanPlanningIOProps),
- // TODO(#1178 Phase 6): resolve scan-planning-mode table
properties here.
- planningMode: ScanPlanningLocal,
- rowFilter: iceberg.AlwaysTrue{},
- selectedFields: []string{"*"},
- caseSensitive: true,
- limit: ScanNoLimit,
- concurrency: runtime.GOMAXPROCS(0),
- reporter: t.MetricsReporter(),
+ directive: directive,
+ directiveErr: directiveErr,
+ planningMode: ScanPlanningLocal,
+ rowFilter: iceberg.AlwaysTrue{},
+ selectedFields: []string{"*"},
+ caseSensitive: true,
+ limit: ScanNoLimit,
+ concurrency: runtime.GOMAXPROCS(0),
+ reporter: t.MetricsReporter(),
}
for _, opt := range opts {
diff --git a/table/transaction.go b/table/transaction.go
index ae1615d04..9b51b322d 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -3523,9 +3523,11 @@ func (t *Transaction) Scan(opts ...ScanOption) (*Scan,
error) {
manifestCache:
newSnapshotManifestCacheForMetadata(updatedMeta),
// Catalog planners can only see committed table state, not
metadata
// staged inside this transaction. Keep transaction scans local
so auto
- // mode cannot silently return stale tasks.
- planner: nil,
- // TODO(#1178 Phase 6): resolve scan-planning-mode table
properties here.
+ // mode cannot silently return stale tasks. The catalog's
+ // scan-planning-mode directive is not applied either: the
server cannot
+ // plan staged state, and committing a transaction reads the
table's
+ // manifests locally whatever the directive says.
+ planner: nil,
planningMode: ScanPlanningLocal,
rowFilter: iceberg.AlwaysTrue{},
selectedFields: []string{"*"},
@@ -3564,6 +3566,7 @@ func (t *Transaction) StagedTable() (*StagedTable, error)
{
t.tbl.cat,
withReporterState(t.tbl.reporter, t.tbl.reporterSet),
WithScanPlanningIOProperties(t.tbl.scanPlanningIOProps),
+ WithScanPlanningDirective(t.tbl.scanPlanningDirective,
t.tbl.hasScanPlanningDirective),
WithLabels(t.tbl.labels),
WithSavedConfig(t.tbl.savedConfig),
),
