iceberg: support table register (#10067)

* s3tables: add RegisterTable op

* iceberg: support table register

* iceberg: test register table

* iceberg: parse engine-written metadata version from location

* iceberg: test metadata version parsing for both filename forms

* iceberg: map register errors through wrapped manager error

* iceberg: validate register metadata-location bucket and reject traversal

* iceberg: log register metadata load failure

weed/s3api/iceberg/handlers_table.go

@@ -323,6 +323,96 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, result)
 }
 
+// handleRegisterTable registers an existing metadata.json under a new catalog
+// entry without generating new metadata.
+func (s *Server) handleRegisterTable(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 RegisterTableRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body")
+		return
+	}
+	if req.Name == "" {
+		writeError(w, http.StatusBadRequest, "BadRequestException", errTableNameRequired.Error())
+		return
+	}
+	if req.MetadataLocation == "" {
+		writeError(w, http.StatusBadRequest, "BadRequestException", "metadata-location is required")
+		return
+	}
+
+	bucketName := getBucketFromPrefix(r)
+	bucketARN := buildTableBucketARN(bucketName)
+	identityName := s3_constants.GetIdentityNameFromContext(r)
+
+	// Read the existing metadata object before registering, so a bad location
+	// is rejected (400) without leaving a dangling catalog entry.
+	metadataBucket, tablePath, err := parseS3Location(tableLocationFromMetadataLocation(req.MetadataLocation))
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid metadata-location: "+err.Error())
+		return
+	}
+	// metadata-location is client-supplied; confine the read to the authorized
+	// catalog bucket and reject traversal segments so path.Join in
+	// loadMetadataFile cannot escape into another bucket.
+	if metadataBucket != bucketName {
+		writeError(w, http.StatusBadRequest, "BadRequestException", "metadata-location must be within bucket "+bucketName)
+		return
+	}
+	if !isValidTablePath(tablePath) {
+		writeError(w, http.StatusBadRequest, "BadRequestException", "invalid metadata-location path")
+		return
+	}
+	metadataFileName := path.Base(req.MetadataLocation)
+	metadataBytes, err := s.loadMetadataFile(r.Context(), metadataBucket, tablePath, metadataFileName)
+	if err != nil {
+		glog.V(1).Infof("Iceberg: RegisterTable load metadata at %s: %v", req.MetadataLocation, err)
+		writeError(w, http.StatusBadRequest, "BadRequestException", "Cannot read metadata at "+req.MetadataLocation)
+		return
+	}
+
+	registerReq := &s3tables.RegisterTableRequest{
+		TableBucketARN:   bucketARN,
+		Namespace:        namespace,
+		Name:             req.Name,
+		MetadataLocation: req.MetadataLocation,
+	}
+	var registerResp s3tables.RegisterTableResponse
+	err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		mgrClient := s3tables.NewManagerClient(client)
+		return s.tablesManager.Execute(r.Context(), mgrClient, "RegisterTable", registerReq, &registerResp, identityName)
+	})
+	if err != nil {
+		var tableErr *s3tables.S3TablesError
+		if errors.As(err, &tableErr) {
+			switch tableErr.Type {
+			case s3tables.ErrCodeNoSuchNamespace:
+				writeError(w, http.StatusNotFound, "NoSuchNamespaceException", fmt.Sprintf("Namespace does not exist: %v", namespace))
+				return
+			case s3tables.ErrCodeTableAlreadyExists:
+				writeError(w, http.StatusConflict, "AlreadyExistsException", fmt.Sprintf("Table already exists: %s", req.Name))
+				return
+			}
+		}
+		glog.V(1).Infof("Iceberg: RegisterTable error: %v", err)
+		writeManagerError(w, err)
+		return
+	}
+
+	getResp := s3tables.GetTableResponse{
+		MetadataLocation: req.MetadataLocation,
+		Metadata:         &s3tables.TableMetadata{FullMetadata: json.RawMessage(metadataBytes)},
+	}
+	result := s.buildLoadTableResult(getResp, bucketName, namespace, req.Name)
+	writeJSON(w, http.StatusOK, result)
+}
+
 // handleLoadTable loads table metadata.
 func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)

