diff --git a/weed/s3api/iceberg/iceberg_create_table_conflict_test.go b/weed/s3api/iceberg/iceberg_create_table_conflict_test.go index 53b0477de9..12e5122688 100644 --- a/weed/s3api/iceberg/iceberg_create_table_conflict_test.go +++ b/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) { diff --git a/weed/s3api/iceberg/metadata_files.go b/weed/s3api/iceberg/metadata_files.go index 635d36daf6..ad68aa7a6a 100644 --- a/weed/s3api/iceberg/metadata_files.go +++ b/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, diff --git a/weed/s3api/iceberg/stage_create.go b/weed/s3api/iceberg/stage_create.go index d94ee72c09..7af71bc038 100644 --- a/weed/s3api/iceberg/stage_create.go +++ b/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, diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index abfd70e624..8e8c0798b8 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/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 } diff --git a/weed/s3api/s3tables/filer_ops_race_test.go b/weed/s3api/s3tables/filer_ops_race_test.go index a1f6533862..e1c75eba8d 100644 --- a/weed/s3api/s3tables/filer_ops_race_test.go +++ b/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])) +}