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: [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]
