zeroshade commented on code in PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#discussion_r3937950400
##########
catalog/sql/sql_test.go:
##########
@@ -1911,6 +1917,384 @@ 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
+ // insertSucceeds: the insert commits, exercising the savepoint happy
path
+ // (SAVEPOINT -> insert -> RELEASE SAVEPOINT) on the Postgres branch.
+ insertSucceeds
+)
+
+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
+}
+
+// These match bun's emitted SQL: an "INSERT"/"EXISTS" statement naming the
+// namespace table, and its "ROLLBACK TO SAVEPOINT" prefix.
+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)
+ switch c.drv.mode {
+ case insertSucceeds:
+ return execer.ExecContext(ctx, query, args)
+ case failUnrelated:
+ c.drv.aborted.Store(true)
+
+ return nil, errEmulatedUnrelated
+ default:
+ c.drv.aborted.Store(true)
+
+ return nil, errEmulatedUnique
+ }
+ }
+
+ 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
+ }
+
+ // Postgres probes information_schema for schema-version detection;
sqlite has
+ // no such view, so answer as the V1 table this suite creates (column
present).
+ if strings.Contains(query, "information_schema") {
+ return &boolRows{val: true}, nil
+ }
+
+ // 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 {
+ // bun's Exists() scans SELECT EXISTS via QueryRowContext.Scan(&bool),
so it
+ // keys on the value: emit one 0/1 row, not an empty result (that is
ErrNoRows).
+ if r.done {
+ return io.EOF
+ }
+ r.done = true
+ if r.val {
+ dest[0] = int64(1)
+ } else {
+ dest[0] = int64(0)
+ }
+
+ return nil
+}
+
+func (s *SqliteCatalogTestSuite) newAbortCatalog(mode insertFailure)
(*sqlcat.Catalog, *pgAbortDriver) {
+ base, err := sql.Open(sqliteshim.ShimName, ":memory:")
+ s.Require().NoError(err)
+ s.Require().NoError(base.Close())
+
+ // drvName derives from databaseName() (unique per call); sql.Register
panics
+ // on a duplicate name, so this must not become a fixed string.
+ drvName := "sqlite-pgabort-" + databaseName()
+ drv := &pgAbortDriver{base: base.Driver(), mode: mode}
+ sql.Register(drvName, drv)
+
+ sqldb, err := sql.Open(drvName, s.catalogUri())
+ s.Require().NoError(err)
+ s.T().Cleanup(func() { _ = sqldb.Close() })
+ // Single connection so the driver's abort state spans the pre-check
and the
+ // transaction that follows it.
+ sqldb.SetMaxOpenConns(1)
+
+ // Postgres dialect so the savepoint-backed recovery path runs; the
driver
+ // above emulates Postgres aborting the tx over an sqlite base.
+ cat, err := sqlcat.NewCatalog("default", sqldb, sqlcat.Postgres,
iceberg.Properties{"warehouse": "file://" + s.warehouse})
+ s.Require().NoError(err)
+
+ return cat, drv
+}
+
+// dupInsertDriver emulates a dialect that rolls back only the failing
statement
+// (SQLite): the namespace insert trips a unique violation without poisoning
the
+// tx, so the re-check runs on the same tx and recovers the sentinel.
+type dupInsertDriver struct {
+ base driver.Driver
+ insertAttempted atomic.Bool
+}
+
+func (d *dupInsertDriver) Open(dsn string) (driver.Conn, error) {
+ conn, err := d.base.Open(dsn)
+ if err != nil {
+ return nil, err
+ }
+
+ return &dupInsertConn{Conn: conn, drv: d}, nil
+}
+
+type dupInsertConn struct {
+ driver.Conn
+ drv *dupInsertDriver
+}
+
+func (c *dupInsertConn) 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 isNamespaceInsert(query) {
+ c.drv.insertAttempted.Store(true)
+
+ return nil, errEmulatedUnique
+ }
+
+ return execer.ExecContext(ctx, query, args)
+}
+
+func (c *dupInsertConn) QueryContext(ctx context.Context, query string, args
[]driver.NamedValue) (driver.Rows, error) {
+ queryer, ok := c.Conn.(driver.QueryerContext)
+ if !ok {
+ return nil, driver.ErrSkip
+ }
+ // After the insert, the namespace exists on the re-check; before it,
absent.
+ if isNamespaceExistsProbe(query) {
+ return &boolRows{val: c.drv.insertAttempted.Load()}, nil
+ }
+
+ return queryer.QueryContext(ctx, query, args)
+}
+
+func (c *dupInsertConn) 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 *dupInsertConn) 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)
+}
+
+func (s *SqliteCatalogTestSuite) newDupCatalog() (*sqlcat.Catalog,
*dupInsertDriver) {
+ base, err := sql.Open(sqliteshim.ShimName, ":memory:")
+ s.Require().NoError(err)
+ s.Require().NoError(base.Close())
+
+ drvName := "sqlite-dupinsert-" + databaseName()
+ drv := &dupInsertDriver{base: base.Driver()}
+ sql.Register(drvName, drv)
+
+ sqldb, err := sql.Open(drvName, s.catalogUri())
+ s.Require().NoError(err)
+ s.T().Cleanup(func() { _ = sqldb.Close() })
+ sqldb.SetMaxOpenConns(1)
+
+ // SQLite dialect so the non-Postgres recovery branch (plain insert +
re-check)
+ // runs; the driver rolls back only the failing statement, so no
savepoint.
+ cat, err := sqlcat.NewCatalog("default", sqldb, sqlcat.SQLite,
iceberg.Properties{"warehouse": "file://" + s.warehouse})
+ s.Require().NoError(err)
+
+ return cat, drv
+}
+
+// Pins the non-Postgres branch: on SQLite a lost insert race recovers the
+// sentinel via the re-check without a savepoint.
+func (s *SqliteCatalogTestSuite) TestCreateNamespaceLosesInsertRaceSQLite() {
+ namespace := table.Identifier{databaseName()}
+ cat, drv := s.newDupCatalog()
+
+ err := cat.CreateNamespace(context.Background(), namespace, nil)
+ s.ErrorIs(err, catalog.ErrNamespaceAlreadyExists)
+ s.Contains(err.Error(), namespace[0])
+ s.True(drv.insertAttempted.Load(), "the emulated insert must have run")
+}
+
+// Exercises the Postgres savepoint happy path: SAVEPOINT -> insert -> RELEASE.
+func (s *SqliteCatalogTestSuite) TestCreateNamespaceSucceedsThroughSavepoint()
{
+ namespace := table.Identifier{databaseName()}
+ cat, drv := s.newAbortCatalog(insertSucceeds)
+
+ err := cat.CreateNamespace(context.Background(), namespace, nil)
+ s.Require().NoError(err)
+ s.True(drv.insertAttempted.Load(), "the emulated insert must have run")
Review Comment:
**major** — Savepoint happy-path test cannot distinguish RELEASE SAVEPOINT
from ROLLBACK TO SAVEPOINT — a silent data-loss mutant survives
TestCreateNamespaceSucceedsThroughSavepoint asserts only
s.Require().NoError(err) and drv.insertAttempted, neither of which depends on
whether the savepoint was released or rolled back. The whole point of the new
Postgres branch is that sp.Commit() emits RELEASE SAVEPOINT (bun v1.2.18
db.go:670) so the insert survives into the outer tx; if it instead emitted
ROLLBACK TO SAVEPOINT (db.go:691) CreateNamespace would return nil having
created nothing — silent data loss on the hot path of every successful Postgres
CreateNamespace, with no real Postgres anywhere in CI. Fix is one assertion:
the pgAbortDriver only intercepts EXISTS probes (isNamespaceExistsProbe), so a
plain COUNT(*) on iceberg_namespace_properties, or CheckNamespaceExists via a
non-intercepting handle, passes through to the real sqlite base and pins it.
<details><summary>Evidence</summary>
```text
Mutation `sp.Commit()` -> `sp.Rollback()` at sql.go:1303. PR's own test: 'ok
github.com/apache/iceberg-go/catalog/sql 0.753s'; full suite also 'ok ...
0.980s'. Probe asserting persistence: unmutated 'PROBE: persisted namespace
property rows = 1' PASS; mutated 'PROBE: persisted namespace property rows = 0'
/ '"0" is not positive' / 'CreateNamespace returned nil but persisted no
namespace rows' FAIL. Confirmed bun Tx.BeginTx sets a non-empty savepoint name
(db.go:753-776), so Commit/Rollback really are RELEASE/ROLLBACK TO, not
whole-tx operations.
```
</details>
--
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]
