Files
seaweedfs/weed/s3api/s3tables/handler_view.go
T
Chris LuandClaude Opus 5.5 4f24b756ba s3tables, iceberg: give tables created without a location their own directory (#12475)
* s3tables: give a table created without a location its own directory

An Iceberg table created without a location kept its files under its name
path, so a rename left them where the next table of that name would land.
Such a table now gets s3://<bucket>/<namespace>/<name>-<uuid>, stored as the
record's warehouseLocation and returned by GetTable; its first commit writes
there without moving the root. Records without one keep resolving to their
name path, so existing tables need no rewrite. Lance datasets stay at their
name path, where Lance clients and the vector engine look for them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: create tables and views in a directory of their own

A table or view created without a location took its name path, or a
unique one only when that path was already taken. It now always gets
<name>-<uuid>, so a later rename never leaves its files under a name
another table can take.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: load and commit a table at its warehouse location

A table created through S3 Tables without a location has no metadata until
its first commit; loading or committing it now uses the directory recorded
for it, and only a record without one falls back to its name path.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: fail a stage-create whose marker is not saved

The marker is how the commit finds the location the stage-create handed
out. Without it the commit fell back to the name path while the client's
files sat under the returned location.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: expect tables created without a location in their own directory

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: rename a table that keeps its files elsewhere before reusing its name

Enterprise keeps the storage marker on a renamed table's files, so the name
of a table whose files sit under its name path stays taken there.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3tables: remove the table directory when a create backs out

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: answer a create that lost the race with the registered table

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: answer a view create that lost the race with the registered view

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: fail a staged commit whose name another create registered

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: remove the directory of a create whose name another table took

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: keep the metadata of a staged commit whose name is taken

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: write the lakekeeper table's data under its warehouse location

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: commit a staged table at the location of its own stage

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: never continue another stage when a commit's own marker is gone

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: test filer honors exclusive creates and token-guarded deletes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-07 16:37:31 -08:00

828 lines
30 KiB
Go

package s3tables
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"strings"
"time"
"github.com/seaweedfs/seaweedfs/weed/glog"
"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)
namespaceMetadata, err := h.loadNamespaceMetadata(r.Context(), filerClient, bucketName, namespaceName)
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
}
// A view is Iceberg metadata, so it belongs only in a bucket that holds
// Iceberg tables.
if bucketMetadata.Format != "" && bucketMetadata.Format != FormatIceberg {
message := fmt.Sprintf("table bucket %s holds %s tables and cannot hold views", bucketName, bucketMetadata.Format)
h.writeError(w, http.StatusConflict, ErrCodeConflict, message)
return fmt.Errorf("%s", message)
}
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
}
// The view is bound to the namespace record it was authorized against.
isNamespace := isNamespaceCreatedAt(namespaceMetadata.CreatedAt)
if err := h.refuseClaimedNamespace(w, r, filerClient, namespacePath, namespaceName, isNamespace); err != nil {
return err
}
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 isUnfinishedCreate(entry.Extended) {
// Another create wrote this record and may still back it out, so
// it is not an existing view; the claim below refuses the name.
return filer_pb.ErrNotFound
}
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
}
if err := ValidateMetadataLocation(req.MetadataLocation, bucketName, namespaceName); err != nil {
h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
return err
}
root := CatalogRecordRoot(bucketName, viewPath, req.MetadataLocation, "")
if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, viewPath, root); err != nil {
return err
}
now := time.Now()
versionToken := generateVersionToken()
metadata := &tableMetadataInternal{
Name: viewName,
Namespace: namespaceName,
Format: FormatIceberg,
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
}
attrs := map[string][]byte{
ExtendedKeyMetadata: metadataBytes,
ExtendedKeyEntryType: []byte(EntryTypeView),
ExtendedKeyDeleting: []byte(createClaim),
}
if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, viewPath, root, attrs, viewPath); err != nil {
return err
}
return h.confirmParent(r.Context(), client, namespacePath, isNamespace, viewPath, attrs, viewPath)
})
if err != nil {
if errors.Is(err, ErrNameTaken) {
h.writeError(w, http.StatusConflict, ErrCodeViewAlreadyExists, fmt.Sprintf("view %s already exists", viewName))
return err
}
if errors.Is(err, ErrLocationInUse) {
h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error())
return err
}
if errors.Is(err, ErrConcurrentUpdate) {
h.writeError(w, http.StatusConflict, ErrCodeConflict, "view changed during create, retry the request")
return err
}
if h.writeNamespaceClosedError(w, err, namespaceName) {
return err
}
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
}
// A view whose create has not confirmed its namespace may still be
// backed out, so it is reported absent.
if EntryType(entry.Extended) != EntryTypeView || isUnfinishedCreate(entry.Extended) {
return filer_pb.ErrNotFound
}
data, ok := entry.Extended[ExtendedKeyMetadata]
if !ok {
return fmt.Errorf("%w: %w: %s", filer_pb.ErrNotFound, 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 {
if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) {
return err
}
if !h.authorizeAbsentTarget(r, &singleClient{client}, "ListViews", bucketName, namespaceName, "", "") {
return ErrAccessDenied
}
return filer_pb.ErrNotFound
}
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, untagged entries, and a
// record its create has not confirmed.
if EntryType(entry.Entry.Extended) != EntryTypeView || isUnfinishedCreate(entry.Entry.Extended) {
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
}
// updateViewTarget is the view state an update was authorized against: the
// metadata the version-token check reads, the stored values the
// optimistic-concurrency check compares and the fields the response echoes.
type updateViewTarget struct {
bucketName string
namespaceName string
viewName string
viewPath string
viewARN string
metadata tableMetadataInternal
storedMetadata []byte
storedPolicy []byte
}
// authorizeUpdateView validates the names an update names, loads the view and
// the policies governing it, and checks the caller may update it, writing the
// error response itself. Credential vending asks the same question without
// updating anything, so it runs this gate through Manager.AuthorizeUpdateView.
func (h *S3TablesHandler) authorizeUpdateView(w http.ResponseWriter, r *http.Request, filerClient FilerClient, req *UpdateViewRequest) (*updateViewTarget, error) {
bucketName, namespaceName, viewName, err := h.parseViewRef(req.TableBucketARN, req.Namespace, req.Name)
if err != nil {
h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
return nil, err
}
if err := ValidateMetadataLocation(req.MetadataLocation, bucketName, namespaceName); err != nil {
h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
return nil, err
}
viewPath := GetTablePath(bucketName, namespaceName, viewName)
var metadata tableMetadataInternal
var storedMetadata []byte
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 || isUnfinishedCreate(entry.Extended) {
return filer_pb.ErrNotFound
}
data, ok := entry.Extended[ExtendedKeyMetadata]
if !ok {
return fmt.Errorf("%w: %w: %s", filer_pb.ErrNotFound, ErrAttributeNotFound, ExtendedKeyMetadata)
}
storedMetadata = data
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 nil, 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 nil, 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 nil, NewAuthError("UpdateView", accountID, "not authorized to update view")
}
return &updateViewTarget{
bucketName: bucketName,
namespaceName: namespaceName,
viewName: viewName,
viewPath: viewPath,
viewARN: viewARN,
metadata: metadata,
storedMetadata: storedMetadata,
storedPolicy: []byte(viewPolicy),
}, 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
}
target, err := h.authorizeUpdateView(w, r, filerClient, &req)
if err != nil {
return err
}
metadata := target.metadata
viewPath := target.viewPath
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
}
}
newRoot := CatalogRecordRoot(target.bucketName, viewPath, req.MetadataLocation, metadata.WarehouseLocation)
movesRoot := req.MetadataLocation != "" && newRoot != metadata.root(target.bucketName, viewPath)
if movesRoot {
if err := h.refuseClaimedLocation(w, r, filerClient, target.bucketName, viewPath, newRoot); err != nil {
return err
}
}
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
}
// Conditional on the record this request read and authorized against, as
// UpdateTable's write is: a concurrent update would otherwise be dropped,
// and a view deleted and created again under the name would take a write
// nobody authorized for it.
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
write := func(extended map[string][]byte) error {
if !bytes.Equal(extended[ExtendedKeyMetadata], target.storedMetadata) {
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata)
}
if !bytes.Equal(extended[ExtendedKeyPolicy], target.storedPolicy) {
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy)
}
extended[ExtendedKeyMetadata] = metadataBytes
delete(extended, ExtendedKeyMoving)
return nil
}
if !movesRoot {
return h.mutateCatalogRecord(r.Context(), client, viewPath, write)
}
return h.moveCatalogRecord(r.Context(), client, target.bucketName, viewPath, newRoot, target.storedMetadata, metadataBytes, write)
})
if err != nil {
if errors.Is(err, ErrLocationInUse) {
h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error())
return err
}
if errors.Is(err, ErrConcurrentUpdate) {
h.writeError(w, http.StatusConflict, ErrCodeConflict, "view changed concurrently, retry the request")
return err
}
if errors.Is(err, filer_pb.ErrNotFound) {
h.writeError(w, http.StatusNotFound, ErrCodeNoSuchView, fmt.Sprintf("view %s not found", target.viewName))
return err
}
h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to update metadata")
return err
}
h.writeJSON(w, http.StatusOK, &UpdateViewResponse{
ViewARN: target.viewARN,
MetadataLocation: metadata.MetadataLocation,
VersionToken: metadata.VersionToken,
})
return nil
}
// completedViewMetadata returns entry's metadata if it is a view whose create
// has finished, checking token against it when token is set.
func completedViewMetadata(entry *filer_pb.Entry, token string) (tableMetadataInternal, error) {
var metadata tableMetadataInternal
if EntryType(entry.Extended) != EntryTypeView || isUnfinishedCreate(entry.Extended) {
return metadata, filer_pb.ErrNotFound
}
data, ok := entry.Extended[ExtendedKeyMetadata]
if !ok {
return metadata, fmt.Errorf("%w: %w: %s", filer_pb.ErrNotFound, ErrAttributeNotFound, ExtendedKeyMetadata)
}
if err := json.Unmarshal(data, &metadata); err != nil {
return metadata, err
}
if token != "" && metadata.VersionToken != token {
return metadata, ErrVersionTokenMismatch
}
return metadata, 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
}
metadata, err = completedViewMetadata(entry, req.VersionToken)
return err
})
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 {
entry, err := h.lookupEntry(r.Context(), client, viewPath)
if err != nil {
return err
}
current, err := completedViewMetadata(entry, req.VersionToken)
if err != nil {
return err
}
// Authorization ran on the first read; only that same record may be deleted.
if current.VersionToken != metadata.VersionToken {
return ErrConcurrentUpdate
}
claimed := snapshotExtended(entry.Extended)
claimed[ExtendedKeyDeleting] = []byte("delete")
if err := h.claimCatalogRecord(r.Context(), client, viewPath, snapshotExtended(entry.Extended), "delete"); err != nil {
return err
}
if err := h.deleteDirectoryIfMatches(r.Context(), client, viewPath, &filer_pb.Entry{Extended: claimed}); err != nil {
if relErr := h.releaseCatalogRecord(context.WithoutCancel(r.Context()), client, viewPath, claimed); relErr != nil {
glog.Warningf("delete view %s: release claim: %v", viewPath, relErr)
}
return err
}
return nil
})
if err != nil {
if errors.Is(err, ErrConcurrentUpdate) {
h.writeError(w, http.StatusConflict, ErrCodeConflict, "view changed during delete, retry the request")
return err
}
if errors.Is(err, filer_pb.ErrNotFound) {
h.writeError(w, http.StatusNotFound, ErrCodeNoSuchView, fmt.Sprintf("view %s not found", viewName))
return err
}
if errors.Is(err, ErrVersionTokenMismatch) {
h.writeError(w, http.StatusConflict, ErrCodeConflict, "version token mismatch")
return err
}
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{
TrustedIssuers: h.trustedIssuers(r),
IsLocalAccountID: h.localAccountIDs,
TableBucketName: bucketName,
Namespace: namespace,
TableName: viewName,
TableBucketTags: bucketTags,
IdentityActions: identityActions,
DefaultAllow: h.defaultAllowFor(r),
}
bucketARN := h.generateTableBucketARN(bucketOwner, bucketName)
return CheckTableOrBucketPermission(operation, principal,
resourceOwner, resourcePolicy, resourceARN,
bucketOwner, bucketPolicy, bucketARN, ctx)
}
// 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)
}