diff --git a/weed/s3api/iceberg/commit_helpers.go b/weed/s3api/iceberg/commit_helpers.go index 8dd19818ec..9ef7eb68c8 100644 --- a/weed/s3api/iceberg/commit_helpers.go +++ b/weed/s3api/iceberg/commit_helpers.go @@ -35,18 +35,66 @@ func sleepBeforeCommitRetry(attempt int) { // decides which commit won. This mirrors how the maintenance worker stages its // own metadata. func (s *Server) stageCommitMetadata(ctx context.Context, metadataBucket, metadataPath, location, metadataFileName string, metadataBytes []byte) (string, string, error) { - err := s.saveMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, true) + name, metadataLocation, _, err := s.stageOwnedMetadata(ctx, metadataBucket, metadataPath, location, metadataFileName, metadataBytes) + return name, metadataLocation, err +} + +// stageOwnedMetadata is stageCommitMetadata that also returns the token the +// file was tagged with, which discardStagedMetadata needs to remove it. +func (s *Server) stageOwnedMetadata(ctx context.Context, metadataBucket, metadataPath, location, metadataFileName string, metadataBytes []byte) (string, string, string, error) { + token := uuid.NewString() + extended := map[string][]byte{stageTokenKey: []byte(token)} + err := s.saveMetadataBlob(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, "application/json", true, extended) if errors.Is(err, filer_pb.ErrEntryAlreadyExists) { // v{N}-{uuid}, a form both the catalog and the maintenance worker // still read the version from. metadataFileName = fmt.Sprintf("%s-%s.metadata.json", strings.TrimSuffix(metadataFileName, ".metadata.json"), uuid.NewString()) glog.V(1).Infof("Iceberg: metadata file already staged for %s, using %s", location, metadataFileName) - err = s.saveMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, true) + err = s.saveMetadataBlob(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, "application/json", true, extended) } if err != nil { - return "", "", err + return "", "", "", err } - return metadataFileName, fmt.Sprintf("%s/metadata/%s", strings.TrimSuffix(location, "/"), metadataFileName), nil + return metadataFileName, fmt.Sprintf("%s/metadata/%s", strings.TrimSuffix(location, "/"), metadataFileName), token, nil +} + +// discardStagedMetadata removes a metadata file this request staged and the +// catalog will never point at. metadataFileName and token must be what +// stageOwnedMetadata returned. The name alone does not identify the file: the +// path can be reused once this request's file is gone — the table dropped and +// its directory purged, then a new create staging the same v1.metadata.json — +// so the delete goes through only while the entry still carries this +// request's token, and a replacement is left alone. +func (s *Server) discardStagedMetadata(ctx context.Context, metadataBucket, metadataPath, metadataFileName, metadataLocation, token string) { + err := s.deleteStagedMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, token) + if errors.Is(err, errStagedFileReplaced) { + glog.V(1).Infof("Iceberg: keeping metadata file %s, it is no longer the one this request staged: %v", metadataLocation, err) + return + } + if err != nil { + glog.V(1).Infof("Iceberg: failed to cleanup metadata file %s after the catalog refused it: %v", metadataLocation, err) + } +} + +// registrationRefused reports whether a failed registration was refused +// before any record pointing at the staged file became visible, so the file can +// go. A name already taken, a conflict, and any other 4xx are refused that way: +// the create wrote nothing, or backed out what it wrote, or lost its +// still-hidden record to another create. One conflict is not: a create whose +// record was already visible, and was updated or renamed away before the +// create could back it out, leaves that record in place — under the original +// name or another — still pointing at the file. A transport failure or 5xx is +// not a refusal either: the record may have been written before the error +// surfaced. +func registrationRefused(err error) bool { + if s3tables.IsPublishedTableChanged(err) { + return false + } + if isS3TablesConflict(err) || isS3TablesAlreadyExists(err) { + return true + } + status, _, _ := classifyManagerError(err) + return status >= 400 && status < 500 } type icebergRequestError struct { @@ -102,16 +150,20 @@ func hasAssertCreateRequirement(requirements table.Requirements) bool { return false } +// isS3TablesAlreadyExists reports whether err says the name is already taken. +// A typed error says so through its code alone: a server fault whose message +// happens to mention an existing entry is still a server fault, and taking it +// for a refusal would answer 409 and discard a file the catalog may point at. +// Only an untyped error is read by its message. func isS3TablesAlreadyExists(err error) bool { if err == nil { return false } - if strings.Contains(strings.ToLower(err.Error()), "already exists") { - return true - } var tableErr *s3tables.S3TablesError - return errors.As(err, &tableErr) && - (tableErr.Type == s3tables.ErrCodeTableAlreadyExists || tableErr.Type == s3tables.ErrCodeNamespaceAlreadyExists || strings.Contains(strings.ToLower(tableErr.Message), "already exists")) + if errors.As(err, &tableErr) { + return tableErr.Type == s3tables.ErrCodeTableAlreadyExists || tableErr.Type == s3tables.ErrCodeNamespaceAlreadyExists + } + return strings.Contains(strings.ToLower(err.Error()), "already exists") } func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommitInput) (*CommitTableResponse, *icebergRequestError) { @@ -147,7 +199,6 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi metadataVersion = 1 } metadataFileName := fmt.Sprintf("v%d.metadata.json", metadataVersion) - newMetadataLocation := fmt.Sprintf("%s/metadata/%s", strings.TrimSuffix(input.location, "/"), metadataFileName) metadataBytes, err := json.Marshal(newMetadata) if err != nil { @@ -194,7 +245,11 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi message: err.Error(), } } - if err := s.saveMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, false); err != nil { + // Two creators racing for the same table both pick this name. Staging it + // exclusively gives each its own file, so the loser can discard its file + // below without deleting the one the winner's record points at. + metadataFileName, newMetadataLocation, stageToken, err := s.stageOwnedMetadata(ctx, metadataBucket, metadataPath, input.location, metadataFileName, metadataBytes) + if err != nil { return nil, &icebergRequestError{ status: http.StatusInternalServerError, errType: "InternalServerError", @@ -216,20 +271,22 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi MetadataVersion: metadataVersion, MetadataLocation: newMetadataLocation, } + var createResp s3tables.CreateTableResponse createErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { mgrClient := s3tables.NewManagerClient(client) - return s.tablesManager.Execute(ctx, mgrClient, "CreateTable", createReq, nil, input.identityName) + return s.tablesManager.Execute(ctx, mgrClient, "CreateTable", createReq, &createResp, input.identityName) }) + tableCreatedConcurrently := &icebergRequestError{ + status: http.StatusConflict, + errType: "CommitFailedException", + message: "Table was created concurrently", + } if createErr != nil { - if cleanupErr := s.deleteMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName); cleanupErr != nil { - glog.V(1).Infof("Iceberg: failed to cleanup metadata file %s after create-on-commit failure: %v", newMetadataLocation, cleanupErr) + if registrationRefused(createErr) { + s.discardStagedMetadata(ctx, metadataBucket, metadataPath, metadataFileName, newMetadataLocation, stageToken) } if isS3TablesConflict(createErr) || isS3TablesAlreadyExists(createErr) { - return nil, &icebergRequestError{ - status: http.StatusConflict, - errType: "CommitFailedException", - message: "Table was created concurrently", - } + return nil, tableCreatedConcurrently } glog.Errorf("Iceberg: CommitTable CreateTable error: %v", createErr) return nil, &icebergRequestError{ @@ -238,6 +295,13 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi message: "Failed to commit table creation: " + createErr.Error(), } } + // A name that already holds a table is answered with that table's own + // location rather than an error: another creator won the race. That + // location is empty for a table registered without one. + if createResp.MetadataLocation != newMetadataLocation { + s.discardStagedMetadata(ctx, metadataBucket, metadataPath, metadataFileName, newMetadataLocation, stageToken) + return nil, tableCreatedConcurrently + } markerBucket := input.markerBucket if markerBucket == "" { diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 63ba543ca0..6299ccbde3 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -269,11 +269,12 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { } // Authoritative existence check: ask the catalog whether a table is registered - // at this name. If it is, short-circuit with the existing table (idempotent - // CreateTable). Any leftover objects at the target path from a previous - // lifecycle of the same name are purged by handleDropTable on the way out — - // we deliberately do not clean storage here to keep CreateTable free of - // destructive side effects. See issue #9074. + // at this name. If it is, refuse with 409 as the Iceberg REST spec requires; + // clients that want create-if-not-exists catch the 409 and load the table. + // Any leftover objects at the target path from a previous lifecycle of the + // same name are purged by handleDropTable on the way out — we deliberately do + // not clean storage here to keep CreateTable free of destructive side + // effects. See issue #9074. existsReq := &s3tables.GetTableRequest{ TableBucketARN: bucketARN, Namespace: namespace, @@ -285,15 +286,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", existsReq, &existsResp, identityName) }) if existsErr == nil { - // Table already registered. Return the existing definition so CTAS/IF NOT - // EXISTS flows see a stable response instead of a 409. - result, buildErr := s.buildLoadTableResult(r, existsResp, bucketName, namespace, tableName) - if buildErr != nil { - glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr) - writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") - return - } - writeLoadResult(w, http.StatusOK, result) + writeTableAlreadyExists(w, namespace, tableName) return } if !isNoSuchTableError(existsErr) { @@ -319,7 +312,9 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { // Stage-create persists metadata in the internal staged area and skips S3Tables registration. if req.StageCreate { stagedTablePath := stageCreateStagedTablePath(namespace, tableName, tableUUID) - if err := s.saveMetadataFile(r.Context(), metadataBucket, stagedTablePath, metadataFileName, metadataBytes, false); err != nil { + // The staged path carries the table UUID, so the name cannot already be + // taken; exclusive keeps it that way if it ever is. + if err := s.saveMetadataFile(r.Context(), metadataBucket, stagedTablePath, metadataFileName, metadataBytes, true); err != nil { writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save staged metadata file: "+err.Error()) return } @@ -340,7 +335,12 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { writeLoadResult(w, http.StatusOK, result) return } - if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes, false); err != nil { + // The location may already hold a v1.metadata.json: a table created there + // earlier and since dropped, an unregistered file, or a request racing this + // one for the same name. Writing it exclusively keeps this request from + // replacing that file, and from deleting it below if registration fails. + metadataFileName, metadataLocation, stageToken, err := s.stageOwnedMetadata(r.Context(), metadataBucket, metadataPath, location, metadataFileName, metadataBytes) + if err != nil { writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save metadata file: "+err.Error()) return } @@ -368,53 +368,17 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { }) if err != nil { - var tableErr *s3tables.S3TablesError - if errors.As(err, &tableErr) && tableErr.Type == s3tables.ErrCodeTableAlreadyExists { - getReq := &s3tables.GetTableRequest{ - TableBucketARN: bucketARN, - Namespace: namespace, - Name: tableName, - } - var getResp s3tables.GetTableResponse - getErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - mgrClient := s3tables.NewManagerClient(client) - return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", getReq, &getResp, identityName) - }) - if getErr != nil { - writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error()) - return - } - result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName) - if buildErr != nil { - glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr) - writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") - return - } - writeLoadResult(w, http.StatusOK, result) - return + // The file is named by this request alone, so it can go once the + // catalog refused the registration before a record naming it became + // visible. After a failure that may have left one (a conflict on a + // record already published, transport, 5xx) it stays: the record, + // under this name or one it was renamed to, could point at it, and + // removing it would leave the table unreadable. + if registrationRefused(err) { + s.discardStagedMetadata(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataLocation, stageToken) } - if strings.Contains(err.Error(), "already exists") { - getReq := &s3tables.GetTableRequest{ - TableBucketARN: bucketARN, - Namespace: namespace, - Name: tableName, - } - var getResp s3tables.GetTableResponse - getErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - mgrClient := s3tables.NewManagerClient(client) - return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", getReq, &getResp, identityName) - }) - if getErr != nil { - writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error()) - return - } - result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName) - if buildErr != nil { - glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr) - writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") - return - } - writeLoadResult(w, http.StatusOK, result) + if isS3TablesAlreadyExists(err) { + writeTableAlreadyExists(w, namespace, tableName) return } glog.V(1).Infof("Iceberg: CreateTable error: %v", err) @@ -422,11 +386,17 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { return } - // Use returned location if available, otherwise fallback to local one - finalLocation := createResp.MetadataLocation - if finalLocation == "" { - finalLocation = metadataLocation + // The catalog answers a create for a name that already holds a table with + // that table's own metadata location rather than an error. The table was + // registered by a request that won the race, not by this one. A winner + // registered without a location comes back empty, which is no match either: + // this request always sends one, and a fresh create echoes it. + if createResp.MetadataLocation != metadataLocation { + s.discardStagedMetadata(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataLocation, stageToken) + writeTableAlreadyExists(w, namespace, tableName) + return } + finalLocation := metadataLocation if markerErr := s.deleteStageCreateMarkers(r.Context(), bucketName, namespace, tableName); markerErr != nil { glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after create: %v", flattenNamespacePath(namespace), tableName, markerErr) } @@ -993,6 +963,12 @@ func (s *Server) handleRenameTable(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } +// writeTableAlreadyExists answers a create for a name that holds a table with +// the 409 the Iceberg REST spec defines for it. +func writeTableAlreadyExists(w http.ResponseWriter, namespace []string, tableName string) { + writeError(w, http.StatusConflict, "AlreadyExistsException", fmt.Sprintf("Table already exists: %s.%s", strings.Join(namespace, "."), tableName)) +} + // isNoSuchTableError reports whether an error from the S3 Tables manager // indicates the target table is not registered in the catalog. func isNoSuchTableError(err error) bool { diff --git a/weed/s3api/iceberg/iceberg_create_table_conflict_test.go b/weed/s3api/iceberg/iceberg_create_table_conflict_test.go new file mode 100644 index 0000000000..a158a26ef6 --- /dev/null +++ b/weed/s3api/iceberg/iceberg_create_table_conflict_test.go @@ -0,0 +1,1013 @@ +package iceberg + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "path" + "regexp" + "sort" + "strings" + "sync" + "testing" + + "github.com/apache/iceberg-go/table" + "github.com/google/uuid" + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const ( + conflictBucket = "warehouse" + conflictNamespace = "ns" + conflictTable = "orders" + conflictLocation = "s3://" + conflictBucket + "/" + conflictNamespace + "/" + conflictTable +) + +// filerHooks lets a test land a competing write, or a failure, at a chosen RPC +// of the handler under test while the in-memory filer still arbitrates them. +type filerHooks struct { + // beforeCreate sees every CreateEntry, including those the filer refuses. + beforeCreate func(*filer_pb.CreateEntryRequest) + // afterCreate runs once the filer has accepted a CreateEntry. + afterCreate func(*filer_pb.CreateEntryRequest) + // updateErr fails an UpdateEntry, before it reaches the filer, when it + // returns an error. + updateErr func(*filer_pb.UpdateEntryRequest) error + // afterUpdate runs once the filer has accepted an UpdateEntry. + afterUpdate func(*filer_pb.UpdateEntryRequest) + // lookupErr fails a LookupDirectoryEntry, before it reaches the filer, + // when it returns an error. + lookupErr func(*filer_pb.LookupDirectoryEntryRequest) error +} + +type hookedFiler struct { + base *s3tables.ManagerClient + hooks *filerHooks +} + +func (h hookedFiler) WithFilerClient(streamingMode bool, fn func(client filer_pb.SeaweedFilerClient) error) error { + return h.base.WithFilerClient(streamingMode, func(client filer_pb.SeaweedFilerClient) error { + return fn(&hookedClient{SeaweedFilerClient: client, hooks: h.hooks}) + }) +} + +type hookedClient struct { + filer_pb.SeaweedFilerClient + hooks *filerHooks +} + +func (c *hookedClient) CreateEntry(ctx context.Context, in *filer_pb.CreateEntryRequest, opts ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) { + if c.hooks.beforeCreate != nil { + c.hooks.beforeCreate(in) + } + resp, err := c.SeaweedFilerClient.CreateEntry(ctx, in, opts...) + if err == nil && resp.GetErrorCode() == filer_pb.FilerError_OK && c.hooks.afterCreate != nil { + c.hooks.afterCreate(in) + } + return resp, err +} + +func (c *hookedClient) UpdateEntry(ctx context.Context, in *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + if c.hooks.updateErr != nil { + if err := c.hooks.updateErr(in); err != nil { + return nil, err + } + } + resp, err := c.SeaweedFilerClient.UpdateEntry(ctx, in, opts...) + if err == nil && c.hooks.afterUpdate != nil { + c.hooks.afterUpdate(in) + } + return resp, err +} + +func (c *hookedClient) LookupDirectoryEntry(ctx context.Context, in *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + if c.hooks.lookupErr != nil { + if err := c.hooks.lookupErr(in); err != nil { + return nil, err + } + } + return c.SeaweedFilerClient.LookupDirectoryEntry(ctx, in, opts...) +} + +// conflictEnv is a catalog over the in-memory filer, which honours O_EXCL and +// the compare-checked updates the real filer's per-path lock provides, and is +// safe for the concurrent creators these tests race against each other. +type conflictEnv struct { + t *testing.T + filer *s3tablestest.MemFiler + server *Server + router *mux.Router + admin *s3tables.Manager + arn string + hooks *filerHooks +} + +func newConflictEnv(t *testing.T) *conflictEnv { + t.Helper() + filer := s3tablestest.Start(t) + hooks := &filerHooks{} + server := NewServer(hookedFiler{base: s3tables.NewManagerClient(filer.Client), hooks: hooks}, openAuthenticator{}) + router := mux.NewRouter().SkipClean(true) + server.RegisterRoutes(router) + + admin := s3tables.NewManager() + admin.SetTrusted(true) + arn, err := s3tables.BuildBucketARN(s3tables.DefaultRegion, s3tables.DefaultAccountID, conflictBucket) + if err != nil { + t.Fatal(err) + } + e := &conflictEnv{t: t, filer: filer, server: server, router: router, admin: admin, arn: arn, hooks: hooks} + e.adminDo("CreateTableBucket", &s3tables.CreateTableBucketRequest{Name: conflictBucket, Format: s3tables.FormatIceberg}, nil) + e.adminDo("CreateNamespace", &s3tables.CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{conflictNamespace}}, nil) + return e +} + +func (e *conflictEnv) adminDo(op string, req, resp interface{}) { + e.t.Helper() + if err := e.admin.Execute(e.t.Context(), s3tables.NewManagerClient(e.filer.Client), op, req, resp, ""); err != nil { + e.t.Fatalf("%s: %v", op, err) + } +} + +// createTable posts a CreateTable body and returns the response. +func (e *conflictEnv) createTable(body string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, "/v1/"+conflictBucket+"/namespaces/"+conflictNamespace+"/tables", strings.NewReader(body)) + rec := httptest.NewRecorder() + e.router.ServeHTTP(rec, req) + return rec +} + +func (e *conflictEnv) record() (*s3tables.GetTableResponse, error) { + var resp s3tables.GetTableResponse + err := e.admin.Execute(e.t.Context(), s3tables.NewManagerClient(e.filer.Client), "GetTable", &s3tables.GetTableRequest{ + TableBucketARN: e.arn, Namespace: []string{conflictNamespace}, Name: conflictTable, + }, &resp, "") + return &resp, err +} + +func (e *conflictEnv) mustRecord() *s3tables.GetTableResponse { + e.t.Helper() + resp, err := e.record() + if err != nil { + e.t.Fatalf("GetTable: %v", err) + } + return resp +} + +func (e *conflictEnv) metadataDir() string { + return metadataDirPath(conflictBucket, path.Join(conflictNamespace, conflictTable)) +} + +// metadataFiles lists the *.metadata.json files in the table's metadata directory. +func (e *conflictEnv) metadataFiles() []string { + e.t.Helper() + var names []string + err := filer_pb.SeaweedList(e.t.Context(), e.filer.Client, e.metadataDir(), "", func(entry *filer_pb.Entry, _ bool) error { + if strings.HasSuffix(entry.Name, ".metadata.json") { + names = append(names, entry.Name) + } + return nil + }, "", false, 0) + if err != nil { + e.t.Fatalf("list %s: %v", e.metadataDir(), err) + } + sort.Strings(names) + return names +} + +// metadataFile returns the Iceberg metadata a metadata-location names, read +// straight from the filer the way a client reading S3 directly would. +func (e *conflictEnv) metadataFile(location string) table.Metadata { + e.t.Helper() + entry := e.filer.Get(e.metadataDir(), path.Base(location)) + if entry == nil { + e.t.Fatalf("no metadata file behind %s", location) + } + metadata, err := table.ParseMetadataBytes(entry.Content) + if err != nil { + e.t.Fatalf("parse %s: %v", location, err) + } + return metadata +} + +// snapshot captures everything under the bucket: every entry's content and +// attributes, which is where the catalog record and its staged files live. +func (e *conflictEnv) snapshot() map[string]string { + e.t.Helper() + out := map[string]string{} + var walk func(dir string) + walk = func(dir string) { + var children []string + err := filer_pb.SeaweedList(e.t.Context(), e.filer.Client, dir, "", func(entry *filer_pb.Entry, _ bool) error { + keys := make([]string, 0, len(entry.Extended)) + for key := range entry.Extended { + keys = append(keys, key) + } + sort.Strings(keys) + var sb strings.Builder + fmt.Fprintf(&sb, "dir=%v content=%q", entry.IsDirectory, entry.Content) + for _, key := range keys { + fmt.Fprintf(&sb, " %s=%q", key, entry.Extended[key]) + } + p := path.Join(dir, entry.Name) + out[p] = sb.String() + if entry.IsDirectory { + children = append(children, p) + } + return nil + }, "", false, 0) + if err != nil { + e.t.Fatalf("list %s: %v", dir, err) + } + for _, child := range children { + walk(child) + } + } + walk(path.Join(s3tables.TablesPath, conflictBucket)) + return out +} + +func assertSnapshotUnchanged(t *testing.T, before, after map[string]string) { + t.Helper() + for p, want := range before { + got, ok := after[p] + if !ok { + t.Errorf("%s was removed", p) + } else if got != want { + t.Errorf("%s changed:\n before %s\n after %s", p, want, got) + } + } + for p := range after { + if _, ok := before[p]; !ok { + t.Errorf("%s was added", p) + } + } +} + +// conflictTableBody is a CreateTable body whose schema carries column, so the +// table a request produced can be told from another's. +func conflictTableBody(column string, extra string) string { + schema := `{"type":"struct","schema-id":0,"fields":[` + + `{"id":1,"name":"id","required":false,"type":"long"},` + + `{"id":2,"name":"` + column + `","required":false,"type":"string"}]}` + body := `{"name":"` + conflictTable + `","schema":` + schema + if extra != "" { + body += "," + extra + } + return body + "}" +} + +func decodeIcebergError(t *testing.T, rec *httptest.ResponseRecorder) ErrorModel { + t.Helper() + var resp ErrorResponse + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode error body %q: %v", rec.Body.String(), err) + } + return resp.Error +} + +func metadataLocationOf(t *testing.T, rec *httptest.ResponseRecorder) string { + t.Helper() + var resp struct { + MetadataLocation string `json:"metadata-location"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode load result %q: %v", rec.Body.String(), err) + } + return resp.MetadataLocation +} + +func hasColumn(metadata table.Metadata, column string) bool { + _, ok := metadata.CurrentSchema().FindFieldByName(column) + return ok +} + +// Creating a table that exists is a 409 in the Iceberg REST spec, whatever +// schema the second request carries. The first table must not move. +func TestCreateTableOnExistingTableIsConflict(t *testing.T) { + e := newConflictEnv(t) + + first := e.createTable(conflictTableBody("first_col", "")) + if first.Code != http.StatusOK { + t.Fatalf("first create = %d (%s)", first.Code, first.Body.String()) + } + record := e.mustRecord() + before := e.snapshot() + + second := e.createTable(conflictTableBody("second_col", "")) + if second.Code != http.StatusConflict { + t.Fatalf("second create = %d (%s), want 409", second.Code, second.Body.String()) + } + if got := decodeIcebergError(t, second); got.Type != "AlreadyExistsException" || got.Code != http.StatusConflict { + t.Errorf("error = %+v, want AlreadyExistsException/409", got) + } + + assertSnapshotUnchanged(t, before, e.snapshot()) + after := e.mustRecord() + if after.MetadataLocation != record.MetadataLocation || after.VersionToken != record.VersionToken { + t.Errorf("record moved from %s/%s to %s/%s", record.MetadataLocation, record.VersionToken, after.MetadataLocation, after.VersionToken) + } + if got := e.metadataFile(after.MetadataLocation); !hasColumn(got, "first_col") || hasColumn(got, "second_col") { + t.Error("the first table's metadata file was replaced") + } +} + +// Stage-create on a name that is taken is refused at once, rather than after +// the engine has written its data: nothing is staged and no marker is left. +func TestStageCreateOnExistingTableIsConflict(t *testing.T) { + // On by default; pinned so a disabling environment cannot turn the 409 into + // the 501 for a disabled stage-create. + t.Setenv("ICEBERG_ENABLE_STAGE_CREATE", "true") + e := newConflictEnv(t) + if rec := e.createTable(conflictTableBody("first_col", "")); rec.Code != http.StatusOK { + t.Fatalf("create = %d (%s)", rec.Code, rec.Body.String()) + } + before := e.snapshot() + + rec := e.createTable(conflictTableBody("second_col", `"stage-create":true`)) + if rec.Code != http.StatusConflict { + t.Fatalf("stage-create = %d (%s), want 409", rec.Code, rec.Body.String()) + } + if got := decodeIcebergError(t, rec); got.Type != "AlreadyExistsException" { + t.Errorf("error type = %q, want AlreadyExistsException", got.Type) + } + assertSnapshotUnchanged(t, before, e.snapshot()) +} + +// A caller may name a location that already holds a metadata file the catalog +// does not know about, whether a table's leftover or another table's. The file +// stays as it is and the new table gets a name of its own beside it. +func TestCreateTableKeepsUnregisteredMetadataFile(t *testing.T) { + e := newConflictEnv(t) + seed := []byte(`{"not":"this table"}`) + if err := e.server.saveMetadataFile(t.Context(), conflictBucket, path.Join(conflictNamespace, conflictTable), "v1.metadata.json", seed, true); err != nil { + t.Fatal(err) + } + + rec := e.createTable(conflictTableBody("col", `"location":"`+conflictLocation+`"`)) + if rec.Code != http.StatusOK { + t.Fatalf("create = %d (%s)", rec.Code, rec.Body.String()) + } + + got := metadataLocationOf(t, rec) + if !regexp.MustCompile(`^` + regexp.QuoteMeta(conflictLocation) + `/metadata/v1-[0-9a-f-]{36}\.metadata\.json$`).MatchString(got) { + t.Errorf("metadata-location = %s, want %s/metadata/v1-.metadata.json", got, conflictLocation) + } + if kept := e.filer.Get(e.metadataDir(), "v1.metadata.json"); kept == nil || string(kept.Content) != string(seed) { + t.Errorf("the file already at v1.metadata.json was replaced: %+v", kept) + } + record := e.mustRecord() + if record.MetadataLocation != got { + t.Errorf("catalog points at %s, response said %s", record.MetadataLocation, got) + } + if record.MetadataVersion != 1 { + t.Errorf("catalog version = %d, want 1", record.MetadataVersion) + } + if !hasColumn(e.metadataFile(record.MetadataLocation), "col") { + t.Error("the file the catalog points at is not this table's metadata") + } +} + +// Many creators racing for one name: exactly one wins, and the metadata file +// the catalog points at is the winner's own, whichever name each tried first. +// The losers must not leave files behind or take the winner's. +func TestConcurrentCreateTableHasOneWinner(t *testing.T) { + const creators = 8 + e := newConflictEnv(t) + + codes := make([]int, creators) + var ready, done sync.WaitGroup + start := make(chan struct{}) + ready.Add(creators) + done.Add(creators) + for i := range creators { + go func() { + defer done.Done() + ready.Done() + <-start + rec := e.createTable(conflictTableBody(fmt.Sprintf("col_%d", i), `"location":"`+conflictLocation+`"`)) + codes[i] = rec.Code + }() + } + ready.Wait() + close(start) + done.Wait() + + winner := -1 + for i, code := range codes { + switch code { + case http.StatusOK: + if winner != -1 { + t.Fatalf("creators %d and %d both got 200: %v", winner, i, codes) + } + winner = i + case http.StatusConflict: + default: + t.Errorf("creator %d got %d, want 200 or 409: %v", i, code, codes) + } + } + if winner == -1 { + t.Fatalf("no creator won: %v", codes) + } + + record := e.mustRecord() + metadata := e.metadataFile(record.MetadataLocation) + if record.Metadata == nil || record.Metadata.Iceberg == nil { + t.Fatalf("record carries no Iceberg metadata: %+v", record) + } + if got, want := metadata.TableUUID().String(), record.Metadata.Iceberg.TableUUID; got != want { + t.Errorf("file behind %s has table-uuid %s, catalog says %s", record.MetadataLocation, got, want) + } + for i := range creators { + if has := hasColumn(metadata, fmt.Sprintf("col_%d", i)); has != (i == winner) { + t.Errorf("column col_%d present = %v, but creator %d won", i, has, winner) + } + } + if files := e.metadataFiles(); len(files) != 1 || files[0] != path.Base(record.MetadataLocation) { + t.Errorf("metadata files = %v, want only %s", files, path.Base(record.MetadataLocation)) + } +} + +// Two creates overlapping at the canonical name, in a fixed order rather than +// at the scheduler's whim: the second stages while the first has staged +// v1.metadata.json and not yet registered it. Both stage exclusively, the +// second falls back to v1-, and neither file is overwritten. The second +// registers first and wins; the first loses and removes only its own file. +func TestOverlappingCreatesStageApart(t *testing.T) { + e := newConflictEnv(t) + body := func(column string) string { + return conflictTableBody(column, `"location":"`+conflictLocation+`"`) + } + isMetadataFile := func(in *filer_pb.CreateEntryRequest) bool { + return !in.Entry.IsDirectory && strings.HasSuffix(in.Entry.Name, ".metadata.json") + } + columnIn := func(name, column string) bool { + entry := e.filer.Get(e.metadataDir(), name) + if entry == nil { + return false + } + metadata, err := table.ParseMetadataBytes(entry.Content) + return err == nil && hasColumn(metadata, column) + } + + var attempts []*filer_pb.CreateEntryRequest + e.hooks.beforeCreate = func(in *filer_pb.CreateEntryRequest) { + if isMetadataFile(in) { + attempts = append(attempts, in) + } + } + var second *httptest.ResponseRecorder + fallback, started, overlapped := "", false, false + e.hooks.afterCreate = func(in *filer_pb.CreateEntryRequest) { + if !isMetadataFile(in) || overlapped { + return + } + if !started { + // The first create has staged the canonical name; the second runs + // to completion before the first registers. Only once, so a + // staging that stops being exclusive fails here instead of nesting. + started = true + second = e.createTable(body("second_col")) + return + } + // What the second create staged: both files are staged, none registered. + fallback, overlapped = in.Entry.Name, true + if !columnIn("v1.metadata.json", "first_col") { + t.Error("v1.metadata.json does not hold the first create's metadata while both are staged") + } + if !columnIn(fallback, "second_col") { + t.Errorf("%s does not hold the second create's metadata", fallback) + } + } + + first := e.createTable(body("first_col")) + if !overlapped || second == nil { + t.Fatalf("overlapped=%v second ran=%v, want the second create staged inside the first", overlapped, second != nil) + } + + if len(attempts) != 3 { + t.Fatalf("metadata staging attempts = %d, want 3 (first, second at the canonical name, second's fallback)", len(attempts)) + } + for i, want := range []string{"v1.metadata.json", "v1.metadata.json", fallback} { + if got := attempts[i]; got.Entry.Name != want || !got.OExcl { + t.Errorf("attempt %d = %s exclusive=%v, want %s exclusive", i, got.Entry.Name, got.OExcl, want) + } + } + if !regexp.MustCompile(`^v1-[0-9a-f-]{36}\.metadata\.json$`).MatchString(fallback) { + t.Errorf("fallback name = %s, want v1-.metadata.json", fallback) + } + + if second.Code != http.StatusOK { + t.Fatalf("second create = %d (%s), want 200", second.Code, second.Body.String()) + } + if first.Code != http.StatusConflict { + t.Fatalf("first create = %d (%s), want 409", first.Code, first.Body.String()) + } + if record := e.mustRecord(); path.Base(record.MetadataLocation) != fallback { + t.Errorf("catalog points at %s, want the winner's %s", record.MetadataLocation, fallback) + } + if files := e.metadataFiles(); len(files) != 1 || files[0] != fallback { + t.Errorf("metadata files = %v, want only the winner's %s", files, fallback) + } + if !columnIn(fallback, "second_col") || columnIn(fallback, "first_col") { + t.Errorf("%s no longer holds only the winner's metadata", fallback) + } +} + +// A creator whose request reaches the catalog after another's has registered +// the name is answered with that table by the catalog, not with an error. It +// is still the loser: 409, and the file it staged goes, not the winner's. The +// S3 Tables API registers a table without a metadata location, and the +// catalog then answers with an empty one. +func TestCreateTableDiscardsFileWhenNameWasTakenMidFlight(t *testing.T) { + for name, winnerLocation := range map[string]string{ + "winner has a location": conflictLocation + "/metadata/v1-winner.metadata.json", + "winner has none": "", + } { + t.Run(name, func(t *testing.T) { + e := newConflictEnv(t) + winnerUUID := uuid.NewString() + registered := false + e.hooks.afterCreate = func(in *filer_pb.CreateEntryRequest) { + if registered || in.Entry.IsDirectory || !strings.HasSuffix(in.Entry.Name, ".metadata.json") { + return + } + registered = true + e.registerWinner(winnerLocation, winnerUUID) + } + + rec := e.createTable(conflictTableBody("col", `"location":"`+conflictLocation+`"`)) + if !registered { + t.Fatal("the competing registration never ran") + } + if rec.Code != http.StatusConflict { + t.Fatalf("create = %d (%s), want 409", rec.Code, rec.Body.String()) + } + if files := e.metadataFiles(); len(files) != 0 { + t.Errorf("the losing request left %v behind", files) + } + e.assertWinnerRecord(winnerLocation, winnerUUID) + }) + } +} + +// A losing create's cleanup removes its file by path, and the path can be +// reused before the cleanup runs: the table is dropped, its directory purged, +// and another create stages the same v1.metadata.json and registers it. The +// cleanup must leave that replacement alone. +func TestLosingCreateKeepsReplacementAtReusedPath(t *testing.T) { + for name, create := range map[string]func(t *testing.T, e *conflictEnv){ + "create table": func(t *testing.T, e *conflictEnv) { + if rec := e.createTable(conflictTableBody("col", `"location":"`+conflictLocation+`"`)); rec.Code != http.StatusConflict { + t.Fatalf("create = %d (%s), want 409", rec.Code, rec.Body.String()) + } + }, + "create on commit": func(t *testing.T, e *conflictEnv) { + if _, reqErr := e.server.finalizeCreateOnCommit(t.Context(), e.finalizeInput(uuid.New())); reqErr == nil || reqErr.status != http.StatusConflict { + t.Fatalf("finalize = %+v, want a 409", reqErr) + } + }, + } { + t.Run(name, func(t *testing.T) { + e := newConflictEnv(t) + replacement := []byte(`{"replacement":true}`) + registered, replaced := false, false + e.hooks.afterCreate = func(in *filer_pb.CreateEntryRequest) { + if registered || in.Entry.IsDirectory || in.Entry.Name != "v1.metadata.json" { + return + } + registered = true + e.registerWinner(conflictLocation+"/metadata/v1-winner.metadata.json", uuid.NewString()) + e.filer.BeforeDelete = func() { + replaced = true + // Stands in for drop, purge and a new create staging the name. + if err := e.server.saveMetadataFile(t.Context(), conflictBucket, path.Join(conflictNamespace, conflictTable), "v1.metadata.json", replacement, false); err != nil { + t.Errorf("replace staged file: %v", err) + } + } + } + + create(t, e) + if !registered || !replaced { + t.Fatalf("registered=%v replaced=%v, want the race to have run", registered, replaced) + } + entry := e.filer.Get(e.metadataDir(), "v1.metadata.json") + if entry == nil { + t.Fatal("the losing request deleted the replacement at its old path") + } + if !bytes.Equal(entry.Content, replacement) { + t.Errorf("v1.metadata.json = %s, want the replacement", entry.Content) + } + }) + } +} + +// registerWinner registers the shared test table through the S3 Tables API, +// the way a competing creator would. +func (e *conflictEnv) registerWinner(metadataLocation, tableUUID string) { + e.t.Helper() + e.adminDo("CreateTable", &s3tables.CreateTableRequest{ + TableBucketARN: e.arn, Namespace: []string{conflictNamespace}, Name: conflictTable, Format: s3tables.FormatIceberg, + MetadataLocation: metadataLocation, MetadataVersion: 1, + Metadata: &s3tables.TableMetadata{Iceberg: &s3tables.IcebergMetadata{TableUUID: tableUUID}}, + }, nil) +} + +func (e *conflictEnv) assertWinnerRecord(metadataLocation, tableUUID string) { + e.t.Helper() + record := e.mustRecord() + if record.MetadataLocation != metadataLocation { + e.t.Errorf("catalog points at %q, want the winner's %q", record.MetadataLocation, metadataLocation) + } + if record.Metadata == nil || record.Metadata.Iceberg == nil || record.Metadata.Iceberg.TableUUID != tableUUID { + e.t.Errorf("catalog record %+v, want the winner's table-uuid %s", record.Metadata, tableUUID) + } +} + +// The staged file goes when the catalog refuses the registration outright, and +// stays when the outcome is unknown: the record may have been written, and a +// catalog pointing at a deleted file cannot be read. +func TestCreateTableStagedFileOnRegistrationFailure(t *testing.T) { + t.Run("transport failure keeps the file", func(t *testing.T) { + e := newConflictEnv(t) + e.hooks.updateErr = func(in *filer_pb.UpdateEntryRequest) error { + if in.Entry.Name == conflictTable { + return status.Error(codes.Unavailable, "filer unavailable") + } + return nil + } + + rec := e.createTable(conflictTableBody("col", `"location":"`+conflictLocation+`"`)) + if rec.Code != http.StatusInternalServerError { + t.Fatalf("create = %d (%s), want 500", rec.Code, rec.Body.String()) + } + if files := e.metadataFiles(); len(files) != 1 { + t.Errorf("metadata files = %v, want the staged file kept", files) + } + }) + + t.Run("refusal removes the file", func(t *testing.T) { + e := newConflictEnv(t) + revoked := false + e.hooks.afterCreate = func(in *filer_pb.CreateEntryRequest) { + if revoked || in.Entry.IsDirectory || !strings.HasSuffix(in.Entry.Name, ".metadata.json") { + return + } + revoked = true + // The namespace is gone by the time the catalog registers the table. + e.adminDo("DeleteNamespace", &s3tables.DeleteNamespaceRequest{TableBucketARN: e.arn, Namespace: []string{conflictNamespace}}, nil) + } + + rec := e.createTable(conflictTableBody("col", `"location":"`+conflictLocation+`"`)) + if !revoked { + t.Fatal("the namespace was never removed") + } + if rec.Code < 400 || rec.Code >= 500 { + t.Fatalf("create = %d (%s), want a 4xx", rec.Code, rec.Body.String()) + } + if files := e.metadataFiles(); len(files) != 0 { + t.Errorf("metadata files = %v, want the refused file removed", files) + } + }) +} + +const renamedTable = "renamed" + +// armSurvivingConflict makes the next create of the test table end in a +// conflict that leaves its record in place: once the catalog confirms the +// record, compete changes it without moving its metadata location, and the +// namespace recheck that follows fails. The catalog then cannot back the +// changed record out. It reports whether compete ran. +func (e *conflictEnv) armSurvivingConflict(compete func(e *conflictEnv)) *bool { + namespacePath := s3tables.GetNamespacePath(conflictBucket, conflictNamespace) + updated, failRecheck := false, false + e.hooks.afterUpdate = func(in *filer_pb.UpdateEntryRequest) { + if _, marked := in.Entry.Extended[s3tables.ExtendedKeyDeleting]; updated || marked || in.Entry.Name != conflictTable { + return + } + updated = true + compete(e) + failRecheck = true + } + e.hooks.lookupErr = func(in *filer_pb.LookupDirectoryEntryRequest) error { + if failRecheck && path.Join(in.Directory, in.Name) == namespacePath { + failRecheck = false + return status.Error(codes.Unavailable, "filer unavailable") + } + return nil + } + return &updated +} + +// A conflict does not always mean the create left nothing behind: the record +// it could not back out still points at the staged file, so the file stays — +// also when the record was renamed away and the original name holds nothing. +func TestCreateKeepsFileOfRecordThatSurvivedConflict(t *testing.T) { + creates := map[string]func(t *testing.T, e *conflictEnv) (int, string){ + "create table": func(t *testing.T, e *conflictEnv) (int, string) { + rec := e.createTable(conflictTableBody("col", `"location":"`+conflictLocation+`"`)) + if rec.Code == http.StatusOK { + return rec.Code, "" + } + return rec.Code, decodeIcebergError(t, rec).Type + }, + "create on commit": func(t *testing.T, e *conflictEnv) (int, string) { + _, reqErr := e.server.finalizeCreateOnCommit(t.Context(), e.finalizeInput(uuid.New())) + if reqErr == nil { + return http.StatusOK, "" + } + return reqErr.status, reqErr.errType + }, + } + competitors := map[string]struct { + compete func(e *conflictEnv) + name string + }{ + "updated": {func(e *conflictEnv) { + e.adminDo("UpdateTable", &s3tables.UpdateTableRequest{ + TableBucketARN: e.arn, Namespace: []string{conflictNamespace}, Name: conflictTable, + }, nil) + }, conflictTable}, + "renamed": {func(e *conflictEnv) { + e.adminDo("RenameTable", &s3tables.RenameTableRequest{ + TableBucketARN: e.arn, + SourceNamespace: []string{conflictNamespace}, SourceName: conflictTable, + DestNamespace: []string{conflictNamespace}, DestName: renamedTable, + }, nil) + }, renamedTable}, + } + for createName, create := range creates { + for competeName, competitor := range competitors { + t.Run(createName+"/"+competeName, func(t *testing.T) { + e := newConflictEnv(t) + ran := e.armSurvivingConflict(competitor.compete) + + code, errType := create(t, e) + if !*ran { + t.Fatal("the competing write never ran") + } + if code != http.StatusConflict || errType != "CommitFailedException" { + t.Fatalf("create = %d %s, want the catalog's 409 CommitFailedException", code, errType) + } + var record s3tables.GetTableResponse + e.adminDo("GetTable", &s3tables.GetTableRequest{ + TableBucketARN: e.arn, Namespace: []string{conflictNamespace}, Name: competitor.name, + }, &record) + if want := conflictLocation + "/metadata/v1.metadata.json"; record.MetadataLocation != want { + t.Fatalf("surviving record points at %s, want the staged %s", record.MetadataLocation, want) + } + if files := e.metadataFiles(); len(files) != 1 || files[0] != path.Base(record.MetadataLocation) { + t.Errorf("metadata files = %v, want the surviving record's %s kept", files, path.Base(record.MetadataLocation)) + } + }) + } + } +} + +// A create's record stays hidden until the mark comes off. When taking the +// mark off fails and another create adopts the still-marked record, the first +// create's file is unreferenced only if its write never went out: a lookup +// failing before the write leaves the mark on, so the file goes. A write that +// went out and failed without a refusal may have published the record first, +// so the file stays. +func TestCreateFailingToUnmarkRecordAnotherAdopted(t *testing.T) { + const winnerLocation = conflictLocation + "/metadata/v1-winner.metadata.json" + tablePath := s3tables.GetTablePath(conflictBucket, conflictNamespace, conflictTable) + for name, tc := range map[string]struct { + // arm fails the first create's mark removal, running adopt first. + arm func(e *conflictEnv, adopt func()) + keepFile bool + }{ + "lookup fails before the write": {func(e *conflictEnv, adopt func()) { + e.hooks.lookupErr = func(in *filer_pb.LookupDirectoryEntryRequest) error { + if path.Join(in.Directory, in.Name) != tablePath { + return nil + } + e.hooks.lookupErr = nil + adopt() + return status.Error(codes.Unavailable, "filer unavailable") + } + }, false}, + "write fails without a refusal": {func(e *conflictEnv, adopt func()) { + e.hooks.updateErr = func(in *filer_pb.UpdateEntryRequest) error { + if _, marked := in.Entry.Extended[s3tables.ExtendedKeyDeleting]; marked || in.Entry.Name != conflictTable { + return nil + } + e.hooks.updateErr = nil + adopt() + return status.Error(codes.Unavailable, "filer unavailable") + } + }, true}, + } { + t.Run(name, func(t *testing.T) { + e := newConflictEnv(t) + winnerUUID := uuid.NewString() + adopted := false + adopt := func() { + adopted = true + e.registerWinner(winnerLocation, winnerUUID) + } + e.hooks.afterUpdate = func(in *filer_pb.UpdateEntryRequest) { + // The first create's claim has landed: its record is marked. + if in.Entry.Name == conflictTable && string(in.Entry.Extended[s3tables.ExtendedKeyDeleting]) == "create" { + e.hooks.afterUpdate = nil + tc.arm(e, adopt) + } + } + + rec := e.createTable(conflictTableBody("col", `"location":"`+conflictLocation+`"`)) + if !adopted { + t.Fatal("the competing create never adopted the marked record") + } + if rec.Code != http.StatusConflict { + t.Fatalf("create = %d (%s), want 409", rec.Code, rec.Body.String()) + } + e.assertWinnerRecord(winnerLocation, winnerUUID) + files := e.metadataFiles() + if tc.keepFile && len(files) != 1 { + t.Errorf("metadata files = %v, want the staged file kept", files) + } + if !tc.keepFile && len(files) != 0 { + t.Errorf("metadata files = %v, want the unreferenced staged file removed", files) + } + }) + } +} + +func TestRegistrationRefused(t *testing.T) { + typed := func(code string) error { return &s3tables.S3TablesError{Type: code, Message: code} } + for name, tc := range map[string]struct { + err error + want bool + }{ + "name taken": {typed(s3tables.ErrCodeTableAlreadyExists), true}, + "conflict": {typed(s3tables.ErrCodeConflict), true}, + "version mismatch": {s3tables.ErrVersionTokenMismatch, true}, + "published then changed": {&s3tables.S3TablesError{Type: s3tables.ErrCodeConflict, Message: s3tables.PublishedTableChangedMessage}, false}, + "wrapped published then changed": {fmt.Errorf("create: %w", s3tables.ErrPublishedRecordChanged), false}, + "access denied": {typed(s3tables.ErrCodeAccessDenied), true}, + "no such namespace": {typed(s3tables.ErrCodeNoSuchNamespace), true}, + "invalid request": {typed(s3tables.ErrCodeInvalidRequest), true}, + "internal error": {typed(s3tables.ErrCodeInternalError), false}, + "transport error": {errors.New("rpc error: code = Unavailable desc = filer unavailable"), false}, + "wrapped name taken": {fmt.Errorf("create: %w", typed(s3tables.ErrCodeTableAlreadyExists)), true}, + "untyped name taken": {errors.New("table orders already exists"), true}, + "wrapped server fault": {fmt.Errorf("create: %w", typed(s3tables.ErrCodeInternalError)), false}, + "server fault naming an existing entry": {&s3tables.S3TablesError{Type: s3tables.ErrCodeInternalError, Message: "failed to check table: entry already exists"}, false}, + "wrapped server fault naming an entry": {fmt.Errorf("table already exists: %w", typed(s3tables.ErrCodeInternalError)), false}, + } { + t.Run(name, func(t *testing.T) { + if got := registrationRefused(tc.err); got != tc.want { + t.Errorf("registrationRefused(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +// finalizeInput is a create-on-commit for the shared test table, as one of +// several writers racing to create it would issue. +func (e *conflictEnv) finalizeInput(tableUUID uuid.UUID) createOnCommitInput { + e.t.Helper() + base, err := newTableMetadata(tableUUID, conflictLocation, nil, nil, nil, nil) + if err != nil { + e.t.Fatal(err) + } + return createOnCommitInput{ + bucketARN: e.arn, + markerBucket: conflictBucket, + namespace: []string{conflictNamespace}, + tableName: conflictTable, + identityName: s3tables.DefaultAccountID, + location: conflictLocation, + tableUUID: tableUUID, + baseMetadata: base, + } +} + +// A create-on-commit that loses the race must remove its own metadata file, +// not the v1.metadata.json the winner's catalog record points at. +func TestFinalizeCreateOnCommitLosesToExistingTable(t *testing.T) { + e := newConflictEnv(t) + if rec := e.createTable(conflictTableBody("winner_col", `"location":"`+conflictLocation+`"`)); rec.Code != http.StatusOK { + t.Fatalf("create = %d (%s)", rec.Code, rec.Body.String()) + } + record := e.mustRecord() + if base := path.Base(record.MetadataLocation); base != "v1.metadata.json" { + t.Fatalf("winner's file = %s, want v1.metadata.json", base) + } + before := e.snapshot() + + _, reqErr := e.server.finalizeCreateOnCommit(t.Context(), e.finalizeInput(uuid.New())) + if reqErr == nil || reqErr.status != http.StatusConflict || reqErr.errType != "CommitFailedException" { + t.Fatalf("finalize = %+v, want 409 CommitFailedException", reqErr) + } + + assertSnapshotUnchanged(t, before, e.snapshot()) + if after := e.mustRecord(); after.MetadataLocation != record.MetadataLocation { + t.Errorf("catalog moved from %s to %s", record.MetadataLocation, after.MetadataLocation) + } +} + +// The winner may be a table registered through the S3 Tables API without a +// metadata location, which the catalog answers with an empty one. The commit +// still lost: it must not report the table as created by its own metadata. +func TestFinalizeCreateOnCommitLosesToTableWithoutLocation(t *testing.T) { + e := newConflictEnv(t) + winnerUUID := uuid.NewString() + e.registerWinner("", winnerUUID) + + _, reqErr := e.server.finalizeCreateOnCommit(t.Context(), e.finalizeInput(uuid.New())) + if reqErr == nil || reqErr.status != http.StatusConflict || reqErr.errType != "CommitFailedException" { + t.Fatalf("finalize = %+v, want 409 CommitFailedException", reqErr) + } + if files := e.metadataFiles(); len(files) != 0 { + t.Errorf("the losing commit left %v behind", files) + } + e.assertWinnerRecord("", winnerUUID) +} + +// 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) { + const creators = 8 + e := newConflictEnv(t) + + uuids := make([]uuid.UUID, creators) + results := make([]*icebergRequestError, creators) + var ready, done sync.WaitGroup + start := make(chan struct{}) + ready.Add(creators) + done.Add(creators) + for i := range creators { + uuids[i] = uuid.New() + go func() { + defer done.Done() + ready.Done() + <-start + _, results[i] = e.server.finalizeCreateOnCommit(t.Context(), e.finalizeInput(uuids[i])) + }() + } + ready.Wait() + close(start) + done.Wait() + + winner := -1 + for i, reqErr := range results { + switch { + case reqErr == nil: + if winner != -1 { + t.Fatalf("creators %d and %d both created the table", winner, i) + } + winner = i + case reqErr.status != http.StatusConflict || reqErr.errType != "CommitFailedException": + t.Errorf("creator %d: %+v, want 409 CommitFailedException", i, reqErr) + } + } + if winner == -1 { + t.Fatal("no creator won") + } + + record := e.mustRecord() + if got := e.metadataFile(record.MetadataLocation).TableUUID(); got != uuids[winner] { + t.Errorf("file behind %s has table-uuid %s, want the winner's %s", record.MetadataLocation, got, uuids[winner]) + } + if got := record.Metadata.Iceberg.TableUUID; got != uuids[winner].String() { + t.Errorf("catalog table-uuid = %s, want the winner's %s", got, uuids[winner]) + } + if files := e.metadataFiles(); len(files) != 1 || files[0] != path.Base(record.MetadataLocation) { + t.Errorf("metadata files = %v, want only %s", files, path.Base(record.MetadataLocation)) + } +} + +// A create answers 409 AlreadyExists only for a typed name-taken code; a typed +// server fault keeps its 500 whatever its message says. +func TestIsS3TablesAlreadyExistsTrustsTypedCode(t *testing.T) { + for name, tc := range map[string]struct { + err error + want bool + }{ + "table taken": {&s3tables.S3TablesError{Type: s3tables.ErrCodeTableAlreadyExists, Message: "table t already exists"}, true}, + "namespace taken": {&s3tables.S3TablesError{Type: s3tables.ErrCodeNamespaceAlreadyExists, Message: "ns"}, true}, + "internal error mentions it": {&s3tables.S3TablesError{Type: s3tables.ErrCodeInternalError, Message: "failed to check table: entry already exists"}, false}, + "conflict mentions it": {&s3tables.S3TablesError{Type: s3tables.ErrCodeConflict, Message: "already exists"}, false}, + "untyped": {errors.New("table orders already exists"), true}, + "untyped other": {errors.New("filer unavailable"), false}, + "nil": {nil, false}, + } { + t.Run(name, func(t *testing.T) { + if got := isS3TablesAlreadyExists(tc.err); got != tc.want { + t.Errorf("isS3TablesAlreadyExists(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} diff --git a/weed/s3api/iceberg/manifest_repair.go b/weed/s3api/iceberg/manifest_repair.go index e1326d9d0d..d2a6e8141c 100644 --- a/weed/s3api/iceberg/manifest_repair.go +++ b/weed/s3api/iceberg/manifest_repair.go @@ -556,7 +556,7 @@ func (st *serverManifestStore) saveFile(ctx context.Context, location string, da if err != nil { return err } - return st.server.saveMetadataBlob(ctx, bucket, tablePath, fileName, data, "application/avro", false) + return st.server.saveMetadataBlob(ctx, bucket, tablePath, fileName, data, "application/avro", false, nil) } // repairAddSnapshotManifests is the server entry point used by the commit diff --git a/weed/s3api/iceberg/metadata_files.go b/weed/s3api/iceberg/metadata_files.go index 64e4a21484..635d36daf6 100644 --- a/weed/s3api/iceberg/metadata_files.go +++ b/weed/s3api/iceberg/metadata_files.go @@ -10,9 +10,18 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) +// stageTokenKey is the extended attribute stageOwnedMetadata tags a staged +// metadata file with: a value unique to the request that wrote it, which the +// request's own cleanup must find before it deletes the file. The internal +// prefix keeps it out of the headers S3 reads return. +const stageTokenKey = s3_constants.SeaweedFSInternalPrefix + "iceberg-stage-token" + // metadataDirPath returns the filer directory holding a table's metadata files. func metadataDirPath(bucketName, tablePath string) string { dir := path.Join(s3tables.TablesPath, bucketName) @@ -29,11 +38,16 @@ func metadataDirPath(bucketName, tablePath string) string { // same v{N} name, and letting the second overwrite the first would leave the // winner's catalog pointer aimed at the loser's metadata. func (s *Server) saveMetadataFile(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte, exclusive bool) error { - return s.saveMetadataBlob(ctx, bucketName, tablePath, metadataFileName, content, "application/json", exclusive) + return s.saveMetadataBlob(ctx, bucketName, tablePath, metadataFileName, content, "application/json", exclusive, nil) } -// saveMetadataBlob saves a file into the table's metadata directory. -func (s *Server) saveMetadataBlob(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte, mimeType string, exclusive bool) error { +// saveMetadataBlob saves a file into the table's metadata directory, with +// extended stored on the entry alongside its mime type. +func (s *Server) saveMetadataBlob(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte, mimeType string, exclusive bool, extended map[string][]byte) error { + attrs := map[string][]byte{"Mime-Type": []byte(mimeType)} + for key, value := range extended { + attrs[key] = value + } // Create context with timeout for file operations opCtx, cancel := context.WithTimeout(ctx, 30*time.Second) @@ -115,10 +129,8 @@ func (s *Server) saveMetadataBlob(ctx context.Context, bucketName, tablePath, me FileMode: uint32(0644), FileSize: uint64(len(content)), }, - Content: content, - Extended: map[string][]byte{ - "Mime-Type": []byte(mimeType), - }, + Content: content, + Extended: attrs, }, }) if err != nil { @@ -144,6 +156,39 @@ func (s *Server) deleteMetadataFile(ctx context.Context, bucketName, tablePath, }) } +// errStagedFileReplaced reports that the entry at a staged file's path no +// longer carries the token it was staged with: it is another writer's file. +var errStagedFileReplaced = errors.New("staged metadata file was replaced") + +// deleteStagedMetadataFile removes a file stageOwnedMetadata wrote, only while +// the entry at its path still carries token. The filer checks the token under +// the path lock the delete runs under, so a file that took the path since — +// the table was dropped and another create staged the same name — is refused +// with errStagedFileReplaced instead of deleted. A path that holds nothing is +// not an error. +func (s *Server) deleteStagedMetadataFile(ctx context.Context, bucketName, tablePath, metadataFileName, token string) error { + if token == "" { + return fmt.Errorf("%w: %s has no stage token", errStagedFileReplaced, metadataFileName) + } + opCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + metadataDir := metadataDirPath(bucketName, tablePath) + return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + _, err := filer_pb.DoRemoveRequest(opCtx, client, &filer_pb.DeleteEntryRequest{ + Directory: metadataDir, + Name: metadataFileName, + IsDeleteData: true, + IgnoreRecursiveError: true, + ExpectedExtended: map[string][]byte{stageTokenKey: []byte(token)}, + }) + if status.Code(err) == codes.FailedPrecondition { + return fmt.Errorf("%w: %s/%s: %v", errStagedFileReplaced, metadataDir, metadataFileName, err) + } + return err + }) +} + func (s *Server) loadMetadataFile(ctx context.Context, bucketName, tablePath, metadataFileName string) ([]byte, error) { opCtx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index cd7159e68f..17252ee6c8 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -17,6 +17,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -26,6 +27,11 @@ var ( ErrConcurrentUpdate = errors.New("attribute changed concurrently") ErrNameTaken = errors.New("catalog name is taken") ErrLocationInUse = errors.New("location is in use by another table or view") + // ErrPublishedRecordChanged is the ErrConcurrentUpdate a create returns + // when the record it wrote was already visible, and another caller changed + // it — updated it, or renamed it away — before the create could back it + // out. Whatever the create staged for that record may still be in use. + ErrPublishedRecordChanged = fmt.Errorf("published record changed before the create could back it out: %w", ErrConcurrentUpdate) ) // extendedAttributeUpdateAttempts bounds the read-modify-write retry loop in @@ -799,11 +805,18 @@ func (h *S3TablesHandler) confirmParentOnce(ctx context.Context, client filer_pb func (h *S3TablesHandler) confirmChild(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, recheck bool, childPath string, written map[string][]byte, emptyDirs []string) error { parent, err := h.openParent(ctx, client, parentPath, isParent) - confirmed := false + // mayBeVisible is whether the mark may have come off: it did, or a write + // taking it off was sent and failed without the filer saying it refused + // it. A failure before any such write — the lookup that precedes it, say — + // leaves the mark on, and the record as invisible as it was. + confirmed, mayBeVisible := false, false if err == nil { // The mark comes off only while the record still holds what this // create wrote; a failed write backs the record out like a closed parent. - if err = h.removeExtendedAttributesIf(ctx, client, childPath, written, ExtendedKeyDeleting); err == nil { + writes := &markRemovalWrites{SeaweedFilerClient: client} + err = h.removeExtendedAttributesIf(ctx, writes, childPath, written, ExtendedKeyDeleting) + mayBeVisible = err == nil || writes.uncertain + if err == nil { confirmed = true if recheck { err = h.parentStillOpen(ctx, client, parentPath, isParent, parent) @@ -813,7 +826,7 @@ func (h *S3TablesHandler) confirmChild(ctx context.Context, client filer_pb.Seaw if err == nil { return nil } - err = h.backOutChild(ctx, client, childPath, written, confirmed, emptyDirs, err) + err = h.backOutChild(ctx, client, childPath, written, confirmed, mayBeVisible, emptyDirs, err) if errors.Is(err, errParentGone) { h.removeLeftoverDirectory(context.WithoutCancel(ctx), client, parentPath) } @@ -821,8 +834,11 @@ func (h *S3TablesHandler) confirmChild(ctx context.Context, client filer_pb.Seaw } // backOutChild removes what a create wrote onto childPath, after the create -// failed with err, and returns err, or why the record stays. -func (h *S3TablesHandler) backOutChild(ctx context.Context, client filer_pb.SeaweedFilerClient, childPath string, written map[string][]byte, confirmed bool, emptyDirs []string, err error) error { +// failed with err, and returns err, or why the record stays. mayBeVisible is +// whether readers may have seen the record: a record that moved on is then +// reported as ErrPublishedRecordChanged, since what was staged for it may +// still be in use. +func (h *S3TablesHandler) backOutChild(ctx context.Context, client filer_pb.SeaweedFilerClient, childPath string, written map[string][]byte, confirmed, mayBeVisible bool, emptyDirs []string, err error) error { ctx = context.WithoutCancel(ctx) expected := written keys := make([]string, 0, len(written)) @@ -845,7 +861,12 @@ func (h *S3TablesHandler) backOutChild(ctx context.Context, client filer_pb.Seaw glog.Warningf("back out %s: %v", childPath, backOutErr) if errors.Is(backOutErr, ErrConcurrentUpdate) { // The record moved past this create — adopted or rewritten — so - // there is nothing of its own left to remove. + // there is nothing of its own left to remove. While it was still + // marked, only another create could take it over; once readers + // could see it, an update or a rename may have built on it. + if mayBeVisible { + return fmt.Errorf("%w: %s", ErrPublishedRecordChanged, childPath) + } return fmt.Errorf("%w: %s changed during create", ErrConcurrentUpdate, childPath) } return fmt.Errorf("%w: %s: %v (%v)", errBackOutFailed, childPath, backOutErr, err) @@ -856,6 +877,24 @@ func (h *S3TablesHandler) backOutChild(ctx context.Context, client filer_pb.Seaw return err } +// markRemovalWrites passes a create's mark-removal calls through to the filer +// and records whether an UpdateEntry was sent whose outcome is not a refusal: +// it succeeded, or it failed in a way that does not say whether it landed. A +// compare the filer refused (FailedPrecondition) or an entry it no longer has +// (NotFound) changed nothing. +type markRemovalWrites struct { + filer_pb.SeaweedFilerClient + uncertain bool +} + +func (w *markRemovalWrites) UpdateEntry(ctx context.Context, in *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + resp, err := w.SeaweedFilerClient.UpdateEntry(ctx, in, opts...) + if code := status.Code(err); code != codes.FailedPrecondition && code != codes.NotFound { + w.uncertain = true + } + return resp, err +} + // dirListPageSize is how many entries dirHoldsFiles reads per ListEntries call. const dirListPageSize = 1024 @@ -1055,7 +1094,8 @@ func (h *S3TablesHandler) claimCatalogEntryAt(ctx context.Context, client filer_ return err } if err := checkLocationUnclaimed(ctx, client, bucketName, path, metadataLocation); err != nil { - return h.backOutChild(ctx, client, path, attrs, false, emptyDirs, err) + // The claim still carries its mark, so no reader has seen the record. + return h.backOutChild(ctx, client, path, attrs, false, false, emptyDirs, err) } return nil } diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 77c0fdbe28..a19d9cd8f5 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -178,6 +178,10 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("table %s already exists", tableName)) return err } + if errors.Is(err, ErrPublishedRecordChanged) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, PublishedTableChangedMessage) + return err + } if errors.Is(err, ErrConcurrentUpdate) { h.writeError(w, http.StatusConflict, ErrCodeConflict, "table changed during create, retry the request") return err diff --git a/weed/s3api/s3tables/table_data_dir_test.go b/weed/s3api/s3tables/table_data_dir_test.go index c702f14e30..16f956965d 100644 --- a/weed/s3api/s3tables/table_data_dir_test.go +++ b/weed/s3api/s3tables/table_data_dir_test.go @@ -24,6 +24,7 @@ func TestTableDataDirFromMetadataLocation(t *testing.T) { func TestMetadataVersionFromLocationUniqueSuffix(t *testing.T) { cases := map[string]int{ + "s3://bkt/ns/t/metadata/v1-0a1b2c3d-4e5f-6789-abcd-ef0123456789.metadata.json": 1, "s3://bkt/ns/t/metadata/v4.metadata.json": 4, "s3://bkt/ns/t/metadata/v4-0a1b2c3d-4e5f-6789-abcd-ef0123456789.metadata.json": 4, "s3://bkt/ns/t/metadata/00007-0a1b2c3d.metadata.json": 7, diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index 6d8808ce43..ede7cf6a9c 100644 --- a/weed/s3api/s3tables/types.go +++ b/weed/s3api/s3tables/types.go @@ -2,6 +2,7 @@ package s3tables import ( "encoding/json" + "errors" "strings" "time" ) @@ -589,6 +590,23 @@ func (e *S3TablesError) Error() string { return e.Message } +// PublishedTableChangedMessage is the message of the conflict a CreateTable +// answers when the table it registered was already visible, and another +// caller updated or renamed it before the create could back it out. The +// record survives, under this name or another, so whatever the caller staged +// for it may still be in use. +const PublishedTableChangedMessage = "table was registered and then changed by another request before the create finished" + +// IsPublishedTableChanged reports whether err, as a CreateTable caller +// receives it, is the conflict PublishedTableChangedMessage describes. +func IsPublishedTableChanged(err error) bool { + if errors.Is(err, ErrPublishedRecordChanged) { + return true + } + var tableErr *S3TablesError + return errors.As(err, &tableErr) && tableErr.Type == ErrCodeConflict && tableErr.Message == PublishedTableChangedMessage +} + // Table formats a catalog entry may declare. // // ICEBERG tables carry metadata the catalog maintains and the maintenance