laskoviymishka commented on code in PR #1466:
URL: https://github.com/apache/iceberg-go/pull/1466#discussion_r3601663748
##########
catalog/sql/sql.go:
##########
@@ -293,6 +294,72 @@ func withWriteTx(ctx context.Context, db *bun.DB, fn
func(context.Context, bun.T
})
}
+const (
+ serializableWriteMaxAttempts = 3
+ serializableWriteRetryDelay = 10 * time.Millisecond
+)
+
+func withSerializableWriteTx(ctx context.Context, db *bun.DB, fn
func(context.Context, bun.Tx) error) error {
+ return retrySerializableWriteTx(ctx, func() error {
+ return db.RunInTx(ctx, &sql.TxOptions{Isolation:
sql.LevelSerializable}, func(ctx context.Context, tx bun.Tx) error {
+ return fn(ctx, tx)
+ })
+ })
+}
+
+func retrySerializableWriteTx(ctx context.Context, run func() error) error {
+ var err error
+ for attempt := 0; attempt < serializableWriteMaxAttempts; attempt++ {
+ err = run()
+ if err == nil || !isRetryableSerializableError(err) {
+ return err
+ }
+ if attempt == serializableWriteMaxAttempts-1 {
+ break
+ }
+
+ delay := serializableWriteRetryDelay << attempt
+ timer := time.NewTimer(delay)
+ select {
+ case <-ctx.Done():
+ if !timer.Stop() {
+ <-timer.C
+ }
+
+ return context.Cause(ctx)
+ case <-timer.C:
+ }
+ }
+
+ return err
+}
+
+func isRetryableSerializableError(err error) bool {
+ var stateErr interface{ SQLState() string }
+ if errors.As(err, &stateErr) {
+ switch stateErr.SQLState() {
+ case "40001", "40P01":
+ return true
+ }
+ }
+
Review Comment:
I think this `interface{ Code() int }` branch never matches the real driver.
mattn/go-sqlite3 implements `Code() sqlite3.ErrNo` and modernc.org/sqlite (what
`sqliteshim` wraps) uses its own named `ErrorCode` — both are `int` underneath,
but interface satisfaction is by exact signature, so `Code() ErrNo` doesn't
satisfy `Code() int`. `errors.As` falls through and we're left relying on the
`database is locked` string match.
The unit test only passes because `sqliteCodeTestError.Code()` returns a
bare `int`, so it exercises a shape the real driver never produces. Net effect:
the SQLITE_BUSY/LOCKED retry doesn't actually fire for a production SQLite
catalog.
I'd switch this to `errors.As` against the concrete driver error type (or
drop the `Code()` branch and lean on the string matches, with a comment on
why), and add a test that runs a real driver error through
`isRetryableSerializableError` so this can't silently regress. wdyt?
##########
catalog/sql/sql.go:
##########
@@ -293,6 +294,72 @@ func withWriteTx(ctx context.Context, db *bun.DB, fn
func(context.Context, bun.T
})
}
+const (
+ serializableWriteMaxAttempts = 3
+ serializableWriteRetryDelay = 10 * time.Millisecond
+)
+
+func withSerializableWriteTx(ctx context.Context, db *bun.DB, fn
func(context.Context, bun.Tx) error) error {
+ return retrySerializableWriteTx(ctx, func() error {
+ return db.RunInTx(ctx, &sql.TxOptions{Isolation:
sql.LevelSerializable}, func(ctx context.Context, tx bun.Tx) error {
+ return fn(ctx, tx)
+ })
+ })
+}
+
+func retrySerializableWriteTx(ctx context.Context, run func() error) error {
+ var err error
+ for attempt := 0; attempt < serializableWriteMaxAttempts; attempt++ {
+ err = run()
+ if err == nil || !isRetryableSerializableError(err) {
+ return err
+ }
+ if attempt == serializableWriteMaxAttempts-1 {
+ break
+ }
+
+ delay := serializableWriteRetryDelay << attempt
+ timer := time.NewTimer(delay)
+ select {
+ case <-ctx.Done():
+ if !timer.Stop() {
+ <-timer.C
+ }
+
+ return context.Cause(ctx)
+ case <-timer.C:
+ }
+ }
+
+ return err
+}
+
+func isRetryableSerializableError(err error) bool {
+ var stateErr interface{ SQLState() string }
+ if errors.As(err, &stateErr) {
+ switch stateErr.SQLState() {
+ case "40001", "40P01":
+ return true
+ }
+ }
+
+ var sqliteErr interface{ Code() int }
+ if errors.As(err, &sqliteErr) {
+ switch sqliteErr.Code() & 0xff {
+ case 5, 6: // SQLITE_BUSY, SQLITE_LOCKED
+ return true
+ }
+ }
+
+ msg := err.Error()
+
Review Comment:
These MySQL matches key off the rendered error string (`Error 1213 (...`),
which is go-sql-driver/mysql's formatting rather than a stable contract — a
driver bump, a proxy, or localized text breaks classification silently.
I'd match on the typed error instead: `errors.As` against
`*mysql.MySQLError` and check `Number == 1213 || Number == 1205`, keeping the
substring only as a commented fallback.
##########
catalog/sql/sql.go:
##########
@@ -293,6 +294,72 @@ func withWriteTx(ctx context.Context, db *bun.DB, fn
func(context.Context, bun.T
})
}
+const (
+ serializableWriteMaxAttempts = 3
+ serializableWriteRetryDelay = 10 * time.Millisecond
+)
+
+func withSerializableWriteTx(ctx context.Context, db *bun.DB, fn
func(context.Context, bun.Tx) error) error {
+ return retrySerializableWriteTx(ctx, func() error {
+ return db.RunInTx(ctx, &sql.TxOptions{Isolation:
sql.LevelSerializable}, func(ctx context.Context, tx bun.Tx) error {
+ return fn(ctx, tx)
+ })
+ })
+}
+
+func retrySerializableWriteTx(ctx context.Context, run func() error) error {
+ var err error
+ for attempt := 0; attempt < serializableWriteMaxAttempts; attempt++ {
+ err = run()
+ if err == nil || !isRetryableSerializableError(err) {
+ return err
+ }
+ if attempt == serializableWriteMaxAttempts-1 {
+ break
+ }
+
+ delay := serializableWriteRetryDelay << attempt
+ timer := time.NewTimer(delay)
+ select {
+ case <-ctx.Done():
+ if !timer.Stop() {
+ <-timer.C
+ }
+
+ return context.Cause(ctx)
+ case <-timer.C:
+ }
+ }
+
+ return err
+}
+
+func isRetryableSerializableError(err error) bool {
+ var stateErr interface{ SQLState() string }
+ if errors.As(err, &stateErr) {
+ switch stateErr.SQLState() {
+ case "40001", "40P01":
+ return true
+ }
+ }
+
+ var sqliteErr interface{ Code() int }
+ if errors.As(err, &sqliteErr) {
+ switch sqliteErr.Code() & 0xff {
+ case 5, 6: // SQLITE_BUSY, SQLITE_LOCKED
+ return true
+ }
+ }
+
+ msg := err.Error()
+
+ return strings.Contains(msg, "Error 1213 (") ||
+ strings.Contains(msg, "Error 1205 (") ||
+ strings.Contains(msg, "database is locked") ||
+ strings.Contains(msg, "database table is locked") ||
Review Comment:
While we're matching Oracle errors — under SERIALIZABLE, lock-cycle
deadlocks come back as ORA-00060, not ORA-08177, and Oracle kills one of the
waiters. That victim is exactly the transaction we'd want to retry, but right
now it surfaces as a permanent error.
I'd add `strings.Contains(msg, "ORA-00060")` alongside the ORA-08177 check.
##########
catalog/sql/drop_namespace_internal_test.go:
##########
@@ -0,0 +1,302 @@
+// 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 sql
+
+import (
+ "context"
+ dbsql "database/sql"
+ "errors"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ _ "github.com/apache/iceberg-go/io/gocloud"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
+ "github.com/uptrace/bun"
+ "github.com/uptrace/bun/driver/sqliteshim"
+)
+
+type namespaceRaceContextKey struct{}
+
+type namespaceRaceOperation string
+
+const (
+ dropNamespaceOperation namespaceRaceOperation = "drop"
+ createViewOperation namespaceRaceOperation = "create-view"
+ updatePropertiesOperation namespaceRaceOperation = "update-properties"
+)
+
+type namespaceRaceHook struct {
+ dropAtDelete chan struct{}
+ createAtInsert chan struct{}
+ updateAtInsert chan struct{}
+ releaseDrop chan struct{}
+ releaseUpdate chan struct{}
+ dropAtDeleteOnce sync.Once
+ createAtInsertOnce sync.Once
+ updateAtInsertOnce sync.Once
+}
+
+func (h *namespaceRaceHook) BeforeQuery(ctx context.Context, event
*bun.QueryEvent) context.Context {
+ operation, _ :=
ctx.Value(namespaceRaceContextKey{}).(namespaceRaceOperation)
+ query := event.Query
+
+ switch {
+ case operation == dropNamespaceOperation && event.Operation() ==
"DELETE" && strings.Contains(query, "iceberg_namespace_properties"):
+ h.dropAtDeleteOnce.Do(func() { close(h.dropAtDelete) })
+ <-h.releaseDrop
+ case operation == createViewOperation && event.Operation() == "INSERT"
&& strings.Contains(query, "iceberg_tables"):
+ h.createAtInsertOnce.Do(func() { close(h.createAtInsert) })
+ <-h.releaseDrop
+ // Give the namespace deletion the first opportunity to commit.
A creator
+ // that did not recheck inside its transaction would then
insert an orphan.
+ time.Sleep(50 * time.Millisecond)
Review Comment:
This 50ms sleep (plus the 250ms select timeout further down) is the
load-bearing synchronization for the create/drop race, and it's wall-clock. On
a slow CI runner the drop may not reach commit within 50ms, in which case
create wins and we quietly exercise the create-wins path instead of the
drop-wins one we actually care about.
The new update-properties test uses a clean release channel to sequence
things deterministically — I'd bring the same idea here: a channel closed by
the drop goroutine right after its `withSerializableWriteTx` returns, with the
creator waiting on that instead of sleeping.
##########
catalog/sql/drop_namespace_internal_test.go:
##########
@@ -0,0 +1,302 @@
+// 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 sql
+
+import (
+ "context"
+ dbsql "database/sql"
+ "errors"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ _ "github.com/apache/iceberg-go/io/gocloud"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
+ "github.com/uptrace/bun"
+ "github.com/uptrace/bun/driver/sqliteshim"
+)
+
+type namespaceRaceContextKey struct{}
+
+type namespaceRaceOperation string
+
+const (
+ dropNamespaceOperation namespaceRaceOperation = "drop"
+ createViewOperation namespaceRaceOperation = "create-view"
+ updatePropertiesOperation namespaceRaceOperation = "update-properties"
+)
+
+type namespaceRaceHook struct {
+ dropAtDelete chan struct{}
+ createAtInsert chan struct{}
+ updateAtInsert chan struct{}
+ releaseDrop chan struct{}
+ releaseUpdate chan struct{}
+ dropAtDeleteOnce sync.Once
+ createAtInsertOnce sync.Once
+ updateAtInsertOnce sync.Once
+}
+
+func (h *namespaceRaceHook) BeforeQuery(ctx context.Context, event
*bun.QueryEvent) context.Context {
+ operation, _ :=
ctx.Value(namespaceRaceContextKey{}).(namespaceRaceOperation)
+ query := event.Query
+
+ switch {
+ case operation == dropNamespaceOperation && event.Operation() ==
"DELETE" && strings.Contains(query, "iceberg_namespace_properties"):
+ h.dropAtDeleteOnce.Do(func() { close(h.dropAtDelete) })
+ <-h.releaseDrop
+ case operation == createViewOperation && event.Operation() == "INSERT"
&& strings.Contains(query, "iceberg_tables"):
+ h.createAtInsertOnce.Do(func() { close(h.createAtInsert) })
+ <-h.releaseDrop
+ // Give the namespace deletion the first opportunity to commit.
A creator
+ // that did not recheck inside its transaction would then
insert an orphan.
+ time.Sleep(50 * time.Millisecond)
+ case operation == updatePropertiesOperation && event.Operation() ==
"INSERT" && strings.Contains(query, "iceberg_namespace_properties"):
+ h.updateAtInsertOnce.Do(func() { close(h.updateAtInsert) })
+ <-h.releaseUpdate
+ }
+
+ return ctx
+}
+
+func (*namespaceRaceHook) AfterQuery(context.Context, *bun.QueryEvent) {}
+
+func newNamespaceRaceCatalog(t *testing.T) *Catalog {
+ t.Helper()
+
+ dbPath := filepath.Join(t.TempDir(), "catalog.db")
+ db, err := dbsql.Open(sqliteshim.ShimName, dbPath)
+ require.NoError(t, err)
+ t.Cleanup(func() { require.NoError(t, db.Close()) })
+ db.SetMaxOpenConns(4)
+ _, err = db.Exec("PRAGMA journal_mode=WAL")
+ require.NoError(t, err)
+
+ warehouse := filepath.Join(t.TempDir(), "warehouse")
+ cat, err := NewCatalog("default", db, SQLite, iceberg.Properties{
+ "warehouse": "file://" + filepath.ToSlash(warehouse),
+ })
+ require.NoError(t, err)
+
+ return cat
+}
+
+type sqlStateTestError string
+
+func (e sqlStateTestError) Error() string { return string(e) }
+func (e sqlStateTestError) SQLState() string { return string(e) }
+
+type sqliteCodeTestError int
+
+func (e sqliteCodeTestError) Error() string { return "sqlite test error" }
+func (e sqliteCodeTestError) Code() int { return int(e) }
+
+func TestRetrySerializableWriteTx(t *testing.T) {
+ t.Run("retries transient errors", func(t *testing.T) {
+ attempts := 0
+ err := retrySerializableWriteTx(context.Background(), func()
error {
+ attempts++
+ if attempts == 1 {
+ return sqlStateTestError("40001")
+ }
+
+ return nil
+ })
+ require.NoError(t, err)
+ require.Equal(t, 2, attempts)
+ })
+
+ t.Run("does not retry permanent errors", func(t *testing.T) {
+ attempts := 0
+ permanentErr := errors.New("permanent error")
+ err := retrySerializableWriteTx(context.Background(), func()
error {
+ attempts++
+
+ return permanentErr
+ })
+ require.ErrorIs(t, err, permanentErr)
+ require.Equal(t, 1, attempts)
+ })
+
+ t.Run("returns transient error after attempts are exhausted", func(t
*testing.T) {
+ attempts := 0
+ transientErr := sqlStateTestError("40001")
+ err := retrySerializableWriteTx(context.Background(), func()
error {
+ attempts++
+
+ return transientErr
+ })
+ require.ErrorIs(t, err, transientErr)
+ require.Equal(t, serializableWriteMaxAttempts, attempts)
+ })
+
+ t.Run("stops when context is canceled", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ attempts := 0
+ err := retrySerializableWriteTx(ctx, func() error {
+ attempts++
+
+ return sqliteCodeTestError(5)
+ })
+ require.ErrorIs(t, err, context.Canceled)
+ require.Equal(t, 1, attempts)
+ })
+}
+
+func TestRetryableSerializableError(t *testing.T) {
+ tests := []struct {
+ name string
+ err error
+ }{
+ {name: "postgres serialization", err:
sqlStateTestError("40001")},
+ {name: "postgres deadlock", err: sqlStateTestError("40P01")},
+ {name: "sqlite busy", err: sqliteCodeTestError(5)},
+ {name: "sqlite locked extended", err: sqliteCodeTestError(6 |
1<<8)},
+ {name: "mysql deadlock", err: errors.New("Error 1213 (40001):
Deadlock found when trying to get lock")},
+ {name: "mysql lock timeout", err: errors.New("Error 1205
(HY000): Lock wait timeout exceeded")},
+ {name: "sqlite3 busy", err: errors.New("database is locked")},
+ {name: "oracle serialization", err: errors.New("ORA-08177:
can't serialize access for this transaction")},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ require.True(t, isRetryableSerializableError(tt.err))
+ })
+ }
+
+ require.False(t, isRetryableSerializableError(errors.New("constraint
violation")))
+}
+
+func TestDropNamespaceDoesNotRaceWithCreateView(t *testing.T) {
+ cat := newNamespaceRaceCatalog(t)
+
+ namespace := table.Identifier{"race_namespace"}
+ viewID := table.Identifier{"race_namespace", "race_view"}
+ require.NoError(t, cat.CreateNamespace(context.Background(), namespace,
nil))
+
Review Comment:
Good to see the update-properties race covered now. CreateTable and the
CommitTable create branch got the same serializable + in-tx recheck treatment
as CreateView, though, and neither has a race test — a regression on either
wouldn't be caught here.
Could we parameterize this over the create operation (CreateTable /
CreateView / CommitTable-create) so all three run against the same drop race?
wdyt?
##########
catalog/sql/sql.go:
##########
@@ -745,6 +852,10 @@ func (c *Catalog) CommitTable(ctx context.Context, ident
table.Identifier, reqs
return nil
})
if err != nil {
+ if current == nil && isRetryableSerializableError(err) {
+ return nil, "", fmt.Errorf("%w: %w",
table.ErrCommitFailed, err)
Review Comment:
Two things on this wrap. `fmt.Errorf("%w: %w", ...)` makes the result
satisfy both `errors.Is(_, table.ErrCommitFailed)` and `errors.Is(_, err)` — so
the raw driver serialization error becomes a first-class sentinel, and a driver
swap silently changes what callers match. I'd wrap only the sentinel:
`fmt.Errorf("%w: %v", table.ErrCommitFailed, err)`.
Separately, handing the exhausted serialization failure back as
`ErrCommitFailed` re-enters `table.doCommit`'s retry loop, which re-runs
`withSerializableWriteTx` for another 3 inner attempts — up to 9 total. That
outer loop exists for CAS-on-metadata conflicts, and on the create branch
(`current == nil`) there's no version skew to resolve anyway. I'd return the
serialization error directly, or under a distinct sentinel, rather than
conflating "someone beat me to the CAS" with "the DB couldn't serialize under
load."
##########
catalog/sql/drop_namespace_internal_test.go:
##########
@@ -0,0 +1,302 @@
+// 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 sql
+
+import (
+ "context"
+ dbsql "database/sql"
+ "errors"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ _ "github.com/apache/iceberg-go/io/gocloud"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
+ "github.com/uptrace/bun"
+ "github.com/uptrace/bun/driver/sqliteshim"
+)
+
+type namespaceRaceContextKey struct{}
+
+type namespaceRaceOperation string
+
+const (
+ dropNamespaceOperation namespaceRaceOperation = "drop"
+ createViewOperation namespaceRaceOperation = "create-view"
+ updatePropertiesOperation namespaceRaceOperation = "update-properties"
+)
+
+type namespaceRaceHook struct {
+ dropAtDelete chan struct{}
+ createAtInsert chan struct{}
+ updateAtInsert chan struct{}
+ releaseDrop chan struct{}
+ releaseUpdate chan struct{}
+ dropAtDeleteOnce sync.Once
+ createAtInsertOnce sync.Once
+ updateAtInsertOnce sync.Once
+}
+
+func (h *namespaceRaceHook) BeforeQuery(ctx context.Context, event
*bun.QueryEvent) context.Context {
+ operation, _ :=
ctx.Value(namespaceRaceContextKey{}).(namespaceRaceOperation)
+ query := event.Query
+
+ switch {
+ case operation == dropNamespaceOperation && event.Operation() ==
"DELETE" && strings.Contains(query, "iceberg_namespace_properties"):
+ h.dropAtDeleteOnce.Do(func() { close(h.dropAtDelete) })
+ <-h.releaseDrop
+ case operation == createViewOperation && event.Operation() == "INSERT"
&& strings.Contains(query, "iceberg_tables"):
+ h.createAtInsertOnce.Do(func() { close(h.createAtInsert) })
+ <-h.releaseDrop
+ // Give the namespace deletion the first opportunity to commit.
A creator
+ // that did not recheck inside its transaction would then
insert an orphan.
+ time.Sleep(50 * time.Millisecond)
+ case operation == updatePropertiesOperation && event.Operation() ==
"INSERT" && strings.Contains(query, "iceberg_namespace_properties"):
+ h.updateAtInsertOnce.Do(func() { close(h.updateAtInsert) })
+ <-h.releaseUpdate
+ }
+
+ return ctx
+}
+
+func (*namespaceRaceHook) AfterQuery(context.Context, *bun.QueryEvent) {}
+
+func newNamespaceRaceCatalog(t *testing.T) *Catalog {
+ t.Helper()
+
+ dbPath := filepath.Join(t.TempDir(), "catalog.db")
+ db, err := dbsql.Open(sqliteshim.ShimName, dbPath)
+ require.NoError(t, err)
+ t.Cleanup(func() { require.NoError(t, db.Close()) })
+ db.SetMaxOpenConns(4)
+ _, err = db.Exec("PRAGMA journal_mode=WAL")
+ require.NoError(t, err)
+
+ warehouse := filepath.Join(t.TempDir(), "warehouse")
+ cat, err := NewCatalog("default", db, SQLite, iceberg.Properties{
+ "warehouse": "file://" + filepath.ToSlash(warehouse),
+ })
+ require.NoError(t, err)
+
+ return cat
+}
+
+type sqlStateTestError string
+
+func (e sqlStateTestError) Error() string { return string(e) }
+func (e sqlStateTestError) SQLState() string { return string(e) }
+
+type sqliteCodeTestError int
+
+func (e sqliteCodeTestError) Error() string { return "sqlite test error" }
+func (e sqliteCodeTestError) Code() int { return int(e) }
+
+func TestRetrySerializableWriteTx(t *testing.T) {
+ t.Run("retries transient errors", func(t *testing.T) {
+ attempts := 0
+ err := retrySerializableWriteTx(context.Background(), func()
error {
+ attempts++
+ if attempts == 1 {
+ return sqlStateTestError("40001")
+ }
+
+ return nil
+ })
+ require.NoError(t, err)
+ require.Equal(t, 2, attempts)
+ })
+
+ t.Run("does not retry permanent errors", func(t *testing.T) {
+ attempts := 0
+ permanentErr := errors.New("permanent error")
+ err := retrySerializableWriteTx(context.Background(), func()
error {
+ attempts++
+
+ return permanentErr
+ })
+ require.ErrorIs(t, err, permanentErr)
+ require.Equal(t, 1, attempts)
+ })
+
+ t.Run("returns transient error after attempts are exhausted", func(t
*testing.T) {
+ attempts := 0
+ transientErr := sqlStateTestError("40001")
+ err := retrySerializableWriteTx(context.Background(), func()
error {
+ attempts++
+
+ return transientErr
+ })
+ require.ErrorIs(t, err, transientErr)
+ require.Equal(t, serializableWriteMaxAttempts, attempts)
+ })
+
+ t.Run("stops when context is canceled", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ attempts := 0
+ err := retrySerializableWriteTx(ctx, func() error {
+ attempts++
+
+ return sqliteCodeTestError(5)
+ })
+ require.ErrorIs(t, err, context.Canceled)
+ require.Equal(t, 1, attempts)
+ })
+}
+
+func TestRetryableSerializableError(t *testing.T) {
+ tests := []struct {
+ name string
+ err error
+ }{
+ {name: "postgres serialization", err:
sqlStateTestError("40001")},
+ {name: "postgres deadlock", err: sqlStateTestError("40P01")},
+ {name: "sqlite busy", err: sqliteCodeTestError(5)},
+ {name: "sqlite locked extended", err: sqliteCodeTestError(6 |
1<<8)},
+ {name: "mysql deadlock", err: errors.New("Error 1213 (40001):
Deadlock found when trying to get lock")},
+ {name: "mysql lock timeout", err: errors.New("Error 1205
(HY000): Lock wait timeout exceeded")},
+ {name: "sqlite3 busy", err: errors.New("database is locked")},
+ {name: "oracle serialization", err: errors.New("ORA-08177:
can't serialize access for this transaction")},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ require.True(t, isRetryableSerializableError(tt.err))
+ })
+ }
+
+ require.False(t, isRetryableSerializableError(errors.New("constraint
violation")))
+}
+
+func TestDropNamespaceDoesNotRaceWithCreateView(t *testing.T) {
+ cat := newNamespaceRaceCatalog(t)
+
+ namespace := table.Identifier{"race_namespace"}
+ viewID := table.Identifier{"race_namespace", "race_view"}
+ require.NoError(t, cat.CreateNamespace(context.Background(), namespace,
nil))
+
+ hook := &namespaceRaceHook{
+ dropAtDelete: make(chan struct{}),
+ createAtInsert: make(chan struct{}),
+ releaseDrop: make(chan struct{}),
+ }
+ cat.db.AddQueryHook(hook)
+
+ dropCtx := context.WithValue(context.Background(),
namespaceRaceContextKey{}, dropNamespaceOperation)
+ createCtx := context.WithValue(context.Background(),
namespaceRaceContextKey{}, createViewOperation)
+ dropErrCh := make(chan error, 1)
+ createErrCh := make(chan error, 1)
+
+ go func() { dropErrCh <- cat.DropNamespace(dropCtx, namespace) }()
+ select {
+ case <-hook.dropAtDelete:
+ case <-time.After(5 * time.Second):
+ t.Fatal("drop did not reach namespace deletion")
+ }
+
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64,
Required: true,
+ })
+ go func() { createErrCh <- cat.CreateView(createCtx, viewID, schema,
"SELECT 1", nil) }()
+
+ select {
+ case <-hook.createAtInsert:
+ case <-time.After(250 * time.Millisecond):
+ // Some SQLite drivers wait to begin the creator transaction
while the
+ // drop is active. Releasing here lets it recheck after the
drop commits.
+ }
+ close(hook.releaseDrop)
+
+ dropErr := <-dropErrCh
+ createErr := <-createErrCh
+ require.False(t, dropErr == nil && createErr == nil, "drop and create
must not both report success")
+
+ viewExists, err := cat.CheckViewExists(context.Background(), viewID)
+ require.NoError(t, err)
+ namespaceExists, err := cat.CheckNamespaceExists(context.Background(),
namespace)
+ require.NoError(t, err)
+
+ if createErr == nil {
Review Comment:
The `if createErr == nil / else if dropErr == nil` pair has no `else`, so
when SQLite serialization makes *both* fail, the test runs no state assertions
and passes vacuously — an orphan (view row present, namespace props gone) with
both erroring would slip right through.
I'd add the else and assert self-consistency there: namespace exists iff the
view exists. If both-fail is an accepted outcome of the retry design, a
`t.Logf` of both errors plus that consistency check would still guard it.
##########
catalog/sql/sql.go:
##########
@@ -869,7 +980,15 @@ func (c *Catalog) RenameTable(ctx context.Context, from,
to table.Identifier) (*
return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
toNs)
}
- err = withWriteTx(ctx, c.db, func(ctx context.Context, tx bun.Tx) error
{
+ err = withSerializableWriteTx(ctx, c.db, func(ctx context.Context, tx
bun.Tx) error {
+ namespaceExists, err := c.namespaceExistsInTx(ctx, tx, toNs)
Review Comment:
This rechecks the destination namespace inside the tx, which is the race the
PR is closing — but the source side still relies on the UPDATE's RowsAffected,
and a concurrent DropNamespace of the *source* namespace only deletes its
property rows, not the `iceberg_tables` row. So the UPDATE still matches and
we'd leave a table row pointing at a namespace that no longer exists.
Narrower than the to-namespace case, but since we're here I'd add a
`namespaceExistsInTx(ctx, tx, fromNs)` check before the UPDATE too.
##########
catalog/sql/sql.go:
##########
@@ -293,6 +294,72 @@ func withWriteTx(ctx context.Context, db *bun.DB, fn
func(context.Context, bun.T
})
}
+const (
Review Comment:
My concern with this retry budget is that it only helps if the DB actually
blocks and then hands back a retryable error. On the production SQLite path it
won't: `NewCatalog` takes a pre-configured `*sql.DB` with no WAL and no
busy_timeout, so a contended writer gets SQLITE_BUSY back immediately, and
three attempts at 10/20ms burn through the budget faster than a single disk
round-trip. The race tests only survive because they set `PRAGMA
journal_mode=WAL` and `SetMaxOpenConns(4)`, which isn't how the catalog is
actually built.
So the retry reads as transparent but for the default SQLite catalog it
mostly isn't. I'd either set WAL + a busy_timeout for SQLite in `NewCatalog`
(or use `BEGIN IMMEDIATE` instead of SERIALIZABLE there), or document that
those pragmas are required for the retry to mean anything. Thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
