s3tables, iceberg: create missing directories exclusively so a concurrent create keeps its record (#12600)

* s3tables: create a missing directory exclusively in ensureDirectory

ensureDirectory looked the path up and, when it was missing, created it with a
plain CreateEntry. A table create that lost the lookup race then replaced the
directory another create had made, claimed and confirmed since, erasing that
table's record, so a third create could claim the name and win as well.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: create missing metadata parent directories exclusively

saveMetadataBlob created each missing directory on the way to the metadata
directory with a plain CreateEntry after a lookup. A table directory another
create had made and written its record onto since the lookup was replaced by a
bare directory, erasing that record.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: create missing stage-create marker directories exclusively

The marker path starts at the table bucket directory, which carries the bucket
record; a plain create after a missed lookup replaces a directory another writer
made in between, with whatever it carries.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/s3api/iceberg/iceberg_create_table_conflict_test.go

@@ -940,6 +940,46 @@ func TestFinalizeCreateOnCommitLosesToTableWithoutLocation(t *testing.T) {
 	e.assertWinnerRecord("", winnerUUID)
 }
 
+// A table directory another create made and wrote its record onto after this
+// writer's lookup missed it keeps that record when a metadata file is staged.
+func TestSaveMetadataBlobKeepsATableDirectoryCreatedSinceTheLookup(t *testing.T) {
+	e := newConflictEnv(t)
+	namespaceDir := path.Join(s3tables.TablesPath, conflictBucket, conflictNamespace)
+	record := `{"Name":"` + conflictTable + `"}`
+	e.hooks.beforeCreate = func(req *filer_pb.CreateEntryRequest) {
+		if req.Directory == namespaceDir && req.Entry.Name == conflictTable {
+			e.hooks.beforeCreate = nil
+			e.filer.Put(namespaceDir, conflictTable, map[string][]byte{s3tables.ExtendedKeyMetadata: []byte(record)})
+		}
+	}
+	if err := e.server.saveMetadataBlob(t.Context(), conflictBucket, path.Join(conflictNamespace, conflictTable), "v1.metadata.json", []byte("{}"), "application/json", true, nil); err != nil {
+		t.Fatalf("save metadata: %v", err)
+	}
+	if got := e.filer.Get(namespaceDir, conflictTable); got == nil || string(got.Extended[s3tables.ExtendedKeyMetadata]) != record {
+		t.Errorf("table record replaced by the directory create: %+v", got)
+	}
+}
+
+// A stage-create marker write keeps the attributes of a directory on its path
+// that another writer created after the marker's lookup missed it.
+func TestStageCreateMarkerKeepsADirectoryCreatedSinceTheLookup(t *testing.T) {
+	e := newConflictEnv(t)
+	bucketDir := path.Join(s3tables.TablesPath, conflictBucket)
+	attrs := map[string][]byte{"probe": []byte("kept")}
+	e.hooks.beforeCreate = func(req *filer_pb.CreateEntryRequest) {
+		if req.Directory == bucketDir && req.Entry.Name == stageCreateMarkerDirName {
+			e.hooks.beforeCreate = nil
+			e.filer.Put(bucketDir, stageCreateMarkerDirName, attrs)
+		}
+	}
+	if err := e.server.writeStageCreateMarker(t.Context(), conflictBucket, []string{conflictNamespace}, conflictTable, uuid.New(), conflictLocation, ""); err != nil {
+		t.Fatalf("write stage-create marker: %v", err)
+	}
+	if got := e.filer.Get(bucketDir, stageCreateMarkerDirName); got == nil || string(got.Extended["probe"]) != "kept" {
+		t.Errorf("directory replaced by the marker's directory create: %+v", got)
+	}
+}
+
 // Two stage-create commits for one name: one creates the table and the other is
 // told it lost, and the file the catalog points at belongs to the winner.
 func TestConcurrentFinalizeCreateOnCommitHasOneWinner(t *testing.T) {

weed/s3api/iceberg/metadata_files.go

@@ -68,9 +68,11 @@ func (s *Server) saveMetadataBlob(ctx context.Context, bucketName, tablePath, me
 				return fmt.Errorf("lookup %s failed: %w", errorContext, err)
 			}
 
-			// If lookup fails with ErrNotFound, try to create the directory.
+			// Exclusive, so a directory created since the lookup keeps the
+			// record a table create may have written onto it.
 			resp, createErr := client.CreateEntry(opCtx, &filer_pb.CreateEntryRequest{
 				Directory: parent,
+				OExcl:     true,
 				Entry: &filer_pb.Entry{
 					Name:        name,
 					IsDirectory: true,

weed/s3api/iceberg/stage_create.go

@@ -191,6 +191,7 @@ func (s *Server) writeStageCreateMarker(ctx context.Context, bucketName string,
 
 			resp, createErr := client.CreateEntry(opCtx, &filer_pb.CreateEntryRequest{
 				Directory: parent,
+				OExcl:     true,
 				Entry: &filer_pb.Entry{
 					Name:        name,
 					IsDirectory: true,

weed/s3api/s3tables/filer_ops.go

@@ -72,7 +72,9 @@ func (h *S3TablesHandler) createDirectoryEntry(ctx context.Context, client filer
 	})
 }
 
-// ensureDirectory ensures a directory exists at the specified path
+// ensureDirectory ensures a directory exists at the specified path. The create
+// is exclusive: a directory another writer made since the lookup may already
+// carry a record, which a plain create would replace with a bare directory.
 func (h *S3TablesHandler) ensureDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
 	dir, name := splitPath(path)
 	_, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
@@ -83,7 +85,9 @@ func (h *S3TablesHandler) ensureDirectory(ctx context.Context, client filer_pb.S
 		return nil
 	}
 	if errors.Is(err, filer_pb.ErrNotFound) {
-		return h.createDirectory(ctx, client, path)
+		if err = h.createDirectoryExclusive(ctx, client, path, nil); errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
+			return nil
+		}
 	}
 	return err
 }

weed/s3api/s3tables/filer_ops_race_test.go

@@ -87,3 +87,16 @@ func TestCreateNamespaceLostRaceKeepsTheFirstCreate(t *testing.T) {
 	require.NotNil(t, survivor)
 	assert.Equal(t, `{"OwnerAccountID":"competitor"}`, string(survivor.Extended[ExtendedKeyMetadata]))
 }
+
+func TestEnsureDirectoryKeepsADirectoryCreatedSinceTheLookup(t *testing.T) {
+	fs := s3tablestest.Start(t)
+	tablePath := GetTablePath("bucket", "ns", "orders")
+	dir, name := splitPath(tablePath)
+	client := &competingCreateClient{SeaweedFilerClient: fs.Client, f: fs, dir: dir, name: name, plant: true}
+
+	require.NoError(t, NewS3TablesHandler().ensureDirectory(context.Background(), client, tablePath))
+
+	survivor := fs.Get(dir, name)
+	require.NotNil(t, survivor)
+	assert.Equal(t, `{"OwnerAccountID":"competitor"}`, string(survivor.Extended[ExtendedKeyMetadata]))
+}