diff --git a/weed/s3api/iceberg/handlers_namespace.go b/weed/s3api/iceberg/handlers_namespace.go index 832509573..2c2a937c5 100644 --- a/weed/s3api/iceberg/handlers_namespace.go +++ b/weed/s3api/iceberg/handlers_namespace.go @@ -2,6 +2,7 @@ package iceberg import ( "encoding/json" + "errors" "fmt" "net/http" "strings" @@ -193,6 +194,113 @@ func (s *Server) handleGetNamespace(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, result) } +// applyNamespacePropertyUpdates applies removals and updates to a copy of +// current and returns the merged property map alongside the Iceberg REST +// summary of which keys were removed, updated, or were missing. +func applyNamespacePropertyUpdates(current map[string]string, removals []string, updates map[string]string) (map[string]string, UpdateNamespacePropertiesResponse) { + properties := make(map[string]string, len(current)) + for k, v := range current { + properties[k] = v + } + + summary := UpdateNamespacePropertiesResponse{Removed: []string{}, Updated: []string{}, Missing: []string{}} + seen := make(map[string]struct{}, len(removals)) + for _, key := range removals { + if _, dup := seen[key]; dup { + continue + } + seen[key] = struct{}{} + if _, ok := properties[key]; ok { + delete(properties, key) + summary.Removed = append(summary.Removed, key) + } else { + summary.Missing = append(summary.Missing, key) + } + } + for key, value := range updates { + properties[key] = value + summary.Updated = append(summary.Updated, key) + } + return properties, summary +} + +// handleUpdateNamespaceProperties applies a set of removals and updates to a +// namespace's properties and returns which keys were removed, updated, or +// missing, per the Iceberg REST spec. +func (s *Server) handleUpdateNamespaceProperties(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + namespace := parseNamespace(vars["namespace"]) + if len(namespace) == 0 { + writeError(w, http.StatusBadRequest, "BadRequestException", "Namespace is required") + return + } + + var req UpdateNamespacePropertiesRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body") + return + } + + // A key cannot be both removed and updated in the same request. + for _, key := range req.Removals { + if _, ok := req.Updates[key]; ok { + writeError(w, http.StatusUnprocessableEntity, "UnprocessableEntityException", + fmt.Sprintf("Cannot remove and update the same key: %s", key)) + return + } + } + + bucketName := getBucketFromPrefix(r) + bucketARN := buildTableBucketARN(bucketName) + identityName := s3_constants.GetIdentityNameFromContext(r) + + // Load the current properties so we can compute the summary and the merged map. + getReq := &s3tables.GetNamespaceRequest{TableBucketARN: bucketARN, Namespace: namespace} + var getResp s3tables.GetNamespaceResponse + err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "GetNamespace", getReq, &getResp, identityName) + }) + if err != nil { + writeNamespaceManagerError(w, err, namespace) + return + } + + properties, summary := applyNamespacePropertyUpdates(getResp.Properties, req.Removals, req.Updates) + + updReq := &s3tables.UpdateNamespaceRequest{TableBucketARN: bucketARN, Namespace: namespace, Properties: properties} + var updResp s3tables.UpdateNamespaceResponse + err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "UpdateNamespace", updReq, &updResp, identityName) + }) + if err != nil { + writeNamespaceManagerError(w, err, namespace) + return + } + + writeJSON(w, http.StatusOK, summary) +} + +// writeNamespaceManagerError maps an s3tables manager error to the matching +// Iceberg REST response. A namespace dropped between read and write surfaces +// as 404 and a denied caller as 403; anything else is a 500. +func writeNamespaceManagerError(w http.ResponseWriter, err error, namespace Namespace) { + var s3Err *s3tables.S3TablesError + if errors.As(err, &s3Err) { + switch s3Err.Type { + case s3tables.ErrCodeNoSuchNamespace: + writeError(w, http.StatusNotFound, "NoSuchNamespaceException", fmt.Sprintf("Namespace does not exist: %v", namespace)) + return + case s3tables.ErrCodeAccessDenied: + writeError(w, http.StatusForbidden, "ForbiddenException", "Not authorized to update namespace properties") + return + } + } + glog.V(1).Infof("Iceberg: UpdateNamespaceProperties error: %v", err) + writeError(w, http.StatusInternalServerError, "InternalServerError", err.Error()) +} + // handleNamespaceExists checks if a namespace exists. func (s *Server) handleNamespaceExists(w http.ResponseWriter, r *http.Request) { vars := mux.Vars(r) diff --git a/weed/s3api/iceberg/iceberg_update_namespace_properties_test.go b/weed/s3api/iceberg/iceberg_update_namespace_properties_test.go new file mode 100644 index 000000000..eabb01208 --- /dev/null +++ b/weed/s3api/iceberg/iceberg_update_namespace_properties_test.go @@ -0,0 +1,87 @@ +package iceberg + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + + "github.com/gorilla/mux" +) + +func TestApplyNamespacePropertyUpdates(t *testing.T) { + current := map[string]string{"foo": "bar", "prop": "yes"} + + props, summary := applyNamespacePropertyUpdates(current, []string{"abc"}, map[string]string{"prop": "no"}) + + // Removal of a missing key reports missing, not removed. + if !reflect.DeepEqual(summary.Removed, []string{}) { + t.Errorf("Removed = %v, want []", summary.Removed) + } + if !reflect.DeepEqual(summary.Updated, []string{"prop"}) { + t.Errorf("Updated = %v, want [prop]", summary.Updated) + } + if !reflect.DeepEqual(summary.Missing, []string{"abc"}) { + t.Errorf("Missing = %v, want [abc]", summary.Missing) + } + if props["prop"] != "no" || props["foo"] != "bar" { + t.Errorf("merged properties = %v, want foo=bar prop=no", props) + } + + // Input map must not be mutated. + if current["prop"] != "yes" { + t.Errorf("source map mutated: prop = %q, want yes", current["prop"]) + } +} + +func TestApplyNamespacePropertyUpdatesRemoveExisting(t *testing.T) { + props, summary := applyNamespacePropertyUpdates( + map[string]string{"a": "1", "b": "2"}, []string{"a"}, nil) + + if !reflect.DeepEqual(summary.Removed, []string{"a"}) { + t.Errorf("Removed = %v, want [a]", summary.Removed) + } + if _, ok := props["a"]; ok { + t.Errorf("key a should have been removed: %v", props) + } + if len(summary.Missing) != 0 || len(summary.Updated) != 0 { + t.Errorf("unexpected summary: %+v", summary) + } +} + +func TestApplyNamespacePropertyUpdatesDuplicateRemovals(t *testing.T) { + // A repeated removal key must not be reported as both removed and missing. + _, summary := applyNamespacePropertyUpdates( + map[string]string{"a": "1"}, []string{"a", "a"}, nil) + + if !reflect.DeepEqual(summary.Removed, []string{"a"}) { + t.Errorf("Removed = %v, want [a]", summary.Removed) + } + if len(summary.Missing) != 0 { + t.Errorf("Missing = %v, want []", summary.Missing) + } +} + +// A key in both removals and updates is rejected before any backend call, so +// the conflict path is reachable without a filer. +func TestHandleUpdateNamespacePropertiesConflict(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/v1/namespaces/ns/properties", + strings.NewReader(`{"removals":["x"],"updates":{"x":"1"}}`)) + req = mux.SetURLVars(req, map[string]string{"namespace": "ns"}) + rec := httptest.NewRecorder() + + (&Server{}).handleUpdateNamespaceProperties(rec, req) + + if rec.Code != http.StatusUnprocessableEntity { + t.Fatalf("status = %d, want 422", rec.Code) + } + var resp ErrorResponse + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode: %v", err) + } + if resp.Error.Type != "UnprocessableEntityException" { + t.Fatalf("error type = %q, want UnprocessableEntityException", resp.Error.Type) + } +} diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go index 043b20096..da892896d 100644 --- a/weed/s3api/iceberg/server.go +++ b/weed/s3api/iceberg/server.go @@ -97,6 +97,7 @@ func (s *Server) RegisterRoutes(router *mux.Router) { router.HandleFunc("/v1/namespaces/{namespace}", s.Auth(s.handleGetNamespace)).Methods(http.MethodGet) router.HandleFunc("/v1/namespaces/{namespace}", s.Auth(s.handleNamespaceExists)).Methods(http.MethodHead) router.HandleFunc("/v1/namespaces/{namespace}", s.Auth(s.handleDropNamespace)).Methods(http.MethodDelete) + router.HandleFunc("/v1/namespaces/{namespace}/properties", s.Auth(s.handleUpdateNamespaceProperties)).Methods(http.MethodPost) // Table endpoints - wrapped with Auth middleware router.HandleFunc("/v1/namespaces/{namespace}/tables", s.Auth(s.handleListTables)).Methods(http.MethodGet) @@ -112,6 +113,7 @@ func (s *Server) RegisterRoutes(router *mux.Router) { router.HandleFunc("/v1/{prefix}/namespaces/{namespace}", s.Auth(s.handleGetNamespace)).Methods(http.MethodGet) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}", s.Auth(s.handleNamespaceExists)).Methods(http.MethodHead) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}", s.Auth(s.handleDropNamespace)).Methods(http.MethodDelete) + router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/properties", s.Auth(s.handleUpdateNamespaceProperties)).Methods(http.MethodPost) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables", s.Auth(s.handleListTables)).Methods(http.MethodGet) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables", s.Auth(s.handleCreateTable)).Methods(http.MethodPost) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables/{table}", s.Auth(s.handleLoadTable)).Methods(http.MethodGet) diff --git a/weed/s3api/iceberg/types.go b/weed/s3api/iceberg/types.go index 2aa399174..7c56c083d 100644 --- a/weed/s3api/iceberg/types.go +++ b/weed/s3api/iceberg/types.go @@ -62,6 +62,19 @@ type GetNamespaceResponse struct { Properties map[string]string `json:"properties"` } +// UpdateNamespacePropertiesRequest is sent to POST /v1/namespaces/{namespace}/properties. +type UpdateNamespacePropertiesRequest struct { + Removals []string `json:"removals,omitempty"` + Updates map[string]string `json:"updates,omitempty"` +} + +// UpdateNamespacePropertiesResponse is returned by POST /v1/namespaces/{namespace}/properties. +type UpdateNamespacePropertiesResponse struct { + Removed []string `json:"removed"` + Updated []string `json:"updated"` + Missing []string `json:"missing"` +} + // ListTablesResponse is returned by GET /v1/namespaces/{namespace}/tables. type ListTablesResponse struct { NextPageToken string `json:"next-page-token,omitempty"` diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index 60386da83..ed639b9f2 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -132,6 +132,8 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request, err = h.handleCreateNamespace(w, r, filerClient) case "GetNamespace": err = h.handleGetNamespace(w, r, filerClient) + case "UpdateNamespace": + err = h.handleUpdateNamespace(w, r, filerClient) case "ListNamespaces": err = h.handleListNamespaces(w, r, filerClient) case "DeleteNamespace": diff --git a/weed/s3api/s3tables/handler_namespace.go b/weed/s3api/s3tables/handler_namespace.go index 61b4ae896..8a825c510 100644 --- a/weed/s3api/s3tables/handler_namespace.go +++ b/weed/s3api/s3tables/handler_namespace.go @@ -276,6 +276,110 @@ func (h *S3TablesHandler) handleGetNamespace(w http.ResponseWriter, r *http.Requ return nil } +// handleUpdateNamespace replaces the stored properties of an existing namespace. +// AWS S3 Tables namespaces have no properties; this backs the Iceberg REST +// catalog's namespace-property updates, which carry the merged map. +func (h *S3TablesHandler) handleUpdateNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + var req UpdateNamespaceRequest + if err := h.readRequestBody(r, &req); err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + if req.TableBucketARN == "" { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "tableBucketARN is required") + return fmt.Errorf("tableBucketARN is required") + } + + namespaceName, err := validateNamespace(req.Namespace) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + bucketName, err := parseBucketNameFromARN(req.TableBucketARN) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + namespacePath := GetNamespacePath(bucketName, namespaceName) + bucketPath := GetTableBucketPath(bucketName) + + var metadata namespaceMetadata + var bucketPolicy string + var bucketTags map[string]string + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata) + if err != nil { + return err + } + if err := json.Unmarshal(data, &metadata); err != nil { + return 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: %v", 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, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", flattenNamespace(req.Namespace))) + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to get namespace: %v", err)) + } + return err + } + + bucketARN := h.generateTableBucketARN(metadata.OwnerAccountID, bucketName) + principal := h.getAccountID(r) + identityActions := getIdentityActions(r) + if !CheckPermissionWithContext("UpdateNamespace", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{ + TableBucketName: bucketName, + Namespace: namespaceName, + TableBucketTags: bucketTags, + IdentityActions: identityActions, + DefaultAllow: h.defaultAllowFor(r), + }) { + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to update namespace") + return ErrAccessDenied + } + + metadata.Properties = req.Properties + metadataBytes, err := json.Marshal(&metadata) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal namespace metadata") + return fmt.Errorf("failed to marshal metadata: %w", err) + } + + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return h.setExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata, metadataBytes) + }) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to update namespace") + return err + } + + resp := &UpdateNamespaceResponse{ + Namespace: metadata.Namespace, + TableBucketARN: req.TableBucketARN, + Properties: metadata.Properties, + } + + h.writeJSON(w, http.StatusOK, resp) + return nil +} + // handleListNamespaces lists all namespaces in a table bucket func (h *S3TablesHandler) handleListNamespaces(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { var req ListNamespacesRequest diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index 05fb661a8..0174b48b3 100644 --- a/weed/s3api/s3tables/types.go +++ b/weed/s3api/s3tables/types.go @@ -100,6 +100,18 @@ type GetNamespaceRequest struct { Namespace []string `json:"namespace"` } +type UpdateNamespaceRequest struct { + TableBucketARN string `json:"tableBucketARN"` + Namespace []string `json:"namespace"` + Properties map[string]string `json:"properties,omitempty"` +} + +type UpdateNamespaceResponse struct { + Namespace []string `json:"namespace"` + TableBucketARN string `json:"tableBucketARN"` + Properties map[string]string `json:"properties,omitempty"` +} + type GetNamespaceResponse struct { Namespace []string `json:"namespace"` CreatedAt time.Time `json:"createdAt"`