weed/s3api/iceberg/iceberg_register_table_test.go

@@ -0,0 +1,61 @@
+package iceberg
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/gorilla/mux"
+)
+
+// TestRegisterTableRequestWireShape pins the field names to the iceberg-go REST
+// client payload: {"name", "metadata-location"}.
+func TestRegisterTableRequestWireShape(t *testing.T) {
+	body := `{"name":"orders","metadata-location":"s3://bucket/ns/orders/metadata/v1.metadata.json"}`
+	var req RegisterTableRequest
+	if err := json.Unmarshal([]byte(body), &req); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if req.Name != "orders" {
+		t.Errorf("Name = %q, want orders", req.Name)
+	}
+	if req.MetadataLocation != "s3://bucket/ns/orders/metadata/v1.metadata.json" {
+		t.Errorf("MetadataLocation = %q", req.MetadataLocation)
+	}
+}
+
+// TestRegisterTableRejectsOutOfBoundsLocation confirms a metadata-location that
+// points outside the request's table bucket, or contains a traversal segment, is
+// rejected with 400 before any metadata read.
+func TestRegisterTableRejectsOutOfBoundsLocation(t *testing.T) {
+	cases := []struct {
+		name             string
+		metadataLocation string
+	}{
+		{"cross bucket", "s3://other/ns/orders/metadata/v1.metadata.json"},
+		{"traversal", "s3://bucket/ns/../../etc/metadata/v1.metadata.json"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			body := `{"name":"orders","metadata-location":"` + c.metadataLocation + `"}`
+			r := httptest.NewRequest(http.MethodPost, "/v1/bucket/namespaces/ns/register", strings.NewReader(body))
+			r = mux.SetURLVars(r, map[string]string{"prefix": "bucket", "namespace": "ns"})
+			rec := httptest.NewRecorder()
+
+			(&Server{}).handleRegisterTable(rec, r)
+
+			if rec.Code != http.StatusBadRequest {
+				t.Fatalf("status = %d, want %d", rec.Code, http.StatusBadRequest)
+			}
+			var resp ErrorResponse
+			if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
+				t.Fatalf("decode response: %v", err)
+			}
+			if resp.Error.Type != "BadRequestException" {
+				t.Fatalf("error type = %q, want BadRequestException", resp.Error.Type)
+			}
+		})
+	}
+}

weed/s3api/iceberg/server.go

@@ -102,6 +102,7 @@ func (s *Server) RegisterRoutes(router *mux.Router) {
 	// Table endpoints - wrapped with Auth middleware
 	router.HandleFunc("/v1/namespaces/{namespace}/tables", s.Auth(s.handleListTables)).Methods(http.MethodGet)
 	router.HandleFunc("/v1/namespaces/{namespace}/tables", s.Auth(s.handleCreateTable)).Methods(http.MethodPost)
+	router.HandleFunc("/v1/namespaces/{namespace}/register", s.Auth(s.handleRegisterTable)).Methods(http.MethodPost)
 	router.HandleFunc("/v1/namespaces/{namespace}/tables/{table}", s.Auth(s.handleLoadTable)).Methods(http.MethodGet)
 	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)
@@ -116,6 +117,7 @@ func (s *Server) RegisterRoutes(router *mux.Router) {
 	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}/register", s.Auth(s.handleRegisterTable)).Methods(http.MethodPost)
 	router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables/{table}", s.Auth(s.handleLoadTable)).Methods(http.MethodGet)
 	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)

weed/s3api/iceberg/types.go

@@ -93,6 +93,12 @@ type CreateTableRequest struct {
 	Properties    iceberg.Properties     `json:"properties,omitempty"`
 }
 
