zeroshade commented on code in PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#discussion_r3732123497
##########
catalog/sql/sql.go:
##########
@@ -1271,6 +1271,12 @@ func (c *Catalog) CreateNamespace(ctx context.Context,
namespace table.Identifie
_, err := tx.NewInsert().Model(&toInsert).Exec(ctx)
if err != nil {
+ // A concurrent writer may have inserted since the
check above; if the
+ // re-check itself fails, fall through to the insert
error.
+ if _, exists, checkErr :=
c.resolveNamespaceKeyInTx(ctx, tx, namespace); checkErr == nil && exists {
Review Comment:
**Blocking — this branch cannot fire on PostgreSQL, so the bug is unfixed
for that dialect.**
The re-check runs on `tx`, the same transaction whose `INSERT` just failed
on line 1272. PostgreSQL aborts a transaction on *any* statement error
(SQLSTATE 25P02); every subsequent command in that transaction fails with
`current transaction is aborted, commands ignored until end of transaction
block`.
So on Postgres `checkErr != nil` always, the `&& exists` branch is never
taken, and control falls through to line 1280 — returning the raw duplicate-key
error, which is precisely the behavior this PR sets out to fix. Postgres is a
first-class dialect here (`sql.go:58`, `sql.go:243`), not an exotic one.
MySQL and SQLite don't abort the transaction on a duplicate-key error, so
the fix works as intended there. Net effect: correct for two of three dialects,
quietly inert on the third, and nothing in the test suite would catch it.
**Suggested fix, either approach:**
- Wrap the insert in a savepoint — `tx.Exec("SAVEPOINT ns_insert")` before,
`ROLLBACK TO SAVEPOINT ns_insert` on failure — which restores the transaction
to a usable state on Postgres and is a harmless no-op on the others.
- Or do the re-check outside the failed transaction, on `c.db` via
`resolveNamespaceKey` rather than `resolveNamespaceKeyInTx`. A fresh connection
isn't poisoned by the aborted transaction. Slightly weaker isolation for the
read, but for an "does it exist now" check after a failure that is fine.
The savepoint version is the more faithful of the two, since it keeps the
read inside the same transactional context.
##########
catalog/sql/sql_test.go:
##########
@@ -1911,6 +1916,148 @@ 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)
+}
+
+// plantingDriver wraps the sqlite driver and, once, inserts the namespace row
+// on the same connection just before the catalog's own insert.
+type plantingDriver struct {
+ base driver.Driver
+ namespace string
+ planted atomic.Bool
+}
+
+func (d *plantingDriver) Open(dsn string) (driver.Conn, error) {
+ conn, err := d.base.Open(dsn)
+ if err != nil {
+ return nil, err
+ }
+
+ return &plantingConn{Conn: conn, drv: d}, nil
+}
+
+type plantingConn struct {
+ driver.Conn
+ drv *plantingDriver
+}
+
+func (c *plantingConn) ExecContext(ctx context.Context, query string, args
[]driver.NamedValue) (driver.Result, error) {
+ execer, ok := c.Conn.(driver.ExecerContext)
+ if !ok {
+ return nil, driver.ErrSkip
+ }
+ // Same statement would be rolled back with the failing insert, so the
plant
+ // goes in as its own statement first.
+ if strings.Contains(query, "iceberg_namespace_properties") &&
strings.HasPrefix(strings.ToUpper(strings.TrimSpace(query)), "INSERT") &&
c.drv.planted.CompareAndSwap(false, true) {
+ if _, err := execer.ExecContext(ctx, "INSERT INTO
iceberg_namespace_properties (catalog_name, namespace, property_key,
property_value) VALUES (?, ?, ?, ?)",
+ []driver.NamedValue{
+ {Ordinal: 1, Value: "default"},
+ {Ordinal: 2, Value: c.drv.namespace},
+ {Ordinal: 3, Value: "exists"},
+ {Ordinal: 4, Value: "true"},
+ }); err != nil {
+ return nil, err
+ }
+ }
+
+ return execer.ExecContext(ctx, query, args)
+}
+
+func (c *plantingConn) QueryContext(ctx context.Context, query string, args
[]driver.NamedValue) (driver.Rows, error) {
+ queryer, ok := c.Conn.(driver.QueryerContext)
+ if !ok {
+ return nil, driver.ErrSkip
+ }
+
+ return queryer.QueryContext(ctx, query, args)
+}
+
+func (c *plantingConn) BeginTx(ctx context.Context, opts driver.TxOptions)
(driver.Tx, error) {
+ return c.Conn.(driver.ConnBeginTx).BeginTx(ctx, opts)
+}
+
+func (c *plantingConn) PrepareContext(ctx context.Context, query string)
(driver.Stmt, error) {
+ return c.Conn.(driver.ConnPrepareContext).PrepareContext(ctx, query)
+}
+
+// The concurrent test above cannot tell whether a loser returned from the
check
+// before the insert or from the recovery after it. This one only passes if the
+// insert ran and failed.
+func (s *SqliteCatalogTestSuite) TestCreateNamespaceLosesInsertRace() {
+ ctx := context.Background()
+ namespace := table.Identifier{databaseName()}
+
+ base, err := sql.Open(sqliteshim.ShimName, ":memory:")
+ s.Require().NoError(err)
+ s.Require().NoError(base.Close())
+
+ drvName := "sqlite-planting-" + namespace[0]
+ sql.Register(drvName, &plantingDriver{base: base.Driver(), namespace:
namespace[0]})
Review Comment:
`sql.Register` installs a driver name into a process-global registry and
panics with `Register called twice for driver ...` on a duplicate. This is safe
here only because `drvName` derives from `databaseName()` (`sql_test.go:152`),
which is unique per call.
That's a load-bearing property of a helper defined ~1900 lines away, and the
failure mode is a panic that takes the whole test binary down rather than a
single failing test.
**Suggested fix:** a one-line comment noting the dependency on
`databaseName()` uniqueness would keep someone from later switching to a fixed
driver name for readability.
##########
catalog/sql/sql.go:
##########
@@ -1271,6 +1271,12 @@ func (c *Catalog) CreateNamespace(ctx context.Context,
namespace table.Identifie
_, err := tx.NewInsert().Model(&toInsert).Exec(ctx)
if err != nil {
+ // A concurrent writer may have inserted since the
check above; if the
+ // re-check itself fails, fall through to the insert
error.
+ if _, exists, checkErr :=
c.resolveNamespaceKeyInTx(ctx, tx, namespace); checkErr == nil && exists {
+ return fmt.Errorf("%w: %s",
catalog.ErrNamespaceAlreadyExists, strings.Join(namespace, "."))
+ }
+
return fmt.Errorf("error inserting namespace properties
for namespace '%s': %w", namespace, err)
Review Comment:
Minor: when the re-check succeeds and reports the namespace exists, the
original insert error is discarded. That's defensible — the observable state
*is* "already exists," and the caller gets the sentinel they need — but it does
mean an insert that failed for some unrelated reason, on a namespace that
happens to exist, reports a cause that isn't the real one.
**Suggested fix:** consider `errors.Join`-ing the driver error into the
returned error on the recovery branch above, so `errors.Is(err,
catalog.ErrNamespaceAlreadyExists)` still works for callers while operators
keep the underlying cause in logs. Not blocking either way.
--
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]
