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 c37ed1a84 feat(table): distinguish rejected commits from unknown state
(#2350)
c37ed1a84 is described below
commit c37ed1a8401c325fbc3291a04360a2686dd939d8
Author: Harrison Crosse <[email protected]>
AuthorDate: Mon Nov 2 15:31:09 2026 -0500
feat(table): distinguish rejected commits from unknown state (#2350)
* feat(table): distinguish rejected commits from unknown state
I added `table.ErrCommitRejected` for commits that the catalog rejects
before writing anything. Rejected transactions stay committable and clean
up files that they cannot reuse. Conflicts still retry.
Previously, these failures counted as unknown state, which skipped cleanup
and prevented another `Commit` even when nothing was written.
I added transaction, catalog and end-to-end tests. The full suite, race
tests and lint pass.
Signed-off-by: Harrison Crosse <[email protected]>
* docs(table): note rejected-commit compatibility and the REST 400
divergence
Signed-off-by: Harrison Crosse <[email protected]>
* test(catalog): cover failures after metadata writes
I added Glue, Hive, and SQL tests that distinguish rejected commits from
failures after metadata writes. I also routed the remaining Hadoop
rejection paths through `internal.RejectUnlessConflict` and added the
`hadoop catalog:` prefix to requirement errors.
Part of #2341.
Signed-off-by: Harrison Crosse <[email protected]>
---------
Signed-off-by: Harrison Crosse <[email protected]>
---
catalog/glue/glue_test.go | 44 +++++++++++++++++++++++
catalog/hadoop/hadoop_test.go | 23 ++++++++++--
catalog/hadoop/tables.go | 10 +++---
catalog/hive/hive_test.go | 41 +++++++++++++++++++++
catalog/internal/update_stage_test.go | 63 ++++++++++++++++++++++++++++++++
catalog/internal/utils.go | 18 ++++++++--
catalog/rest/rest.go | 11 +++++-
catalog/rest/rest_test.go | 67 ++++++++++++++++++++++++++++++++++
catalog/sql/sql_test.go | 68 +++++++++++++++++++++++++++++++++++
table/commit_retry_test.go | 57 +++++++++++++++++++++++++++++
table/retry_manifest_cleanup_test.go | 58 ++++++++++++++++++++++++++++++
table/rewrite_manifests_test.go | 50 +++++++++++++++++++++++++-
table/table.go | 16 +++++++++
website/src/concurrent-writes.md | 7 ++++
website/src/releases.md | 1 +
15 files changed, 523 insertions(+), 11 deletions(-)
diff --git a/catalog/glue/glue_test.go b/catalog/glue/glue_test.go
index 3838da9bd..e33d19ad6 100644
--- a/catalog/glue/glue_test.go
+++ b/catalog/glue/glue_test.go
@@ -2442,6 +2442,50 @@ func
TestGlueCommitTableValidatesRequirementsForMissingTable(t *testing.T) {
mockGlueSvc.AssertExpectations(t)
}
+func TestGlueCommitTableRejectsOnlyFailuresBeforeMetadataWrite(t *testing.T) {
+ ctx := context.Background()
+ ident := TableIdentifier("test_database", "test_table")
+ newCatalog := func(t *testing.T) (*Catalog, *mockGlueClient) {
+ tableLocation := filepath.Join(t.TempDir(), "test_table")
+ metadataLocation := filepath.Join(tableLocation, "metadata",
"v1.metadata.json")
+ writeGluePurgeTableMetadata(t, iceio.LocalFS{}, tableLocation,
metadataLocation, nil)
+ glueTable := gluePurgeTable(metadataLocation)
+ glueTable.VersionId = aws.String("v1")
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("GetTable", mock.Anything, &glue.GetTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.GetTableOutput{Table:
&glueTable}, nil).Once()
+
+ return &Catalog{glueSvc: mockGlueSvc}, mockGlueSvc
+ }
+
+ t.Run("staging failure", func(t *testing.T) {
+ glueCatalog, mockGlueSvc := newCatalog(t)
+
+ _, _, err := glueCatalog.CommitTable(ctx, ident, nil,
[]table.Update{table.NewSetCurrentSchemaUpdate(99)})
+ require.ErrorIs(t, err, table.ErrCommitRejected)
+ require.NotErrorIs(t, err, table.ErrCommitFailed)
+ mockGlueSvc.AssertNotCalled(t, "UpdateTable", mock.Anything,
mock.Anything, mock.Anything)
+ mockGlueSvc.AssertExpectations(t)
+ })
+
+ t.Run("glue update failure after metadata write", func(t *testing.T) {
+ glueCatalog, mockGlueSvc := newCatalog(t)
+ mockGlueSvc.On("UpdateTable", mock.Anything, mock.Anything,
mock.Anything).
+ Return(&glue.UpdateTableOutput{}, errors.New("glue
unavailable")).Once()
+
+ _, _, err := glueCatalog.CommitTable(ctx, ident, nil,
[]table.Update{
+ table.NewSetPropertiesUpdate(iceberg.Properties{"key":
"value"}),
+ })
+ require.ErrorContains(t, err, "glue unavailable")
+ require.NotErrorIs(t, err, table.ErrCommitRejected)
+ require.NotErrorIs(t, err, table.ErrCommitFailed)
+ mockGlueSvc.AssertExpectations(t)
+ })
+}
+
func cleanupTable(t *testing.T, ctlg catalog.Catalog, tbIdent
table.Identifier, awsCfg aws.Config) {
t.Helper()
diff --git a/catalog/hadoop/hadoop_test.go b/catalog/hadoop/hadoop_test.go
index 9725bc78c..f35bb2fe8 100644
--- a/catalog/hadoop/hadoop_test.go
+++ b/catalog/hadoop/hadoop_test.go
@@ -2822,12 +2822,25 @@ func (s *HadoopCatalogTestSuite)
TestCommitTableRequirementFailure() {
s.Contains(err.Error(), "UUID")
}
+func (s *HadoopCatalogTestSuite) metadataDirEntries(ident table.Identifier)
[]string {
+ entries, err := os.ReadDir(s.cat.metadataDir(ident))
+ s.Require().NoError(err)
+ names := make([]string, 0, len(entries))
+ for _, e := range entries {
+ names = append(names, e.Name())
+ }
+
+ return names
+}
+
func (s *HadoopCatalogTestSuite) TestCommitTableLocationChange() {
ctx := context.Background()
s.createTestTable("ns", "tbl")
+ ident := []string{"ns", "tbl"}
+ before := s.metadataDirEntries(ident)
_, _, err := s.cat.CommitTable(
- ctx, []string{"ns", "tbl"},
+ ctx, ident,
nil,
[]table.Update{
table.NewSetLocationUpdate("/some/other/location"),
@@ -2835,14 +2848,18 @@ func (s *HadoopCatalogTestSuite)
TestCommitTableLocationChange() {
)
s.Require().Error(err)
s.Contains(err.Error(), "location cannot be changed")
+ s.ErrorIs(err, table.ErrCommitRejected)
+ s.Equal(before, s.metadataDirEntries(ident), "a rejected commit must
not write metadata")
}
func (s *HadoopCatalogTestSuite) TestCommitTableWriteMetadataLocation() {
ctx := context.Background()
s.createTestTable("ns", "tbl")
+ ident := []string{"ns", "tbl"}
+ before := s.metadataDirEntries(ident)
_, _, err := s.cat.CommitTable(
- ctx, []string{"ns", "tbl"},
+ ctx, ident,
nil,
[]table.Update{
table.NewSetPropertiesUpdate(iceberg.Properties{
@@ -2852,6 +2869,8 @@ func (s *HadoopCatalogTestSuite)
TestCommitTableWriteMetadataLocation() {
)
s.Require().Error(err)
s.Contains(err.Error(), "write.metadata.location")
+ s.ErrorIs(err, table.ErrCommitRejected)
+ s.Equal(before, s.metadataDirEntries(ident), "a rejected commit must
not write metadata")
}
func (s *HadoopCatalogTestSuite) TestCommitTableNoOrphanedTempFiles() {
diff --git a/catalog/hadoop/tables.go b/catalog/hadoop/tables.go
index 7c7ae6055..783fc5005 100644
--- a/catalog/hadoop/tables.go
+++ b/catalog/hadoop/tables.go
@@ -195,7 +195,7 @@ func (c *Catalog) CommitTable(ctx context.Context, ident
table.Identifier, reqs
currentMetadata = current.Metadata()
}
if err := internal.ValidateRequirements(ident, currentMetadata, reqs);
err != nil {
- return nil, "", err
+ return nil, "",
internal.RejectUnlessConflict(fmt.Errorf("hadoop catalog: %w", err))
}
// Step 3: Apply updates to produce new metadata.
@@ -208,23 +208,23 @@ func (c *Catalog) CommitTable(ctx context.Context, ident
table.Identifier, reqs
} else {
baseMeta, err = table.NewMetadata(iceberg.NewSchema(0), nil,
table.UnsortedSortOrder, "", nil)
if err != nil {
- return nil, "", fmt.Errorf("hadoop catalog: failed to
create base metadata: %w", err)
+ return nil, "",
internal.RejectUnlessConflict(fmt.Errorf("hadoop catalog: failed to create base
metadata: %w", err))
}
}
updated, err := internal.UpdateTableMetadata(baseMeta, updates,
currentMetadataLoc)
if err != nil {
- return nil, "", fmt.Errorf("hadoop catalog: failed to apply
updates: %w", err)
+ return nil, "",
internal.RejectUnlessConflict(fmt.Errorf("hadoop catalog: failed to apply
updates: %w", err))
}
// Step 4: Validate table location has not changed.
if current != nil && updated.Location() != current.Location() {
- return nil, "", errors.New("hadoop catalog: table location
cannot be changed")
+ return nil, "",
internal.RejectUnlessConflict(errors.New("hadoop catalog: table location cannot
be changed"))
}
// Step 5: Reject write.metadata.location property.
if v := updated.Properties().Get(table.WriteMetadataLocationKey, ""); v
!= "" {
- return nil, "", fmt.Errorf("hadoop catalog: %s property is not
supported", table.WriteMetadataLocationKey)
+ return nil, "",
internal.RejectUnlessConflict(fmt.Errorf("hadoop catalog: %s property is not
supported", table.WriteMetadataLocationKey))
}
// Step 6: Determine next version number.
diff --git a/catalog/hive/hive_test.go b/catalog/hive/hive_test.go
index f818bd997..073386d77 100644
--- a/catalog/hive/hive_test.go
+++ b/catalog/hive/hive_test.go
@@ -847,6 +847,47 @@ func TestHiveCommitTableSynchronizesHMSMetadata(t
*testing.T) {
mockClient.AssertExpectations(t)
}
+func TestHiveCommitTableRejectsOnlyFailuresBeforeMetadataWrite(t *testing.T) {
+ ctx := context.Background()
+ ident := TableIdentifier("test_database", "test_table")
+ newCatalog := func(t *testing.T) (*Catalog, *mockHiveClient) {
+ tableLocation := filepath.Join(t.TempDir(), "test_table")
+ metadataLocation := filepath.Join(tableLocation, "metadata",
"v1.metadata.json")
+ writeHivePurgeTableMetadata(t, iceio.LocalFS{}, tableLocation,
metadataLocation, nil)
+
+ mockClient := &mockHiveClient{}
+ expectImmediateTableLock(mockClient, 1)
+ mockClient.On("GetTable", mock.Anything, "test_database",
"test_table").
+ Return(hivePurgeTable(metadataLocation, tableLocation),
nil).Once()
+
+ return NewCatalogWithClient(mockClient, iceberg.Properties{}),
mockClient
+ }
+
+ t.Run("staging failure", func(t *testing.T) {
+ cat, mockClient := newCatalog(t)
+
+ _, _, err := cat.CommitTable(ctx, ident, nil,
[]table.Update{table.NewSetCurrentSchemaUpdate(99)})
+ require.ErrorIs(t, err, table.ErrCommitRejected)
+ require.NotErrorIs(t, err, table.ErrCommitFailed)
+ mockClient.AssertNotCalled(t, "AlterTable", mock.Anything,
mock.Anything, mock.Anything, mock.Anything)
+ mockClient.AssertExpectations(t)
+ })
+
+ t.Run("hms alter failure after metadata write", func(t *testing.T) {
+ cat, mockClient := newCatalog(t)
+ mockClient.On("AlterTable", mock.Anything, "test_database",
"test_table", mock.AnythingOfType("*hive_metastore.Table")).
+ Return(errors.New("metastore unavailable")).Once()
+
+ _, _, err := cat.CommitTable(ctx, ident, nil, []table.Update{
+ table.NewSetPropertiesUpdate(iceberg.Properties{"key":
"value"}),
+ })
+ require.ErrorContains(t, err, "metastore unavailable")
+ require.NotErrorIs(t, err, table.ErrCommitRejected)
+ require.NotErrorIs(t, err, table.ErrCommitFailed)
+ mockClient.AssertExpectations(t)
+ })
+}
+
func TestHiveDropTableLockFailureIsRetryable(t *testing.T) {
mockClient := &mockHiveClient{}
mockClient.On("Lock", mock.Anything,
mock.AnythingOfType("*hive_metastore.LockRequest")).
diff --git a/catalog/internal/update_stage_test.go
b/catalog/internal/update_stage_test.go
index 0c81b2205..c62631bd8 100644
--- a/catalog/internal/update_stage_test.go
+++ b/catalog/internal/update_stage_test.go
@@ -92,6 +92,69 @@ func
TestUpdateAndStageTableRejectsEmptyMetadataLocationForExistingTable(t *test
require.Contains(t, err.Error(), "invalid metadata location")
}
+func TestUpdateAndStageTableClassifiesFailures(t *testing.T) {
+ metadata, err := table.NewMetadata(iceberg.NewSchema(0), nil,
table.UnsortedSortOrder, "file:///tmp/table",
+ iceberg.Properties{table.PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+ current := table.New(table.Identifier{"db", "tbl"}, metadata,
"file:///tmp/table/metadata/v1.metadata.json", nil, nil)
+ parent := int64(1)
+ errTag := errors.New("requirement failed: ref \"t\" is a tag")
+
+ tests := []struct {
+ name string
+ current *table.Table
+ reqs []table.Requirement
+ updates []table.Update
+ want []error
+ notWant error
+ }{
+ {
+ name: "metadata builder error",
+ current: current,
+ updates:
[]table.Update{table.NewSetCurrentSchemaUpdate(99)},
+ want: []error{table.ErrCommitRejected},
+ notWant: table.ErrCommitFailed,
+ },
+ {
+ name: "requirement failure that is not a conflict",
+ current: current,
+ reqs:
[]table.Requirement{failingRequirement{errTag}},
+ want: []error{table.ErrCommitRejected, errTag},
+ notWant: table.ErrCommitFailed,
+ },
+ {
+ name: "requirement failure",
+ current: current,
+ reqs:
[]table.Requirement{table.AssertCurrentSchemaID(99)},
+ want: []error{table.ErrCommitFailed},
+ notWant: table.ErrCommitRejected,
+ },
+ {
+ name: "stale sequence number",
+ current: current,
+ updates:
[]table.Update{table.NewAddSnapshotUpdate(&table.Snapshot{
+ SnapshotID: 2,
+ ParentSnapshotID: &parent,
+ SequenceNumber: metadata.LastSequenceNumber(),
+ TimestampMs: metadata.LastUpdatedMillis(),
+ ManifestList:
"file:///tmp/table/metadata/snap-2.avro",
+ })},
+ want: []error{table.ErrCommitFailed},
+ notWant: table.ErrCommitRejected,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ staged, err := UpdateAndStageTable(t.Context(), nil,
tt.current, table.Identifier{"db", "tbl"}, tt.reqs, tt.updates, nil)
+ require.Nil(t, staged)
+ for _, want := range tt.want {
+ require.ErrorIs(t, err, want)
+ }
+ require.NotErrorIs(t, err, tt.notWant)
+ })
+ }
+}
+
func TestParseMetadataVersionAcceptsLegacyMetadataLocation(t *testing.T) {
require.Equal(t, 1,
ParseMetadataVersion("file:///tmp/table/metadata/v1.metadata.json"))
require.Equal(t, 2,
ParseMetadataVersion("file:///tmp/table/metadata/v2.gz.metadata.json"))
diff --git a/catalog/internal/utils.go b/catalog/internal/utils.go
index 3ca437076..4fc67cb0f 100644
--- a/catalog/internal/utils.go
+++ b/catalog/internal/utils.go
@@ -281,7 +281,22 @@ func ValidateRequirements(ident table.Identifier, current
table.Metadata, reqs [
return nil
}
-func UpdateAndStageTable(ctx context.Context, catprops iceberg.Properties,
current *table.Table, ident table.Identifier, reqs []table.Requirement, updates
[]table.Update, cat table.CatalogIO) (*table.StagedTable, error) {
+// RejectUnlessConflict wraps non-conflict errors with table.ErrCommitRejected.
+// Use it only before the catalog has written anything.
+func RejectUnlessConflict(err error) error {
+ if err == nil || errors.Is(err, table.ErrCommitFailed) {
+ return err
+ }
+
+ return fmt.Errorf("%w: %w", table.ErrCommitRejected, err)
+}
+
+// UpdateAndStageTable applies updates to the current table without writing
+// anything. Every error it returns matches table.ErrCommitRejected unless it
+// matches table.ErrCommitFailed.
+func UpdateAndStageTable(ctx context.Context, catprops iceberg.Properties,
current *table.Table, ident table.Identifier, reqs []table.Requirement, updates
[]table.Update, cat table.CatalogIO) (_ *table.StagedTable, err error) {
+ defer func() { err = RejectUnlessConflict(err) }()
+
var (
baseMeta table.Metadata
currentMetadata table.Metadata
@@ -293,7 +308,6 @@ func UpdateAndStageTable(ctx context.Context, catprops
iceberg.Properties, curre
baseMeta = currentMetadata
metadataLoc = current.MetadataLocation()
} else {
- var err error
baseMeta, err = table.NewMetadata(iceberg.NewSchema(0), nil,
table.UnsortedSortOrder, "", nil)
if err != nil {
return nil, err
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index 1f3c007b9..304fa03df 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -116,6 +116,12 @@ var (
ErrCommitStateUnknown = fmt.Errorf("%w: commit failed due to unknown
reason", ErrRESTError)
ErrOAuthError = fmt.Errorf("%w: oauth error", ErrRESTError)
+ // These statuses mean the catalog did not change the table.
+ errCommitBadRequest = fmt.Errorf("%w: %w", ErrBadRequest,
table.ErrCommitRejected)
+ errCommitUnauthorized = fmt.Errorf("%w: %w", ErrUnauthorized,
table.ErrCommitRejected)
+ errCommitForbidden = fmt.Errorf("%w: %w", ErrForbidden,
table.ErrCommitRejected)
+ errCommitNoSuchTable = fmt.Errorf("%w: %w", catalog.ErrNoSuchTable,
table.ErrCommitRejected)
+
// errNotModified reports a 304 to a request sent with withIfNoneMatch.
errNotModified = errors.New("not modified")
)
@@ -1803,7 +1809,10 @@ func (r *Catalog) CommitTable(ctx context.Context, ident
table.Identifier, requi
ret, err := doPost[payload, commitTableResponse](ctx, r.baseURI, path,
payload{Identifier: restIdentifier, Requirements: requirements,
Updates: updates}, r.cl,
map[int]error{
- http.StatusNotFound: catalog.ErrNoSuchTable,
+ http.StatusBadRequest: errCommitBadRequest,
+ http.StatusUnauthorized: errCommitUnauthorized,
+ http.StatusForbidden: errCommitForbidden,
+ http.StatusNotFound: errCommitNoSuchTable,
http.StatusConflict: ErrCommitFailed,
http.StatusInternalServerError: ErrCommitStateUnknown,
http.StatusBadGateway: ErrCommitStateUnknown,
diff --git a/catalog/rest/rest_test.go b/catalog/rest/rest_test.go
index a46756c12..9eae1cbf5 100644
--- a/catalog/rest/rest_test.go
+++ b/catalog/rest/rest_test.go
@@ -4389,6 +4389,73 @@ func (r *RestCatalogSuite)
TestCommitTableErrorBodyKeepsCommitStateUnknown() {
}
}
+func (r *RestCatalogSuite) TestCommitTableClassifiesRejections() {
+ var statusCode int
+ var body []byte
+
+ r.mux.HandleFunc("/v1/oauth/tokens", func(w http.ResponseWriter, req
*http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(map[string]any{
+ "access_token": TestToken, "token_type": "Bearer",
"expires_in": 3600,
+ })
+ })
+
+ r.mux.HandleFunc("/v1/namespaces/db/tables/tbl", func(w
http.ResponseWriter, req *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(statusCode)
+ w.Write(body)
+ })
+
+ cat, err := rest.NewCatalog(context.Background(), "rest", r.srv.URL,
+ rest.WithCredential(TestCreds))
+ r.Require().NoError(err)
+
+ for _, tc := range []struct {
+ status int
+ errType string
+ want []error
+ rejected bool
+ }{
+ {http.StatusBadRequest, "BadRequestException",
[]error{rest.ErrBadRequest, rest.ErrRESTError}, true},
+ {http.StatusUnauthorized, "NotAuthorizedException",
[]error{rest.ErrUnauthorized, rest.ErrRESTError}, true},
+ {http.StatusForbidden, "ForbiddenException",
[]error{rest.ErrForbidden, rest.ErrRESTError}, true},
+ {http.StatusNotFound, "NoSuchTableException",
[]error{catalog.ErrNoSuchTable}, true},
+ {http.StatusConflict, "CommitFailedException",
[]error{table.ErrCommitFailed, rest.ErrRESTError}, false},
+ {http.StatusInternalServerError, "CommitStateUnknownException",
[]error{rest.ErrCommitStateUnknown}, false},
+ } {
+ jsonBody, err := json.Marshal(map[string]any{
+ "error": map[string]any{"message":
http.StatusText(tc.status), "type": tc.errType, "code": tc.status},
+ })
+ r.Require().NoError(err)
+
+ for _, b := range []struct {
+ name string
+ body []byte
+ }{
+ {"JSON body", jsonBody},
+ {"undecodable body",
[]byte(`{"error":{"message":"boom`)},
+ } {
+ statusCode, body = tc.status, b.body
+ r.Run(strconv.Itoa(tc.status)+"/"+b.name, func() {
+ _, _, err :=
cat.CommitTable(context.Background(),
+ table.Identifier{"db", "tbl"},
+ []table.Requirement{},
+ []table.Update{},
+ )
+ r.Require().Error(err)
+ for _, want := range tc.want {
+ r.ErrorIs(err, want)
+ }
+ if tc.rejected {
+ r.ErrorIs(err, table.ErrCommitRejected)
+ } else {
+ r.NotErrorIs(err,
table.ErrCommitRejected)
+ }
+ })
+ }
+ }
+}
+
func (r *RestCatalogSuite) TestUpdateTableErrCommitStateUnknown() {
var statusCode int
diff --git a/catalog/sql/sql_test.go b/catalog/sql/sql_test.go
index b8a39360e..747b10608 100644
--- a/catalog/sql/sql_test.go
+++ b/catalog/sql/sql_test.go
@@ -1732,6 +1732,74 @@ func (s *SqliteCatalogTestSuite)
TestStaleSchemaUpdateIsRefusedOnRecommit() {
s.Equal(peerMeta.MetadataLocation(), current.MetadataLocation(),
"nothing may be committed")
}
+func (s *SqliteCatalogTestSuite)
TestRejectedUpdateLeavesTransactionResumable() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.randomTableIdentifier()
+ s.Require().NoError(cat.CreateNamespace(ctx,
catalog.NamespaceFromIdent(tblID), nil))
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String,
Required: true,
+ })
+ stale, err := cat.CreateTable(ctx, tblID, schema,
catalog.WithProperties(iceberg.Properties{
+ table.PropertyFormatVersion: "1",
+ }))
+ s.Require().NoError(err)
+
+ tx := stale.NewTransaction()
+ s.Require().NoError(tx.UpgradeFormatVersion(2))
+
+ peer := stale.NewTransaction()
+ s.Require().NoError(peer.UpgradeFormatVersion(3))
+ peerTbl, err := peer.Commit(ctx)
+ s.Require().NoError(err)
+
+ // The catalog's metadata builder refuses to downgrade the peer's table.
+ _, err = tx.Commit(ctx)
+ s.Require().ErrorIs(err, table.ErrCommitRejected)
+ s.Require().ErrorIs(err, iceberg.ErrInvalidArgument)
+ s.Require().NotErrorIs(err, table.ErrTransactionUnusable)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal(peerTbl.MetadataLocation(), current.MetadataLocation(),
"nothing may be committed")
+
+ _, err = tx.Commit(ctx)
+ s.Require().ErrorIs(err, table.ErrCommitRejected)
+ s.Require().NotErrorIs(err, table.ErrTransactionUnusable)
+}
+
+func (s *SqliteCatalogTestSuite)
TestCatalogUpdateFailureAfterMetadataWriteEndsTransaction() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.randomTableIdentifier()
+ s.Require().NoError(cat.CreateNamespace(ctx,
catalog.NamespaceFromIdent(tblID), nil))
+ tbl, err := cat.CreateTable(ctx, tblID, tableSchemaNested)
+ s.Require().NoError(err)
+
+ sqlDB := s.getDB()
+ defer sqlDB.Close()
+ _, err = sqlDB.Exec(`
+ CREATE TRIGGER fail_catalog_update
+ BEFORE UPDATE ON iceberg_tables
+ BEGIN
+ SELECT RAISE(FAIL, 'forced update failure');
+ END
+ `)
+ s.Require().NoError(err)
+
+ tx := tbl.NewTransaction()
+ s.Require().NoError(tx.SetProperties(iceberg.Properties{"key":
"value"}))
+ _, err = tx.Commit(ctx)
+ s.Require().ErrorContains(err, "forced update failure")
+ s.Require().NotErrorIs(err, table.ErrCommitRejected)
+ s.Require().NotErrorIs(err, table.ErrCommitFailed)
+
+ _, err = tx.Commit(ctx)
+ s.Require().EqualError(err, "transaction has already been committed")
+}
+
func (s *SqliteCatalogTestSuite) TestConflictClearedBeforeRecommitLands() {
ctx := context.Background()
cat := s.getCatalogSqlite()
diff --git a/table/commit_retry_test.go b/table/commit_retry_test.go
index f4f5b5d96..3501e0d3f 100644
--- a/table/commit_retry_test.go
+++ b/table/commit_retry_test.go
@@ -231,6 +231,23 @@ func TestDoCommit_DoesNotRetryUnrelatedError(t *testing.T)
{
assert.Equal(t, int32(1), cat.attempts.Load())
}
+func TestDoCommit_RetriesConflictThatIsAlsoRejected(t *testing.T) {
+ cat := &flakyCatalog{
+ failUntilAttempt: 1,
+ failWith: fmt.Errorf("%w: %w", ErrCommitFailed,
ErrCommitRejected),
+ }
+ tbl := newRetryTestTable(t, cat, iceberg.Properties{
+ CommitNumRetriesKey: "2",
+ CommitMinRetryWaitMsKey: "1",
+ CommitMaxRetryWaitMsKey: "2",
+ })
+ cat.metadata = tbl.Metadata()
+
+ _, err := tbl.doCommit(t.Context(), nil, nil)
+ require.NoError(t, err)
+ assert.Equal(t, int32(2), cat.attempts.Load(), "a rejection that is
also a conflict must be retried")
+}
+
func TestDoCommit_RespectsContextCancellation(t *testing.T) {
cat := &flakyCatalog{
failUntilAttempt: 100,
@@ -1270,6 +1287,46 @@ func TestTransactionCommit_ResumeOfNonReplayableCommit(t
*testing.T) {
})
}
+func TestTransactionCommit_ResumeOfRejectedNonReplayableCommit(t *testing.T) {
+ head := int64(100)
+ rejected := fmt.Errorf("%w: simulated metadata update failure",
ErrCommitRejected)
+ newRejectedTx := func(t *testing.T) (*Transaction, *sequentialCatalog,
Metadata) {
+ base := newConflictTestMetadataWithProps(t, &head, nil)
+ cat := &sequentialCatalog{metadata: base, errs:
[]error{rejected}}
+ tbl := New(Identifier{"db", "no-replay"}, base, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, cat)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"k":
"v"}))
+ tx.noReplay = true
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitRejected)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+
+ return tx, cat, base
+ }
+
+ t.Run("branch unchanged lands", func(t *testing.T) {
+ tx, cat, _ := newRejectedTx(t)
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, "v", committed.Properties()["k"])
+ assert.Equal(t, int32(2), cat.attempts.Load())
+ })
+
+ t.Run("branch moved is unusable", func(t *testing.T) {
+ tx, cat, base := newRejectedTx(t)
+ cat.loadMeta = graftSnapshotOnto(t, base, MainBranch, 200)
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ assert.EqualError(t, err, "transaction cannot be committed
again: requirement no longer holds after refresh ("+noReplayHint+"): "+
+ `requirement failed: branch "main" has changed:
expected id 100, found 200`)
+ assert.Equal(t, int32(1), cat.attempts.Load(), "the resume must
stop before the catalog")
+ })
+}
+
// doCommit exhausts its own retry loop on ErrCommitFailed, the transaction
must still be left retriable.
func TestTransactionCommit_RetriableAfterExhaustedInternalRetries(t
*testing.T) {
// numRetries == 2 → doCommit makes 3 attempts, all clean conflicts, so
it
diff --git a/table/retry_manifest_cleanup_test.go
b/table/retry_manifest_cleanup_test.go
index 864678751..dc1ae36f7 100644
--- a/table/retry_manifest_cleanup_test.go
+++ b/table/retry_manifest_cleanup_test.go
@@ -20,6 +20,7 @@ package table
import (
"context"
"errors"
+ "fmt"
"maps"
"path/filepath"
"slices"
@@ -327,6 +328,63 @@ func
TestFailedCommitKeepsOnlyStagedFilesUntilResumeLands(t *testing.T) {
}
}
+func TestRejectedCommitDeletesItsRebuildsAndStaysResumable(t *testing.T) {
+ for _, p := range retryCleanupProducers {
+ t.Run(p.name, func(t *testing.T) {
+ f := newRetryCleanupFixture(t, "2", p.props)
+ fs := &unreadableListIO{memIO: f.wfs}
+ rejected := fmt.Errorf("%w: simulated metadata update
failure", ErrCommitRejected)
+ cat := &peerHeadsCatalog{current: f.h0, heads:
[]Metadata{f.h1, f.h2}, errs: []error{ErrCommitFailed, rejected}}
+ tx := newOCCTable(t, f.h0, fs, cat).NewTransaction()
+ p.stage(t, tx)
+ staged := f.newFiles()
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitRejected)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+ var unknown *commitStateUnknownError
+ require.NotErrorAs(t, err, &unknown)
+ require.Equal(t, 2, cat.attempts, "a rejection must not
be retried")
+ assert.Equal(t, staged, f.newFiles(), "only the staged
files may remain, and all of them")
+ var lists, manifests int
+ for _, path := range fs.removed {
+ if strings.HasPrefix(filepath.Base(path),
"snap-") {
+ lists++
+ } else {
+ manifests++
+ }
+ }
+ assert.NotZero(t, lists, "the rebuilt manifest list
must be deleted")
+ assert.NotZero(t, manifests, "the rebuilt manifests
must be deleted")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.Equal(t, 3, cat.attempts)
+ head, headManifestFiles := headManifests(t, committed,
f.wfs)
+ p.checkHead(t, head, headManifestFiles)
+ f.assertOnlyCommittedFilesAdded(t, committed)
+ })
+ }
+}
+
+func TestRejectedExpiryDeletesNothing(t *testing.T) {
+ f := newRetryCleanupFixture(t, "2", nil)
+ rejected := fmt.Errorf("%w: simulated metadata update failure",
ErrCommitRejected)
+ cat := &peerHeadsCatalog{current: f.h2, heads: []Metadata{f.h2}, errs:
[]error{rejected}}
+ tx := newOCCTable(t, f.h2, f.wfs, cat).NewTransaction()
+ expired := f.h0.CurrentSnapshot()
+ require.NoError(t,
tx.apply([]Update{NewRemoveSnapshotsUpdate([]int64{expired.SnapshotID}, true)},
nil))
+ before := slices.Sorted(maps.Keys(f.wfs.files))
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitRejected)
+ assert.Equal(t, before, slices.Sorted(maps.Keys(f.wfs.files)), "a
rejected expiry must not run its post-commit deletion")
+
+ _, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.NotContains(t, f.wfs.files, expired.ManifestList, "the expiry
deletes once its commit lands")
+}
+
// Cleanup must not guess which files a landed commit references when it
// cannot read a committed manifest list.
func TestCommitCleanupSkippedWhenCommittedListUnreadable(t *testing.T) {
diff --git a/table/rewrite_manifests_test.go b/table/rewrite_manifests_test.go
index 87de65987..32445a82a 100644
--- a/table/rewrite_manifests_test.go
+++ b/table/rewrite_manifests_test.go
@@ -865,12 +865,14 @@ func mergedManifestsCreated(track *trackingFS,
preExisting map[string]struct{})
// LoadTable advances the head to the next pre-built metadata, so the rewrite
// must rebuild — and supersede the previous attempt's merged manifest — on
// every retry, not just the first (which a static head would not exercise). It
-// returns ErrCommitFailed for the first `failures` CommitTable calls.
+// returns ErrCommitFailed for the first `failures` CommitTable calls, then
+// returns reject once if it is non-nil.
type stagedHeadCatalog struct {
current table.Metadata
heads []table.Metadata
headIdx int
failures int
+ reject error
commitTableCalls atomic.Int32
location string
fsF func(context.Context) (iceio.IO, error)
@@ -891,6 +893,9 @@ func (c *stagedHeadCatalog) CommitTable(_ context.Context,
_ table.Identifier, _
if int(n) <= c.failures {
return nil, "", fmt.Errorf("%w: simulated 409 conflict",
table.ErrCommitFailed)
}
+ if int(n) == c.failures+1 && c.reject != nil {
+ return nil, "", c.reject
+ }
meta, err := table.UpdateTableMetadata(c.current, updates, "")
if err != nil {
@@ -1046,6 +1051,49 @@ func
TestRewriteManifestsKeepsMergedManifestsUntilResumeLands(t *testing.T) {
}
}
+func TestRewriteManifestsRejectionKeepsMergedManifests(t *testing.T) {
+ ctx := context.Background()
+ dir := t.TempDir()
+ track := &trackingFS{}
+ fsF := func(_ context.Context) (iceio.IO, error) { return track, nil }
+
+ h0, h1, _ := stagedRewriteHeads(t, ctx, dir, "2", fsF, track)
+
+ rejected := fmt.Errorf("%w: simulated metadata update failure",
table.ErrCommitRejected)
+ cat := &stagedHeadCatalog{current: h0, heads: []table.Metadata{h1},
failures: 1, reject: rejected, location: dir, fsF: fsF}
+ tbl := table.New(table.Identifier{"default", "staged"}, h0,
dir+"/metadata/00000.json", fsF, cat)
+
+ preExisting := track.snapshotCreated()
+
+ txn := tbl.NewTransaction()
+ res, err := txn.RewriteManifests(ctx)
+ require.NoError(t, err)
+ staged := track.snapshotCreated()
+ _, err = txn.Commit(ctx)
+ require.ErrorIs(t, err, table.ErrCommitRejected)
+ require.NotErrorIs(t, err, table.ErrTransactionUnusable)
+ require.EqualValues(t, 2, cat.commitTableCalls.Load(), "a rejection
must not be retried")
+
+ merged := mergedManifestsCreated(track, preExisting)
+ require.Len(t, merged, 2, "the staged rewrite and its rebuild each
write one merged manifest")
+ for _, p := range merged {
+ assert.Falsef(t, track.wasRemoved(p), "merged manifest %s must
be kept while the transaction can commit", p)
+ }
+ var rebuiltLists int
+ for p := range track.snapshotCreated() {
+ if _, ok := staged[p]; ok ||
!strings.HasPrefix(filepath.Base(p), "snap-") {
+ continue
+ }
+ rebuiltLists++
+ assert.Truef(t, track.wasRemoved(p), "rebuilt manifest list %s
must be deleted", p)
+ }
+ assert.Equal(t, 1, rebuiltLists)
+
+ committed, err := txn.Commit(ctx)
+ require.NoError(t, err)
+ assertCommittedManifestsExist(t, committed, track, res)
+}
+
func committedManifestPaths(t *testing.T, committed *table.Table, fs iceio.IO)
map[string]struct{} {
t.Helper()
manifests, err := committed.CurrentSnapshot().Manifests(fs)
diff --git a/table/table.go b/table/table.go
index 4033709c3..8d337aad1 100644
--- a/table/table.go
+++ b/table/table.go
@@ -62,6 +62,18 @@ import (
// stops.
var ErrCommitFailed = errors.New("commit failed, refresh and try again")
+// ErrCommitRejected means the catalog rejected a commit without writing
+// anything. It is the Go equivalent of Java's CleanableFailure.
+// Unlike Java, the REST catalog treats every 400 commit response as a
+// rejection, including one with error type IllegalArgumentException, since the
+// server rejected the request either way.
+//
+// Transaction.Commit does not retry it, but a later Commit can try again.
+// Staged files are retained. Files created during retries are deleted unless
+// the transaction can reuse them. Errors that also match ErrCommitFailed
+// are treated as conflicts.
+var ErrCommitRejected = errors.New("commit rejected")
+
// ErrTransactionUnusable means this transaction can't be committed again,
// because after a refresh the commit fails a check that every later attempt
// would fail too. An error can match both ErrTransactionUnusable and
@@ -902,7 +914,11 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// may have actually succeeded — retrying could duplicate work.
// Suppress orphan cleanup for the same reason: one of the
orphaned
// manifest-list files may actually be the snapshot the catalog
accepted.
+ // Rejected commits cannot reference these files, so cleanup is
safe.
if !errors.Is(err, ErrCommitFailed) {
+ if errors.Is(err, ErrCommitRejected) {
+ return nil, err
+ }
cleanupOrphans = false
return nil, &commitStateUnknownError{err: err}
diff --git a/website/src/concurrent-writes.md b/website/src/concurrent-writes.md
index f3b0756cc..2dbca534a 100644
--- a/website/src/concurrent-writes.md
+++ b/website/src/concurrent-writes.md
@@ -55,9 +55,16 @@ case errors.Is(err, table.ErrCommitDiverged), errors.Is(err,
table.ErrTransactio
// unrecoverable: rebuild the operation from the latest table
case errors.Is(err, table.ErrCommitFailed):
// retriable: refresh and try again (the retry loop does this for you)
+case errors.Is(err, table.ErrCommitRejected):
+ // the catalog wrote nothing: fix the cause, then call Commit again
}
```
+A commit can also fail with `table.ErrCommitRejected`, which means the catalog
+rejected it before writing anything, for example an update the catalog can't
+apply or a REST 400, 401, 403 or 404. The commit is not retried, and the
+transaction can still be committed.
+
## Isolation levels
Delete and update operations are validated under an isolation level, set per
diff --git a/website/src/releases.md b/website/src/releases.md
index bc247b2c0..dd9e3eda6 100644
--- a/website/src/releases.md
+++ b/website/src/releases.md
@@ -96,6 +96,7 @@ Per-release notes (highlights, breaking changes,
contributors) are published on
- **Row-lineage column predicate renamed.** Use `iceberg.IsRowLineageColumn`
instead of `iceberg.IsMetadataColumn` to identify `_row_id` and
`_last_updated_sequence_number`. Use `iceberg.IsReservedFieldID` to check the
full reserved metadata field ID range.
- **File writes and directory walks now take an operation context.** Update
`WriteFileIO` and `ListableIO` implementations and calls to `Create`,
`WriteFile`, and `WalkDir` to pass `context.Context`; cloud writes and listings
use it for cancellation.
- **In and NotIn follow SQL semantics for NULL.** A multi-valued `NotIn`,
`NOT(In(...))` and `NOT(NotIn(...))` no longer match rows where the column is
NULL, consistent with `NotEqual`. A predicate under NOT on a column missing
from an older data file no longer matches every row of that file; those rows
match as a stored NULL would. Java's `Evaluator` still matches NULL for `notEq`
and `notIn`, see
[apache/iceberg#18701](https://github.com/apache/iceberg/issues/18701).
+- **Commit rejections now match `table.ErrCommitRejected`.** SQL, Glue, Hive
and Hadoop failures while preparing metadata, plus REST commit responses 400,
401, 403 and 404, leave transactions committable (after a REST 404, the next
`Commit` fails when it reloads the table). Existing `errors.Is` matches keep
working, but SQL, Glue, Hive and Hadoop rejection messages now start with
`commit rejected: `, so code that matches error text may need updating.
## Verifying and producing releases
