[PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["iremcaginyurtturk (via GitHub)" <gi...@apache.org>]

iremcaginyurtturk opened a new pull request, #1635:
URL: https://github.com/apache/iceberg-go/pull/1635

   ## Problem
   
   `CreateNamespace` in the sql catalog resolves the namespace key, returns `ErrNamespaceAlreadyExists` if it is present, and otherwise inserts the properties. Two callers creating the same namespace both pass that check, and the loser's insert violates the primary key on `iceberg_namespace_properties`. The driver's error is returned unwrapped:
   
   ```
   error inserting namespace properties for namespace '[raw]': constraint failed:
   UNIQUE constraint failed: iceberg_namespace_properties.catalog_name,
   iceberg_namespace_properties.namespace, iceberg_namespace_properties.property_key (1555)
   ```
   
   Callers that write `errors.Is(err, catalog.ErrNamespaceAlreadyExists)` — the contract the other implementations honour, and the natural way to write create-if-absent — see a failure for a namespace that does exist. It surfaces as an intermittent first-run failure in any tool that loads two tables into a new namespace concurrently; that is how I ran into it.
   
   ## Fix
   
   Re-resolve the namespace key when the insert fails, and return `ErrNamespaceAlreadyExists` if it is present now. An insert that failed for any other reason is returned exactly as before.
   
   I kept the pre-check rather than replacing it with an upsert: it keeps the common path a single read, and the behaviour on the raced path is then identical to the non-raced one.
   
   ## Testing
   
   `TestCreateNamespaceConcurrent` starts eight goroutines that create the same namespace behind a barrier, then asserts exactly one `nil` and seven `ErrNamespaceAlreadyExists`.
   
   It reproduces the bug deterministically — on `main` it fails on the first run with the `UNIQUE constraint failed` error above, and passes with the fix. The test's catalog sets a sqlite `busy_timeout` so the writers queue on the file lock rather than failing with `SQLITE_BUSY`, which is separate contention and not what this test is about.
   
   `go test ./catalog/...` passes, `gofmt` clean, `golangci-lint run catalog/sql/...` reports 0 issues.
   
   I did not open an issue first, per CONTRIBUTING — happy to file one and link it if you would prefer that order.


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["laskoviymishka (via GitHub)" <gi...@apache.org>]

laskoviymishka commented on code in PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#discussion_r3719182685


##########
catalog/sql/sql_test.go:
##########
@@ -1911,6 +1914,58 @@ 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
+	for err := range errs {
+		switch {
+		case err == nil:
+			created++
+		case errors.Is(err, catalog.ErrNamespaceAlreadyExists):
+			alreadyExists++
+		default:
+			s.Failf("unexpected error", "want nil or ErrNamespaceAlreadyExists, got %v", err)

Review Comment:
   `s.Failf` marks the test failed but doesn't stop the loop, so one unexpected error leaves `created` and `alreadyExists` short and you get three failures instead of one, with the message that actually matters buried under two count mismatches.
   
   I'd collect and assert after the drain:
   
   ```go
   var unexpected []error
   // ... default: unexpected = append(unexpected, err)
   s.Empty(unexpected, "want nil or ErrNamespaceAlreadyExists")
   s.Equal(1, created)
   s.Equal(writers-1, alreadyExists)
   ```



##########
catalog/sql/sql_test.go:
##########
@@ -1911,6 +1914,58 @@ 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
+	for err := range errs {
+		switch {
+		case err == nil:
+			created++
+		case errors.Is(err, catalog.ErrNamespaceAlreadyExists):
+			alreadyExists++
+		default:
+			s.Failf("unexpected error", "want nil or ErrNamespaceAlreadyExists, got %v", err)
+		}
+	}
+
+	s.Equal(1, created)
+	s.Equal(writers-1, alreadyExists)

Review Comment:
   This assertion can't tell whether the new branch ran.
   
   The pre-check at line 1246 sits outside `withWriteTx`, which doesn't open until 1261. So a loser has two ways to get here: it sees `exists == true` at 1251 and returns before touching the insert, or it passes the pre-check, loses the write-lock race, and reaches the new code at 1276. Both return `fmt.Errorf("%w: %s", catalog.ErrNamespaceAlreadyExists, strings.Join(namespace, "."))`, byte for byte, so `alreadyExists == 7` holds either way and nothing in the drain loop can distinguish them. Which path a given goroutine takes is down to scheduling.
   
   I'd add a deterministic test next to this one that forces the insert to fail: create the namespace first, then drive the insert path with the row already present, so 1276-1278 are actually covered. Keeping this one as a concurrency smoke test is fine, I'd just not lean on it as the regression test for the fix.



##########
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 can insert between the check above and here, and
+			// the driver's duplicate-key error is not what callers match on.
+			if _, exists, checkErr := c.resolveNamespaceKey(ctx, namespace); checkErr == nil && exists {

Review Comment:
   non-blocking: the comment covers the race but not the `checkErr == nil` half of the guard. I'd add a clause saying that if the re-check itself fails we fall through and return the original insert error, since that's the non-obvious part. 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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["iremcaginyurtturk (via GitHub)" <gi...@apache.org>]

iremcaginyurtturk commented on PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#issuecomment-5191744286

   Thanks — you were right that the concurrent test proved nothing about the new branch. Fixed in `358e302`.
   
   **Deterministic coverage.** `TestCreateNamespaceLosesInsertRace` only passes if the insert ran and failed: a driver wrapper plants the namespace row on the same connection immediately before the catalog's insert. It has to be a separate statement — sqlite rolls back the failing statement's own changes, and a `BEFORE INSERT` trigger goes with it, which is where I started and why that approach does not work. Verified both ways: with the recovery removed it fails on `Target error should be in err chain`, with it in place it passes.
   
   **The re-check moved into the transaction**, `resolveNamespaceKeyInTx(ctx, tx, ...)`. Two reasons, both found writing the test above. On a separate connection it cannot see a winner that has not committed yet, and with a constrained pool it deadlocks outright — the new test sets `SetMaxOpenConns(1)` and the old code hung there rather than failing. Thanks for pushing on this; the original would have been unreliable in exactly the situation it was meant to handle.
   
   **Error collection** in the concurrent test now accumulates and asserts after the drain, as suggested — one stray error reports once instead of three times. I kept that test as the smoke test, with a comment saying the deterministic one is the regression test.
   
   **Comment** now covers the `checkErr == nil` fall-through.
   
   On the follow-up: agreed, and `CreateNamespace` being the only create path with its check outside the transaction is the real fix — this recovers from the race rather than closing it. Happy to open an issue for moving it to `withSerializableWriteTx` alongside `CreateTable` and `CreateView`, unless you would rather fold it into this PR.
   
   `go test ./catalog/...` passes, `golangci-lint run catalog/sql/...` reports 0 issues.


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["laskoviymishka (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["laskoviymishka (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["iremcaginyurtturk (via GitHub)" <gi...@apache.org>]

iremcaginyurtturk commented on PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#issuecomment-5237008631

   Thanks for the detailed pass. Pushed the fixes:
   
   - **Error shape:** the race-loser now returns the same `fmt.Errorf("%w: %s", …)` as the serial pre-check, so the two paths are byte-identical and `errors.As` no longer unwraps to the driver error only on the race path.
   - **Dropped causes:** the rollback-failure branch now `errors.Join(err, rbErr)` and the re-check-failure branch `errors.Join(err, checkErr)`, so a poisoned tx / failed re-check isn't reported as a plain insert error.
   - **mssql:** confirmed bun emits `SAVE TRANSACTION` for the MSSavepoint feature, so the savepoint path holds across the declared dialects.
   - **Emulation robustness:** added a comment naming the bun SQL shape the matchers depend on.
   
   On the `boolRows` EOF suggestion — I went a different way, and wanted to flag why: bun's `Exists()` runs `SELECT EXISTS(...)` and scans the value via `QueryRowContext(...).Scan(&bool)`, so an empty result comes back as `sql.ErrNoRows` rather than "false" and the probe errors out (I hit this — the tests fail). So I kept `boolRows` emitting an explicit `0`/`1` row and documented that it must, rather than signalling EOF. I did confirm the pinning holds: reverting the savepoint makes `TestCreateNamespaceLosesInsertRace` fail, and the negative case would fail if the re-check ever reported "exists" for the unrelated-failure path — so both are genuinely pinned. Happy to revisit if you'd prefer a different shape there.


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["laskoviymishka (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["zeroshade (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["zeroshade (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["zeroshade (via GitHub)" <gi...@apache.org>]

zeroshade commented on code in PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#discussion_r3960483440


##########
catalog/sql/sql.go:
##########
@@ -1269,9 +1269,46 @@ func (c *Catalog) CreateNamespace(ctx context.Context, namespace table.Identifie
 			})
 		}
 
-		_, err := tx.NewInsert().Model(&toInsert).Exec(ctx)
-		if err != nil {
-			return fmt.Errorf("error inserting namespace properties for namespace '%s': %w", namespace, err)
+		// A concurrent writer may have won the race; return the sentinel in the same
+		// form as the pre-check above. This must stay the first read on the outer tx
+		// so MySQL's REPEATABLE READ snapshot lands after the winner committed.
+		recheck := func(insertErr error) error {
+			_, 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(insertErr, checkErr))
+			}
+
+			return fmt.Errorf("error inserting namespace properties for namespace '%s': %w", namespace, insertErr)
+		}
+
+		// Postgres aborts the whole tx on a failed insert, so only it needs the
+		// savepoint for the re-check; SQLite/MySQL don't, and Oracle can't RELEASE it.
+		if tx.Dialect().Name() == dialect.PG {
+			sp, err := tx.BeginTx(ctx, nil)

Review Comment:
   **nit** — Postgres-only gate on the savepoint branch is unpinned by any test
   
   The dialect gate exists so Oracle (no RELEASE SAVEPOINT) and SQLite/MySQL (statement-level rollback) skip the savepoint. Removing it is behaviourally invisible to the suite because sqlite supports savepoints, so the emulation cannot distinguish gated from ungated. Cheaply pinnable: have dupInsertConn.ExecContext (sql_test.go:2188) record whether any statement matching a SAVEPOINT prefix was seen, and assert it was NOT in TestCreateNamespaceLosesInsertRaceSQLite. Follow-up, not a merge blocker.



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["zeroshade (via GitHub)" <gi...@apache.org>]

zeroshade commented on code in PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635#discussion_r3926371927


##########
catalog/sql/sql.go:
##########
@@ -1269,9 +1269,46 @@ func (c *Catalog) CreateNamespace(ctx context.Context, namespace table.Identifie
 			})
 		}
 
-		_, err := tx.NewInsert().Model(&toInsert).Exec(ctx)
-		if err != nil {
-			return fmt.Errorf("error inserting namespace properties for namespace '%s': %w", namespace, err)
+		// A concurrent writer may have won the race; return the sentinel in the same
+		// form as the pre-check above. This must stay the first read on the outer tx
+		// so MySQL's REPEATABLE READ snapshot lands after the winner committed.
+		recheck := func(insertErr error) error {
+			_, 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(insertErr, checkErr))
+			}
+
+			return fmt.Errorf("error inserting namespace properties for namespace '%s': %w", namespace, insertErr)
+		}
+
+		// Only Postgres aborts the whole tx on a failed insert, so only it needs a
+		// savepoint to keep the re-check runnable; Oracle has no RELEASE SAVEPOINT.
+		if tx.Dialect().Name() == dialect.PG {
+			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, errors.Join(err, rbErr))
+				}
+
+				return recheck(err)
+			}
+
+			if err = sp.Commit(); err != nil {
+				return fmt.Errorf("error releasing savepoint for namespace '%s': %w", namespace, err)

Review Comment:
   **minor** — Postgres happy path (SAVEPOINT/RELEASE SAVEPOINT) is never executed by any test
   
   Both pgAbortDriver tests force the insert to fail, so sp.Commit() at 1303 and the 'return nil' at 1307 never run. This is newly added code on the hot path of every successful CreateNamespace against Postgres, and there is no real Postgres anywhere in CI (grep over .github/workflows and dev/ finds none). A success-path assertion through the existing pg-dialect-over-sqlite harness would cover it cheaply.



##########
catalog/sql/sql.go:
##########
@@ -1269,9 +1269,46 @@ func (c *Catalog) CreateNamespace(ctx context.Context, namespace table.Identifie
 			})
 		}
 
-		_, err := tx.NewInsert().Model(&toInsert).Exec(ctx)
-		if err != nil {
-			return fmt.Errorf("error inserting namespace properties for namespace '%s': %w", namespace, err)
+		// A concurrent writer may have won the race; return the sentinel in the same
+		// form as the pre-check above. This must stay the first read on the outer tx
+		// so MySQL's REPEATABLE READ snapshot lands after the winner committed.
+		recheck := func(insertErr error) error {
+			_, 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(insertErr, checkErr))
+			}
+
+			return fmt.Errorf("error inserting namespace properties for namespace '%s': %w", namespace, insertErr)
+		}
+
+		// Only Postgres aborts the whole tx on a failed insert, so only it needs a
+		// savepoint to keep the re-check runnable; Oracle has no RELEASE SAVEPOINT.
+		if tx.Dialect().Name() == dialect.PG {
+			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, errors.Join(err, rbErr))
+				}
+
+				return recheck(err)
+			}
+
+			if err = sp.Commit(); err != nil {
+				return fmt.Errorf("error releasing savepoint for namespace '%s': %w", namespace, err)
+			}
+
+			return nil
+		}
+
+		if _, err := tx.NewInsert().Model(&toInsert).Exec(ctx); err != nil {
+			return recheck(err)

Review Comment:
   **major** — Non-Postgres recovery branch has no deterministic test; the PR's own repro dialect is unpinned
   
   The Postgres gate added in 25fd793 moved the deterministic driver test entirely onto the PG branch, so the SQLite/MySQL/Oracle/MSSQL dispatch at 1310-1312 is covered only by the scheduling-dependent TestCreateNamespaceConcurrent. This re-opens `laskoviymishka`'s round-1 blocker for exactly the dialect in the PR description's repro (the quoted error is SQLite's 1555). Add a SQLite-dialect deterministic test using a NON-aborting driver. Note a dialect parameter on newAbortCatalog will NOT work: pgAbortDriver unconditionally sets `aborted` on the insert and only ROLLBACK TO SAVEPOINT clears it, so a SQLite-dialect run never clears it and the re-check fails with errEmulatedAborted. It needs a separate ~40-line driver reusing isNamespaceInsert/isNamespaceExistsProbe/boolRows.
   
   <details><summary>Evidence</summary>
   
   ```text
   go tool cover: 'catalog/sql/sql.go:1311.4,1312.1 1 0' (count 0). Reverting 1310-1312 to the pre-PR raw error leaves 'go test ./catalog/sql/ -count=1' => ok. Detection rate measured over 15 independent single runs: 13/15. Throwaway probe with a non-aborting dupDriver at SQLite dialect: unmodified head => 'race-loser err = namespace already exists: my-iceberg-db-mpifixpfynaunvszosha' PASS; reverted mutant => 'error inserting namespace properties ...: UNIQUE constraint failed' FAIL.
   ```
   
   </details>



##########
catalog/sql/sql.go:
##########
@@ -1269,9 +1269,46 @@ func (c *Catalog) CreateNamespace(ctx context.Context, namespace table.Identifie
 			})
 		}
 
