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) + } +}