laskoviymishka commented on code in PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#discussion_r3779433620
##########
catalog/sql/sql_test.go:
##########
@@ -1911,6 +1917,253 @@ 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
+}
+
+// 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) {
Review Comment:
One robustness gap in the emulation: if the wrapped conn ever doesn't
implement `ExecerContext` (same for `QueryerContext` just below), this returns
`driver.ErrSkip`, and database/sql falls back to prepare-then-exec, which skips
the interceptor entirely. The injected failure never fires, the test sees a
clean `CreateNamespace`, and it fails on a confusing nil-vs-sentinel mismatch
instead of "the simulation didn't run." sqliteshim implements both today so
it's not live, but nothing pins it.
Asserting `drv.insertAttempted.Load()` after the call in both race tests
would catch that, and doubles as proof the savepoint path actually ran rather
than getting skipped.
##########
catalog/sql/sql.go:
##########
@@ -1269,11 +1269,35 @@ 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 {
+ // A failed rollback poisons the tx (Postgres);
keep both causes.
+ return fmt.Errorf("error inserting namespace
properties for namespace '%s': %w", namespace, errors.Join(err, rbErr))
+ }
+ // A concurrent writer may have won the race; return
the sentinel in the
+ // same form as the pre-check above so both paths are
identical.
+ _, exists, checkErr := c.resolveNamespaceKeyInTx(ctx,
tx, namespace)
Review Comment:
While we're here: this re-check being correct on MySQL leans on it being the
first consistent read in the outer tx. Under REPEATABLE READ the snapshot is
taken at the first non-locking SELECT, and since the insert is the only
statement before it, the snapshot lands after the race winner committed and
`exists` comes back true. Safe today.
It's a quiet invariant though. If anyone later adds an existence SELECT
before the insert inside `withWriteTx`, the snapshot would establish too early
and the re-check could miss the winner. A one-line comment that this must stay
the first read on the outer tx would keep it from breaking silently. Not a
blocker.
##########
catalog/sql/sql.go:
##########
@@ -1269,11 +1269,35 @@ 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 {
+ // A failed rollback poisons the tx (Postgres);
keep both causes.
+ return fmt.Errorf("error inserting namespace
properties for namespace '%s': %w", namespace, errors.Join(err, rbErr))
+ }
+ // A concurrent writer may have won the race; return
the sentinel in the
+ // same form as the pre-check above so both paths are
identical.
+ _, exists, checkErr := c.resolveNamespaceKeyInTx(ctx,
tx, namespace)
+ if checkErr == nil && exists {
+ return fmt.Errorf("%w: %s",
catalog.ErrNamespaceAlreadyExists, strings.Join(namespace, "."))
+ }
+ if checkErr != nil {
+ return fmt.Errorf("error inserting namespace
properties for namespace '%s': %w", namespace, errors.Join(err, checkErr))
+ }
+
return fmt.Errorf("error inserting namespace properties
for namespace '%s': %w", namespace, err)
}
+ if err = sp.Commit(); err != nil {
Review Comment:
This is the other leg of the multi-dialect savepoint worry from last round.
I checked mssql's `SAVE TRANSACTION` on the create side, but the commit side
has its own gap: `sp.Commit()` on a savepoint-backed tx emits `RELEASE
SAVEPOINT`, and Oracle has no such statement. It supports `SAVEPOINT` and
`ROLLBACK TO SAVEPOINT` but frees savepoints implicitly. bun only skips the
RELEASE for the mssql feature flag as far as I can tell.
If that's right, this regresses Oracle on the happy path: `CreateNamespace`
was a plain `Exec` before this PR with no savepoint, so the commit would now
fail and the `RunInTx` defer rolls back the insert that just succeeded.
Oracle's a declared dialect, so it's not hypothetical.
Worth confirming against bun's oracle dialect before merge. If it doesn't
emit a no-op RELEASE, I'd gate the savepoint to Postgres, the only dialect that
actually needs the aborted-transaction recovery, rather than take it on every
path. 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]