-		_, err := tx.NewInsert().Model(&toInsert).Exec(ctx)
-		if err != nil {
-			return fmt.Errorf("error inserting namespace properties for namespace '%s': %w", namespace, err)
+		// A concurrent writer may have won the race; return the sentinel in the same
+		// form as the pre-check above. This must stay the first read on the outer tx
+		// so MySQL's REPEATABLE READ snapshot lands after the winner committed.
+		recheck := func(insertErr error) error {
+			_, 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(insertErr, checkErr))
+			}
+
+			return fmt.Errorf("error inserting namespace properties for namespace '%s': %w", namespace, insertErr)
+		}
+
+		// Only Postgres aborts the whole tx on a failed insert, so only it needs a
+		// savepoint to keep the re-check runnable; Oracle has no RELEASE SAVEPOINT.

Review Comment:
   **nit** — Gate comment conflates two independent justifications
   
   'Only Postgres aborts the whole tx on a failed insert, so only it needs a savepoint to keep the re-check runnable; Oracle has no RELEASE SAVEPOINT.' These are two separate facts: the first justifies why PG needs the savepoint, the second why it must not be applied universally. As written, Oracle reads like the reason for a Postgres-only gate. Also, the 'only Postgres' claim is asserted for MSSQL without evidence (SQL Server's behavior depends on XACT_ABORT); scoping the sentence to the dialects actually reasoned about would be more honest.



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] fix(catalog/sql): return ErrNamespaceAlreadyExists on a concurrent create [iceberg-go]

["laskoviymishka (via GitHub)" <gi...@apache.org>]

laskoviymishka merged PR #1635:
URL: https://github.com/apache/iceberg-go/pull/1635


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org