From 628ce5762572e8c2068628725f3257cd8ed40a7a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 23 Jun 2026 14:07:13 -0700 Subject: [PATCH] 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 | 90 ++++++++ .../iceberg/iceberg_register_table_test.go | 61 +++++ weed/s3api/iceberg/server.go | 2 + weed/s3api/iceberg/types.go | 6 + weed/s3api/s3tables/handler.go | 2 + weed/s3api/s3tables/handler_table.go | 216 ++++++++++++++++++ weed/s3api/s3tables/handler_table_test.go | 25 ++ weed/s3api/s3tables/types.go | 13 ++ 8 files changed, 415 insertions(+) create mode 100644 weed/s3api/iceberg/iceberg_register_table_test.go create mode 100644 weed/s3api/s3tables/handler_table_test.go diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index acad64a01..05659874d 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/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, ®isterResp, 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) diff --git a/weed/s3api/iceberg/iceberg_register_table_test.go b/weed/s3api/iceberg/iceberg_register_table_test.go new file mode 100644 index 000000000..da847bd09 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go index da892896d..78792fd9c 100644 --- a/weed/s3api/iceberg/server.go +++ b/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) diff --git a/weed/s3api/iceberg/types.go b/weed/s3api/iceberg/types.go index 7c56c083d..2f54189e8 100644 --- a/weed/s3api/iceberg/types.go +++ b/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"` diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index ed639b9f2..dbc71f1d2 100644 --- a/weed/s3api/s3tables/handler.go +++ b/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": diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index ecfe3eafb..31b35192d 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/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 { diff --git a/weed/s3api/s3tables/handler_table_test.go b/weed/s3api/s3tables/handler_table_test.go new file mode 100644 index 000000000..aacd11f80 --- /dev/null +++ b/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) + } + } +} diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index 0174b48b3..21722451c 100644 --- a/weed/s3api/s3tables/types.go +++ b/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"`