iceberg: support namespace property updates (#10052)

* iceberg: support namespace property updates

Add POST /v1/namespaces/{namespace}/properties to the REST catalog. It
applies the request's removals and updates and returns the removed/updated/
missing summary the spec defines. A new UpdateNamespace op on the S3 Tables
manager rewrites the stored namespace properties; AWS S3 Tables namespaces
have no properties, so this is the SeaweedFS-side backing for the catalog.

* iceberg: dedup namespace property removals

A key repeated in removals was deleted on its first occurrence, then
reported as missing on the next — landing in both removed and missing.
Skip keys already processed.

* iceberg: map namespace-update backend errors to REST statuses

UpdateNamespaceProperties returned 500 for every manager failure, masking
the namespace being dropped between read and write, or a denied caller.
Inspect the typed S3TablesError and answer 404/403 accordingly, 500 only
for the rest. Also replaces the GetNamespace not-found string match.

* iceberg: test the namespace-properties conflict path

Cover the 422 returned when a key appears in both removals and updates.
The check runs before any backend call, so it needs no filer.

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)

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

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)

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"`

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":

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

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"`