laskoviymishka commented on code in PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#discussion_r3723143703
##########
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:
This guard can't fire on Postgres. A statement error there marks the whole
transaction aborted and every command after it gets `25P02` until a rollback,
so once the insert fails with a `23505` unique violation the `SELECT` that
`resolveNamespaceKeyInTx` runs on this same `tx` comes back `checkErr != nil`
rather than `exists`. We fall through and return the raw insert error, which is
exactly what we returned before this PR. `postgres` is one of the five dialects
this file declares, so it's a supported target.
SQLite rolls back only the offending statement, which is why the new test is
green. MySQL and Oracle behave like SQLite here; Postgres is the outlier, and
it's the one most deployments run against.
Two shapes hold across dialects: a `SAVEPOINT` around the insert so the
re-check has a live transaction to run in, or matching the driver's
unique-violation code directly and dropping the re-query entirely. Moving the
existence check inside a `withSerializableWriteTx`, the way `CreateTable` and
`DropNamespace` do, is worth doing regardless, but on Postgres unique-index
enforcement isn't covered by SSI predicate locking, so a loser can still come
back with `23505` instead of a serialization failure. It narrows the window
rather than replacing this block. wdyt?
##########
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
Review Comment:
This comment has the mechanism slightly off, and the wrong version is what
hides the dialect gap. The plant isn't "its own statement" in any transactional
sense; it's in the same transaction as the catalog's insert and never
separately committed. What actually saves it is that SQLite rolls back only the
failing statement, so the plant row survives to be seen by the re-check.
I'd say that part out loud, because on Postgres the same sequence loses both
and the re-check can't run at all (see `sql.go:1276`). Something like "SQLite
rolls back only the failing statement, so the plant stays visible to the
re-check; other dialects differ" keeps the next reader from taking this green
test as multi-dialect coverage.
##########
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)
Review Comment:
non-blocking: these two panic rather than fail if the shim ever stops
implementing the interface, and a panic in a driver callback takes the whole
test binary down with a trace that won't point back here. `ExecContext` and
`QueryContext` just above already do the ok-check, so it's mostly consistency:
```go
beginTx, ok := c.Conn.(driver.ConnBeginTx)
if !ok {
return nil, driver.ErrBadConn
}
return beginTx.BeginTx(ctx, opts)
```
Same for `PrepareContext` below it.
--
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]
