diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 05659874d..14728fd35 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -621,6 +621,64 @@ func (s *Server) handleDropTable(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } +// handleRenameTable moves a table's catalog pointer to a new namespace/name. +func (s *Server) handleRenameTable(w http.ResponseWriter, r *http.Request) { + var req RenameTableRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body") + return + } + + source := parseNamespace(encodeNamespace(req.Source.Namespace)) + dest := parseNamespace(encodeNamespace(req.Destination.Namespace)) + if len(source) == 0 || req.Source.Name == "" || len(dest) == 0 || req.Destination.Name == "" { + writeError(w, http.StatusBadRequest, "BadRequestException", "source and destination namespace and name are required") + return + } + + bucketName := getBucketFromPrefix(r) + bucketARN := buildTableBucketARN(bucketName) + identityName := s3_constants.GetIdentityNameFromContext(r) + + renameReq := &s3tables.RenameTableRequest{ + TableBucketARN: bucketARN, + SourceNamespace: source, + SourceName: req.Source.Name, + DestNamespace: dest, + DestName: req.Destination.Name, + } + + err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "RenameTable", renameReq, nil, identityName) + }) + + if err != nil { + var tableErr *s3tables.S3TablesError + if errors.As(err, &tableErr) { + switch tableErr.Type { + case s3tables.ErrCodeNoSuchTable: + writeError(w, http.StatusNotFound, "NoSuchTableException", fmt.Sprintf("Table does not exist: %s", req.Source.Name)) + return + case s3tables.ErrCodeNoSuchNamespace: + writeError(w, http.StatusNotFound, "NoSuchNamespaceException", fmt.Sprintf("Namespace does not exist: %v", dest)) + return + case s3tables.ErrCodeTableAlreadyExists: + writeError(w, http.StatusConflict, "AlreadyExistsException", fmt.Sprintf("Table already exists: %s", req.Destination.Name)) + return + case s3tables.ErrCodeInvalidRequest: + writeError(w, http.StatusBadRequest, "BadRequestException", tableErr.Message) + return + } + } + glog.V(1).Infof("Iceberg: RenameTable error: %v", err) + writeManagerError(w, err) + return + } + + w.WriteHeader(http.StatusNoContent) +} + // isNoSuchTableError reports whether an error from the S3 Tables manager // indicates the target table is not registered in the catalog. func isNoSuchTableError(err error) bool { diff --git a/weed/s3api/iceberg/iceberg_rename_table_test.go b/weed/s3api/iceberg/iceberg_rename_table_test.go new file mode 100644 index 000000000..0529c75f2 --- /dev/null +++ b/weed/s3api/iceberg/iceberg_rename_table_test.go @@ -0,0 +1,32 @@ +package iceberg + +import ( + "net/http/httptest" + "strings" + "testing" +) + +func TestHandleRenameTableRejectsIncompleteIdentifiers(t *testing.T) { + cases := []struct { + name string + body string + }{ + {"missing source name", `{"source":{"namespace":["ns"]},"destination":{"namespace":["ns"],"name":"t2"}}`}, + {"missing source namespace", `{"source":{"name":"t"},"destination":{"namespace":["ns"],"name":"t2"}}`}, + {"missing destination name", `{"source":{"namespace":["ns"],"name":"t"},"destination":{"namespace":["ns"]}}`}, + {"missing destination namespace", `{"source":{"namespace":["ns"],"name":"t"},"destination":{"name":"t2"}}`}, + {"empty body", `{}`}, + {"malformed json", `{`}, + } + s := &Server{} + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := httptest.NewRequest("POST", "/v1/tables/rename", strings.NewReader(tc.body)) + w := httptest.NewRecorder() + s.handleRenameTable(w, r) + if w.Code != 400 { + t.Fatalf("handleRenameTable() status = %d, want 400", w.Code) + } + }) + } +} diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go index 8f6f29237..366853bf1 100644 --- a/weed/s3api/iceberg/server.go +++ b/weed/s3api/iceberg/server.go @@ -107,6 +107,7 @@ func (s *Server) RegisterRoutes(router *mux.Router) { router.HandleFunc("/v1/namespaces/{namespace}/tables/{table}", s.Auth(s.handleTableExists)).Methods(http.MethodHead) router.HandleFunc("/v1/namespaces/{namespace}/tables/{table}", s.Auth(s.handleDropTable)).Methods(http.MethodDelete) router.HandleFunc("/v1/namespaces/{namespace}/tables/{table}", s.Auth(s.handleUpdateTable)).Methods(http.MethodPost) + router.HandleFunc("/v1/tables/rename", s.Auth(s.handleRenameTable)).Methods(http.MethodPost) // View endpoints - wrapped with Auth middleware router.HandleFunc("/v1/namespaces/{namespace}/views", s.Auth(s.handleListViews)).Methods(http.MethodGet) @@ -133,6 +134,7 @@ func (s *Server) RegisterRoutes(router *mux.Router) { router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables/{table}", s.Auth(s.handleTableExists)).Methods(http.MethodHead) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables/{table}", s.Auth(s.handleDropTable)).Methods(http.MethodDelete) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables/{table}", s.Auth(s.handleUpdateTable)).Methods(http.MethodPost) + router.HandleFunc("/v1/{prefix}/tables/rename", s.Auth(s.handleRenameTable)).Methods(http.MethodPost) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views", s.Auth(s.handleListViews)).Methods(http.MethodGet) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views", s.Auth(s.handleCreateView)).Methods(http.MethodPost) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views/{view}", s.Auth(s.handleLoadView)).Methods(http.MethodGet) diff --git a/weed/s3api/iceberg/types.go b/weed/s3api/iceberg/types.go index 316d3b824..43f37913f 100644 --- a/weed/s3api/iceberg/types.go +++ b/weed/s3api/iceberg/types.go @@ -158,6 +158,12 @@ func (r *LoadTableResult) UnmarshalJSON(data []byte) error { return nil } +// RenameTableRequest is sent to POST /v1/tables/rename. +type RenameTableRequest struct { + Source TableIdentifier `json:"source"` + Destination TableIdentifier `json:"destination"` +} + // CommitTableRequest is sent to POST /v1/namespaces/{namespace}/tables/{table}. type CommitTableRequest struct { Identifier *TableIdentifier `json:"identifier,omitempty"` diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index cf032aa06..d3b8c6916 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -84,6 +84,32 @@ func (h *S3TablesHandler) setExtendedAttribute(ctx context.Context, client filer }) } +// removeExtendedAttributes deletes the given extended attributes from an entry, +// leaving the directory and its children intact. +func (h *S3TablesHandler) removeExtendedAttributes(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, keys ...string) error { + dir, name := splitPath(path) + + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ + Directory: dir, + Name: name, + }) + if err != nil { + return err + } + + entry := resp.Entry + if entry.Extended != nil { + for _, key := range keys { + delete(entry.Extended, key) + } + } + + return filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ + Directory: dir, + Entry: entry, + }) +} + // setExtendedAttributes sets multiple extended attributes on an existing entry // in a single UpdateEntry, so callers don't leave the entry partially tagged. func (h *S3TablesHandler) setExtendedAttributes(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, attrs map[string][]byte) error { diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index 101aec6a6..6013a20b5 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -40,6 +40,7 @@ const ( var ( ErrVersionTokenMismatch = errors.New("version token mismatch") ErrAccessDenied = errors.New("access denied") + ErrTableAlreadyExists = errors.New("table already exists") ) type ResourceType string @@ -158,6 +159,8 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request, err = h.handleUpdateTable(w, r, filerClient) case "DeleteTable": err = h.handleDeleteTable(w, r, filerClient) + case "RenameTable": + err = h.handleRenameTable(w, r, filerClient) // View operations case "CreateView": diff --git a/weed/s3api/s3tables/handler_rename_test.go b/weed/s3api/s3tables/handler_rename_test.go new file mode 100644 index 000000000..4507a0b62 --- /dev/null +++ b/weed/s3api/s3tables/handler_rename_test.go @@ -0,0 +1,354 @@ +package s3tables + +import ( + "context" + "encoding/json" + "net" + "path" + "sort" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" +) + +// memFilerServer is an in-memory filer used to drive Manager operations +// end-to-end without a live cluster. +type memFilerServer struct { + filer_pb.UnimplementedSeaweedFilerServer + entries map[string]map[string]*filer_pb.Entry // dir -> name -> entry + client filer_pb.SeaweedFilerClient +} + +func newMemFilerServer() *memFilerServer { + return &memFilerServer{entries: make(map[string]map[string]*filer_pb.Entry)} +} + +func (f *memFilerServer) getEntry(dir, name string) *filer_pb.Entry { + if d, ok := f.entries[dir]; ok { + return d[name] + } + return nil +} + +func (f *memFilerServer) putEntry(dir, name string, extended map[string][]byte) { + if _, ok := f.entries[dir]; !ok { + f.entries[dir] = make(map[string]*filer_pb.Entry) + } + f.entries[dir][name] = &filer_pb.Entry{Name: name, IsDirectory: true, Extended: extended} +} + +func (f *memFilerServer) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if e := f.getEntry(req.Directory, req.Name); e != nil { + return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil + } + // Carry the sentinel text so filer_pb.LookupEntry maps it to ErrNotFound. + return nil, status.Errorf(codes.NotFound, "%s: %s/%s", filer_pb.ErrNotFound.Error(), req.Directory, req.Name) +} + +func (f *memFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream grpc.ServerStreamingServer[filer_pb.ListEntriesResponse]) error { + d, ok := f.entries[req.Directory] + if !ok { + return nil + } + names := make([]string, 0, len(d)) + for name := range d { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: d[name]}); err != nil { + return err + } + } + return nil +} + +func (f *memFilerServer) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + if _, ok := f.entries[req.Directory]; !ok { + f.entries[req.Directory] = make(map[string]*filer_pb.Entry) + } + f.entries[req.Directory][req.Entry.Name] = req.Entry + return &filer_pb.CreateEntryResponse{}, nil +} + +func (f *memFilerServer) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + if _, ok := f.entries[req.Directory]; !ok { + f.entries[req.Directory] = make(map[string]*filer_pb.Entry) + } + f.entries[req.Directory][req.Entry.Name] = req.Entry + return &filer_pb.UpdateEntryResponse{}, nil +} + +func (f *memFilerServer) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { + if d, ok := f.entries[req.Directory]; ok { + delete(d, req.Name) + } + // Honor recursive data deletion so a regression that wipes the table directory + // also drops its metadata/ and data/ children (the data-loss this guards against). + if req.IsRecursive && req.IsDeleteData { + child := path.Join(req.Directory, req.Name) + for dir := range f.entries { + if dir == child || strings.HasPrefix(dir, child+"/") { + delete(f.entries, dir) + } + } + } + return &filer_pb.DeleteEntryResponse{}, nil +} + +func (f *memFilerServer) Ping(_ context.Context, _ *filer_pb.PingRequest) (*filer_pb.PingResponse, error) { + now := time.Now().UnixNano() + return &filer_pb.PingResponse{StartTimeNs: now, RemoteTimeNs: now, StopTimeNs: now}, nil +} + +func startMemFiler(t *testing.T) *memFilerServer { + t.Helper() + fs := newMemFilerServer() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + server := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(server, fs) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.GracefulStop) + + conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + fs.client = filer_pb.NewSeaweedFilerClient(conn) + deadline := time.Now().Add(5 * time.Second) + for { + pingCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + _, err := fs.client.Ping(pingCtx, &filer_pb.PingRequest{}) + cancel() + if err == nil { + break + } + require.False(t, time.Now().After(deadline), "filer not ready: %v", err) + time.Sleep(10 * time.Millisecond) + } + return fs +} + +const renameTestBucket = "renamebkt" + +func mustBucketARN(t *testing.T) string { + t.Helper() + arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, renameTestBucket) + require.NoError(t, err) + return arn +} + +// startRenameManager seeds a bucket/namespace/table and returns a trusted Manager. +func startRenameManager(t *testing.T) (*memFilerServer, *Manager) { + t.Helper() + fs := startMemFiler(t) + + bucketMeta, _ := json.Marshal(tableBucketMetadata{Name: renameTestBucket, OwnerAccountID: DefaultAccountID}) + fs.putEntry(TablesPath, renameTestBucket, map[string][]byte{ + ExtendedKeyTableBucket: []byte("{}"), + ExtendedKeyMetadata: bucketMeta, + }) + + nsMeta, _ := json.Marshal(namespaceMetadata{Namespace: []string{"ns"}, OwnerAccountID: DefaultAccountID}) + fs.putEntry(GetTableBucketPath(renameTestBucket), "ns", map[string][]byte{ExtendedKeyMetadata: nsMeta}) + + tableMeta, _ := json.Marshal(tableMetadataInternal{ + Name: "t", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + MetadataVersion: 3, + MetadataLocation: "s3://" + renameTestBucket + "/ns/t/metadata/v3.metadata.json", + }) + fs.putEntry(GetNamespacePath(renameTestBucket, "ns"), "t", map[string][]byte{ + ExtendedKeyMetadata: tableMeta, + ExtendedKeyMetadataVersion: []byte("3"), + }) + + // Physical metadata.json and data files live under the table directory. + tablePath := GetTablePath(renameTestBucket, "ns", "t") + fs.putEntry(tablePath, "metadata", nil) + fs.putEntry(tablePath, "data", nil) + fs.putEntry(path.Join(tablePath, "metadata"), "v3.metadata.json", nil) + + m := NewManager() + m.SetTrusted(true) + return fs, m +} + +func runRename(t *testing.T, m *Manager, fs *memFilerServer, req *RenameTableRequest) error { + t.Helper() + return m.Execute(context.Background(), NewManagerClient(fs.client), "RenameTable", req, nil, "") +} + +func runGetTable(t *testing.T, m *Manager, fs *memFilerServer, namespace, name string) (*GetTableResponse, error) { + t.Helper() + resp := &GetTableResponse{} + err := m.Execute(context.Background(), NewManagerClient(fs.client), "GetTable", &GetTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{namespace}, + Name: name, + }, resp, "") + return resp, err +} + +func TestRenameTablePreservesData(t *testing.T) { + fs, m := startRenameManager(t) + + req := &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "t2", + } + require.NoError(t, runRename(t, m, fs, req)) + + // The source directory and its metadata.json/data children must survive: rename + // is catalog-only and the destination still points at the original location. + srcPath := GetTablePath(renameTestBucket, "ns", "t") + assert.NotNil(t, fs.getEntry(srcPath, "metadata"), "source metadata dir must survive") + assert.NotNil(t, fs.getEntry(srcPath, "data"), "source data dir must survive") + assert.NotNil(t, fs.getEntry(path.Join(srcPath, "metadata"), "v3.metadata.json"), "metadata.json must survive") + + // Source catalog xattrs are dropped so the name stops resolving. + src := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t") + require.NotNil(t, src, "source directory must remain to hold the data children") + _, hasMeta := src.Extended[ExtendedKeyMetadata] + assert.False(t, hasMeta, "source table-metadata xattr must be removed") + + _, err := runGetTable(t, m, fs, "ns", "t") + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeNoSuchTable, s3Err.Type) + + // The destination resolves to the preserved (original) MetadataLocation. + got, err := runGetTable(t, m, fs, "ns", "t2") + require.NoError(t, err) + assert.Equal(t, "t2", got.Name) + assert.Equal(t, "s3://"+renameTestBucket+"/ns/t/metadata/v3.metadata.json", got.MetadataLocation) + + dest := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t2") + require.NotNil(t, dest) + assert.Equal(t, []byte("3"), dest.Extended[ExtendedKeyMetadataVersion]) +} + +func TestRenameTableSourceMissing(t *testing.T) { + fs, m := startRenameManager(t) + err := runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "ghost", + DestNamespace: []string{"ns"}, + DestName: "t2", + }) + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeNoSuchTable, s3Err.Type) +} + +func TestRenameTableDestExists(t *testing.T) { + fs, m := startRenameManager(t) + existing, _ := json.Marshal(tableMetadataInternal{Name: "t2", Namespace: "ns", OwnerAccountID: DefaultAccountID}) + fs.putEntry(GetNamespacePath(renameTestBucket, "ns"), "t2", map[string][]byte{ExtendedKeyMetadata: existing}) + + err := runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "t2", + }) + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeTableAlreadyExists, s3Err.Type) + assert.NotNil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t"), "source must be untouched on conflict") +} + +func TestRenameTableDestNamespaceMissing(t *testing.T) { + fs, m := startRenameManager(t) + err := runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"other"}, + DestName: "t2", + }) + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeNoSuchNamespace, s3Err.Type) + assert.NotNil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t"), "source must be untouched") +} + +// A principal allowed to rename the source must still be denied when it cannot +// create a table in the destination namespace. +func TestRenameTableDestNamespaceUnauthorized(t *testing.T) { + fs, m := startRenameManager(t) + m.SetTrusted(false) + m.SetDefaultAllow(false) + + // "mover" may rename the source table but holds no rights on "dest". + srcPolicy, _ := json.Marshal(map[string]interface{}{ + "Version": "2012-10-17", + "Statement": []map[string]interface{}{{ + "Effect": "Allow", + "Principal": "mover", + "Action": "s3tables:RenameTable", + "Resource": "*", + }}, + }) + srcEntry := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t") + require.NotNil(t, srcEntry) + srcEntry.Extended[ExtendedKeyPolicy] = srcPolicy + + destNsMeta, _ := json.Marshal(namespaceMetadata{Namespace: []string{"dest"}, OwnerAccountID: DefaultAccountID}) + fs.putEntry(GetTableBucketPath(renameTestBucket), "dest", map[string][]byte{ExtendedKeyMetadata: destNsMeta}) + + mover := &testIdentity{Name: "mover", Account: &testIdentityAccount{Id: "mover"}} + ctx := s3_constants.SetIdentityInContext(context.Background(), mover) + err := m.Execute(ctx, NewManagerClient(fs.client), "RenameTable", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"dest"}, + DestName: "t2", + }, nil, "mover") + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeAccessDenied, s3Err.Type) + + assert.NotNil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t"), "source must be untouched") + assert.Nil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "dest"), "t2"), "destination must not be written") +} + +func TestRenameTableInvalidName(t *testing.T) { + fs, m := startRenameManager(t) + err := runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "Bad/Name", + }) + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeInvalidRequest, s3Err.Type) +} diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 9e514db3d..e2144a9b2 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -556,7 +556,8 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request, }) if err != nil { - if errors.Is(err, filer_pb.ErrNotFound) { + // A directory without the table-metadata xattr is not a table (e.g. a renamed-away source). + if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) { h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) } else { h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to get table: %v", err)) @@ -1184,6 +1185,14 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque if err := h.deleteDirectory(r.Context(), client, tablePath); err != nil { return err } + // A renamed table keeps its data at the original location, so the catalog + // path no longer holds it; purge the data directory too when it differs. + dataPath := tableDataDirFromMetadataLocation(metadata.MetadataLocation) + if dataPath != "" && dataPath != tablePath && strings.HasPrefix(dataPath+"/", GetTableBucketPath(bucketName)+"/") { + if err := h.deleteDirectory(r.Context(), client, dataPath); err != nil { + return err + } + } return nil }) @@ -1196,6 +1205,262 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque return nil } +// handleRenameTable moves a table's catalog entry to a new namespace/name within +// the same bucket. It is catalog-only: the metadata.json and data files stay put, +// the destination keeps the source's MetadataLocation, and the source name is +// soft-deleted in place (its catalog xattrs are dropped, its data is left intact). +func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + var req RenameTableRequest + if err := h.readRequestBody(r, &req); err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + if req.TableBucketARN == "" || len(req.SourceNamespace) == 0 || req.SourceName == "" || len(req.DestNamespace) == 0 || req.DestName == "" { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "tableBucketARN, sourceNamespace, sourceName, destNamespace, and destName are required") + return fmt.Errorf("missing required parameters") + } + + bucketName, err := parseBucketNameFromARN(req.TableBucketARN) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + srcNamespace, err := validateNamespace(req.SourceNamespace) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + srcName, err := validateTableName(req.SourceName) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + destNamespace, err := validateNamespace(req.DestNamespace) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + destName, err := validateTableName(req.DestName) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + srcPath := GetTablePath(bucketName, srcNamespace, srcName) + destPath := GetTablePath(bucketName, destNamespace, destName) + + var metadata tableMetadataInternal + var metadataVersionXattr []byte + var tablePolicy string + var bucketPolicy string + var bucketTags map[string]string + var tableTags map[string]string + var bucketMetadata tableBucketMetadata + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + data, err := h.getExtendedAttribute(r.Context(), client, srcPath, ExtendedKeyMetadata) + if err != nil { + return err + } + if err := json.Unmarshal(data, &metadata); err != nil { + return fmt.Errorf("failed to unmarshal table metadata: %w", err) + } + + if versionData, err := h.getExtendedAttribute(r.Context(), client, srcPath, ExtendedKeyMetadataVersion); err == nil { + metadataVersionXattr = versionData + } else if !errors.Is(err, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch metadata version: %w", err) + } + + policyData, err := h.getExtendedAttribute(r.Context(), client, srcPath, ExtendedKeyPolicy) + if err == nil { + tablePolicy = string(policyData) + } else if !errors.Is(err, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch table policy: %w", err) + } + tableTags, err = h.readTags(r.Context(), client, srcPath) + if err != nil { + return err + } + + bucketPath := GetTableBucketPath(bucketName) + data, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata) + if err == nil { + if err := json.Unmarshal(data, &bucketMetadata); err != nil { + return fmt.Errorf("failed to unmarshal bucket metadata: %w", err) + } + } else if !errors.Is(err, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch bucket metadata: %w", err) + } + policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy) + if err == nil { + bucketPolicy = string(policyData) + } else if !errors.Is(err, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch bucket policy: %w", err) + } + bucketTags, err = h.readTags(r.Context(), client, bucketPath) + if err != nil { + return err + } + + return nil + }) + + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", srcName)) + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check table: %v", err)) + } + return err + } + + tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, srcNamespace+"/"+srcName) + bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName) + principal := h.getAccountID(r) + identityActions := getIdentityActions(r) + tableAllowed := CheckPermissionWithContext("RenameTable", principal, metadata.OwnerAccountID, tablePolicy, tableARN, &PolicyContext{ + TableBucketName: bucketName, + Namespace: srcNamespace, + TableName: srcName, + TableBucketTags: bucketTags, + ResourceTags: tableTags, + IdentityActions: identityActions, + DefaultAllow: h.defaultAllowFor(r), + }) + bucketAllowed := CheckPermissionWithContext("RenameTable", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{ + TableBucketName: bucketName, + Namespace: srcNamespace, + TableName: srcName, + TableBucketTags: bucketTags, + ResourceTags: tableTags, + IdentityActions: identityActions, + DefaultAllow: h.defaultAllowFor(r), + }) + if !tableAllowed && !bucketAllowed { + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to rename table") + return NewAuthError("RenameTable", principal, "not authorized to rename table") + } + + // Require the destination namespace to exist and the destination table to be free. + destNamespacePath := GetNamespacePath(bucketName, destNamespace) + var destNamespaceMetadata namespaceMetadata + var destNamespacePolicy string + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + data, err := h.getExtendedAttribute(r.Context(), client, destNamespacePath, ExtendedKeyMetadata) + if err != nil { + return err + } + if err := json.Unmarshal(data, &destNamespaceMetadata); err != nil { + return fmt.Errorf("failed to unmarshal destination namespace metadata: %w", err) + } + policyData, err := h.getExtendedAttribute(r.Context(), client, destNamespacePath, ExtendedKeyPolicy) + if err == nil { + destNamespacePolicy = string(policyData) + } else if !errors.Is(err, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch destination namespace policy: %w", err) + } + if _, err := h.getExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadata); err == nil { + return ErrTableAlreadyExists + } else if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) { + return err + } + return nil + }) + + if err != nil { + if errors.Is(err, ErrTableAlreadyExists) { + h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("table %s already exists", destName)) + } else if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", destNamespace)) + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check destination: %v", err)) + } + return err + } + + // Renaming places the table into the destination namespace, so the principal + // must also be allowed to create a table there (the source check alone lets a + // caller move tables into namespaces they don't control). + destNamespaceAllowed := CheckPermissionWithContext("CreateTable", principal, destNamespaceMetadata.OwnerAccountID, destNamespacePolicy, bucketARN, &PolicyContext{ + TableBucketName: bucketName, + Namespace: destNamespace, + TableName: destName, + TableBucketTags: bucketTags, + IdentityActions: identityActions, + DefaultAllow: h.defaultAllowFor(r), + }) + destBucketAllowed := CheckPermissionWithContext("CreateTable", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{ + TableBucketName: bucketName, + Namespace: destNamespace, + TableName: destName, + TableBucketTags: bucketTags, + IdentityActions: identityActions, + DefaultAllow: h.defaultAllowFor(r), + }) + if !destNamespaceAllowed && !destBucketAllowed { + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create table in the destination namespace") + return NewAuthError("RenameTable", principal, "not authorized to create table in the destination namespace") + } + + metadata.Name = destName + metadata.Namespace = destNamespace + metadata.ModifiedAt = time.Now() + + metadataBytes, err := json.Marshal(&metadata) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal table metadata") + return fmt.Errorf("failed to marshal metadata: %w", err) + } + + // Write the destination entry before deleting the source so a mid-rename + // failure can never lose the table. + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + if err := h.createDirectory(r.Context(), client, destPath); err != nil { + return err + } + if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadata, metadataBytes); err != nil { + return err + } + if len(metadataVersionXattr) > 0 { + if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadataVersion, metadataVersionXattr); err != nil { + return err + } + } + if len(tableTags) > 0 { + tagsBytes, err := json.Marshal(tableTags) + if err != nil { + return fmt.Errorf("failed to marshal tags: %w", err) + } + if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyTags, tagsBytes); err != nil { + return err + } + } + if tablePolicy != "" { + if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyPolicy, []byte(tablePolicy)); err != nil { + return err + } + } + // Soft-delete the source catalog identity in place: drop its catalog xattrs + // so the name stops resolving while the metadata/ and data/ children stay put + // (manifests embed absolute paths, so the data must not move). + return h.removeExtendedAttributes(r.Context(), client, srcPath, + ExtendedKeyMetadata, ExtendedKeyMetadataVersion, ExtendedKeyPolicy, ExtendedKeyTags) + }) + + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to rename table") + return err + } + + h.writeJSON(w, http.StatusOK, &RenameTableResponse{ + TableARN: h.generateTableARN(metadata.OwnerAccountID, bucketName, destNamespace+"/"+destName), + MetadataLocation: metadata.MetadataLocation, + }) + return nil +} + // handleUpdateTable updates table metadata func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { var req UpdateTableRequest diff --git a/weed/s3api/s3tables/table_data_dir_test.go b/weed/s3api/s3tables/table_data_dir_test.go new file mode 100644 index 000000000..ad3244958 --- /dev/null +++ b/weed/s3api/s3tables/table_data_dir_test.go @@ -0,0 +1,23 @@ +package s3tables + +import ( + "path" + "testing" +) + +func TestTableDataDirFromMetadataLocation(t *testing.T) { + cases := []struct { + loc string + want string + }{ + {"s3://warehouse/sales/orders/metadata/v1.metadata.json", path.Join(TablesPath, "warehouse/sales/orders")}, + {"s3://warehouse/sales/orders/metadata/00003-9f1c.metadata.json", path.Join(TablesPath, "warehouse/sales/orders")}, + {"s3://warehouse/ns/tbl", path.Join(TablesPath, "warehouse/ns/tbl")}, + {"", ""}, + } + for _, c := range cases { + if got := tableDataDirFromMetadataLocation(c.loc); got != c.want { + t.Errorf("tableDataDirFromMetadataLocation(%q) = %q, want %q", c.loc, got, c.want) + } + } +} diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index c732cebb9..6416cf08f 100644 --- a/weed/s3api/s3tables/types.go +++ b/weed/s3api/s3tables/types.go @@ -271,6 +271,19 @@ type UpdateTableResponse struct { MetadataLocation string `json:"metadataLocation,omitempty"` } +type RenameTableRequest struct { + TableBucketARN string `json:"tableBucketARN"` + SourceNamespace []string `json:"sourceNamespace"` + SourceName string `json:"sourceName"` + DestNamespace []string `json:"destNamespace"` + DestName string `json:"destName"` +} + +type RenameTableResponse struct { + TableARN string `json:"tableARN"` + MetadataLocation string `json:"metadataLocation,omitempty"` +} + // View types // // Views are stored exactly like tables (a filer directory carrying a metadata diff --git a/weed/s3api/s3tables/utils.go b/weed/s3api/s3tables/utils.go index 2cf3683bd..9b6aa10b6 100644 --- a/weed/s3api/s3tables/utils.go +++ b/weed/s3api/s3tables/utils.go @@ -97,6 +97,22 @@ func GetTablePath(bucketName, namespace, tableName string) string { return path.Join(TablesPath, bucketName, namespace, tableName) } +// tableDataDirFromMetadataLocation maps a table's s3:// metadata location to the +// filer directory holding its data. A renamed table is catalog-only, so its data +// stays at the original location while its catalog entry moves; this lets a drop +// purge the real data instead of the now-empty catalog path. +func tableDataDirFromMetadataLocation(metadataLocation string) string { + loc := strings.TrimSuffix(metadataLocation, "/") + if idx := strings.LastIndex(loc, "/metadata/"); idx != -1 { + loc = loc[:idx] + } + loc = strings.TrimPrefix(loc, "s3://") + if loc == "" { + return "" + } + return path.Join(TablesPath, loc) +} + // GetTableObjectRootDir returns the root path for table bucket object storage func GetTableObjectRootDir() string { return path.Join(TablesPath, tableObjectRootDirName)