From 5b1450b06b7394989c04a6310a7eab0e0e64774c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 7 Nov 2026 05:04:09 -0800 Subject: [PATCH] s3tables, iceberg: one catalog record per table location (#12473) * s3tables: refuse a table or view location another record holds A create, register, update or rename could name a location at, under or above another catalog record's files, including the files a renamed table left under its old name. The new record then loaded that table's metadata, had credentials vended for its files, and could purge them on drop. Each of those writes now scans the bucket's records and refuses a root another record claims, and a drop of a record sharing its files is refused. Co-Authored-By: Claude Opus 5.5 * iceberg: check a table's location before writing to it REST create and stage-create write a metadata file, and stage-create vends write credentials, before the catalog registers the table, so the create gate now also validates the location and refuses one another record holds. RegisterTable runs the same gate before it reads the metadata file, instead of reading first. Co-Authored-By: Claude Opus 5.5 * iceberg: keep a dropped location another record holds DropTable and DropView purge the dropped record's location after the catalog delete. A view recorded at a table's location took that table's directory, record included. The purge now leaves a location another record still holds. Co-Authored-By: Claude Opus 5.5 * s3tables: authorize CreateTable before reporting the namespace A caller without CreateTable got 404 for a missing namespace, 403 for an existing one and 500 with decoder text for an unreadable one. The grant is now evaluated against the bucket first, and only a caller that passes learns the namespace is missing or unreadable. Co-Authored-By: Claude Opus 5.5 * iceberg maintenance: skip a table whose files another record shares Orphan removal and snapshot expiry delete what a table's metadata does not reference. For two records over one location that deletes the other's files, so detection leaves such tables alone. Co-Authored-By: Claude Opus 5.5 * s3tables: count another record's catalog entry as claimed A table keeping its files elsewhere still owns its entry directory, and a view recorded there would take that entry with it on drop. The ownership check now refuses a root at, above or under another record's entry too. Maintenance still compares files only, so a table created over a renamed table's old name does not stop that table being maintained. Co-Authored-By: Claude Opus 5.5 * s3tables: check a claimed location again once the record is written Two creates, or a create and an update moving a root, could both read the same root free and both land. A create now checks again after writing its still-marked record and backs out if another record claims the root, and an update that moved its root checks again after the write and puts the previous metadata back. Of two racing writers at least one sees the other. Co-Authored-By: Claude Opus 5.5 * iceberg: enable stage-create in the location ownership test Co-Authored-By: Claude Opus 5.5 * iceberg maintenance: check for a shared location again before removing orphans A job can run long after its table was scheduled, or come from a detector that never checked, so remove_orphans reads the bucket's records itself and leaves the files alone when another record shares the table's location. Co-Authored-By: Claude Opus 5.5 * s3tables: check a renamed record's new root again once it is written A record without a location moves its root with its name, and was checked only before the rename wrote the destination, so a create landing on that root in between could share it. The rename now checks again with the destination written and backs out, like a create. Co-Authored-By: Claude Opus 5.5 * s3tables: put a moved record back when its location cannot be checked again An update that moved its root kept the move when the check after the write failed to list the bucket, so a racing writer that relied on that check could share the root. The move is now undone as it is when the root is taken. Co-Authored-By: Claude Opus 5.5 * s3tables: hold a moved record's old root until the move lands An update moving a record's root wrote the new metadata first and put the old metadata back when the check that followed found the new root taken. The old root was free in between, so a create racing for it could pass its own check and then share it with the restored record. The update now stores the metadata it is about to write beside the current one, so the record holds both roots, checks the new root with that reservation in place, and only then switches the metadata. A move that backs out drops the reservation without touching the metadata, and an update that keeps its root clears a reservation a crashed move left behind. Co-Authored-By: Claude Opus 5.5 * s3tables: test a move racing another update of the record Co-Authored-By: Claude Opus 5.5 * s3tables: drop a move's reservation when a rename or delete takes the record A rename or a decoupled delete that took a record while an update held its move reservation left the reservation on the old name, where the move could no longer clear it, and later creates of that name were refused as taken. Both now clear it with the record's other attributes, and a failed rename does not put it back. Co-Authored-By: Claude Opus 5.5 * test: register a table from metadata under its own location The Spark register test registered a live table's metadata file under a second name, which now fails because both records would own the source's files. It checks that refusal instead, and registers a copy uploaded under the new table's own directory, with its location pointing there, to cover reading and writing a registered table. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- .../spark_register_table_test.go | 103 ++--- weed/s3api/iceberg/handlers_commit.go | 2 +- weed/s3api/iceberg/handlers_table.go | 29 +- .../s3api/iceberg/iceberg_create_view_test.go | 11 +- .../iceberg/iceberg_drop_location_test.go | 57 +++ .../iceberg/iceberg_location_owner_test.go | 122 ++++++ weed/s3api/lance/handlers_test.go | 4 +- weed/s3api/s3tables/filer_ops.go | 217 ++++++++++- weed/s3api/s3tables/handler.go | 3 + .../s3tables/handler_existence_hiding_test.go | 32 ++ .../s3tables/handler_location_owner_test.go | 361 ++++++++++++++++++ weed/s3api/s3tables/handler_table.go | 105 +++-- weed/s3api/s3tables/handler_view.go | 28 +- weed/s3api/s3tables/manager.go | 13 +- weed/worker/tasks/iceberg/detection.go | 30 ++ weed/worker/tasks/iceberg/operations.go | 8 + .../tasks/iceberg/shared_location_test.go | 98 +++++ 17 files changed, 1136 insertions(+), 87 deletions(-) create mode 100644 weed/s3api/iceberg/iceberg_drop_location_test.go create mode 100644 weed/s3api/iceberg/iceberg_location_owner_test.go create mode 100644 weed/s3api/s3tables/handler_location_owner_test.go create mode 100644 weed/worker/tasks/iceberg/shared_location_test.go diff --git a/test/s3tables/catalog_spark/spark_register_table_test.go b/test/s3tables/catalog_spark/spark_register_table_test.go index 1c186c5b61..9963d1615a 100644 --- a/test/s3tables/catalog_spark/spark_register_table_test.go +++ b/test/s3tables/catalog_spark/spark_register_table_test.go @@ -3,6 +3,7 @@ package catalog_spark import ( "bytes" "context" + "encoding/json" "fmt" "io" "path" @@ -16,13 +17,14 @@ import ( "github.com/aws/aws-sdk-go-v2/service/s3" ) -// TestSparkRegisterTableServesItsData registers a table under a new name from -// the current metadata file of another and checks that Spark reads the -// registered rows and can write on top of them. One file is the one the -// catalog wrote; the other is a copy uploaded through the S3 gateway, the way -// the metadata of a table migrated in arrives, which the filer stores in -// chunks rather than inline. A registered table used to load as an empty -// placeholder: SELECT found no rows and INSERT failed its table UUID check. +// TestSparkRegisterTableServesItsData registers a table from a copy of another +// table's current metadata, uploaded through the S3 gateway into the new +// table's own directory the way the metadata of a table migrated in arrives, +// which the filer stores in chunks rather than inline, and checks that Spark +// reads the registered rows and can write on top of them. A registered table +// used to load as an empty placeholder: SELECT found no rows and INSERT failed +// its table UUID check. The source's own file is refused under another name, +// since both tables would then own the source's files. func TestSparkRegisterTableServesItsData(t *testing.T) { if testing.Short() { t.Skip("Skipping integration test in short mode") @@ -30,56 +32,63 @@ func TestSparkRegisterTableServesItsData(t *testing.T) { env, _, tableBucket := setupSparkTestEnv(t) - for _, tc := range []struct { - name string - upload bool - }{ - {"metadata file written by the catalog", false}, - {"metadata file uploaded through S3", true}, - } { - t.Run(tc.name, func(t *testing.T) { - namespace := "register_" + randomString(6) - source := fmt.Sprintf("iceberg.%s.src_%s", namespace, randomString(6)) - table := "reg_" + randomString(6) - ident := fmt.Sprintf("iceberg.%s.%s", namespace, table) - - output := runSparkPySQL(t, env.sparkContainer, fmt.Sprintf(` + namespace := "register_" + randomString(6) + source := fmt.Sprintf("iceberg.%s.src_%s", namespace, randomString(6)) + output := runSparkPySQL(t, env.sparkContainer, fmt.Sprintf(` spark.sql("CREATE NAMESPACE iceberg.%[1]s") spark.sql("CREATE TABLE %[2]s (id INT, name STRING) USING iceberg") spark.sql("INSERT INTO %[2]s VALUES (1, 'alpha'), (2, 'beta'), (3, 'gamma')") meta = spark.sql("SELECT file FROM %[2]s.metadata_log_entries ORDER BY timestamp DESC LIMIT 1").collect()[0]['file'] print(f"metadata file: {meta}") `, namespace, source), env.icebergRestPort, env.s3Port) - match := regexp.MustCompile(`metadata file: (\S+)`).FindStringSubmatch(output) - if match == nil { - t.Fatalf("create/insert of %s failed, output:\n%s", source, output) - } - metadataFile := match[1] - if tc.upload { - metadataFile = uploadMetadataCopy(t, env, tableBucket, metadataFile) - } + match := regexp.MustCompile(`metadata file: (\S+)`).FindStringSubmatch(output) + if match == nil { + t.Fatalf("create/insert of %s failed, output:\n%s", source, output) + } + metadataFile := match[1] - output = runSparkPySQL(t, env.sparkContainer, fmt.Sprintf(` + t.Run("the source's own metadata file is refused", func(t *testing.T) { + output := runSparkPySQL(t, env.sparkContainer, fmt.Sprintf(` +try: + spark.sql("CALL iceberg.system.register_table('%[1]s.alias_%[2]s', '%[3]s')") + print("register accepted") +except Exception as e: + print(f"register refused: {e}") +count = spark.sql("SELECT COUNT(*) AS c FROM %[4]s").collect()[0]['c'] +print(f"source row count: {count}") +`, namespace, randomString(6), metadataFile, source), env.icebergRestPort, env.s3Port) + if !strings.Contains(output, "register refused") || !strings.Contains(output, "location is in use") { + t.Fatalf("registering %s under another name was not refused, output:\n%s", metadataFile, output) + } + if !strings.Contains(output, "source row count: 3") { + t.Fatalf("the source lost its rows, output:\n%s", output) + } + }) + + t.Run("a copy uploaded under the new table's location", func(t *testing.T) { + table := "reg_" + randomString(6) + ident := fmt.Sprintf("iceberg.%s.%s", namespace, table) + copied := uploadMetadataCopy(t, env, tableBucket, metadataFile, path.Join(namespace, table)) + output := runSparkPySQL(t, env.sparkContainer, fmt.Sprintf(` spark.sql("CALL iceberg.system.register_table('%[1]s.%[2]s', '%[3]s')") registered = spark.sql("SELECT COUNT(*) AS c FROM %[4]s").collect()[0]['c'] print(f"registered row count: {registered}") spark.sql("INSERT INTO %[4]s VALUES (4, 'delta')") inserted = spark.sql("SELECT COUNT(*) AS c FROM %[4]s").collect()[0]['c'] print(f"row count after insert: {inserted}") -`, namespace, table, metadataFile, ident), env.icebergRestPort, env.s3Port) - if !strings.Contains(output, "registered row count: 3") { - t.Fatalf("the table registered from %s does not serve its rows, output:\n%s", metadataFile, output) - } - if !strings.Contains(output, "row count after insert: 4") { - t.Fatalf("INSERT into the table registered from %s failed, output:\n%s", metadataFile, output) - } - }) - } +`, namespace, table, copied, ident), env.icebergRestPort, env.s3Port) + if !strings.Contains(output, "registered row count: 3") { + t.Fatalf("the table registered from %s does not serve its rows, output:\n%s", copied, output) + } + if !strings.Contains(output, "row count after insert: 4") { + t.Fatalf("INSERT into the table registered from %s failed, output:\n%s", copied, output) + } + }) } -// uploadMetadataCopy puts a copy of metadataFile next to it through the S3 -// gateway and returns the copy's location. -func uploadMetadataCopy(t *testing.T, env *TestEnvironment, bucket, metadataFile string) string { +// uploadMetadataCopy puts a copy of metadataFile whose location is tableDir +// under tableDir through the S3 gateway and returns the copy's location. +func uploadMetadataCopy(t *testing.T, env *TestEnvironment, bucket, metadataFile, tableDir string) string { t.Helper() key, ok := strings.CutPrefix(metadataFile, "s3://"+bucket+"/") if !ok { @@ -104,8 +113,16 @@ func uploadMetadataCopy(t *testing.T, env *TestEnvironment, bucket, metadataFile if err != nil { t.Fatalf("read %s: %v", key, err) } + var fields map[string]json.RawMessage + if err := json.Unmarshal(content, &fields); err != nil { + t.Fatalf("parse %s: %v", key, err) + } + fields["location"], _ = json.Marshal("s3://" + bucket + "/" + tableDir) + if content, err = json.Marshal(fields); err != nil { + t.Fatalf("encode %s: %v", key, err) + } - copyKey := path.Join(path.Dir(key), "00003-uploaded.metadata.json") + copyKey := path.Join(tableDir, "metadata", "00003-uploaded.metadata.json") if _, err := client.PutObject(ctx, &s3.PutObjectInput{ Bucket: aws.String(bucket), Key: aws.String(copyKey), diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index 1f2aed1ae1..66e5af7515 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/weed/s3api/iceberg/handlers_commit.go @@ -180,7 +180,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { } // From here the commit creates the table, writing its metadata // file before the create that authorizes it. - if authErr := s.authorizeCreateTable(r.Context(), bucketARN, namespace, tableName, identityName); authErr != nil { + if authErr := s.authorizeCreateTable(r.Context(), bucketARN, namespace, tableName, identityName, location); authErr != nil { writeManagerError(w, authErr) return } diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 98debaeb12..63ba543ca0 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -138,14 +138,15 @@ func (s *Server) tablePathOccupied(ctx context.Context, bucketName, tablePath st return occupied, err } -// authorizeCreateTable checks the caller may create the table, for the create -// paths that write to the bucket before the catalog registers it. -func (s *Server) authorizeCreateTable(ctx context.Context, bucketARN string, namespace []string, tableName, identityName string) error { +// authorizeCreateTable checks the caller may create the table at location, for +// the create paths that touch the bucket before the catalog registers it. +func (s *Server) authorizeCreateTable(ctx context.Context, bucketARN string, namespace []string, tableName, identityName, location string) error { return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return s.tablesManager.AuthorizeCreateTable(ctx, s3tables.NewManagerClient(client), &s3tables.CreateTableRequest{ - TableBucketARN: bucketARN, - Namespace: namespace, - Name: tableName, + TableBucketARN: bucketARN, + Namespace: namespace, + Name: tableName, + MetadataLocation: location, }, identityName) }) } @@ -304,7 +305,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { // Both branches below write into the table bucket, and stage-create never // reaches the registration that carries the authorization, so the caller has // to pass the CreateTable gate here. - if authErr := s.authorizeCreateTable(r.Context(), bucketARN, namespace, tableName, identityName); authErr != nil { + if authErr := s.authorizeCreateTable(r.Context(), bucketARN, namespace, tableName, identityName, metadataLocation); authErr != nil { writeManagerError(w, authErr) return } @@ -488,6 +489,11 @@ func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "BadRequestException", "invalid metadata-location path") return } + if authErr := s.authorizeCreateTable(r.Context(), bucketARN, namespace, req.Name, identityName, req.MetadataLocation); authErr != nil { + writeManagerError(w, authErr) + return + } + metadataFileName := path.Base(req.MetadataLocation) // A table migrated in has its metadata uploaded through the S3 gateway, // which stores it in chunks rather than inline, so follow the chunks. @@ -1031,7 +1037,14 @@ func (s *Server) cleanupStaleTableLocation(ctx context.Context, bucketName, tabl return fmt.Errorf("cleanupStaleTableLocation: refusing unsafe table path %q", tablePath) } return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - err := filer_pb.DoRemove(ctx, client, parentDir, name, true, true, true, false, nil) + roots, err := s3tables.CatalogRecordRoots(ctx, client, bucketName) + if err != nil { + return err + } + if s3tables.RootClaimed(roots, "", path.Join(parentDir, name)) { + return s3tables.ErrLocationInUse + } + err = filer_pb.DoRemove(ctx, client, parentDir, name, true, true, true, false, nil) if err == nil || errors.Is(err, filer_pb.ErrNotFound) { return nil } diff --git a/weed/s3api/iceberg/iceberg_create_view_test.go b/weed/s3api/iceberg/iceberg_create_view_test.go index 7f5e641f07..b08b40c5cc 100644 --- a/weed/s3api/iceberg/iceberg_create_view_test.go +++ b/weed/s3api/iceberg/iceberg_create_view_test.go @@ -81,13 +81,20 @@ func (m *memFiler) UpdateEntry(_ context.Context, in *filer_pb.UpdateEntryReques func (m *memFiler) ListEntries(_ context.Context, in *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_ListEntriesClient, error) { var names []string for p := range m.entries { - if path.Dir(p) == in.Directory { - names = append(names, p) + if path.Dir(p) != in.Directory { + continue } + if name := path.Base(p); in.StartFromFileName != "" && (name < in.StartFromFileName || name == in.StartFromFileName && !in.InclusiveStartFrom) { + continue + } + names = append(names, p) } sort.Strings(names) stream := &memListEntries{} for _, name := range names { + if in.Limit > 0 && uint32(len(stream.entries)) >= in.Limit { + break + } stream.entries = append(stream.entries, m.entries[name]) } return stream, nil diff --git a/weed/s3api/iceberg/iceberg_drop_location_test.go b/weed/s3api/iceberg/iceberg_drop_location_test.go new file mode 100644 index 0000000000..4052c5b1b0 --- /dev/null +++ b/weed/s3api/iceberg/iceberg_drop_location_test.go @@ -0,0 +1,57 @@ +package iceberg + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// A view recorded at another table's location, as one created before the +// location rule could be, is dropped without purging that table's files. +func TestDropViewKeepsAnotherRecordsFiles(t *testing.T) { + fc := newMemFiler() + _, ordersPath := seedOrders(t, fc) + viewMeta, err := json.Marshal(map[string]any{ + "name": "alias", + "namespace": "finance", + "ownerAccountId": "mallory", + "versionToken": "v1", + "metadataVersion": 1, + "metadataLocation": "s3://" + ownerTestBucket + "/finance/orders/metadata/v1.metadata.json", + }) + if err != nil { + t.Fatal(err) + } + fc.seed(s3tables.GetTablePath(ownerTestBucket, "finance", "alias"), &filer_pb.Entry{ + Name: "alias", + IsDirectory: true, + Extended: map[string][]byte{ + s3tables.ExtendedKeyMetadata: viewMeta, + s3tables.ExtendedKeyEntryType: []byte(s3tables.EntryTypeView), + }, + }) + s := NewServer(fc, nil) + + r := httptest.NewRequest(http.MethodDelete, "/v1/"+ownerTestBucket+"/namespaces/finance/views/alias", nil) + r = mux.SetURLVars(r, map[string]string{"prefix": ownerTestBucket, "namespace": "finance", "view": "alias"}) + w := httptest.NewRecorder() + s.handleDropView(w, withActions(r, "mallory", creatorActions...)) + + if w.Code != http.StatusNoContent { + t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusNoContent, w.Body.String()) + } + if _, ok := fc.entries[s3tables.GetTablePath(ownerTestBucket, "finance", "alias")]; ok { + t.Errorf("dropped view still recorded") + } + if _, ok := fc.entries[ordersPath]; !ok { + t.Errorf("dropping the view deleted finance.orders") + } + if _, ok := fc.entries[ordersPath+"/data/f.parquet"]; !ok { + t.Errorf("dropping the view deleted the files of finance.orders") + } +} diff --git a/weed/s3api/iceberg/iceberg_location_owner_test.go b/weed/s3api/iceberg/iceberg_location_owner_test.go new file mode 100644 index 0000000000..6438631857 --- /dev/null +++ b/weed/s3api/iceberg/iceberg_location_owner_test.go @@ -0,0 +1,122 @@ +package iceberg + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/apache/iceberg-go/view" + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +const ownerTestBucket = "warehouse" + +var creatorActions = []string{"s3tables:CreateTable", "s3tables:GetTable", "s3tables:UpdateTable", "s3tables:CreateView", "s3tables:GetView", "s3tables:DeleteView"} + +// seedOrders registers finance.orders with a data file, and returns the +// location of its metadata file. +func seedOrders(t *testing.T, fc *memFiler) (string, string) { + t.Helper() + seedNamespace(fc, ownerTestBucket, "finance", "alice") + location, _ := seedRegistrableMetadata(t, fc, "orders") + seedOwnedTable(fc, ownerTestBucket, "finance", "orders", "alice") + ordersPath := s3tables.GetTablePath(ownerTestBucket, "finance", "orders") + fc.seed(ordersPath+"/data/f.parquet", &filer_pb.Entry{Name: "f.parquet", Content: []byte("x")}) + return location, ordersPath +} + +func createViewAt(t *testing.T, name, location string) *http.Request { + t.Helper() + schema := iceberg.NewSchemaWithIdentifiers(0, nil, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true}) + ver, err := view.NewVersionFromSQL(1, schema.ID, "SELECT 1", table.Identifier{"finance"}) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(CreateViewRequest{Name: name, Schema: schema, ViewVersion: ver, Location: location}) + if err != nil { + t.Fatal(err) + } + r := httptest.NewRequest(http.MethodPost, "/v1/"+ownerTestBucket+"/namespaces/finance/views", bytes.NewReader(body)) + return mux.SetURLVars(r, map[string]string{"prefix": ownerTestBucket, "namespace": "finance"}) +} + +// A caller allowed to create tables in finance cannot point a new table or view +// at the files of finance.orders, and gets no credentials or metadata file +// written there. +func TestCreatePathsRefuseAnotherTablesLocation(t *testing.T) { + t.Setenv("ICEBERG_ENABLE_STAGE_CREATE", "true") + ordersLocation := "s3://" + ownerTestBucket + "/finance/orders" + cases := map[string]func(s *Server, metadataLocation string) *httptest.ResponseRecorder{ + "register": func(s *Server, metadataLocation string) *httptest.ResponseRecorder { + r := httptest.NewRequest(http.MethodPost, "/v1/"+ownerTestBucket+"/namespaces/finance/register", + strings.NewReader(`{"name":"alias","metadata-location":"`+metadataLocation+`"}`)) + r = mux.SetURLVars(r, map[string]string{"prefix": ownerTestBucket, "namespace": "finance"}) + w := httptest.NewRecorder() + s.handleRegisterTable(w, withActions(r, "mallory", creatorActions...)) + return w + }, + "create": func(s *Server, _ string) *httptest.ResponseRecorder { + r := newCreateTableRequest(t, ownerTestBucket, "finance", `{"name":"alias","location":"`+ordersLocation+`"}`, "mallory") + w := httptest.NewRecorder() + s.handleCreateTable(w, withActions(r, "mallory", creatorActions...)) + return w + }, + "stage-create": func(s *Server, _ string) *httptest.ResponseRecorder { + r := newCreateTableRequest(t, ownerTestBucket, "finance", `{"name":"alias","location":"`+ordersLocation+`/data","stage-create":true}`, "mallory") + w := httptest.NewRecorder() + s.handleCreateTable(w, withActions(r, "mallory", creatorActions...)) + return w + }, + "create-view": func(s *Server, _ string) *httptest.ResponseRecorder { + w := httptest.NewRecorder() + s.handleCreateView(w, withActions(createViewAt(t, "alias", ordersLocation), "mallory", creatorActions...)) + return w + }, + } + for name, run := range cases { + t.Run(name, func(t *testing.T) { + fc := &writeLogFiler{memFiler: newMemFiler()} + metadataLocation, _ := seedOrders(t, fc.memFiler) + s := NewServer(fc, nil) + + w := run(s, metadataLocation) + + if w.Code != http.StatusConflict { + t.Errorf("status = %d, want %d (body: %s)", w.Code, http.StatusConflict, w.Body.String()) + } + if len(fc.created) != 0 { + t.Errorf("refused create wrote %v", fc.created) + } + if _, ok := fc.entries[s3tables.GetTablePath(ownerTestBucket, "finance", "alias")]; ok { + t.Errorf("refused create left a catalog record") + } + }) + } +} + +// Registration is authorized before the metadata file is read, so a caller +// without CreateTable learns nothing about the files of the bucket. +func TestRegisterTableAuthorizedBeforeRead(t *testing.T) { + fc := newMemFiler() + metadataLocation, _ := seedOrders(t, fc) + s := NewServer(fc, nil) + + for _, location := range []string{metadataLocation, "s3://" + ownerTestBucket + "/finance/orders/metadata/missing.metadata.json"} { + r := httptest.NewRequest(http.MethodPost, "/v1/"+ownerTestBucket+"/namespaces/finance/register", + strings.NewReader(`{"name":"alias","metadata-location":"`+location+`"}`)) + r = mux.SetURLVars(r, map[string]string{"prefix": ownerTestBucket, "namespace": "finance"}) + w := httptest.NewRecorder() + s.handleRegisterTable(w, withActions(r, "reader", "s3tables:GetTable")) + if w.Code != http.StatusForbidden { + t.Errorf("%s: status = %d, want %d (body: %s)", location, w.Code, http.StatusForbidden, w.Body.String()) + } + } +} diff --git a/weed/s3api/lance/handlers_test.go b/weed/s3api/lance/handlers_test.go index 23ea7b610f..32fd3a89a9 100644 --- a/weed/s3api/lance/handlers_test.go +++ b/weed/s3api/lance/handlers_test.go @@ -235,8 +235,10 @@ func TestRegisterOverwriteKeepsTheOldDataset(t *testing.T) { h.createBucket(t, "analytics", s3tables.FormatLance) h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK) h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/declare", `{}`, http.StatusOK) - h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$archive/declare", `{}`, http.StatusOK) + h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$current/declare", `{}`, http.StatusOK) + h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/register", + `{"location":"s3://analytics/sales/current","mode":"Overwrite"}`, http.StatusConflict) h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/register", `{"location":"s3://analytics/sales/archive","mode":"Overwrite"}`, http.StatusOK) diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 80ad13ef25..cd7159e68f 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -7,8 +7,10 @@ import ( "errors" "fmt" "io" + "net/http" "os" "path" + "strings" "time" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -23,6 +25,7 @@ var ( ErrAttributeNotFound = errors.New("attribute not found") 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") ) // extendedAttributeUpdateAttempts bounds the read-modify-write retry loop in @@ -363,6 +366,7 @@ var s3tablesExtendedKeys = []string{ ExtendedKeyMaintenanceStatus, ExtendedKeyEntryType, ExtendedKeyDeleting, + ExtendedKeyMoving, ExtendedKeyVectorPolicy, } @@ -809,6 +813,16 @@ 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) + if errors.Is(err, errParentGone) { + h.removeLeftoverDirectory(context.WithoutCancel(ctx), client, parentPath) + } + return err +} + +// 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 { ctx = context.WithoutCancel(ctx) expected := written keys := make([]string, 0, len(written)) @@ -839,9 +853,6 @@ func (h *S3TablesHandler) confirmChild(ctx context.Context, client filer_pb.Seaw for _, dir := range emptyDirs { h.removeLeftoverDirectory(ctx, client, dir) } - if errors.Is(err, errParentGone) { - h.removeLeftoverDirectory(ctx, client, parentPath) - } return err } @@ -905,6 +916,206 @@ func (h *S3TablesHandler) dirHoldsFiles(ctx context.Context, client filer_pb.Sea return false, nil } +// eachDirEntry calls fn on the entries of dir page by page until fn returns +// false. A missing dir has none. +func eachDirEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, fn func(*filer_pb.Entry) bool) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + lastFileName := "" + for { + resp, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{ + Directory: dir, + Limit: dirListPageSize, + StartFromFileName: lastFileName, + InclusiveStartFrom: lastFileName == "", + }) + if err != nil { + if status.Code(err) == codes.NotFound || errors.Is(err, filer_pb.ErrNotFound) { + return nil + } + return err + } + received := false + for { + entry, err := resp.Recv() + if err != nil { + if err == io.EOF || status.Code(err) == codes.NotFound { + break + } + return err + } + if entry.Entry == nil { + continue + } + received = true + lastFileName = entry.Entry.Name + if !fn(entry.Entry) { + return nil + } + } + if !received { + return nil + } + } +} + +// CatalogRecordRoot returns the directory the catalog record at recordPath +// keeps its files under: the location it names inside the bucket, or its own +// path. +func CatalogRecordRoot(bucketName, recordPath, metadataLocation string) string { + if dir := TableDataDirFromMetadataLocation(metadataLocation); strings.HasPrefix(dir, GetTableBucketPath(bucketName)+"/") { + return dir + } + return recordPath +} + +// CatalogRecordRoots maps every table and view record of the bucket, in any +// namespace, to the directories it keeps its files under: the one its metadata +// gives it and, while an update moves it, the one it moves to. A renamed record +// keeps the directory of its old name. +func CatalogRecordRoots(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName string) (map[string][]string, error) { + var namespaces []string + if err := eachDirEntry(ctx, client, GetTableBucketPath(bucketName), func(entry *filer_pb.Entry) bool { + if entry.IsDirectory && !strings.HasPrefix(entry.Name, ".") { + namespaces = append(namespaces, entry.Name) + } + return true + }); err != nil { + return nil, err + } + roots := map[string][]string{} + for _, namespace := range namespaces { + namespacePath := GetNamespacePath(bucketName, namespace) + if err := eachDirEntry(ctx, client, namespacePath, func(entry *filer_pb.Entry) bool { + if _, ok := entry.Extended[ExtendedKeyMetadata]; !ok || !entry.IsDirectory { + return true + } + recordPath := path.Join(namespacePath, entry.Name) + for _, key := range []string{ExtendedKeyMetadata, ExtendedKeyMoving} { + data, ok := entry.Extended[key] + if !ok { + continue + } + var record struct { + MetadataLocation string `json:"metadataLocation"` + } + _ = json.Unmarshal(data, &record) + roots[recordPath] = append(roots[recordPath], CatalogRecordRoot(bucketName, recordPath, record.MetadataLocation)) + } + return true + }); err != nil { + return nil, err + } + } + return roots, nil +} + +// PathsOverlap reports whether a and b are the same path or one holds the other. +func PathsOverlap(a, b string) bool { + return a == b || strings.HasPrefix(a, b+"/") || strings.HasPrefix(b, a+"/") +} + +// RootClaimed reports whether a record other than selfPath has its entry or its +// files at root, above it or under it. +func RootClaimed(roots map[string][]string, selfPath, root string) bool { + for recordPath, others := range roots { + if recordPath == selfPath { + continue + } + if PathsOverlap(recordPath, root) { + return true + } + for _, other := range others { + if PathsOverlap(other, root) { + return true + } + } + } + return false +} + +// checkLocationUnclaimed returns ErrLocationInUse when the root the record at +// recordPath would have with metadataLocation is claimed by another record. +func checkLocationUnclaimed(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, recordPath, metadataLocation string) error { + roots, err := CatalogRecordRoots(ctx, client, bucketName) + if err != nil { + return err + } + if RootClaimed(roots, recordPath, CatalogRecordRoot(bucketName, recordPath, metadataLocation)) { + return ErrLocationInUse + } + return nil +} + +// claimCatalogEntryAt claims path like claimCatalogEntry, then checks the root +// metadataLocation gives it again with the record written: of two creates +// racing for one root, at least one sees the other and backs out. +func (h *S3TablesHandler) claimCatalogEntryAt(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, path, metadataLocation string, attrs map[string][]byte, emptyDirs ...string) error { + if err := h.claimCatalogEntry(ctx, client, path, attrs); err != nil { + return err + } + if err := checkLocationUnclaimed(ctx, client, bucketName, path, metadataLocation); err != nil { + return h.backOutChild(ctx, client, path, attrs, false, emptyDirs, err) + } + return nil +} + +// moveCatalogRecord applies write, which moves the record at path onto the +// root of moving, only once that root is reserved: moving is first stored +// beside the current metadata, so the record holds both roots while the new +// one is checked again. Of two writers racing for it at least one sees the +// other, and the old root stays held until the move lands. A move that backs +// out drops the reservation and leaves the metadata as it was. +func (h *S3TablesHandler) moveCatalogRecord(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, path, metadataLocation string, stored, moving []byte, write func(extended map[string][]byte) error) error { + if err := h.mutateCatalogRecord(ctx, client, path, func(extended map[string][]byte) error { + if !bytes.Equal(extended[ExtendedKeyMetadata], stored) { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata) + } + if _, ok := extended[ExtendedKeyMoving]; ok { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMoving) + } + extended[ExtendedKeyMoving] = moving + return nil + }); err != nil { + return err + } + err := checkLocationUnclaimed(ctx, client, bucketName, path, metadataLocation) + if err == nil { + err = h.mutateCatalogRecord(ctx, client, path, func(extended map[string][]byte) error { + if !bytes.Equal(extended[ExtendedKeyMoving], moving) { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMoving) + } + return write(extended) + }) + } + if err != nil { + if releaseErr := h.mutateCatalogRecord(context.WithoutCancel(ctx), client, path, func(extended map[string][]byte) error { + if !bytes.Equal(extended[ExtendedKeyMoving], moving) { + return ErrConcurrentUpdate + } + delete(extended, ExtendedKeyMoving) + return nil + }); releaseErr != nil && !errors.Is(releaseErr, ErrConcurrentUpdate) && !errors.Is(releaseErr, filer_pb.ErrNotFound) { + glog.Warningf("release move of %s: %v", path, releaseErr) + } + } + return err +} + +// refuseClaimedLocation answers a request whose record would share its files +// with another table or view of the bucket. +func (h *S3TablesHandler) refuseClaimedLocation(w http.ResponseWriter, r *http.Request, filerClient FilerClient, bucketName, recordPath, metadataLocation string) error { + err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return checkLocationUnclaimed(r.Context(), client, bucketName, recordPath, metadataLocation) + }) + if errors.Is(err, ErrLocationInUse) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) + } else if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to check location") + } + return err +} + // claimTreeDirectory writes a fresh delete claim onto the directory at dirPath // while it still carries expected, and returns the claim with the attributes it // left. A claim seen in expected, one a delete that died left, is replaced like diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index 8e1b932bac..84565439a5 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -38,6 +38,9 @@ const ( // a namespace being deleted carries the random claim token written by // claimForDelete. ExtendedKeyDeleting = s3_constants.ExtS3TablesPrefix + "deleting" + // ExtendedKeyMoving holds the metadata an update moving a record's root is + // about to write, so the record holds both roots until the move lands. + ExtendedKeyMoving = s3_constants.ExtS3TablesPrefix + "moving" // ExtendedKeyVectorPolicy is written by the S3 Vectors service on table // buckets. It lives here rather than in the vectors package so the catalog diff --git a/weed/s3api/s3tables/handler_existence_hiding_test.go b/weed/s3api/s3tables/handler_existence_hiding_test.go index 187ebf093e..63a912e4c5 100644 --- a/weed/s3api/s3tables/handler_existence_hiding_test.go +++ b/weed/s3api/s3tables/handler_existence_hiding_test.go @@ -374,3 +374,35 @@ func TestAbsentTagTargetDenySurvivesForeignAccountARN(t *testing.T) { }) } } + +// A caller refused CreateTable gets the same answer whether the namespace +// exists, is missing, is unreadable, or the bucket is missing. +func TestCreateTableDenialMatchesMissingNamespace(t *testing.T) { + existing, manager := startRenameManager(t) + missing := s3tablestest.Start(t) + manager.SetTrusted(false) + manager.SetDefaultAllow(false) + existing.Put(GetTableBucketPath(renameTestBucket), "broken", map[string][]byte{ExtendedKeyMetadata: []byte("{")}) + + for _, operation := range []string{"CreateTable", "RegisterTable"} { + t.Run(operation, func(t *testing.T) { + request := func(namespace string) interface{} { + if operation == "RegisterTable" { + return &RegisterTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{namespace}, Name: "x", + MetadataLocation: "s3://" + renameTestBucket + "/" + namespace + "/x/metadata/v1.metadata.json"} + } + return &CreateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{namespace}, Name: "x", Format: FormatIceberg} + } + + want := runUnauthorizedRequest(t, manager, existing, operation, request("ns")) + assert.Equal(t, 403, want.status) + assert.Equal(t, want, runUnauthorizedRequest(t, manager, existing, operation, request("nope"))) + assert.Equal(t, want, runUnauthorizedRequest(t, manager, existing, operation, request("broken"))) + assert.Equal(t, want, runUnauthorizedRequest(t, manager, missing, operation, request("ns"))) + + status, resp := runHandler(t, manager, existing.Client, operation, request("nope")) + assert.Equal(t, 404, status) + assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type) + }) + } +} diff --git a/weed/s3api/s3tables/handler_location_owner_test.go b/weed/s3api/s3tables/handler_location_owner_test.go new file mode 100644 index 0000000000..afba1584ff --- /dev/null +++ b/weed/s3api/s3tables/handler_location_owner_test.go @@ -0,0 +1,361 @@ +package s3tables + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "path" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" +) + +func renameLocation(rel string) string { + return "s3://" + renameTestBucket + "/" + rel +} + +// A table's files belong to one catalog record. A create, register, update or +// view naming a location at, under or above another record's files is refused, +// and so is one naming the files a renamed table left under its old name. +func TestLocationOwnedByOneRecord(t *testing.T) { + fs, m := startRenameManager(t) + bucketARN := mustBucketARN(t) + requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{TableBucketARN: bucketARN, Namespace: []string{"ns2"}}) + requireOK(t, m, fs, "CreateTable", &CreateTableRequest{ + TableBucketARN: bucketARN, + Namespace: []string{"ns"}, + Name: "own", + Format: FormatIceberg, + MetadataLocation: renameLocation("ns/own/metadata/v1.metadata.json"), + }) + + refused := func(t *testing.T, operation string, input interface{}) { + t.Helper() + status, resp := runHandler(t, m, fs.Client, operation, input) + assert.Equal(t, http.StatusConflict, status, "%s: %+v", operation, resp) + assert.Equal(t, ErrCodeConflict, resp.Type) + assert.NotContains(t, resp.Message, "ns/t") + } + checkRefusals := func(t *testing.T) { + for _, location := range []string{ + "ns/t/metadata/v3.metadata.json", + "ns/t/data/part/metadata/v1.metadata.json", + "ns/metadata/v1.metadata.json", + } { + refused(t, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "alias", Format: FormatLance, MetadataLocation: renameLocation(location)}) + refused(t, "RegisterTable", &RegisterTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "alias", MetadataLocation: renameLocation(location)}) + refused(t, "CreateView", &CreateViewRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "alias", MetadataLocation: renameLocation(location)}) + refused(t, "UpdateTable", &UpdateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "own", MetadataVersion: 2, MetadataLocation: renameLocation(location)}) + } + assert.Nil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "alias")) + } + + t.Run("live table", checkRefusals) + + requireOK(t, m, fs, "RenameTable", &RenameTableRequest{ + TableBucketARN: bucketARN, + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns2"}, + DestName: "t", + }) + t.Run("renamed table", func(t *testing.T) { + checkRefusals(t) + refused(t, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "t", Format: FormatIceberg}) + }) + + t.Run("free locations", func(t *testing.T) { + requireOK(t, m, fs, "UpdateTable", &UpdateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "own", MetadataVersion: 2, MetadataLocation: renameLocation("ns/own/metadata/v2.metadata.json")}) + requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "t2", Format: FormatIceberg, MetadataLocation: renameLocation("ns/t2/metadata/v1.metadata.json")}) + requireOK(t, m, fs, "RegisterTable", &RegisterTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "t3", MetadataLocation: renameLocation("ns/t3/metadata/v1.metadata.json")}) + }) +} + +// A drop of a record that shares its files with another is refused, so the +// other record keeps them. +func TestDeleteTableKeepsSharedFiles(t *testing.T) { + fs, m := startRenameManager(t) + aliasMeta, err := json.Marshal(tableMetadataInternal{ + Name: "alias", + Namespace: "ns", + Format: FormatIceberg, + OwnerAccountID: DefaultAccountID, + MetadataVersion: 3, + MetadataLocation: renameLocation("ns/t/metadata/v3.metadata.json"), + }) + require.NoError(t, err) + fs.Put(GetNamespacePath(renameTestBucket, "ns"), "alias", map[string][]byte{ + ExtendedKeyMetadata: aliasMeta, + ExtendedKeyEntryType: []byte(EntryTypeTable), + }) + + for _, name := range []string{"alias", "t"} { + status, resp := runHandler(t, m, fs.Client, "DeleteTable", deleteTableRequest(t, name)) + assert.Equal(t, http.StatusConflict, status, "%s: %+v", name, resp) + } + tablePath := GetTablePath(renameTestBucket, "ns", "t") + assert.NotNil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t")) + assert.NotNil(t, fs.Get(path.Join(tablePath, "metadata"), "v3.metadata.json")) +} + +// rootRaceClient lands a rival record over the same root the moment a write +// reaches the record at dir/name, the way a second create that read the root +// free before this record existed publishes its own. +type rootRaceClient struct { + filer_pb.SeaweedFilerClient + f *s3tablestest.MemFiler + dir string + name string + location string + planted bool + then func() +} + +func (c *rootRaceClient) plant() { + if c.planted { + return + } + c.planted = true + if c.then != nil { + defer c.then() + } + rival, _ := json.Marshal(tableMetadataInternal{Name: "rival", Namespace: "ns", Format: FormatIceberg, MetadataLocation: c.location}) + c.f.Put(c.dir, "rival", map[string][]byte{ + ExtendedKeyMetadata: rival, + ExtendedKeyEntryType: []byte(EntryTypeTable), + ExtendedKeyDeleting: []byte(createClaim), + }) +} + +func (c *rootRaceClient) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + resp, err := c.SeaweedFilerClient.UpdateEntry(ctx, req, opts...) + if err == nil && req.Directory == c.dir && req.Entry.Name == c.name { + c.plant() + } + return resp, err +} + +// listFailClient fails every listing once a write reaches the record at +// dir/name, so the check that follows the write cannot run. +type listFailClient struct { + filer_pb.SeaweedFilerClient + dir string + name string + failing bool +} + +func (c *listFailClient) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + resp, err := c.SeaweedFilerClient.UpdateEntry(ctx, req, opts...) + if err == nil && req.Directory == c.dir && req.Entry.Name == c.name { + c.failing = true + } + return resp, err +} + +func (c *listFailClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { + if c.failing { + return nil, errors.New("listing unavailable") + } + return c.SeaweedFilerClient.ListEntries(ctx, in, opts...) +} + +// Two writers that both read a root free each check it again once their record +// is written, so the one that sees the other backs out instead of both landing. +func TestLocationRaceBacksOut(t *testing.T) { + nsPath := GetNamespacePath(renameTestBucket, "ns") + location := renameLocation("ns/shared/metadata/v1.metadata.json") + + t.Run("create", func(t *testing.T) { + fs, m := startRenameManager(t) + client := &rootRaceClient{SeaweedFilerClient: fs.Client, f: fs, dir: nsPath, name: "mine", location: location} + status, resp := runHandler(t, m, client, "CreateTable", &CreateTableRequest{ + TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "mine", Format: FormatIceberg, MetadataLocation: location, + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + require.True(t, client.planted) + if entry := fs.Get(nsPath, "mine"); entry != nil { + assert.NotContains(t, entry.Extended, ExtendedKeyMetadata) + } + }) + + t.Run("update", func(t *testing.T) { + fs, m := startRenameManager(t) + before := fs.Get(nsPath, "t").Extended[ExtendedKeyMetadata] + client := &rootRaceClient{SeaweedFilerClient: fs.Client, f: fs, dir: nsPath, name: "t", location: location} + status, resp := runHandler(t, m, client, "UpdateTable", &UpdateTableRequest{ + TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "t", MetadataVersion: 4, MetadataLocation: location, + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + require.True(t, client.planted) + assert.Equal(t, string(before), string(fs.Get(nsPath, "t").Extended[ExtendedKeyMetadata])) + assert.NotContains(t, fs.Get(nsPath, "t").Extended, ExtendedKeyMoving) + }) + + // A rival reading the old root while the move is checked still finds it + // held, so a move that backs out never hands back a root someone took. + for _, operation := range []string{"UpdateTable", "UpdateView"} { + t.Run(operation+" keeps its old root while moving", func(t *testing.T) { + fs, m := startRenameManager(t) + oldLocation := renameLocation("ns/old/metadata/v1.metadata.json") + var update interface{} + if operation == "UpdateTable" { + requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "a", Format: FormatIceberg, MetadataLocation: oldLocation}) + update = &UpdateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "a", MetadataVersion: 2, MetadataLocation: location} + } else { + requireOK(t, m, fs, "CreateView", &CreateViewRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "a", MetadataLocation: oldLocation}) + update = &UpdateViewRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "a", MetadataVersion: 2, MetadataLocation: location} + } + before := fs.Get(nsPath, "a").Extended[ExtendedKeyMetadata] + var oldRootErr error + client := &rootRaceClient{SeaweedFilerClient: fs.Client, f: fs, dir: nsPath, name: "a", location: location} + client.then = func() { + oldRootErr = checkLocationUnclaimed(context.Background(), fs.Client, renameTestBucket, path.Join(nsPath, "b"), oldLocation) + } + status, resp := runHandler(t, m, client, operation, update) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + require.True(t, client.planted) + assert.ErrorIs(t, oldRootErr, ErrLocationInUse) + assert.Equal(t, string(before), string(fs.Get(nsPath, "a").Extended[ExtendedKeyMetadata])) + assert.NotContains(t, fs.Get(nsPath, "a").Extended, ExtendedKeyMoving) + }) + } + + // Another update of the record landing while a move is checked cannot + // leave the record on the root a racing create took. + for _, other := range []string{"ns/shared/metadata/v2.metadata.json", "ns/old/metadata/v2.metadata.json"} { + t.Run("update moving under an update to "+other, func(t *testing.T) { + fs, m := startRenameManager(t) + requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "a", Format: FormatIceberg, MetadataLocation: renameLocation("ns/old/metadata/v1.metadata.json")}) + client := &rootRaceClient{SeaweedFilerClient: fs.Client, f: fs, dir: nsPath, name: "a", location: location} + client.then = func() { + runHandler(t, m, fs.Client, "UpdateTable", &UpdateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "a", MetadataVersion: 3, MetadataLocation: renameLocation(other)}) + } + status, resp := runHandler(t, m, client, "UpdateTable", &UpdateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "a", MetadataVersion: 2, MetadataLocation: location}) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + require.True(t, client.planted) + assert.False(t, RootClaimed(mustRoots(t, fs), path.Join(nsPath, "rival"), path.Join(nsPath, "shared"))) + assert.NotContains(t, fs.Get(nsPath, "a").Extended, ExtendedKeyMoving) + }) + } + + t.Run("update with a failed check", func(t *testing.T) { + fs, m := startRenameManager(t) + before := fs.Get(nsPath, "t").Extended[ExtendedKeyMetadata] + client := &listFailClient{SeaweedFilerClient: fs.Client, dir: nsPath, name: "t"} + status, resp := runHandler(t, m, client, "UpdateTable", &UpdateTableRequest{ + TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "t", MetadataVersion: 4, MetadataLocation: location, + }) + assert.NotEqual(t, http.StatusOK, status, "%+v", resp) + require.True(t, client.failing) + assert.Equal(t, string(before), string(fs.Get(nsPath, "t").Extended[ExtendedKeyMetadata])) + assert.NotContains(t, fs.Get(nsPath, "t").Extended, ExtendedKeyMoving) + }) + + t.Run("rename", func(t *testing.T) { + fs, m := startRenameManager(t) + requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "c", Format: FormatIceberg}) + client := &rootRaceClient{SeaweedFilerClient: fs.Client, f: fs, dir: nsPath, name: "dest", location: renameLocation("ns/dest/metadata/v1.metadata.json")} + status, resp := runHandler(t, m, client, "RenameTable", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), SourceNamespace: []string{"ns"}, SourceName: "c", DestNamespace: []string{"ns"}, DestName: "dest", + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + require.True(t, client.planted) + assert.Contains(t, fs.Get(nsPath, "c").Extended, ExtendedKeyMetadata) + if entry := fs.Get(nsPath, "dest"); entry != nil { + assert.NotContains(t, entry.Extended, ExtendedKeyMetadata) + } + }) +} + +// A move reserves its new root on the record, so a second move of the record +// waits for it, and an update that keeps the root clears a reservation a +// crashed move left behind. +func TestMoveReservationLeftBehind(t *testing.T) { + fs, m := startRenameManager(t) + nsPath := GetNamespacePath(renameTestBucket, "ns") + bucketARN := mustBucketARN(t) + entry := fs.Get(nsPath, "t") + leftover, _ := json.Marshal(tableMetadataInternal{Name: "t", Namespace: "ns", MetadataLocation: renameLocation("ns/elsewhere/metadata/v4.metadata.json")}) + entry.Extended[ExtendedKeyMoving] = leftover + fs.Put(nsPath, "t", entry.Extended) + + refused := renameLocation("ns/moved/metadata/v1.metadata.json") + assert.True(t, RootClaimed(mustRoots(t, fs), path.Join(nsPath, "other"), path.Join(nsPath, "elsewhere"))) + status, resp := runHandler(t, m, fs.Client, "UpdateTable", &UpdateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "t", MetadataVersion: 4, MetadataLocation: refused}) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + + requireOK(t, m, fs, "UpdateTable", &UpdateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "t", MetadataVersion: 4, MetadataLocation: renameLocation("ns/t/metadata/v4.metadata.json")}) + assert.NotContains(t, fs.Get(nsPath, "t").Extended, ExtendedKeyMoving) + assert.False(t, RootClaimed(mustRoots(t, fs), path.Join(nsPath, "other"), path.Join(nsPath, "elsewhere"))) + + requireOK(t, m, fs, "UpdateTable", &UpdateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "t", MetadataVersion: 5, MetadataLocation: refused}) + assert.NotContains(t, fs.Get(nsPath, "t").Extended, ExtendedKeyMoving) +} + +func mustRoots(t *testing.T, fs *s3tablestest.MemFiler) map[string][]string { + t.Helper() + roots, err := CatalogRecordRoots(context.Background(), fs.Client, renameTestBucket) + require.NoError(t, err) + return roots +} + +// A table that keeps its files elsewhere still owns its catalog entry: a view +// recorded at that entry would take it with the view's files on drop. +func TestLocationRefusesAnotherRecordsEntry(t *testing.T) { + fs, m := startRenameManager(t) + bucketARN := mustBucketARN(t) + requireOK(t, m, fs, "CreateTable", &CreateTableRequest{ + TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "a", Format: FormatIceberg, + MetadataLocation: renameLocation("ns/storage/metadata/v1.metadata.json"), + }) + status, resp := runHandler(t, m, fs.Client, "CreateView", &CreateViewRequest{ + TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "v", MetadataLocation: renameLocation("ns/a/metadata/v1.metadata.json"), + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) +} + +// A record taken away by a rename or a delete while a move holds its +// reservation leaves the old name free: the move can no longer clear the +// reservation once the record is gone. +func TestRecordGoneDropsItsMoveReservation(t *testing.T) { + nsPath := GetNamespacePath(renameTestBucket, "ns") + reserve := func(t *testing.T, fs *s3tablestest.MemFiler, name string) { + t.Helper() + moving, _ := json.Marshal(tableMetadataInternal{Name: name, Namespace: "ns", Format: FormatIceberg, MetadataLocation: renameLocation("ns/target/metadata/v9.metadata.json")}) + fs.Get(nsPath, name).Extended[ExtendedKeyMoving] = moving + } + reuse := func(t *testing.T, m *Manager, fs *s3tablestest.MemFiler, name string) { + t.Helper() + assert.NotContains(t, fs.Get(nsPath, name).Extended, ExtendedKeyMoving) + requireOK(t, m, fs, "CreateTable", &CreateTableRequest{ + TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: name, Format: FormatIceberg, + MetadataLocation: renameLocation("ns/fresh-" + name + "/metadata/v1.metadata.json"), + }) + } + + t.Run("rename", func(t *testing.T) { + fs, m := startRenameManager(t) + reserve(t, fs, "t") + requireOK(t, m, fs, "RenameTable", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), SourceNamespace: []string{"ns"}, SourceName: "t", DestNamespace: []string{"ns"}, DestName: "u", + }) + assert.NotContains(t, fs.Get(nsPath, "u").Extended, ExtendedKeyMoving) + reuse(t, m, fs, "t") + }) + + t.Run("delete", func(t *testing.T) { + fs, m := startRenameManager(t) + requireOK(t, m, fs, "CreateTable", &CreateTableRequest{ + TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "d", Format: FormatIceberg, + MetadataLocation: renameLocation("ns/d-files/metadata/v1.metadata.json"), + }) + reserve(t, fs, "d") + requireOK(t, m, fs, "DeleteTable", &DeleteTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "d"}) + reuse(t, m, fs, "d") + }) +} diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 3d036df2f6..77c0fdbe28 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -112,6 +112,9 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) return err } + if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, tablePath, req.MetadataLocation); err != nil { + return err + } // Create the table now := time.Now() @@ -164,7 +167,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque if len(tagsBytes) > 0 { attrs[ExtendedKeyTags] = tagsBytes } - if err := h.claimCatalogEntry(r.Context(), client, tablePath, attrs); err != nil { + if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, tablePath, req.MetadataLocation, attrs, dataPath, tablePath); err != nil { return err } return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceCreatedAt(target.namespaceCreatedAt), tablePath, attrs, dataPath, tablePath) @@ -179,6 +182,10 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusConflict, ErrCodeConflict, "table changed during create, retry the request") return err } + if errors.Is(err, ErrLocationInUse) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) + return err + } if h.writeNamespaceClosedError(w, err, namespaceName) { return err } @@ -281,6 +288,9 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check table: %v", err)) return err } + if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, tablePath, req.MetadataLocation); err != nil { + return err + } now := time.Now() versionToken := generateVersionToken() @@ -311,7 +321,7 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req if err := h.ensureDirectory(r.Context(), client, tablePath); err != nil { return err } - if err := h.claimCatalogEntry(r.Context(), client, tablePath, attrs); err != nil { + if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, tablePath, req.MetadataLocation, attrs, tablePath); err != nil { return err } return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceCreatedAt(target.namespaceCreatedAt), tablePath, attrs, tablePath) @@ -325,6 +335,10 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req h.writeError(w, http.StatusConflict, ErrCodeConflict, "table changed during register, retry the request") return err } + if errors.Is(err, ErrLocationInUse) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) + return err + } if h.writeNamespaceClosedError(w, err, namespaceName) { return err } @@ -1047,6 +1061,9 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque // Delete the table err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + if err := checkLocationUnclaimed(r.Context(), client, bucketName, tablePath, metadata.MetadataLocation); err != nil { + return err + } dataPath := TableDataDirFromMetadataLocation(metadata.MetadataLocation) if dataPath != "" && dataPath != tablePath && strings.HasPrefix(dataPath+"/", GetTableBucketPath(bucketName)+"/") { // Refuse to purge a data path that is an ancestor of the table's own @@ -1116,7 +1133,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque } err := h.removeExtendedAttributesIf(context.WithoutCancel(r.Context()), client, tablePath, claimed, ExtendedKeyMetadata, ExtendedKeyMetadataVersion, ExtendedKeyPolicy, ExtendedKeyTags, ExtendedKeyEntryType, - ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyDeleting) + ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyDeleting, ExtendedKeyMoving) if err != nil { if errors.Is(err, ErrConcurrentUpdate) { return ErrVersionTokenMismatch @@ -1159,7 +1176,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque } return h.removeExtendedAttributesIf(context.WithoutCancel(r.Context()), client, tablePath, claimed, ExtendedKeyMetadata, ExtendedKeyMetadataVersion, ExtendedKeyPolicy, ExtendedKeyTags, ExtendedKeyEntryType, - ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyDeleting) + ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyDeleting, ExtendedKeyMoving) } // Colocated table: the name path holds the data. A token caller's // delete carries the captured attributes as a filer precondition, so @@ -1218,6 +1235,10 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusConflict, ErrCodeConflict, "table changed during delete, retry the request") return err } + if errors.Is(err, ErrLocationInUse) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) + return err + } if errors.Is(err, filer_pb.ErrNotFound) { h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) return err @@ -1230,8 +1251,9 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque return nil } -// renamedTableAttributes are the catalog attributes a rename carries to the new -// name and clears from the old one. +// renamedTableAttributes are the catalog attributes a rename clears from the +// old name. All but a move's reservation, which the move can no longer clear +// once the record is gone, are carried to the new name. var renamedTableAttributes = []string{ ExtendedKeyMetadata, ExtendedKeyMetadataVersion, @@ -1240,6 +1262,7 @@ var renamedTableAttributes = []string{ ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyEntryType, + ExtendedKeyMoving, } // catalogEntryKind describes the entry a rename operates on, so tables and @@ -1688,6 +1711,11 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ if err := h.refuseClaimedNamespace(w, r, filerClient, destNamespacePath, destNamespace, isDestNamespace); err != nil { return err } + if metadata.MetadataLocation == "" { + if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, destPath, ""); err != nil { + return err + } + } metadata.Name = destName metadata.Namespace = destNamespace @@ -1767,6 +1795,11 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ h.removeLeftoverDirectory(ctx, client, destPath) return releaseClaim(err) } + if metadata.MetadataLocation == "" { + if err := checkLocationUnclaimed(r.Context(), client, bucketName, destPath, ""); err != nil { + return backOutDest(err) + } + } destParent, err := h.openParent(r.Context(), client, destNamespacePath, isDestNamespace) if err != nil { return backOutDest(err) @@ -1832,7 +1865,7 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ } } for key, data := range srcExtended { - if key != ExtendedKeyDeleting { + if key != ExtendedKeyDeleting && key != ExtendedKeyMoving { extended[key] = data } } @@ -1893,6 +1926,10 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ h.writeError(w, http.StatusNotFound, kind.notFoundCode, fmt.Sprintf("%s %s not found", kind.noun, srcName)) return err } + if errors.Is(err, ErrLocationInUse) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) + return err + } if h.writeNamespaceClosedError(w, err, destNamespace) { return err } @@ -2104,6 +2141,12 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque if req.ClearFullMetadata && metadata.Metadata != nil { metadata.Metadata.FullMetadata = nil } + movesRoot := req.MetadataLocation != "" && CatalogRecordRoot(target.bucketName, target.tablePath, req.MetadataLocation) != CatalogRecordRoot(target.bucketName, target.tablePath, metadata.MetadataLocation) + if movesRoot { + if err := h.refuseClaimedLocation(w, r, filerClient, target.bucketName, target.tablePath, req.MetadataLocation); err != nil { + return err + } + } if req.MetadataLocation != "" { metadata.MetadataLocation = req.MetadataLocation } @@ -2127,7 +2170,7 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque // snapshot. mutateEntryExtended retries on a changed entry, so the check // lives in the mutation, where it sees the value that is current now. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.mutateCatalogRecord(r.Context(), client, target.tablePath, func(extended map[string][]byte) error { + write := func(extended map[string][]byte) error { if !bytes.Equal(extended[ExtendedKeyMetadata], target.storedMetadata) { return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata) } @@ -2139,11 +2182,20 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy) } extended[ExtendedKeyMetadata] = metadataBytes + delete(extended, ExtendedKeyMoving) return nil - }) + } + if !movesRoot { + return h.mutateCatalogRecord(r.Context(), client, target.tablePath, write) + } + return h.moveCatalogRecord(r.Context(), client, target.bucketName, target.tablePath, metadata.MetadataLocation, target.storedMetadata, metadataBytes, write) }) if err != nil { + if errors.Is(err, ErrLocationInUse) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) + return err + } if errors.Is(err, ErrConcurrentUpdate) { h.writeError(w, http.StatusConflict, ErrCodeConflict, "table was updated concurrently") return ErrVersionTokenMismatch @@ -2210,23 +2262,13 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re return nil, err } - // Check if namespace exists + // A namespace that is missing or unreadable is authorized against the + // bucket alone, and reported only to a caller that passes. namespacePath := GetNamespacePath(bucketName, namespaceName) - nsMetadata, err := h.loadNamespaceMetadata(r.Context(), filerClient, bucketName, namespaceName) - namespaceExists := err == nil - if err != nil { - if errors.Is(err, filer_pb.ErrNotFound) && allowMissingNamespace { - // An implicit namespace can be authorized before it is declared: - // evaluate the grant against the bucket alone. - err = nil - } else { - if errors.Is(err, filer_pb.ErrNotFound) { - h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", namespaceName)) - } else { - h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check namespace: %v", err)) - } - return nil, err - } + nsMetadata, namespaceErr := h.loadNamespaceMetadata(r.Context(), filerClient, bucketName, namespaceName) + namespaceExists := namespaceErr == nil + if errors.Is(namespaceErr, filer_pb.ErrNotFound) && allowMissingNamespace { + namespaceErr = nil } // Authorize table creation using policy framework (namespace + bucket policies) @@ -2238,6 +2280,11 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re var bucketMetadata tableBucketMetadata err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + if !namespaceExists { + if _, err := h.lookupEntry(r.Context(), client, bucketPath); errors.Is(err, filer_pb.ErrNotFound) { + return nil + } + } // Fetch bucket metadata to use correct owner for bucket policy evaluation data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata) if err == nil { @@ -2309,6 +2356,14 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create table in this namespace") return nil, ErrAccessDenied } + if namespaceErr != nil { + if errors.Is(namespaceErr, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", namespaceName)) + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to check namespace") + } + return nil, namespaceErr + } if namespaceExists { if err := h.refuseClaimedNamespace(w, r, filerClient, namespacePath, namespaceName, isNamespaceCreatedAt(nsMetadata.CreatedAt)); err != nil { diff --git a/weed/s3api/s3tables/handler_view.go b/weed/s3api/s3tables/handler_view.go index e8f6dadb61..b949bc2192 100644 --- a/weed/s3api/s3tables/handler_view.go +++ b/weed/s3api/s3tables/handler_view.go @@ -138,6 +138,9 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) return err } + if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, viewPath, req.MetadataLocation); err != nil { + return err + } now := time.Now() versionToken := generateVersionToken() @@ -169,7 +172,7 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques ExtendedKeyEntryType: []byte(EntryTypeView), ExtendedKeyDeleting: []byte(createClaim), } - if err := h.claimCatalogEntry(r.Context(), client, viewPath, attrs); err != nil { + if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, viewPath, req.MetadataLocation, attrs, viewPath); err != nil { return err } return h.confirmParent(r.Context(), client, namespacePath, isNamespace, viewPath, attrs, viewPath) @@ -179,6 +182,10 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques h.writeError(w, http.StatusConflict, ErrCodeViewAlreadyExists, fmt.Sprintf("view %s already exists", viewName)) return err } + if errors.Is(err, ErrLocationInUse) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) + return err + } if errors.Is(err, ErrConcurrentUpdate) { h.writeError(w, http.StatusConflict, ErrCodeConflict, "view changed during create, retry the request") return err @@ -551,6 +558,12 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques metadata.Metadata.FullMetadata = req.Metadata.FullMetadata } } + movesRoot := req.MetadataLocation != "" && CatalogRecordRoot(target.bucketName, viewPath, req.MetadataLocation) != CatalogRecordRoot(target.bucketName, viewPath, metadata.MetadataLocation) + if movesRoot { + if err := h.refuseClaimedLocation(w, r, filerClient, target.bucketName, viewPath, req.MetadataLocation); err != nil { + return err + } + } if req.MetadataLocation != "" { metadata.MetadataLocation = req.MetadataLocation } @@ -573,7 +586,7 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques // and a view deleted and created again under the name would take a write // nobody authorized for it. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.mutateCatalogRecord(r.Context(), client, viewPath, func(extended map[string][]byte) error { + write := func(extended map[string][]byte) error { if !bytes.Equal(extended[ExtendedKeyMetadata], target.storedMetadata) { return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata) } @@ -581,10 +594,19 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy) } extended[ExtendedKeyMetadata] = metadataBytes + delete(extended, ExtendedKeyMoving) return nil - }) + } + if !movesRoot { + return h.mutateCatalogRecord(r.Context(), client, viewPath, write) + } + return h.moveCatalogRecord(r.Context(), client, target.bucketName, viewPath, metadata.MetadataLocation, target.storedMetadata, metadataBytes, write) }) if err != nil { + if errors.Is(err, ErrLocationInUse) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) + return err + } if errors.Is(err, ErrConcurrentUpdate) { h.writeError(w, http.StatusConflict, ErrCodeConflict, "view changed concurrently, retry the request") return err diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go index 0572d74a36..da32f986ce 100644 --- a/weed/s3api/s3tables/manager.go +++ b/weed/s3api/s3tables/manager.go @@ -128,7 +128,18 @@ func (m *Manager) AuthorizeCreateTable(ctx context.Context, filerClient FilerCli recorder := httptest.NewRecorder() target, authErr := m.handler.authorizeCreateTable(recorder, httpReq, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags, req.AllowMissingNamespace) if authErr == nil { - return m.handler.checkVectorNamespaceWrite(ctx, filerClient, target.bucketName, req.Namespace) + if err := m.handler.checkVectorNamespaceWrite(ctx, filerClient, target.bucketName, req.Namespace); err != nil { + return err + } + if req.MetadataLocation == "" { + return nil + } + tablePath := GetTablePath(target.bucketName, target.namespaceName, target.tableName) + if authErr = ValidateMetadataLocation(req.MetadataLocation, target.bucketName, target.namespaceName); authErr != nil { + m.handler.writeError(recorder, http.StatusBadRequest, ErrCodeInvalidRequest, authErr.Error()) + } else if authErr = m.handler.refuseClaimedLocation(recorder, httpReq, filerClient, target.bucketName, tablePath, req.MetadataLocation); authErr == nil { + return nil + } } if decoded := decodeS3TablesHTTPResponse(recorder, nil); decoded != nil { return decoded diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go index 3737311836..5ae4f8f7f8 100644 --- a/weed/worker/tasks/iceberg/detection.go +++ b/weed/worker/tasks/iceberg/detection.go @@ -77,6 +77,7 @@ func (h *Handler) scanTablesForMaintenance( // List namespaces within the bucket bucketPath := path.Join(bucketsPath, bucketName) + var recordRoots map[string][]string nsEntries, err := listFilerEntries(ctx, filerClient, bucketPath, "") if err != nil { glog.Warningf("iceberg maintenance: failed to list namespaces in bucket %s: %v", bucketName, err) @@ -135,6 +136,18 @@ func (h *Handler) scanTablesForMaintenance( } continue } + // Maintenance deletes what a table's metadata does not reference, + // so a table sharing its files with another record is left alone. + if recordRoots == nil { + if recordRoots, err = s3tables.CatalogRecordRoots(ctx, filerClient, bucketName); err != nil { + glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot list catalog records: %v", bucketName, nsName, tblName, err) + continue + } + } + if filesShared(recordRoots, path.Join(bucketPath, tablePath), path.Join(bucketPath, state.DataPath)) { + glog.V(1).Infof("iceberg maintenance: skipping %s/%s/%s: its location is shared with another table or view", bucketName, nsName, tblName) + continue + } if err := state.loadMetadata(ctx, filerClient, bucketName); err != nil { glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot load iceberg metadata: %v", bucketName, nsName, tblName, err) continue @@ -185,6 +198,23 @@ func (h *Handler) scanTablesForMaintenance( return tables, nil } +// filesShared reports whether a record other than selfPath keeps its files at +// root, above it or under it. A record that only sits on root, as one created +// over a renamed table's old name does, keeps no files there. +func filesShared(roots map[string][]string, selfPath, root string) bool { + for recordPath, others := range roots { + if recordPath == selfPath { + continue + } + for _, other := range others { + if s3tables.PathsOverlap(other, root) { + return true + } + } + } + return false +} + func normalizeDetectionConfig(config Config) Config { config = applyThresholdDefaults(config) if config.SnapshotRetentionMs <= 0 { diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go index 8b7bb22f24..3c0d2b1bf4 100644 --- a/weed/worker/tasks/iceberg/operations.go +++ b/weed/worker/tasks/iceberg/operations.go @@ -312,6 +312,14 @@ func (h *Handler) removeOrphans( if err != nil { return "", nil, fmt.Errorf("load metadata: %w", err) } + roots, err := s3tables.CatalogRecordRoots(ctx, filerClient, bucketName) + if err != nil { + return "", nil, fmt.Errorf("list catalog records: %w", err) + } + bucketPath := path.Join(s3tables.TablesPath, bucketName) + if filesShared(roots, path.Join(bucketPath, tablePath), path.Join(bucketPath, state.DataPath)) { + return "skipped: location shared with another table or view", nil, nil + } orphanCandidates, err := collectOrphanCandidates(ctx, filerClient, bucketName, state.DataPath, state.Metadata, state.MetadataFileName, config.OrphanOlderThanHours) if err != nil { diff --git a/weed/worker/tasks/iceberg/shared_location_test.go b/weed/worker/tasks/iceberg/shared_location_test.go new file mode 100644 index 0000000000..1de4be36f6 --- /dev/null +++ b/weed/worker/tasks/iceberg/shared_location_test.go @@ -0,0 +1,98 @@ +package iceberg + +import ( + "context" + "path" + "testing" + "time" + + "github.com/apache/iceberg-go/table" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "google.golang.org/protobuf/proto" +) + +// Maintenance deletes the files a table's metadata does not reference, so a +// table whose location another record also points at is not maintained. +func TestDetectSkipsSharedLocation(t *testing.T) { + fs, client := startFakeFiler(t) + now := time.Now().UnixMilli() + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "analytics", + TableName: "events", + DataPath: "analytics/events", + Snapshots: []table.Snapshot{ + {SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"}, + {SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"}, + {SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"}, + }, + Age: 200 * time.Hour, + } + populateTable(t, fs, setup) + config := Config{SnapshotRetentionMs: hoursToMs(0), MaxSnapshotsToKeep: 2, MaxCommitRetries: 3} + + tables, err := NewHandler(nil).scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0) + if err != nil || len(tables) != 1 { + t.Fatalf("before the alias: tables = %d, err = %v, want 1 table", len(tables), err) + } + + nsPath := path.Join(s3tables.TablesPath, setup.BucketName, setup.Namespace) + alias := proto.Clone(fs.getEntry(nsPath, setup.TableName)).(*filer_pb.Entry) + alias.Name = "alias" + fs.putEntry(nsPath, alias.Name, alias) + + tables, err = NewHandler(nil).scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0) + if err != nil { + t.Fatalf("scanTablesForMaintenance: %v", err) + } + if len(tables) != 0 { + t.Fatalf("tables sharing one location were scheduled: %+v", tables) + } +} + +// A table created over a renamed table's old name sits on that table's files +// but keeps its own elsewhere, so the renamed table is still maintained. +func TestFilesSharedIgnoresAnEntryOverTheRoot(t *testing.T) { + roots := map[string][]string{ + "/buckets/b/ns2/moved": {"/buckets/b/ns/old"}, + "/buckets/b/ns/old": {"/buckets/b/ns/old-1"}, + } + if filesShared(roots, "/buckets/b/ns2/moved", "/buckets/b/ns/old") { + t.Error("an entry over the root counted as sharing its files") + } + if !filesShared(roots, "/buckets/b/ns/other", "/buckets/b/ns/old/data") { + t.Error("a root inside another record's files was not shared") + } +} + +// A job scheduled before another record came to share the table's location +// leaves the files alone when it runs. +func TestRemoveOrphansSkipsSharedLocation(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "analytics", + TableName: "events", + DataPath: "analytics/events", + Snapshots: []table.Snapshot{{SnapshotID: 1, TimestampMs: time.Now().UnixMilli(), ManifestList: "metadata/snap-1.avro"}}, + } + populateTable(t, fs, setup) + dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data") + fs.putEntry(dataDir, "other.parquet", &filer_pb.Entry{ + Name: "other.parquet", + Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Add(-200 * time.Hour).Unix()}, + }) + nsPath := path.Join(s3tables.TablesPath, setup.BucketName, setup.Namespace) + alias := proto.Clone(fs.getEntry(nsPath, setup.TableName)).(*filer_pb.Entry) + alias.Name = "alias" + fs.putEntry(nsPath, alias.Name, alias) + + result, _, err := NewHandler(nil).removeOrphans(context.Background(), client, setup.BucketName, setup.tablePath(), Config{OrphanOlderThanHours: 72, MaxCommitRetries: 3}) + if err != nil { + t.Fatalf("removeOrphans: %v", err) + } + if fs.getEntry(dataDir, "other.parquet") == nil { + t.Fatalf("a file under a shared location was removed (%q)", result) + } +}