laskoviymishka commented on code in PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#discussion_r3739117266
##########
catalog/sql/sql.go:
##########
@@ -1269,11 +1269,30 @@ func (c *Catalog) CreateNamespace(ctx context.Context,
namespace table.Identifie
})
}
- _, err := tx.NewInsert().Model(&toInsert).Exec(ctx)
+ // Run the insert in a savepoint so a failure can be rolled
back without
+ // aborting the whole transaction (Postgres); the re-check then
still runs.
+ sp, err := tx.BeginTx(ctx, nil)
if err != nil {
+ return fmt.Errorf("error creating savepoint for
namespace '%s': %w", namespace, err)
+ }
+
+ if _, err = sp.NewInsert().Model(&toInsert).Exec(ctx); err !=
nil {
+ if rbErr := sp.Rollback(); rbErr != nil {
+ return fmt.Errorf("error inserting namespace
properties for namespace '%s': %w", namespace, err)
+ }
+ // A concurrent writer may have won the race; if the
re-check confirms
+ // it, return the sentinel joined with the cause, else
the insert error.
+ if _, exists, checkErr :=
c.resolveNamespaceKeyInTx(ctx, tx, namespace); checkErr == nil && exists {
+ return errors.Join(fmt.Errorf("%w: %s",
catalog.ErrNamespaceAlreadyExists, strings.Join(namespace, ".")), err)
Review Comment:
The race-loser here returns `errors.Join(sentinel, err)`, but the pre-check
path a few lines up returns a plain `fmt.Errorf("%w: %s",
catalog.ErrNamespaceAlreadyExists, …)`. `errors.Is` holds either way so the 409
mapping is safe, but `.Error()` now carries the raw insert error on a second
line, and `errors.As` will unwrap to the driver's constraint error only on the
race path — so two paths that mean the same thing produce structurally
different errors depending on scheduling.
I'd drop `err` from the join and return the same form as the pre-check so
they're identical. If we want the cause for logs, `%v` keeps it in the string
without putting it in the unwrap chain.
##########
catalog/sql/sql_test.go:
##########
@@ -1911,6 +1917,249 @@ func (s *SqliteCatalogTestSuite)
TestLoadEmptyNamespaceProperties() {
}
}
+func (s *SqliteCatalogTestSuite) TestCreateNamespaceConcurrent() {
+ // Two callers creating the same namespace: exactly one succeeds and
the other
+ // gets ErrNamespaceAlreadyExists, not the driver's duplicate-key error.
+ const writers = 8
+
+ // A busy timeout so the writers queue on the sqlite lock instead of
failing
+ // with SQLITE_BUSY, which is a different contention problem to this
one.
+ loaded, err := catalog.Load(context.Background(), "default",
iceberg.Properties{
+ "uri": s.catalogUri() + "?_pragma=" +
url.QueryEscape("busy_timeout(10000)"),
+ sqlcat.DriverKey: sqliteshim.ShimName,
+ sqlcat.DialectKey: string(sqlcat.SQLite),
+ "type": "sql",
+ "warehouse": "file://" + s.warehouse,
+ })
+ s.Require().NoError(err)
+
+ cat := loaded.(*sqlcat.Catalog)
+ ctx := context.Background()
+ namespace := table.Identifier{databaseName()}
+
+ start := make(chan struct{})
+ errs := make(chan error, writers)
+
+ var wg sync.WaitGroup
+ for range writers {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ errs <- cat.CreateNamespace(ctx, namespace, nil)
+ }()
+ }
+ close(start)
+ wg.Wait()
+ close(errs)
+
+ created, alreadyExists := 0, 0
+ var unexpected []error
+ for err := range errs {
+ switch {
+ case err == nil:
+ created++
+ case errors.Is(err, catalog.ErrNamespaceAlreadyExists):
+ alreadyExists++
+ default:
+ unexpected = append(unexpected, err)
+ }
+ }
+
+ s.Empty(unexpected, "want nil or ErrNamespaceAlreadyExists")
+ s.Equal(1, created)
+ s.Equal(writers-1, alreadyExists)
+}
+
+// insertFailure selects how the emulated namespace insert fails.
+type insertFailure int
+
+const (
+ // failUnique: insert trips the unique constraint and the namespace
exists on
+ // the re-check -- the raced create-if-absent that the fix must recover.
+ failUnique insertFailure = iota
+ // failUnrelated: insert fails for an unrelated reason and the
namespace does
+ // not exist -- the original error must survive, not become "already
exists".
+ failUnrelated
+)
+
+var (
+ errEmulatedUnique = errors.New("UNIQUE constraint failed:
iceberg_namespace_properties.catalog_name,
iceberg_namespace_properties.namespace,
iceberg_namespace_properties.property_key")
+ errEmulatedUnrelated = errors.New("disk I/O error")
+ errEmulatedAborted = errors.New("current transaction is aborted,
commands ignored until end of transaction block")
+)
+
+// pgAbortDriver emulates Postgres for the create race: after the namespace
+// insert fails it refuses statements until a ROLLBACK TO SAVEPOINT clears it.
+type pgAbortDriver struct {
+ base driver.Driver
+ mode insertFailure
+ insertAttempted atomic.Bool
+ aborted atomic.Bool
+}
+
+func (d *pgAbortDriver) Open(dsn string) (driver.Conn, error) {
+ conn, err := d.base.Open(dsn)
+ if err != nil {
+ return nil, err
+ }
+
+ return &pgAbortConn{Conn: conn, drv: d}, nil
+}
+
+type pgAbortConn struct {
+ driver.Conn
+ drv *pgAbortDriver
+}
+
+func isNamespaceInsert(query string) bool {
+ return strings.HasPrefix(strings.ToUpper(strings.TrimSpace(query)),
"INSERT") &&
+ strings.Contains(query, "iceberg_namespace_properties")
+}
+
+func isNamespaceExistsProbe(query string) bool {
+ return strings.Contains(strings.ToUpper(query), "EXISTS") &&
+ strings.Contains(query, "iceberg_namespace_properties")
+}
+
+func isRollbackToSavepoint(query string) bool {
Review Comment:
Small robustness thing on the emulation: both `isRollbackToSavepoint` and
`isNamespaceExistsProbe` match bun's exact SQL text (the `ROLLBACK TO
SAVEPOINT` prefix, the literal `EXISTS`). Postgres and SQLite also accept
`ROLLBACK TO <name>` without the keyword, and if a bun version ever changes
either shape the `aborted` flag never clears or the probe stops firing, and the
test falls through to an opaque `ErrorIs` failure instead of exercising the
recovery.
Not blocking, but a one-line comment naming the bun format these depend on —
or asserting the savepoint query was actually seen — would save the next person
the debugging. The driver-scoped `aborted`/`insertAttempted` state is fine
given `SetMaxOpenConns(1)` and you've already documented that, so I'd leave it.
##########
catalog/sql/sql_test.go:
##########
@@ -1911,6 +1917,249 @@ func (s *SqliteCatalogTestSuite)
TestLoadEmptyNamespaceProperties() {
}
}
+func (s *SqliteCatalogTestSuite) TestCreateNamespaceConcurrent() {
+ // Two callers creating the same namespace: exactly one succeeds and
the other
+ // gets ErrNamespaceAlreadyExists, not the driver's duplicate-key error.
+ const writers = 8
+
+ // A busy timeout so the writers queue on the sqlite lock instead of
failing
+ // with SQLITE_BUSY, which is a different contention problem to this
one.
+ loaded, err := catalog.Load(context.Background(), "default",
iceberg.Properties{
+ "uri": s.catalogUri() + "?_pragma=" +
url.QueryEscape("busy_timeout(10000)"),
+ sqlcat.DriverKey: sqliteshim.ShimName,
+ sqlcat.DialectKey: string(sqlcat.SQLite),
+ "type": "sql",
+ "warehouse": "file://" + s.warehouse,
+ })
+ s.Require().NoError(err)
+
+ cat := loaded.(*sqlcat.Catalog)
+ ctx := context.Background()
+ namespace := table.Identifier{databaseName()}
+
+ start := make(chan struct{})
+ errs := make(chan error, writers)
+
+ var wg sync.WaitGroup
+ for range writers {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ errs <- cat.CreateNamespace(ctx, namespace, nil)
+ }()
+ }
+ close(start)
+ wg.Wait()
+ close(errs)
+
+ created, alreadyExists := 0, 0
+ var unexpected []error
+ for err := range errs {
+ switch {
+ case err == nil:
+ created++
+ case errors.Is(err, catalog.ErrNamespaceAlreadyExists):
+ alreadyExists++
+ default:
+ unexpected = append(unexpected, err)
+ }
+ }
+
+ s.Empty(unexpected, "want nil or ErrNamespaceAlreadyExists")
+ s.Equal(1, created)
+ s.Equal(writers-1, alreadyExists)
+}
+
+// insertFailure selects how the emulated namespace insert fails.
+type insertFailure int
+
+const (
+ // failUnique: insert trips the unique constraint and the namespace
exists on
+ // the re-check -- the raced create-if-absent that the fix must recover.
+ failUnique insertFailure = iota
+ // failUnrelated: insert fails for an unrelated reason and the
namespace does
+ // not exist -- the original error must survive, not become "already
exists".
+ failUnrelated
+)
+
+var (
+ errEmulatedUnique = errors.New("UNIQUE constraint failed:
iceberg_namespace_properties.catalog_name,
iceberg_namespace_properties.namespace,
iceberg_namespace_properties.property_key")
+ errEmulatedUnrelated = errors.New("disk I/O error")
+ errEmulatedAborted = errors.New("current transaction is aborted,
commands ignored until end of transaction block")
+)
+
+// pgAbortDriver emulates Postgres for the create race: after the namespace
+// insert fails it refuses statements until a ROLLBACK TO SAVEPOINT clears it.
+type pgAbortDriver struct {
+ base driver.Driver
+ mode insertFailure
+ insertAttempted atomic.Bool
+ aborted atomic.Bool
+}
+
+func (d *pgAbortDriver) Open(dsn string) (driver.Conn, error) {
+ conn, err := d.base.Open(dsn)
+ if err != nil {
+ return nil, err
+ }
+
+ return &pgAbortConn{Conn: conn, drv: d}, nil
+}
+
+type pgAbortConn struct {
+ driver.Conn
+ drv *pgAbortDriver
+}
+
+func isNamespaceInsert(query string) bool {
+ return strings.HasPrefix(strings.ToUpper(strings.TrimSpace(query)),
"INSERT") &&
+ strings.Contains(query, "iceberg_namespace_properties")
+}
+
+func isNamespaceExistsProbe(query string) bool {
+ return strings.Contains(strings.ToUpper(query), "EXISTS") &&
+ strings.Contains(query, "iceberg_namespace_properties")
+}
+
+func isRollbackToSavepoint(query string) bool {
+ return strings.HasPrefix(strings.ToUpper(strings.TrimSpace(query)),
"ROLLBACK TO SAVEPOINT")
+}
+
+func (c *pgAbortConn) ExecContext(ctx context.Context, query string, args
[]driver.NamedValue) (driver.Result, error) {
+ execer, ok := c.Conn.(driver.ExecerContext)
+ if !ok {
+ return nil, driver.ErrSkip
+ }
+
+ if c.drv.aborted.Load() {
+ if isRollbackToSavepoint(query) {
+ c.drv.aborted.Store(false)
+
+ return execer.ExecContext(ctx, query, args)
+ }
+
+ return nil, errEmulatedAborted
+ }
+
+ if isNamespaceInsert(query) {
+ c.drv.insertAttempted.Store(true)
+ c.drv.aborted.Store(true)
+ if c.drv.mode == failUnique {
+ return nil, errEmulatedUnique
+ }
+
+ return nil, errEmulatedUnrelated
+ }
+
+ return execer.ExecContext(ctx, query, args)
+}
+
+func (c *pgAbortConn) QueryContext(ctx context.Context, query string, args
[]driver.NamedValue) (driver.Rows, error) {
+ queryer, ok := c.Conn.(driver.QueryerContext)
+ if !ok {
+ return nil, driver.ErrSkip
+ }
+
+ if c.drv.aborted.Load() {
+ return nil, errEmulatedAborted
+ }
+
+ // The namespace exists on the re-check only in the unique-violation
case: a
+ // concurrent winner committed the row. An unrelated failure leaves it
absent.
+ if isNamespaceExistsProbe(query) {
+ return &boolRows{val: c.drv.insertAttempted.Load() &&
c.drv.mode == failUnique}, nil
+ }
+
+ return queryer.QueryContext(ctx, query, args)
+}
+
+func (c *pgAbortConn) BeginTx(ctx context.Context, opts driver.TxOptions)
(driver.Tx, error) {
+ beginTx, ok := c.Conn.(driver.ConnBeginTx)
+ if !ok {
+ return nil, driver.ErrBadConn
+ }
+
+ return beginTx.BeginTx(ctx, opts)
+}
+
+func (c *pgAbortConn) PrepareContext(ctx context.Context, query string)
(driver.Stmt, error) {
+ prepCtx, ok := c.Conn.(driver.ConnPrepareContext)
+ if !ok {
+ return nil, driver.ErrBadConn
+ }
+
+ return prepCtx.PrepareContext(ctx, query)
+}
+
+// boolRows is a single-row, single-column result carrying a SQL EXISTS answer.
+type boolRows struct {
+ val bool
+ done bool
+}
+
+func (r *boolRows) Columns() []string { return []string{"exists"} }
+func (r *boolRows) Close() error { return nil }
+func (r *boolRows) Next(dest []driver.Value) error {
Review Comment:
`Next` returns a row (with `int64(0)`) when `val` is false rather than
signaling EOF, so this only reads as "not exists" if bun's existence check
scans the column value rather than row presence. If it ever keys on presence,
the `failUnrelated` case would see a row, conclude the namespace exists, and
hand back `ErrNamespaceAlreadyExists` — the exact thing that test asserts
against — so it'd fail opaquely instead of catching a regression.
Simplest way to make it robust either way is to return `io.EOF` when `!val`
so "not exists" is genuinely an empty result:
```go
func (r *boolRows) Next(dest []driver.Value) error {
if r.done || !r.val {
return io.EOF
}
r.done = true
dest[0] = int64(1)
return nil
}
```
Worth a quick check that reverting the recovery in `sql.go` still fails
`TestCreateNamespaceInsertFailureSurfacesOriginalError` — if it does, the
negative case is genuinely pinned.
##########
catalog/sql/sql.go:
##########
@@ -1269,11 +1269,30 @@ func (c *Catalog) CreateNamespace(ctx context.Context,
namespace table.Identifie
})
}
- _, err := tx.NewInsert().Model(&toInsert).Exec(ctx)
+ // Run the insert in a savepoint so a failure can be rolled
back without
+ // aborting the whole transaction (Postgres); the re-check then
still runs.
+ sp, err := tx.BeginTx(ctx, nil)
if err != nil {
+ return fmt.Errorf("error creating savepoint for
namespace '%s': %w", namespace, err)
+ }
+
+ if _, err = sp.NewInsert().Model(&toInsert).Exec(ctx); err !=
nil {
+ if rbErr := sp.Rollback(); rbErr != nil {
Review Comment:
Two failure branches in this block drop their cause. On rollback failure we
return the insert `err` and throw away `rbErr` — but a failed `ROLLBACK TO
SAVEPOINT` is a different and worse condition than a failed insert, and on
Postgres it's exactly what leaves the tx poisoned, so that's the diagnostic I'd
most want to keep. Same shape at the re-check just below: when `checkErr !=
nil` we fall straight through to the insert-error return and lose that the
re-check itself failed.
I'd `errors.Join(err, rbErr)` on the rollback path and thread `checkErr`
into the fall-through return so neither cause vanishes. wdyt?
--
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]
