From 0403e47ef6c4ecef421b07e5eb1b87c9b53b1ae4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 23 Jun 2026 15:22:31 -0700 Subject: [PATCH] iceberg: support views (#10069) * s3tables: tag table entries and exclude views from table listings * s3tables: add view CRUD operations * iceberg: support view create, load, exists, drop, and list * iceberg: support view update * iceberg: test view error classification and metadata round-trip * iceberg: pre-check existence and write view metadata only after create * iceberg: map view namespace-not-found to 404 * iceberg: test view create namespace-404 and duplicate no-clobber * s3tables: tag view metadata and entry type atomically CreateView wrote ExtendedKeyMetadata and ExtendedKeyEntryType in two UpdateEntry calls, so a partial failure could leave a view directory untagged. Add setExtendedAttributes to set both in one UpdateEntry. * iceberg: roll back view registration when metadata write fails The metadata file is written after the catalog registers the view. If that write fails, drop the just-created view so it doesn't linger pointing at a missing metadata.json. Reuse the DeleteView path via a shared dropView helper. --- weed/s3api/iceberg/handlers_view.go | 389 +++++++++++ weed/s3api/iceberg/handlers_view_update.go | 162 +++++ .../s3api/iceberg/iceberg_create_view_test.go | 203 ++++++ weed/s3api/iceberg/iceberg_view_test.go | 115 ++++ weed/s3api/iceberg/server.go | 14 + weed/s3api/iceberg/types.go | 55 ++ weed/s3api/s3tables/filer_ops.go | 40 ++ weed/s3api/s3tables/handler.go | 22 + weed/s3api/s3tables/handler_table.go | 37 +- weed/s3api/s3tables/handler_view.go | 648 ++++++++++++++++++ weed/s3api/s3tables/handler_view_test.go | 33 + weed/s3api/s3tables/types.go | 87 +++ weed/s3api/s3tables/utils.go | 13 + 13 files changed, 1815 insertions(+), 3 deletions(-) create mode 100644 weed/s3api/iceberg/handlers_view.go create mode 100644 weed/s3api/iceberg/handlers_view_update.go create mode 100644 weed/s3api/iceberg/iceberg_create_view_test.go create mode 100644 weed/s3api/iceberg/iceberg_view_test.go create mode 100644 weed/s3api/s3tables/handler_view.go create mode 100644 weed/s3api/s3tables/handler_view_test.go diff --git a/weed/s3api/iceberg/handlers_view.go b/weed/s3api/iceberg/handlers_view.go new file mode 100644 index 000000000..daa4151e0 --- /dev/null +++ b/weed/s3api/iceberg/handlers_view.go @@ -0,0 +1,389 @@ +package iceberg + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "os" + "path" + "strings" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/view" + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// handleListViews lists views in a namespace. +func (s *Server) handleListViews(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 + } + + bucketName := getBucketFromPrefix(r) + bucketARN := buildTableBucketARN(bucketName) + identityName := s3_constants.GetIdentityNameFromContext(r) + + pageToken, pageSize, err := parsePagination(r) + if err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", err.Error()) + return + } + + listReq := &s3tables.ListViewsRequest{ + TableBucketARN: bucketARN, + Namespace: namespace, + ContinuationToken: pageToken, + MaxViews: pageSize, + } + var listResp s3tables.ListViewsResponse + err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "ListViews", listReq, &listResp, identityName) + }) + if err != nil { + if strings.Contains(err.Error(), "not found") { + writeError(w, http.StatusNotFound, "NoSuchNamespaceException", fmt.Sprintf("Namespace does not exist: %v", namespace)) + return + } + glog.V(1).Infof("Iceberg: ListViews error: %v", err) + writeManagerError(w, err) + return + } + + identifiers := make([]TableIdentifier, 0, len(listResp.Views)) + for _, v := range listResp.Views { + identifiers = append(identifiers, TableIdentifier{Namespace: namespace, Name: v.Name}) + } + writeJSON(w, http.StatusOK, ListViewsResponse{ + NextPageToken: listResp.ContinuationToken, + Identifiers: identifiers, + }) +} + +// handleCreateView creates a new view and writes its v1 metadata.json. +func (s *Server) handleCreateView(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 CreateViewRequest + 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", "view name is required") + return + } + if req.Schema == nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "view schema is required") + return + } + if req.ViewVersion == nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "view-version is required") + return + } + + bucketName := getBucketFromPrefix(r) + bucketARN := buildTableBucketARN(bucketName) + identityName := s3_constants.GetIdentityNameFromContext(r) + + viewPath := path.Join(flattenNamespacePath(namespace), req.Name) + location := strings.TrimSuffix(req.Location, "/") + if location == "" { + if req.Properties != nil { + if warehouse := strings.TrimSuffix(req.Properties["warehouse"], "/"); warehouse != "" { + location = fmt.Sprintf("%s/%s", warehouse, viewPath) + } + } + if location == "" { + if warehouse := strings.TrimSuffix(os.Getenv("ICEBERG_WAREHOUSE"), "/"); warehouse != "" { + location = fmt.Sprintf("%s/%s", warehouse, viewPath) + } + } + if location == "" { + location = fmt.Sprintf("s3://%s/%s", bucketName, viewPath) + } + } else { + parsedBucket, parsedPath, err := parseS3Location(location) + if err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid view location: "+err.Error()) + return + } + if parsedPath == "" { + location = fmt.Sprintf("s3://%s/%s", parsedBucket, viewPath) + } + } + + metadataBucket, metadataPath, err := parseS3Location(location) + if err != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid view location: "+err.Error()) + return + } + if metadataBucket != bucketName { + writeError(w, http.StatusBadRequest, "BadRequestException", "view location must be within bucket "+bucketName) + return + } + if !isValidTablePath(metadataPath) { + writeError(w, http.StatusBadRequest, "BadRequestException", "invalid view location path") + return + } + + metadata, err := view.NewMetadata(req.ViewVersion, req.Schema, location, viewProperties(req.Properties)) + if err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Failed to build view metadata: "+err.Error()) + return + } + + metadataFileName := "v1.metadata.json" + metadataLocation := fmt.Sprintf("%s/metadata/%s", location, metadataFileName) + metadataBytes, err := json.Marshal(metadata) + if err != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to serialize view metadata: "+err.Error()) + return + } + + // Authoritative existence check before touching storage: a registered view + // short-circuits with its stored definition (idempotent CreateView) so we + // never overwrite the persisted metadata of an existing view. + if existsResp, existsErr := s.getView(r, namespace, req.Name); existsErr == nil { + result, buildErr := s.buildViewResponse(existsResp, bucketName, namespace, req.Name) + if buildErr != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", buildErr.Error()) + return + } + writeJSON(w, http.StatusOK, result) + return + } else if !isViewNotFound(existsErr) { + glog.V(1).Infof("Iceberg: CreateView existence check failed for %s.%s: %v", flattenNamespacePath(namespace), req.Name, existsErr) + writeManagerError(w, existsErr) + return + } + + createReq := &s3tables.CreateViewRequest{ + TableBucketARN: bucketARN, + Namespace: namespace, + Name: req.Name, + Metadata: &s3tables.TableMetadata{ + Iceberg: &s3tables.IcebergMetadata{TableUUID: metadata.ViewUUID().String()}, + FullMetadata: metadataBytes, + }, + MetadataLocation: metadataLocation, + MetadataVersion: 1, + } + var createResp s3tables.CreateViewResponse + err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "CreateView", createReq, &createResp, identityName) + }) + if err != nil { + if isViewAlreadyExists(err) { + writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error()) + return + } + var viewErr *s3tables.S3TablesError + if errors.As(err, &viewErr) && viewErr.Type == s3tables.ErrCodeNoSuchNamespace { + writeError(w, http.StatusNotFound, "NoSuchNamespaceException", fmt.Sprintf("Namespace does not exist: %v", namespace)) + return + } + glog.V(1).Infof("Iceberg: CreateView error: %v", err) + writeManagerError(w, err) + return + } + + // Persist the metadata file only after the catalog registers the view, so a + // missing namespace or name collision fails before any bytes hit storage. + if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil { + // Roll back the registered view so it doesn't linger pointing at metadata + // that was never written. + if dropErr := s.dropView(r, namespace, req.Name); dropErr != nil { + glog.V(1).Infof("Iceberg: failed to roll back view %s.%s after metadata write error: %v", flattenNamespacePath(namespace), req.Name, dropErr) + } + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save view metadata file: "+err.Error()) + return + } + + finalLocation := createResp.MetadataLocation + if finalLocation == "" { + finalLocation = metadataLocation + } + writeJSON(w, http.StatusOK, ViewResponse{ + MetadataLocation: finalLocation, + Metadata: metadata, + Config: s.buildFileIOConfig(), + }) +} + +// handleLoadView loads view metadata. +func (s *Server) handleLoadView(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + namespace := parseNamespace(vars["namespace"]) + viewName := vars["view"] + if len(namespace) == 0 || viewName == "" { + writeError(w, http.StatusBadRequest, "BadRequestException", "Namespace and view name are required") + return + } + + getResp, err := s.getView(r, namespace, viewName) + if err != nil { + if isViewNotFound(err) { + writeError(w, http.StatusNotFound, "NoSuchViewException", fmt.Sprintf("View does not exist: %s", viewName)) + return + } + glog.V(1).Infof("Iceberg: LoadView error: %v", err) + writeManagerError(w, err) + return + } + + result, err := s.buildViewResponse(getResp, getBucketFromPrefix(r), namespace, viewName) + if err != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", err.Error()) + return + } + writeJSON(w, http.StatusOK, result) +} + +// handleViewExists checks if a view exists. +func (s *Server) handleViewExists(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + namespace := parseNamespace(vars["namespace"]) + viewName := vars["view"] + if len(namespace) == 0 || viewName == "" { + w.WriteHeader(http.StatusBadRequest) + return + } + if _, err := s.getView(r, namespace, viewName); err != nil { + w.WriteHeader(http.StatusNotFound) + return + } + w.WriteHeader(http.StatusNoContent) +} + +// handleDropView deletes a view. +func (s *Server) handleDropView(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + namespace := parseNamespace(vars["namespace"]) + viewName := vars["view"] + if len(namespace) == 0 || viewName == "" { + writeError(w, http.StatusBadRequest, "BadRequestException", "Namespace and view name are required") + return + } + + var storedMetadataLocation string + if getResp, err := s.getView(r, namespace, viewName); err == nil { + storedMetadataLocation = getResp.MetadataLocation + } + + err := s.dropView(r, namespace, viewName) + if err != nil { + if isViewNotFound(err) { + writeError(w, http.StatusNotFound, "NoSuchViewException", fmt.Sprintf("View does not exist: %s", viewName)) + return + } + glog.V(1).Infof("Iceberg: DropView error: %v", err) + writeManagerError(w, err) + return + } + + if storedMetadataLocation != "" { + if viewLoc := tableLocationFromMetadataLocation(storedMetadataLocation); viewLoc != "" { + if dataBucket, dataPath, parseErr := parseS3Location(viewLoc); parseErr == nil { + if cleanupErr := s.cleanupStaleTableLocation(r.Context(), dataBucket, dataPath); cleanupErr != nil { + glog.V(1).Infof("Iceberg: failed to purge dropped view location s3://%s/%s: %v", dataBucket, dataPath, cleanupErr) + } + } + } + } + + w.WriteHeader(http.StatusNoContent) +} + +// dropView removes a view registration from the catalog. +func (s *Server) dropView(r *http.Request, namespace []string, viewName string) error { + deleteReq := &s3tables.DeleteViewRequest{ + TableBucketARN: buildTableBucketARN(getBucketFromPrefix(r)), + Namespace: namespace, + Name: viewName, + } + identityName := s3_constants.GetIdentityNameFromContext(r) + return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "DeleteView", deleteReq, nil, identityName) + }) +} + +// getView fetches a view's stored metadata pointer from the catalog. +func (s *Server) getView(r *http.Request, namespace []string, viewName string) (s3tables.GetViewResponse, error) { + getReq := &s3tables.GetViewRequest{ + TableBucketARN: buildTableBucketARN(getBucketFromPrefix(r)), + Namespace: namespace, + Name: viewName, + } + var getResp s3tables.GetViewResponse + identityName := s3_constants.GetIdentityNameFromContext(r) + err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "GetView", getReq, &getResp, identityName) + }) + return getResp, err +} + +// buildViewResponse parses the stored view metadata into a ViewResponse. +func (s *Server) buildViewResponse(getResp s3tables.GetViewResponse, bucketName string, namespace []string, viewName string) (ViewResponse, error) { + if getResp.Metadata == nil || len(getResp.Metadata.FullMetadata) == 0 { + return ViewResponse{}, fmt.Errorf("view %s has no metadata", viewName) + } + metadata, err := view.ParseMetadataBytes(getResp.Metadata.FullMetadata) + if err != nil { + return ViewResponse{}, fmt.Errorf("failed to parse view metadata: %w", err) + } + return ViewResponse{ + MetadataLocation: getResp.MetadataLocation, + Metadata: metadata, + Config: s.buildFileIOConfig(), + }, nil +} + +// viewProperties returns a non-nil copy of props for view metadata construction. +func viewProperties(props iceberg.Properties) iceberg.Properties { + out := make(iceberg.Properties, len(props)) + for k, v := range props { + out[k] = v + } + return out +} + +func isViewNotFound(err error) bool { + if err == nil { + return false + } + var viewErr *s3tables.S3TablesError + if errors.As(err, &viewErr) { + if viewErr.Type == s3tables.ErrCodeNoSuchView || viewErr.Type == s3tables.ErrCodeNoSuchNamespace { + return true + } + } + return strings.Contains(strings.ToLower(err.Error()), "not found") +} + +func isViewAlreadyExists(err error) bool { + if err == nil { + return false + } + var viewErr *s3tables.S3TablesError + if errors.As(err, &viewErr) && viewErr.Type == s3tables.ErrCodeViewAlreadyExists { + return true + } + return strings.Contains(strings.ToLower(err.Error()), "already exists") +} diff --git a/weed/s3api/iceberg/handlers_view_update.go b/weed/s3api/iceberg/handlers_view_update.go new file mode 100644 index 000000000..001a641f3 --- /dev/null +++ b/weed/s3api/iceberg/handlers_view_update.go @@ -0,0 +1,162 @@ +package iceberg + +import ( + "encoding/json" + "fmt" + "math/rand/v2" + "net/http" + "strings" + "time" + + "github.com/apache/iceberg-go/view" + "github.com/google/uuid" + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// handleUpdateView applies requirements and updates to a view, writes a new +// metadata.json, and flips the stored pointer. Mirrors the table commit flow. +func (s *Server) handleUpdateView(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + namespace := parseNamespace(vars["namespace"]) + viewName := vars["view"] + if len(namespace) == 0 || viewName == "" { + writeError(w, http.StatusBadRequest, "BadRequestException", "Namespace and view name are required") + return + } + + var req UpdateViewRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body: "+err.Error()) + return + } + + bucketName := getBucketFromPrefix(r) + bucketARN := buildTableBucketARN(bucketName) + identityName := s3_constants.GetIdentityNameFromContext(r) + + const maxCommitAttempts = 3 + for attempt := 1; attempt <= maxCommitAttempts; attempt++ { + getResp, err := s.getView(r, namespace, viewName) + if err != nil { + if isViewNotFound(err) { + writeError(w, http.StatusNotFound, "NoSuchViewException", fmt.Sprintf("View does not exist: %s", viewName)) + return + } + glog.V(1).Infof("Iceberg: UpdateView GetView error: %v", err) + writeManagerError(w, err) + return + } + if getResp.Metadata == nil || len(getResp.Metadata.FullMetadata) == 0 { + writeError(w, http.StatusInternalServerError, "InternalServerError", "view has no metadata") + return + } + + currentMetadata, err := view.ParseMetadataBytes(getResp.Metadata.FullMetadata) + if err != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to parse current view metadata: "+err.Error()) + return + } + + for _, requirement := range req.Requirements { + if err := requirement.Validate(currentMetadata); err != nil { + writeError(w, http.StatusConflict, "CommitFailedException", "Requirement failed: "+err.Error()) + return + } + } + + builder, err := view.MetadataBuilderFromBase(currentMetadata) + if err != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to create view metadata builder: "+err.Error()) + return + } + for _, update := range req.Updates { + if err := update.Apply(builder); err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Failed to apply update: "+err.Error()) + return + } + } + newMetadata, err := builder.Build() + if err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Failed to build new view metadata: "+err.Error()) + return + } + + location := tableLocationFromMetadataLocation(getResp.MetadataLocation) + if location == "" { + location = newMetadata.Location() + } + metadataVersion := getResp.MetadataVersion + 1 + metadataFileName := fmt.Sprintf("v%d.metadata.json", metadataVersion) + newMetadataLocation := fmt.Sprintf("%s/metadata/%s", strings.TrimSuffix(location, "/"), metadataFileName) + + metadataBytes, err := json.Marshal(newMetadata) + if err != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to serialize view metadata: "+err.Error()) + return + } + + metadataBucket, metadataPath, err := parseS3Location(location) + if err != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid view location: "+err.Error()) + return + } + if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil { + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save view metadata file: "+err.Error()) + return + } + + tableUUID := newMetadata.ViewUUID() + if tableUUID == uuid.Nil { + tableUUID = currentMetadata.ViewUUID() + } + updateReq := &s3tables.UpdateViewRequest{ + TableBucketARN: bucketARN, + Namespace: namespace, + Name: viewName, + VersionToken: getResp.VersionToken, + Metadata: &s3tables.TableMetadata{ + Iceberg: &s3tables.IcebergMetadata{TableUUID: tableUUID.String()}, + FullMetadata: metadataBytes, + }, + MetadataVersion: metadataVersion, + MetadataLocation: newMetadataLocation, + } + err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "UpdateView", updateReq, nil, identityName) + }) + if err == nil { + writeJSON(w, http.StatusOK, ViewResponse{ + MetadataLocation: newMetadataLocation, + Metadata: newMetadata, + Config: s.buildFileIOConfig(), + }) + return + } + + if isS3TablesConflict(err) { + if cleanupErr := s.deleteMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName); cleanupErr != nil { + glog.V(1).Infof("Iceberg: failed to cleanup view metadata file %s on conflict: %v", newMetadataLocation, cleanupErr) + } + if attempt < maxCommitAttempts { + glog.V(1).Infof("Iceberg: UpdateView conflict for %s (attempt %d/%d), retrying", viewName, attempt, maxCommitAttempts) + jitter := time.Duration(rand.Int64N(int64(25 * time.Millisecond))) + time.Sleep(time.Duration(50*attempt)*time.Millisecond + jitter) + continue + } + writeError(w, http.StatusConflict, "CommitFailedException", "Version token mismatch") + return + } + + if cleanupErr := s.deleteMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName); cleanupErr != nil { + glog.V(1).Infof("Iceberg: failed to cleanup view metadata file %s after update failure: %v", newMetadataLocation, cleanupErr) + } + glog.Errorf("Iceberg: UpdateView error: %v", err) + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to commit view update: "+err.Error()) + return + } +} diff --git a/weed/s3api/iceberg/iceberg_create_view_test.go b/weed/s3api/iceberg/iceberg_create_view_test.go new file mode 100644 index 000000000..bc7c6a74e --- /dev/null +++ b/weed/s3api/iceberg/iceberg_create_view_test.go @@ -0,0 +1,203 @@ +package iceberg + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "path" + "strings" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/apache/iceberg-go/view" + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "google.golang.org/grpc" +) + +// memFiler is a minimal in-memory SeaweedFilerClient backing the view handlers +// end-to-end through the s3tables manager. Only the entry operations the create +// path touches are implemented; anything else panics so a missing dependency is +// loud rather than silently passing. +type memFiler struct { + filer_pb.SeaweedFilerClient + entries map[string]*filer_pb.Entry + // failFileCreate, when set, makes CreateEntry fail for non-directory entries, + // simulating a metadata-file write error. + failFileCreate error +} + +func newMemFiler() *memFiler { + return &memFiler{entries: map[string]*filer_pb.Entry{}} +} + +func (m *memFiler) WithFilerClient(_ bool, fn func(client filer_pb.SeaweedFilerClient) error) error { + return fn(m) +} + +func (m *memFiler) seed(p string, entry *filer_pb.Entry) { + m.entries[p] = entry +} + +func (m *memFiler) LookupDirectoryEntry(_ context.Context, in *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + entry, ok := m.entries[path.Join(in.Directory, in.Name)] + if !ok { + return nil, filer_pb.ErrNotFound + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil +} + +func (m *memFiler) CreateEntry(_ context.Context, in *filer_pb.CreateEntryRequest, _ ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) { + if m.failFileCreate != nil && !in.Entry.IsDirectory { + return nil, m.failFileCreate + } + m.entries[path.Join(in.Directory, in.Entry.Name)] = in.Entry + return &filer_pb.CreateEntryResponse{}, nil +} + +func (m *memFiler) DeleteEntry(_ context.Context, in *filer_pb.DeleteEntryRequest, _ ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) { + prefix := path.Join(in.Directory, in.Name) + for p := range m.entries { + if p == prefix || strings.HasPrefix(p, prefix+"/") { + delete(m.entries, p) + } + } + return &filer_pb.DeleteEntryResponse{}, nil +} + +func (m *memFiler) UpdateEntry(_ context.Context, in *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + m.entries[path.Join(in.Directory, in.Entry.Name)] = in.Entry + return &filer_pb.UpdateEntryResponse{}, nil +} + +func newCreateViewRequest(t *testing.T, namespace, name, sql string) *http.Request { + t.Helper() + schema := iceberg.NewSchemaWithIdentifiers(0, nil, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + ) + ver, err := view.NewVersionFromSQL(1, schema.ID, sql, table.Identifier{namespace}) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(CreateViewRequest{Name: name, Schema: schema, ViewVersion: ver}) + if err != nil { + t.Fatal(err) + } + r := httptest.NewRequest(http.MethodPost, "/v1/namespaces/"+namespace+"/views", bytes.NewReader(body)) + r = mux.SetURLVars(r, map[string]string{"namespace": namespace}) + r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), s3_constants.AccountAdminId)) + return r +} + +// seedNamespace registers a bucket and namespace so the s3tables existence and +// auth-context lookups pass; ownership is irrelevant since the admin principal +// is always allowed. +func seedNamespace(fc *memFiler, bucket, namespace string) { + fc.seed(s3tables.GetTableBucketPath(bucket), &filer_pb.Entry{Name: bucket, IsDirectory: true}) + meta, _ := json.Marshal(map[string]any{"namespace": []string{namespace}, "ownerAccountId": s3_constants.AccountAdminId}) + fc.seed(s3tables.GetNamespacePath(bucket, namespace), &filer_pb.Entry{ + Name: namespace, + IsDirectory: true, + Extended: map[string][]byte{s3tables.ExtendedKeyMetadata: meta}, + }) +} + +func TestCreateViewMissingNamespaceReturns404(t *testing.T) { + fc := newMemFiler() + s := NewServer(fc, nil) + + w := httptest.NewRecorder() + s.handleCreateView(w, newCreateViewRequest(t, "ns", "v", "SELECT 1")) + + if w.Code != http.StatusNotFound { + t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusNotFound, w.Body.String()) + } + var errResp ErrorResponse + if err := json.Unmarshal(w.Body.Bytes(), &errResp); err != nil { + t.Fatalf("unmarshal error body: %v", err) + } + if errResp.Error.Type != "NoSuchNamespaceException" { + t.Fatalf("error type = %q, want NoSuchNamespaceException", errResp.Error.Type) + } +} + +func TestCreateViewTagsEntryAsView(t *testing.T) { + const bucket = "warehouse" + fc := newMemFiler() + seedNamespace(fc, bucket, "ns") + s := NewServer(fc, nil) + + w := httptest.NewRecorder() + s.handleCreateView(w, newCreateViewRequest(t, "ns", "v", "SELECT 1")) + if w.Code != http.StatusOK { + t.Fatalf("create status = %d, want 200 (body: %s)", w.Code, w.Body.String()) + } + + entry, ok := fc.entries[s3tables.GetTablePath(bucket, "ns", "v")] + if !ok { + t.Fatalf("view entry not created") + } + if got := string(entry.Extended[s3tables.ExtendedKeyEntryType]); got != s3tables.EntryTypeView { + t.Fatalf("entryType = %q, want %q", got, s3tables.EntryTypeView) + } + if _, ok := entry.Extended[s3tables.ExtendedKeyMetadata]; !ok { + t.Fatalf("view entry missing metadata attribute") + } +} + +func TestCreateViewDuplicateDoesNotClobberMetadata(t *testing.T) { + const bucket = "warehouse" + fc := newMemFiler() + seedNamespace(fc, bucket, "ns") + s := NewServer(fc, nil) + + w := httptest.NewRecorder() + s.handleCreateView(w, newCreateViewRequest(t, "ns", "v", "SELECT 1")) + if w.Code != http.StatusOK { + t.Fatalf("first create status = %d, want 200 (body: %s)", w.Code, w.Body.String()) + } + + metadataKey := path.Join(s3tables.GetTablePath(bucket, "ns", "v"), "metadata", "v1.metadata.json") + first, ok := fc.entries[metadataKey] + if !ok { + t.Fatalf("metadata file not written at %s", metadataKey) + } + original := append([]byte(nil), first.Content...) + + // Re-create the same view with different SQL: the existence pre-check must + // short-circuit so the persisted v1.metadata.json stays byte-for-byte intact. + w = httptest.NewRecorder() + s.handleCreateView(w, newCreateViewRequest(t, "ns", "v", "SELECT 2")) + if w.Code != http.StatusOK { + t.Fatalf("duplicate create status = %d, want 200 (body: %s)", w.Code, w.Body.String()) + } + if got := fc.entries[metadataKey].Content; !bytes.Equal(got, original) { + t.Fatalf("duplicate create clobbered stored metadata:\n got %s\nwant %s", got, original) + } +} + +func TestCreateViewRollsBackEntryWhenMetadataWriteFails(t *testing.T) { + const bucket = "warehouse" + fc := newMemFiler() + seedNamespace(fc, bucket, "ns") + fc.failFileCreate = errors.New("disk full") + s := NewServer(fc, nil) + + w := httptest.NewRecorder() + s.handleCreateView(w, newCreateViewRequest(t, "ns", "v", "SELECT 1")) + if w.Code != http.StatusInternalServerError { + t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusInternalServerError, w.Body.String()) + } + + // The registered view must be rolled back so it doesn't linger pointing at + // metadata that was never written. + if _, ok := fc.entries[s3tables.GetTablePath(bucket, "ns", "v")]; ok { + t.Fatalf("view entry left behind after metadata write failure") + } +} diff --git a/weed/s3api/iceberg/iceberg_view_test.go b/weed/s3api/iceberg/iceberg_view_test.go new file mode 100644 index 000000000..69de9fdfe --- /dev/null +++ b/weed/s3api/iceberg/iceberg_view_test.go @@ -0,0 +1,115 @@ +package iceberg + +import ( + "encoding/json" + "errors" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/apache/iceberg-go/view" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +func TestIsViewNotFound(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {nil, false}, + {&s3tables.S3TablesError{Type: s3tables.ErrCodeNoSuchView, Message: "view x not found"}, true}, + {&s3tables.S3TablesError{Type: s3tables.ErrCodeNoSuchNamespace, Message: "namespace x not found"}, true}, + {errors.New("view x not found"), true}, + {&s3tables.S3TablesError{Type: s3tables.ErrCodeViewAlreadyExists, Message: "already exists"}, false}, + {errors.New("connection refused"), false}, + } + for _, c := range cases { + if got := isViewNotFound(c.err); got != c.want { + t.Errorf("isViewNotFound(%v) = %v, want %v", c.err, got, c.want) + } + } +} + +func TestIsViewAlreadyExists(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {nil, false}, + {&s3tables.S3TablesError{Type: s3tables.ErrCodeViewAlreadyExists, Message: "already exists"}, true}, + {errors.New("view already exists"), true}, + {&s3tables.S3TablesError{Type: s3tables.ErrCodeNoSuchView, Message: "not found"}, false}, + {errors.New("connection refused"), false}, + } + for _, c := range cases { + if got := isViewAlreadyExists(c.err); got != c.want { + t.Errorf("isViewAlreadyExists(%v) = %v, want %v", c.err, got, c.want) + } + } +} + +func TestViewPropertiesCopiesAndIsNonNil(t *testing.T) { + if got := viewProperties(nil); got == nil { + t.Fatal("viewProperties(nil) = nil, want non-nil") + } + src := iceberg.Properties{"k": "v"} + out := viewProperties(src) + out["k2"] = "v2" + if _, ok := src["k2"]; ok { + t.Fatal("viewProperties did not copy: mutation leaked into source") + } + if out["k"] != "v" { + t.Fatalf("viewProperties dropped key: got %q", out["k"]) + } +} + +// TestViewResponseRoundTrip verifies a built view metadata marshals into the +// ViewResponse wire shape and parses back via the iceberg-go view parser. +func TestViewResponseRoundTrip(t *testing.T) { + schema := iceberg.NewSchemaWithIdentifiers(0, nil, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + ) + ver, err := view.NewVersionFromSQL(1, schema.ID, "SELECT 1", table.Identifier{"ns"}) + if err != nil { + t.Fatal(err) + } + md, err := view.NewMetadata(ver, schema, "s3://bucket/ns/v", iceberg.Properties{}) + if err != nil { + t.Fatal(err) + } + + resp := ViewResponse{ + MetadataLocation: "s3://bucket/ns/v/metadata/v1.metadata.json", + Metadata: md, + Config: iceberg.Properties{}, + } + data, err := json.Marshal(resp) + if err != nil { + t.Fatalf("marshal: %v", err) + } + + var raw map[string]json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + t.Fatalf("unmarshal envelope: %v", err) + } + if _, ok := raw["metadata-location"]; !ok { + t.Fatal("response missing metadata-location") + } + if _, ok := raw["metadata"]; !ok { + t.Fatal("response missing metadata") + } + + var back ViewResponse + if err := json.Unmarshal(data, &back); err != nil { + t.Fatalf("unmarshal back: %v", err) + } + if back.Metadata == nil { + t.Fatal("metadata nil after round-trip") + } + if back.Metadata.CurrentVersion().VersionID != 1 { + t.Fatalf("current version = %d, want 1", back.Metadata.CurrentVersion().VersionID) + } + if back.MetadataLocation != resp.MetadataLocation { + t.Fatalf("metadata-location = %q, want %q", back.MetadataLocation, resp.MetadataLocation) + } +} diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go index 98ce0cac4..8f6f29237 100644 --- a/weed/s3api/iceberg/server.go +++ b/weed/s3api/iceberg/server.go @@ -108,6 +108,14 @@ func (s *Server) RegisterRoutes(router *mux.Router) { 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) + // View endpoints - wrapped with Auth middleware + router.HandleFunc("/v1/namespaces/{namespace}/views", s.Auth(s.handleListViews)).Methods(http.MethodGet) + router.HandleFunc("/v1/namespaces/{namespace}/views", s.Auth(s.handleCreateView)).Methods(http.MethodPost) + router.HandleFunc("/v1/namespaces/{namespace}/views/{view}", s.Auth(s.handleLoadView)).Methods(http.MethodGet) + router.HandleFunc("/v1/namespaces/{namespace}/views/{view}", s.Auth(s.handleViewExists)).Methods(http.MethodHead) + router.HandleFunc("/v1/namespaces/{namespace}/views/{view}", s.Auth(s.handleDropView)).Methods(http.MethodDelete) + router.HandleFunc("/v1/namespaces/{namespace}/views/{view}", s.Auth(s.handleUpdateView)).Methods(http.MethodPost) + // Multi-table transaction commit - wrapped with Auth middleware router.HandleFunc("/v1/transactions/commit", s.Auth(s.handleCommitTransaction)).Methods(http.MethodPost) @@ -125,6 +133,12 @@ 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}/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) + router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views/{view}", s.Auth(s.handleViewExists)).Methods(http.MethodHead) + router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views/{view}", s.Auth(s.handleDropView)).Methods(http.MethodDelete) + router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views/{view}", s.Auth(s.handleUpdateView)).Methods(http.MethodPost) router.HandleFunc("/v1/{prefix}/transactions/commit", s.Auth(s.handleCommitTransaction)).Methods(http.MethodPost) // Catch-all for debugging diff --git a/weed/s3api/iceberg/types.go b/weed/s3api/iceberg/types.go index 2f54189e8..316d3b824 100644 --- a/weed/s3api/iceberg/types.go +++ b/weed/s3api/iceberg/types.go @@ -7,6 +7,7 @@ import ( "github.com/apache/iceberg-go" "github.com/apache/iceberg-go/table" + "github.com/apache/iceberg-go/view" ) // CatalogConfig is returned by GET /v1/config. @@ -214,3 +215,57 @@ func (r *CommitTableResponse) UnmarshalJSON(data []byte) error { return nil } + +// ListViewsResponse is returned by GET /v1/namespaces/{namespace}/views. +type ListViewsResponse struct { + NextPageToken string `json:"next-page-token,omitempty"` + Identifiers []TableIdentifier `json:"identifiers"` +} + +// CreateViewRequest is sent to POST /v1/namespaces/{namespace}/views. +type CreateViewRequest struct { + Name string `json:"name"` + Schema *iceberg.Schema `json:"schema"` + Location string `json:"location,omitempty"` + Properties iceberg.Properties `json:"properties,omitempty"` + ViewVersion *view.Version `json:"view-version"` +} + +// ViewResponse is returned by view create/update and load operations. +type ViewResponse struct { + MetadataLocation string `json:"metadata-location"` + Metadata view.Metadata `json:"metadata"` + Config iceberg.Properties `json:"config"` +} + +// viewResponseAlias is used for custom JSON unmarshaling. +type viewResponseAlias struct { + MetadataLocation string `json:"metadata-location"` + RawMetadata json.RawMessage `json:"metadata"` + Config iceberg.Properties `json:"config,omitempty"` +} + +// UnmarshalJSON parses view.Metadata using iceberg-go's view parser. +func (r *ViewResponse) UnmarshalJSON(data []byte) error { + var alias viewResponseAlias + if err := json.Unmarshal(data, &alias); err != nil { + return err + } + r.MetadataLocation = alias.MetadataLocation + r.Config = alias.Config + if len(alias.RawMetadata) > 0 { + metadata, err := view.ParseMetadataBytes(alias.RawMetadata) + if err != nil { + return err + } + r.Metadata = metadata + } + return nil +} + +// UpdateViewRequest is sent to POST /v1/namespaces/{namespace}/views/{view}. +type UpdateViewRequest struct { + Identifier *TableIdentifier `json:"identifier,omitempty"` + Requirements view.Requirements `json:"requirements"` + Updates view.Updates `json:"updates"` +} diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 7a0ad66ff..cf032aa06 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -84,6 +84,33 @@ func (h *S3TablesHandler) setExtendedAttribute(ctx context.Context, client filer }) } +// 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 { + 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 { + entry.Extended = make(map[string][]byte) + } + for key, data := range attrs { + entry.Extended[key] = data + } + + return filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ + Directory: dir, + Entry: entry, + }) +} + // getExtendedAttribute gets an extended attribute from an entry func (h *S3TablesHandler) getExtendedAttribute(ctx context.Context, client filer_pb.SeaweedFilerClient, path, key string) ([]byte, error) { dir, name := splitPath(path) @@ -107,6 +134,19 @@ func (h *S3TablesHandler) getExtendedAttribute(ctx context.Context, client filer return data, nil } +// lookupEntry returns the filer entry at the given path. +func (h *S3TablesHandler) lookupEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) (*filer_pb.Entry, error) { + dir, name := splitPath(path) + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ + Directory: dir, + Name: name, + }) + if err != nil { + return nil, err + } + return resp.Entry, nil +} + // deleteExtendedAttribute deletes an extended attribute from an entry func (h *S3TablesHandler) deleteExtendedAttribute(ctx context.Context, client filer_pb.SeaweedFilerClient, path, key string) error { dir, name := splitPath(path) diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index dbc71f1d2..101aec6a6 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -26,6 +26,12 @@ const ( ExtendedKeyMetadataVersion = "s3tables.metadataVersion" ExtendedKeyPolicy = "s3tables.policy" ExtendedKeyTags = "s3tables.tags" + ExtendedKeyEntryType = "s3tables.entryType" + + // Entry-type marker values for ExtendedKeyEntryType. Absent or "table" means + // a table; views are stored like tables but tagged "view". + EntryTypeTable = "table" + EntryTypeView = "view" // Maximum request body size (10MB) maxRequestBodySize = 10 * 1024 * 1024 @@ -153,6 +159,18 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request, case "DeleteTable": err = h.handleDeleteTable(w, r, filerClient) + // View operations + case "CreateView": + err = h.handleCreateView(w, r, filerClient) + case "GetView": + err = h.handleGetView(w, r, filerClient) + case "ListViews": + err = h.handleListViews(w, r, filerClient) + case "UpdateView": + err = h.handleUpdateView(w, r, filerClient) + case "DeleteView": + err = h.handleDeleteView(w, r, filerClient) + // Table Policy operations case "PutTablePolicy": err = h.handlePutTablePolicy(w, r, filerClient) @@ -363,6 +381,10 @@ func (h *S3TablesHandler) generateTableARN(ownerAccountID, bucketName, tableID s return fmt.Sprintf("arn:aws:s3tables:%s:%s:bucket/%s/table/%s", h.region, ownerAccountID, bucketName, tableID) } +func (h *S3TablesHandler) generateViewARN(ownerAccountID, bucketName, viewID string) string { + return fmt.Sprintf("arn:aws:s3tables:%s:%s:bucket/%s/view/%s", h.region, ownerAccountID, bucketName, viewID) +} + func isAuthError(err error) bool { var authErr *AuthError return errors.As(err, &authErr) || errors.Is(err, ErrAccessDenied) diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 31b35192d..9e514db3d 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -164,13 +164,23 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque tablePath := GetTablePath(bucketName, namespaceName, tableName) - // Check if table already exists + // Check if a table or view already exists at this name. Names are unique + // across tables and views in a namespace. var existingMetadata tableMetadataInternal + var existingIsView bool err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata) + entry, err := h.lookupEntry(r.Context(), client, tablePath) if err != nil { return err } + if entryType(entry.Extended) == EntryTypeView { + existingIsView = true + return nil + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata) + } if unmarshalErr := json.Unmarshal(data, &existingMetadata); unmarshalErr != nil { return fmt.Errorf("failed to parse existing table metadata: %w", unmarshalErr) } @@ -178,6 +188,10 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque }) if err == nil { + if existingIsView { + h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("a view named %s already exists", tableName)) + return fmt.Errorf("view name conflict: %s", tableName) + } tableARN := h.generateTableARN(existingMetadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName) h.writeJSON(w, http.StatusOK, &CreateTableResponse{ TableARN: tableARN, @@ -230,6 +244,11 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque return err } + // Tag the entry as a table so view listings can exclude it. + if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyEntryType, []byte(EntryTypeTable)); err != nil { + return err + } + // Set tags if provided if len(req.Tags) > 0 { tagsBytes, err := json.Marshal(req.Tags) @@ -519,10 +538,17 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request, var metadata tableMetadataInternal err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata) + entry, err := h.lookupEntry(r.Context(), client, tablePath) if err != nil { return err } + if entryType(entry.Extended) == EntryTypeView { + return filer_pb.ErrNotFound + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata) + } if err := json.Unmarshal(data, &metadata); err != nil { return fmt.Errorf("failed to unmarshal table metadata: %w", err) } @@ -886,6 +912,11 @@ func (h *S3TablesHandler) listTablesWithClient(r *http.Request, client filer_pb. continue } + // Views share the table layout; exclude them from table listings. + if entryType(entry.Entry.Extended) == EntryTypeView { + continue + } + // Read table metadata from extended attribute data, ok := entry.Entry.Extended[ExtendedKeyMetadata] if !ok { diff --git a/weed/s3api/s3tables/handler_view.go b/weed/s3api/s3tables/handler_view.go new file mode 100644 index 000000000..79a4e8f0c --- /dev/null +++ b/weed/s3api/s3tables/handler_view.go @@ -0,0 +1,648 @@ +package s3tables + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// handleCreateView creates a new view in a namespace. Views share the table +// storage layout but are tagged ExtendedKeyEntryType="view". Names are unique +// across tables and views in a namespace. +func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + var req CreateViewRequest + 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") + } + + bucketName, err := parseBucketNameFromARN(req.TableBucketARN) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + viewName, err := validateTableName(req.Name) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + // Check if namespace exists + 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 + } + return json.Unmarshal(data, &namespaceMetadata) + }) + 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 + } + + accountID := h.getAccountID(r) + bucketPath := GetTableBucketPath(bucketName) + namespacePolicy, bucketPolicy, bucketTags, bucketMetadata, err := h.loadAuthContext(r, filerClient, namespacePath, bucketPath) + 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) + if !h.authorizeViewOp(r, "CreateView", accountID, namespaceMetadata.OwnerAccountID, bucketMetadata.OwnerAccountID, namespacePolicy, bucketPolicy, bucketARN, bucketName, namespaceName, viewName, bucketTags) { + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create view in this namespace") + return ErrAccessDenied + } + + viewPath := GetTablePath(bucketName, namespaceName, viewName) + + // Reject if a table or view already exists at this name. + var existingMetadata tableMetadataInternal + var existingIsTable bool + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + entry, err := h.lookupEntry(r.Context(), client, viewPath) + if err != nil { + return err + } + if entryType(entry.Extended) != EntryTypeView { + existingIsTable = true + return nil + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata) + } + return json.Unmarshal(data, &existingMetadata) + }) + if err == nil { + if existingIsTable { + h.writeError(w, http.StatusConflict, ErrCodeViewAlreadyExists, fmt.Sprintf("a table named %s already exists", viewName)) + return fmt.Errorf("table name conflict: %s", viewName) + } + viewARN := h.generateViewARN(existingMetadata.OwnerAccountID, bucketName, namespaceName+"/"+viewName) + h.writeJSON(w, http.StatusOK, &CreateViewResponse{ + ViewARN: viewARN, + VersionToken: existingMetadata.VersionToken, + MetadataLocation: existingMetadata.MetadataLocation, + }) + return nil + } else if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check view: %v", err)) + return err + } + + now := time.Now() + versionToken := generateVersionToken() + metadata := &tableMetadataInternal{ + Name: viewName, + Namespace: namespaceName, + Format: "ICEBERG", + CreatedAt: now, + ModifiedAt: now, + OwnerAccountID: namespaceMetadata.OwnerAccountID, + VersionToken: versionToken, + MetadataVersion: max(req.MetadataVersion, 1), + MetadataLocation: req.MetadataLocation, + Metadata: req.Metadata, + } + + metadataBytes, err := json.Marshal(metadata) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal view metadata") + return err + } + + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + if err := h.ensureDirectory(r.Context(), client, viewPath); err != nil { + return err + } + return h.setExtendedAttributes(r.Context(), client, viewPath, map[string][]byte{ + ExtendedKeyMetadata: metadataBytes, + ExtendedKeyEntryType: []byte(EntryTypeView), + }) + }) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create view") + return err + } + + viewARN := h.generateViewARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+viewName) + h.writeJSON(w, http.StatusOK, &CreateViewResponse{ + ViewARN: viewARN, + VersionToken: versionToken, + MetadataLocation: metadata.MetadataLocation, + }) + return nil +} + +// handleGetView returns the metadata pointer for a view. +func (h *S3TablesHandler) handleGetView(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + var req GetViewRequest + if err := h.readRequestBody(r, &req); err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + bucketName, namespace, viewName, err := h.parseViewRef(req.TableBucketARN, req.Namespace, req.Name) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + viewPath := GetTablePath(bucketName, namespace, viewName) + var metadata tableMetadataInternal + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + entry, err := h.lookupEntry(r.Context(), client, viewPath) + if err != nil { + return err + } + if entryType(entry.Extended) != EntryTypeView { + return filer_pb.ErrNotFound + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata) + } + return json.Unmarshal(data, &metadata) + }) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchView, fmt.Sprintf("view %s not found", viewName)) + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to get view: %v", err)) + } + return err + } + + accountID := h.getAccountID(r) + viewPolicy, bucketPolicy, bucketTags, bucketMetadata, err := h.loadAuthContext(r, filerClient, viewPath, GetTableBucketPath(bucketName)) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to fetch policies: %v", err)) + return err + } + viewARN := h.generateViewARN(metadata.OwnerAccountID, bucketName, namespace+"/"+viewName) + if !h.authorizeViewOp(r, "GetView", accountID, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, viewPolicy, bucketPolicy, viewARN, bucketName, namespace, viewName, bucketTags) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchView, fmt.Sprintf("view %s not found", viewName)) + return ErrAccessDenied + } + + h.writeJSON(w, http.StatusOK, &GetViewResponse{ + Name: metadata.Name, + ViewARN: viewARN, + Namespace: expandNamespace(metadata.Namespace), + CreatedAt: metadata.CreatedAt, + ModifiedAt: metadata.ModifiedAt, + OwnerAccountID: metadata.OwnerAccountID, + MetadataLocation: metadata.MetadataLocation, + MetadataVersion: metadata.MetadataVersion, + VersionToken: metadata.VersionToken, + Metadata: metadata.Metadata, + }) + return nil +} + +// handleListViews lists views in a namespace, excluding table entries. +func (h *S3TablesHandler) handleListViews(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + var req ListViewsRequest + 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") + } + + bucketName, err := parseBucketNameFromARN(req.TableBucketARN) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + maxViews := req.MaxViews + if maxViews <= 0 { + maxViews = 100 + } + const maxViewsLimit = 1000 + if maxViews > maxViewsLimit { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "MaxViews exceeds maximum allowed value") + return fmt.Errorf("invalid maxViews value: %d", maxViews) + } + + namespaceName, err := validateNamespace(req.Namespace) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + var views []ViewSummary + var paginationToken string + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + namespacePath := GetNamespacePath(bucketName, namespaceName) + bucketPath := GetTableBucketPath(bucketName) + var nsMeta namespaceMetadata + data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata) + if err != nil { + return err + } + if err := json.Unmarshal(data, &nsMeta); err != nil { + return err + } + + namespacePolicy, bucketPolicy, bucketTags, bucketMeta, err := h.loadAuthContext(r, &singleClient{client}, namespacePath, bucketPath) + if err != nil { + return err + } + bucketARN := h.generateTableBucketARN(bucketMeta.OwnerAccountID, bucketName) + if !h.authorizeViewOp(r, "ListViews", h.getAccountID(r), nsMeta.OwnerAccountID, bucketMeta.OwnerAccountID, namespacePolicy, bucketPolicy, bucketARN, bucketName, namespaceName, "", bucketTags) { + return ErrAccessDenied + } + + views, paginationToken, err = h.listViewsInNamespace(r, client, bucketName, namespaceName, req.Prefix, req.ContinuationToken, maxViews) + return err + }) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + views = []ViewSummary{} + paginationToken = "" + } else if isAuthError(err) { + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "Access Denied") + return err + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to list views: %v", err)) + return err + } + } + + h.writeJSON(w, http.StatusOK, &ListViewsResponse{ + Views: views, + ContinuationToken: paginationToken, + }) + return nil +} + +func (h *S3TablesHandler) listViewsInNamespace(r *http.Request, client filer_pb.SeaweedFilerClient, bucketName, namespaceName, prefix, continuationToken string, maxViews int) ([]ViewSummary, string, error) { + namespacePath := GetNamespacePath(bucketName, namespaceName) + var views []ViewSummary + lastFileName := continuationToken + ctx := r.Context() + + for len(views) < maxViews { + resp, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{ + Directory: namespacePath, + Limit: uint32(maxViews * 2), + StartFromFileName: lastFileName, + InclusiveStartFrom: lastFileName == "" || lastFileName == continuationToken, + }) + if err != nil { + return nil, "", err + } + + hasMore := false + for { + entry, respErr := resp.Recv() + if respErr != nil { + if respErr == io.EOF { + break + } + return nil, "", respErr + } + if entry.Entry == nil { + continue + } + if len(views) == 0 && continuationToken != "" && entry.Entry.Name == continuationToken { + continue + } + + hasMore = true + lastFileName = entry.Entry.Name + + if !entry.Entry.IsDirectory || strings.HasPrefix(entry.Entry.Name, ".") { + continue + } + if prefix != "" && !strings.HasPrefix(entry.Entry.Name, prefix) { + continue + } + // Only include view entries; skip tables and untagged entries. + if entryType(entry.Entry.Extended) != EntryTypeView { + continue + } + data, ok := entry.Entry.Extended[ExtendedKeyMetadata] + if !ok { + continue + } + var metadata tableMetadataInternal + if err := json.Unmarshal(data, &metadata); err != nil { + continue + } + + views = append(views, ViewSummary{ + Name: entry.Entry.Name, + ViewARN: h.generateViewARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+entry.Entry.Name), + Namespace: expandNamespace(namespaceName), + CreatedAt: metadata.CreatedAt, + ModifiedAt: metadata.ModifiedAt, + }) + + if len(views) >= maxViews { + return views, lastFileName, nil + } + } + + if !hasMore { + break + } + } + + if len(views) < maxViews { + lastFileName = "" + } + return views, lastFileName, nil +} + +// handleUpdateView replaces the metadata pointer for a view, mirroring UpdateTable. +func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + var req UpdateViewRequest + if err := h.readRequestBody(r, &req); err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + bucketName, namespaceName, viewName, err := h.parseViewRef(req.TableBucketARN, req.Namespace, req.Name) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + viewPath := GetTablePath(bucketName, namespaceName, viewName) + var metadata tableMetadataInternal + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + entry, err := h.lookupEntry(r.Context(), client, viewPath) + if err != nil { + return err + } + if entryType(entry.Extended) != EntryTypeView { + return filer_pb.ErrNotFound + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata) + } + return json.Unmarshal(data, &metadata) + }) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchView, "view not found") + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, err.Error()) + } + return err + } + + accountID := h.getAccountID(r) + viewPolicy, bucketPolicy, bucketTags, bucketMetadata, err := h.loadAuthContext(r, filerClient, viewPath, GetTableBucketPath(bucketName)) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, err.Error()) + return err + } + viewARN := h.generateViewARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+viewName) + if !h.authorizeViewOp(r, "UpdateView", accountID, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, viewPolicy, bucketPolicy, viewARN, bucketName, namespaceName, viewName, bucketTags) { + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to update view") + return NewAuthError("UpdateView", accountID, "not authorized to update view") + } + + if req.VersionToken != "" && req.VersionToken != metadata.VersionToken { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "Version token mismatch") + return ErrVersionTokenMismatch + } + + if req.Metadata != nil { + if metadata.Metadata == nil { + metadata.Metadata = &TableMetadata{} + } + if req.Metadata.Iceberg != nil { + if metadata.Metadata.Iceberg == nil { + metadata.Metadata.Iceberg = &IcebergMetadata{} + } + if req.Metadata.Iceberg.TableUUID != "" { + metadata.Metadata.Iceberg.TableUUID = req.Metadata.Iceberg.TableUUID + } + } + if len(req.Metadata.FullMetadata) > 0 { + metadata.Metadata.FullMetadata = req.Metadata.FullMetadata + } + } + if req.MetadataLocation != "" { + metadata.MetadataLocation = req.MetadataLocation + } + if req.MetadataVersion > 0 { + metadata.MetadataVersion = req.MetadataVersion + } else if metadata.MetadataVersion == 0 { + metadata.MetadataVersion = 1 + } + metadata.ModifiedAt = time.Now() + metadata.VersionToken = generateVersionToken() + + metadataBytes, err := json.Marshal(metadata) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal metadata") + return err + } + + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return h.setExtendedAttribute(r.Context(), client, viewPath, ExtendedKeyMetadata, metadataBytes) + }) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to update metadata") + return err + } + + h.writeJSON(w, http.StatusOK, &UpdateViewResponse{ + ViewARN: viewARN, + MetadataLocation: metadata.MetadataLocation, + VersionToken: metadata.VersionToken, + }) + return nil +} + +// handleDeleteView deletes a view. +func (h *S3TablesHandler) handleDeleteView(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + var req DeleteViewRequest + if err := h.readRequestBody(r, &req); err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + bucketName, namespaceName, viewName, err := h.parseViewRef(req.TableBucketARN, req.Namespace, req.Name) + if err != nil { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) + return err + } + + viewPath := GetTablePath(bucketName, namespaceName, viewName) + var metadata tableMetadataInternal + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + entry, err := h.lookupEntry(r.Context(), client, viewPath) + if err != nil { + return err + } + if entryType(entry.Extended) != EntryTypeView { + return filer_pb.ErrNotFound + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata) + } + if err := json.Unmarshal(data, &metadata); err != nil { + return err + } + if req.VersionToken != "" && metadata.VersionToken != req.VersionToken { + return ErrVersionTokenMismatch + } + return nil + }) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchView, fmt.Sprintf("view %s not found", viewName)) + } else if errors.Is(err, ErrVersionTokenMismatch) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "version token mismatch") + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check view: %v", err)) + } + return err + } + + accountID := h.getAccountID(r) + viewPolicy, bucketPolicy, bucketTags, bucketMetadata, err := h.loadAuthContext(r, filerClient, viewPath, GetTableBucketPath(bucketName)) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, err.Error()) + return err + } + viewARN := h.generateViewARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+viewName) + if !h.authorizeViewOp(r, "DeleteView", accountID, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, viewPolicy, bucketPolicy, viewARN, bucketName, namespaceName, viewName, bucketTags) { + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to delete view") + return NewAuthError("DeleteView", accountID, "not authorized to delete view") + } + + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return h.deleteDirectory(r.Context(), client, viewPath) + }) + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete view") + return err + } + + h.writeJSON(w, http.StatusOK, nil) + return nil +} + +// parseViewRef validates and resolves a (bucketARN, namespace, name) triple. +func (h *S3TablesHandler) parseViewRef(bucketARN string, namespace []string, name string) (bucketName, namespaceName, viewName string, err error) { + if bucketARN == "" || len(namespace) == 0 || name == "" { + return "", "", "", fmt.Errorf("tableBucketARN, namespace, and name are required") + } + if bucketName, err = parseBucketNameFromARN(bucketARN); err != nil { + return "", "", "", err + } + if namespaceName, err = validateNamespace(namespace); err != nil { + return "", "", "", err + } + if viewName, err = validateTableName(name); err != nil { + return "", "", "", err + } + return bucketName, namespaceName, viewName, nil +} + +// loadAuthContext fetches the resource policy, bucket policy, bucket tags and +// bucket metadata used to authorize a view operation. +func (h *S3TablesHandler) loadAuthContext(r *http.Request, filerClient FilerClient, resourcePath, bucketPath string) (resourcePolicy, bucketPolicy string, bucketTags map[string]string, bucketMetadata tableBucketMetadata, err error) { + bucketTags = map[string]string{} + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + data, e := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata) + if e == nil { + if e := json.Unmarshal(data, &bucketMetadata); e != nil { + return fmt.Errorf("failed to unmarshal bucket metadata: %w", e) + } + } else if !errors.Is(e, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch bucket metadata: %w", e) + } + + policyData, e := h.getExtendedAttribute(r.Context(), client, resourcePath, ExtendedKeyPolicy) + if e == nil { + resourcePolicy = string(policyData) + } else if !errors.Is(e, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch resource policy: %w", e) + } + + policyData, e = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy) + if e == nil { + bucketPolicy = string(policyData) + } else if !errors.Is(e, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch bucket policy: %w", e) + } + if tags, e := h.readTags(r.Context(), client, bucketPath); e != nil { + return e + } else if tags != nil { + bucketTags = tags + } + return nil + }) + return resourcePolicy, bucketPolicy, bucketTags, bucketMetadata, err +} + +// authorizeViewOp evaluates the resource and bucket policies for a view operation. +func (h *S3TablesHandler) authorizeViewOp(r *http.Request, operation, principal, resourceOwner, bucketOwner, resourcePolicy, bucketPolicy, resourceARN, bucketName, namespace, viewName string, bucketTags map[string]string) bool { + identityActions := getIdentityActions(r) + ctx := &PolicyContext{ + TableBucketName: bucketName, + Namespace: namespace, + TableName: viewName, + TableBucketTags: bucketTags, + IdentityActions: identityActions, + DefaultAllow: h.defaultAllowFor(r), + } + resourceAllowed := CheckPermissionWithContext(operation, principal, resourceOwner, resourcePolicy, resourceARN, ctx) + bucketARN := h.generateTableBucketARN(bucketOwner, bucketName) + bucketAllowed := CheckPermissionWithContext(operation, principal, bucketOwner, bucketPolicy, bucketARN, ctx) + return resourceAllowed || bucketAllowed +} + +// singleClient adapts a bound filer client to the FilerClient interface so the +// shared helpers can be reused inside an existing WithFilerClient closure. +type singleClient struct { + client filer_pb.SeaweedFilerClient +} + +func (s *singleClient) WithFilerClient(streamingMode bool, fn func(client filer_pb.SeaweedFilerClient) error) error { + return fn(s.client) +} diff --git a/weed/s3api/s3tables/handler_view_test.go b/weed/s3api/s3tables/handler_view_test.go new file mode 100644 index 000000000..0a9cc0e5a --- /dev/null +++ b/weed/s3api/s3tables/handler_view_test.go @@ -0,0 +1,33 @@ +package s3tables + +import "testing" + +func TestEntryType(t *testing.T) { + cases := []struct { + name string + extended map[string][]byte + want string + }{ + {"nil extended is table", nil, EntryTypeTable}, + {"absent marker is table", map[string][]byte{ExtendedKeyMetadata: []byte("{}")}, EntryTypeTable}, + {"empty marker is table", map[string][]byte{ExtendedKeyEntryType: {}}, EntryTypeTable}, + {"table marker", map[string][]byte{ExtendedKeyEntryType: []byte(EntryTypeTable)}, EntryTypeTable}, + {"view marker", map[string][]byte{ExtendedKeyEntryType: []byte(EntryTypeView)}, EntryTypeView}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := entryType(c.extended); got != c.want { + t.Fatalf("entryType() = %q, want %q", got, c.want) + } + }) + } +} + +func TestGenerateViewARN(t *testing.T) { + h := NewS3TablesHandler() + got := h.generateViewARN(DefaultAccountID, "warehouse", "ns/v") + want := "arn:aws:s3tables:us-east-1:000000000000:bucket/warehouse/view/ns/v" + if got != want { + t.Fatalf("generateViewARN() = %q, want %q", got, want) + } +} diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index 21722451c..c732cebb9 100644 --- a/weed/s3api/s3tables/types.go +++ b/weed/s3api/s3tables/types.go @@ -271,6 +271,91 @@ type UpdateTableResponse struct { MetadataLocation string `json:"metadataLocation,omitempty"` } +// View types +// +// Views are stored exactly like tables (a filer directory carrying a metadata +// pointer xattr) but tagged with ExtendedKeyEntryType="view". They reuse +// TableMetadata for the metadata pointer. + +type CreateViewRequest struct { + TableBucketARN string `json:"tableBucketARN"` + Namespace []string `json:"namespace"` + Name string `json:"name"` + Metadata *TableMetadata `json:"metadata,omitempty"` + MetadataVersion int `json:"metadataVersion,omitempty"` + MetadataLocation string `json:"metadataLocation,omitempty"` +} + +type CreateViewResponse struct { + ViewARN string `json:"viewARN"` + VersionToken string `json:"versionToken"` + MetadataLocation string `json:"metadataLocation,omitempty"` +} + +type GetViewRequest struct { + TableBucketARN string `json:"tableBucketARN"` + Namespace []string `json:"namespace"` + Name string `json:"name"` +} + +type GetViewResponse struct { + Name string `json:"name"` + ViewARN string `json:"viewARN"` + Namespace []string `json:"namespace"` + CreatedAt time.Time `json:"createdAt"` + ModifiedAt time.Time `json:"modifiedAt"` + OwnerAccountID string `json:"ownerAccountId"` + MetadataLocation string `json:"metadataLocation,omitempty"` + VersionToken string `json:"versionToken"` + MetadataVersion int `json:"metadataVersion"` + Metadata *TableMetadata `json:"metadata,omitempty"` +} + +type ListViewsRequest struct { + TableBucketARN string `json:"tableBucketARN"` + Namespace []string `json:"namespace,omitempty"` + Prefix string `json:"prefix,omitempty"` + ContinuationToken string `json:"continuationToken,omitempty"` + MaxViews int `json:"maxViews,omitempty"` +} + +type ViewSummary struct { + Name string `json:"name"` + ViewARN string `json:"viewARN"` + Namespace []string `json:"namespace"` + CreatedAt time.Time `json:"createdAt"` + ModifiedAt time.Time `json:"modifiedAt"` + MetadataLocation string `json:"metadataLocation,omitempty"` +} + +type ListViewsResponse struct { + Views []ViewSummary `json:"views"` + ContinuationToken string `json:"continuationToken,omitempty"` +} + +type UpdateViewRequest struct { + TableBucketARN string `json:"tableBucketARN"` + Namespace []string `json:"namespace"` + Name string `json:"name"` + VersionToken string `json:"versionToken,omitempty"` + Metadata *TableMetadata `json:"metadata,omitempty"` + MetadataVersion int `json:"metadataVersion,omitempty"` + MetadataLocation string `json:"metadataLocation,omitempty"` +} + +type UpdateViewResponse struct { + ViewARN string `json:"viewARN"` + VersionToken string `json:"versionToken"` + MetadataLocation string `json:"metadataLocation,omitempty"` +} + +type DeleteViewRequest struct { + TableBucketARN string `json:"tableBucketARN"` + Namespace []string `json:"namespace"` + Name string `json:"name"` + VersionToken string `json:"versionToken,omitempty"` +} + // Table policy types type PutTablePolicyRequest struct { @@ -337,6 +422,8 @@ const ( ErrCodeNamespaceAlreadyExists = "NamespaceAlreadyExists" ErrCodeNamespaceNotEmpty = "NamespaceNotEmpty" ErrCodeTableAlreadyExists = "TableAlreadyExists" + ErrCodeNoSuchView = "NoSuchView" + ErrCodeViewAlreadyExists = "ViewAlreadyExists" ErrCodeAccessDenied = "AccessDenied" ErrCodeInvalidRequest = "InvalidRequest" ErrCodeInternalError = "InternalError" diff --git a/weed/s3api/s3tables/utils.go b/weed/s3api/s3tables/utils.go index 2aedefa2b..2cf3683bd 100644 --- a/weed/s3api/s3tables/utils.go +++ b/weed/s3api/s3tables/utils.go @@ -146,6 +146,19 @@ func IsTableBucketEntry(entry *filer_pb.Entry) bool { return ok } +// entryType returns the entry-type marker for a catalog entry. Tables and views +// share the same on-disk layout; the marker distinguishes them. An absent marker +// means table for back-compat. +func entryType(extended map[string][]byte) string { + if extended == nil { + return EntryTypeTable + } + if v, ok := extended[ExtendedKeyEntryType]; ok && len(v) > 0 { + return string(v) + } + return EntryTypeTable +} + // Utility functions // validateBucketName validates bucket name and returns an error if invalid.