+// RegisterTableRequest is sent to POST /v1/namespaces/{namespace}/register.
+type RegisterTableRequest struct {
+	Name             string `json:"name"`
+	MetadataLocation string `json:"metadata-location"`
+}
+
 type LoadTableResult struct {
 	MetadataLocation string             `json:"metadata-location,omitempty"`
 	Metadata         table.Metadata     `json:"metadata"`

weed/s3api/s3tables/handler.go

@@ -142,6 +142,8 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request,
 	// Table operations
 	case "CreateTable":
 		err = h.handleCreateTable(w, r, filerClient)
+	case "RegisterTable":
+		err = h.handleRegisterTable(w, r, filerClient)
 	case "GetTable":
 		err = h.handleGetTable(w, r, filerClient)
 	case "ListTables":

weed/s3api/s3tables/handler_table.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -259,6 +260,221 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 	return nil
 }
 
+// metadataVersionFromLocation parses the version N from a metadata location.
+// SeaweedFS writes v{N}.metadata.json; Iceberg engines (Spark/Trino/Flink/Java)
+// write {NNNNN}-{uuid}.metadata.json with a zero-padded leading version. Returns
+// 1 when no version can be parsed.
+func metadataVersionFromLocation(metadataLocation string) int {
+	name := metadataLocation
+	if idx := strings.LastIndex(name, "/"); idx != -1 {
+		name = name[idx+1:]
+	}
+	name = strings.TrimSuffix(name, ".metadata.json")
+	// v{N} form
+	if v, err := strconv.Atoi(strings.TrimPrefix(name, "v")); err == nil && v > 0 {
+		return v
+	}
+	// {NNNNN}-{uuid} form: the leading integer before the first '-'
+	if idx := strings.IndexByte(name, '-'); idx != -1 {
+		if v, err := strconv.Atoi(name[:idx]); err == nil && v > 0 {
+			return v
+		}
+	}
+	return 1
+}
+
+// handleRegisterTable registers an existing Iceberg metadata.json under a new
+// catalog entry. Unlike CreateTable it does not generate metadata: it points the
+// table at the caller-supplied MetadataLocation.
+func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error {
+
+	var req RegisterTableRequest
+	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
+	}
+
+	if req.Name == "" {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "name is required")
+		return fmt.Errorf("name is required")
+	}
+
+	if req.MetadataLocation == "" {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "metadataLocation is required")
+		return fmt.Errorf("metadataLocation is required")
+	}
+
+	bucketName, err := parseBucketNameFromARN(req.TableBucketARN)
+	if err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
+	tableName, err := validateTableName(req.Name)
+	if err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
+	// Namespace must exist.
+	namespacePath := GetNamespacePath(bucketName, namespaceName)
+	var namespaceMetadata namespaceMetadata
+	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, &namespaceMetadata); err != nil {
+			return fmt.Errorf("failed to unmarshal namespace metadata: %w", 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", namespaceName))
+		} else {
+			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check namespace: %v", err))
+		}
+		return err
+	}
+
+	// Authorize using policy framework (namespace + bucket policies).
+	accountID := h.getAccountID(r)
+	bucketPath := GetTableBucketPath(bucketName)
+	namespacePolicy := ""
+	bucketPolicy := ""
+	bucketTags := map[string]string{}
+	var bucketMetadata tableBucketMetadata
+
+	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		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: %v", err)
+		}
+
+		policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy)
+		if err == nil {
+			namespacePolicy = string(policyData)
+		} else if !errors.Is(err, ErrAttributeNotFound) {
+			return fmt.Errorf("failed to fetch namespace policy: %v", 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)
+		}
+		if tags, err := h.readTags(r.Context(), client, bucketPath); err != nil {
+			return err
+		} else if tags != nil {
+			bucketTags = tags
+		}
+
+		return nil
+	})
+	if err != nil {
+		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to fetch policies: %v", err))
+		return err
+	}
+
+	bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
+	identityActions := getIdentityActions(r)
+	nsAllowed := CheckPermissionWithContext("CreateTable", accountID, namespaceMetadata.OwnerAccountID, namespacePolicy, bucketARN, &PolicyContext{
+		TableBucketName: bucketName,
+		Namespace:       namespaceName,
+		TableName:       tableName,
+		TableBucketTags: bucketTags,
+		IdentityActions: identityActions,
+		DefaultAllow:    h.defaultAllowFor(r),
+	})
+	bucketAllowed := CheckPermissionWithContext("CreateTable", accountID, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
+		TableBucketName: bucketName,
+		Namespace:       namespaceName,
+		TableName:       tableName,
+		TableBucketTags: bucketTags,
+		IdentityActions: identityActions,
+		DefaultAllow:    h.defaultAllowFor(r),
+	})
+	if !nsAllowed && !bucketAllowed {
+		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to register table in this namespace")
+		return ErrAccessDenied
+	}
+
+	tablePath := GetTablePath(bucketName, namespaceName, tableName)
+
+	// Table must be absent.
+	var existingMetadata tableMetadataInternal
+	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
+		if err != nil {
+			return err
+		}
+		return json.Unmarshal(data, &existingMetadata)
+	})
+	if err == nil {
+		h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("table %s already exists", tableName))
+		return fmt.Errorf("table %s already exists", tableName)
+	} else if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) {
+		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check table: %v", err))
+		return err
+	}
+
+	now := time.Now()
+	versionToken := generateVersionToken()
+	metadata := &tableMetadataInternal{
+		Name:             tableName,
+		Namespace:        namespaceName,
+		Format:           "ICEBERG",
+		CreatedAt:        now,
+		ModifiedAt:       now,
+		OwnerAccountID:   namespaceMetadata.OwnerAccountID,
+		VersionToken:     versionToken,
+		MetadataVersion:  metadataVersionFromLocation(req.MetadataLocation),
+		MetadataLocation: req.MetadataLocation,
+	}
+
+	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)
+	}
+
+	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		if err := h.ensureDirectory(r.Context(), client, tablePath); err != nil {
+			return err
+		}
+		return h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata, metadataBytes)
+	})
+	if err != nil {
+		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to register table")
+		return err
+	}
+
+	tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName)
+	h.writeJSON(w, http.StatusOK, &RegisterTableResponse{
+		TableARN:         tableARN,
+		VersionToken:     versionToken,
+		MetadataLocation: metadata.MetadataLocation,
+	})
+	return nil
+}
+
 // handleGetTable gets details of a table
 func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error {
 

weed/s3api/s3tables/handler_table_test.go

@@ -0,0 +1,25 @@
+package s3tables
+
+import "testing"
+
+func TestMetadataVersionFromLocation(t *testing.T) {
+	cases := []struct {
+		location string
+		want     int
+	}{
+		{"s3://bucket/ns/tbl/metadata/v1.metadata.json", 1},
+		{"s3://bucket/ns/tbl/metadata/v7.metadata.json", 7},
+		{"v42.metadata.json", 42},
+		{"s3://bucket/ns/tbl/metadata/00003-9f1c2b3a-4d5e-6f70-8192-a3b4c5d6e7f8.metadata.json", 3},
+		{"00012-abcdef.metadata.json", 12},
+		{"s3://bucket/ns/tbl/metadata/00000-abc.metadata.json", 1},
+		{"s3://bucket/ns/tbl/metadata/v0.metadata.json", 1},
+		{"s3://bucket/ns/tbl/metadata/garbage.metadata.json", 1},
+		{"", 1},
+	}
+	for _, c := range cases {
+		if got := metadataVersionFromLocation(c.location); got != c.want {
+			t.Errorf("metadataVersionFromLocation(%q) = %d, want %d", c.location, got, c.want)
+		}
+	}
+}

weed/s3api/s3tables/types.go

@@ -192,6 +192,19 @@ type CreateTableResponse struct {
 	MetadataLocation string `json:"metadataLocation,omitempty"`
 }
 
+type RegisterTableRequest struct {
+	TableBucketARN   string   `json:"tableBucketARN"`
+	Namespace        []string `json:"namespace"`
+	Name             string   `json:"name"`
+	MetadataLocation string   `json:"metadataLocation"`
+}
+
+type RegisterTableResponse struct {
+	TableARN         string `json:"tableARN"`
+	VersionToken     string `json:"versionToken"`
+	MetadataLocation string `json:"metadataLocation,omitempty"`
+}
+
 type GetTableRequest struct {
 	TableBucketARN string   `json:"tableBucketARN,omitempty"`
 	Namespace      []string `json:"namespace,omitempty"`