From 195f8156de5bdfd1a71d88aeaac5eb01dbc26345 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nguy=E1=BB=85n=20L=E1=BB=99c=20Ph=C3=BAc?= <112325531+omninguyen@users.noreply.github.com> Date: Sun, 8 Nov 2026 01:54:04 +0700 Subject: [PATCH] fix(iceberg): keep a losing table create from deleting the winner's metadata (#12474) * iceberg: keep a losing table create from deleting the winner's metadata * iceberg: keep a staged file whose published record was renamed away A create answers a conflict when its record was already visible and another request changed it before the create could back it out. The cleanup then looked only at the original name, so a record renamed in between read as absent and the file it still points at was deleted. S3 Tables now reports that case with a conflict of its own, and both create paths keep the staged file on it instead of probing the name. A conflict over a still-hidden record, lost to another create, still removes the file. * iceberg: discard a losing create's metadata only while it is still its own The cleanup deleted the staged file by name. The path can be reused before it runs: the table is dropped and purged, and another create stages the same v1.metadata.json and registers it, which the losing request then deleted. Staged files now carry a per-request token in an internal extended attribute, and the cleanup deletes with that token as the filer's ExpectedExtended precondition, checked under the path lock the delete runs under. A file that took the path since is left in place. * iceberg: classify a typed S3 Tables error by its code, not its message isS3TablesAlreadyExists matched "already exists" in the message before looking at the code, so an InternalError whose text mentioned an existing entry read as the name being taken: CreateTable answered 409 instead of 500, and registrationRefused discarded a staged file the catalog may still point at. A typed error now decides by its code; only an untyped one is matched by message. * iceberg: pin stage-create on in the stage-create conflict test Stage-create is on by default, but a shell that sets ICEBERG_ENABLE_STAGE_CREATE=false turned the expected 409 into a 501. Setting it in the test keeps the result independent of the environment. * iceberg: test two creates overlapping at the canonical metadata name The concurrent-creators test only lines the creators up before they start, so whether two of them ever collide at v1.metadata.json is up to the scheduler. The new test fixes the order: the second create runs while the first has staged v1.metadata.json and not yet registered it. It checks that every staging attempt is exclusive, that the second falls back to v1-, that both files hold their own metadata while both are staged, and that the loser removes only its own file. * s3tables: a create whose unmark write never went out stayed hidden confirmChild counted any failure to take a create's mark off as possibly published, the lookup that precedes the write included. That lookup failing leaves the mark on, so the record stayed hidden; if another create then adopted it, the back-out reported a published record that changed, and the Iceberg catalog kept a staged file no record names. The mark-removal calls now go through a wrapper that notes whether an UpdateEntry went out without the filer refusing it. Only then may the record have been seen. A refused compare, a missing entry, or a failure before any write leaves the create's record hidden. * s3tables: pass mayBeVisible into backOutChild Fixes the build after rebasing onto the backOutChild split. --- weed/s3api/iceberg/commit_helpers.go | 102 +- weed/s3api/iceberg/handlers_table.go | 108 +- .../iceberg_create_table_conflict_test.go | 1013 +++++++++++++++++ weed/s3api/iceberg/manifest_repair.go | 2 +- weed/s3api/iceberg/metadata_files.go | 59 +- weed/s3api/s3tables/filer_ops.go | 54 +- weed/s3api/s3tables/handler_table.go | 4 + weed/s3api/s3tables/table_data_dir_test.go | 1 + weed/s3api/s3tables/types.go | 18 + 9 files changed, 1261 insertions(+), 100 deletions(-) create mode 100644 weed/s3api/iceberg/iceberg_create_table_conflict_test.go 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