Files
seaweedfs/weed/s3api/iceberg/handlers_table.go
T
Chris Lu ac6f3c92ef s3api/iceberg: report the reason a table schema was rejected (#10473)
* s3api/iceberg: report the reason a table schema was rejected

newTableMetadata swallowed the iceberg-go error and returned nil, so every
schema the metadata builder refused came back as a bare 500 "Failed to build
table metadata". A v3-only column type is the common case: creating a table
with a variant field but no format-version 3 property leaves the client with
nothing, while "variant is not supported until v3" sits in the server log.

Return the error instead and classify it. Schema, spec and argument failures
are the caller's input, so they answer 400 with the underlying reason; the
rest stay 500. Paths that build placeholder metadata with no schema keep
their existing 500 via newEmptyTableMetadata.

* s3api/iceberg: fail LoadTable when placeholder metadata cannot be built

buildLoadTableResult dropped a nil from the placeholder path straight into
the response. That serializes as "metadata":null under HTTP 200, which no
Iceberg client can parse -- a worse outcome than the 500 the nil was meant
to signal.

Return an error instead and let the five callers answer 500. The nil-return
convention goes away with it, so the commit and transaction paths check an
error rather than a sentinel.

* s3api/iceberg: route rejected schemas through writeManagerError

The two helpers added here duplicated work the package already does.
writeManagerError is the canonical error-to-response mapper -- it already
downgrades client-input failures to 400 and defaults the rest to 500 -- so
teach it the iceberg-go schema and spec sentinels instead of standing up a
parallel classifier. The placeholder wrapper was a pure alias for
newTableMetadata with nil arguments; call that directly.

No behavior change beyond the 500 message, which now reads err.Error()
like every other manager error rather than carrying its own prefix.
2026-07-28 14:44:56 -07:00

848 lines
31 KiB
Go

package iceberg
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"os"
"path"
"strings"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/table"
"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"
)
var errTableNameRequired = errors.New("table name is required")
func validateCreateTableRequest(req CreateTableRequest) error {
if req.Name == "" {
return errTableNameRequired
}
return nil
}
// handleListTables lists tables in a namespace.
func (s *Server) handleListTables(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)
// Extract identity from context
identityName := s3_constants.GetIdentityNameFromContext(r)
pageToken, pageSize, err := parsePagination(r)
if err != nil {
writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
return
}
listReq := &s3tables.ListTablesRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
ContinuationToken: pageToken,
MaxTables: pageSize,
}
var listResp s3tables.ListTablesResponse
err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "ListTables", 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: ListTables error: %v", err)
writeManagerError(w, err)
return
}
// Convert to Iceberg format
identifiers := make([]TableIdentifier, 0, len(listResp.Tables))
for _, t := range listResp.Tables {
identifiers = append(identifiers, TableIdentifier{
Namespace: namespace,
Name: t.Name,
})
}
result := ListTablesResponse{
NextPageToken: listResp.ContinuationToken,
Identifiers: identifiers,
}
writeJSON(w, http.StatusOK, result)
}
// tablePathOccupied reports whether a filer entry already exists at the catalog
// name path. A leftover directory there (e.g. data kept when another table was
// renamed to this name) means a table created at the default location would
// overwrite it, so the caller routes the new table to a unique location. A
// lookup failure other than not-found is returned so the caller can fail the
// create rather than fall back to the default path on a transient filer error.
func (s *Server) tablePathOccupied(ctx context.Context, bucketName, tablePath string) (bool, error) {
full := path.Join(s3tables.TablesPath, bucketName, tablePath)
dir, name := path.Split(full)
dir = strings.TrimSuffix(dir, "/")
occupied := false
err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
Directory: dir,
Name: name,
})
if errors.Is(err, filer_pb.ErrNotFound) {
return nil
}
if err != nil {
return err
}
occupied = true
return nil
})
return occupied, err
}
// handleCreateTable creates a new table.
func (s *Server) handleCreateTable(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 CreateTableRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body")
return
}
if err := validateCreateTableRequest(req); err != nil {
writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
return
}
if req.StageCreate && !isStageCreateEnabled() {
writeError(w, http.StatusNotImplemented, "NotImplementedException", "stage-create is disabled")
return
}
bucketName := getBucketFromPrefix(r)
bucketARN := buildTableBucketARN(bucketName)
// Extract identity from context
identityName := s3_constants.GetIdentityNameFromContext(r)
// Generate UUID for the new table
tableUUID := uuid.New()
tablePath := path.Join(flattenNamespacePath(namespace), req.Name)
location := strings.TrimSuffix(req.Location, "/")
if location == "" {
// If a leftover directory already occupies the default location (data
// kept when another table was renamed to this name), route this table
// to a unique location so it cannot overwrite that table's files.
occupied, err := s.tablePathOccupied(r.Context(), bucketName, tablePath)
if err != nil {
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to check table location: "+err.Error())
return
}
if occupied {
tablePath = path.Join(flattenNamespacePath(namespace), req.Name+"-"+tableUUID.String())
}
if req.Properties != nil {
if warehouse := strings.TrimSuffix(req.Properties["warehouse"], "/"); warehouse != "" {
location = fmt.Sprintf("%s/%s", warehouse, tablePath)
}
}
if location == "" {
if warehouse := strings.TrimSuffix(os.Getenv("ICEBERG_WAREHOUSE"), "/"); warehouse != "" {
location = fmt.Sprintf("%s/%s", warehouse, tablePath)
}
}
if location == "" {
location = fmt.Sprintf("s3://%s/%s", bucketName, tablePath)
}
} else {
parsedBucket, parsedPath, err := parseS3Location(location)
if err != nil {
writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid table location: "+err.Error())
return
}
if parsedPath == "" {
location = fmt.Sprintf("s3://%s/%s", parsedBucket, tablePath)
}
}
// Build proper Iceberg table metadata using iceberg-go types
metadata, err := newTableMetadata(tableUUID, location, req.Schema, req.PartitionSpec, req.WriteOrder, req.Properties)
if err != nil {
glog.V(1).Infof("Iceberg: CreateTable %s metadata error: %v", req.Name, err)
writeManagerError(w, err)
return
}
// Serialize metadata to JSON
metadataBytes, err := json.Marshal(metadata)
if err != nil {
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to serialize metadata: "+err.Error())
return
}
// Backfill Iceberg spec-required fields that iceberg-go v0.5.0 omits when
// empty (e.g. current-snapshot-id), so the persisted v*.metadata.json is
// readable by strict clients (Java/Spark/Trino) that go directly to S3.
metadataBytes = ensureMetadataSpecCompliance(metadataBytes)
tableName := req.Name
metadataFileName := "v1.metadata.json" // Initial version is always 1
metadataLocation := fmt.Sprintf("%s/metadata/%s", location, metadataFileName)
metadataBucket, metadataPath, err := parseS3Location(location)
if err != nil {
writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid table location: "+err.Error())
return
}
// location/name are client-supplied; confine the metadata write to the
// authorized catalog bucket and reject traversal segments so path.Join in
// saveMetadataFile cannot escape into another bucket.
if metadataBucket != bucketName {
writeError(w, http.StatusBadRequest, "BadRequestException", "table location must be within bucket "+bucketName)
return
}
if !isValidTablePath(metadataPath) {
writeError(w, http.StatusBadRequest, "BadRequestException", "invalid table location path")
return
}
// Authoritative existence check: ask the catalog whether a table is registered
// at this name. If it is, short-circuit with the existing table (idempotent
// CreateTable). Any leftover objects at the target path from a previous
// lifecycle of the same name are purged by handleDropTable on the way out —
// we deliberately do not clean storage here to keep CreateTable free of
// destructive side effects. See issue #9074.
existsReq := &s3tables.GetTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: tableName,
}
var existsResp s3tables.GetTableResponse
existsErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", existsReq, &existsResp, identityName)
})
if existsErr == nil {
// Table already registered. Return the existing definition so CTAS/IF NOT
// EXISTS flows see a stable response instead of a 409.
result, buildErr := s.buildLoadTableResult(existsResp, bucketName, namespace, tableName)
if buildErr != nil {
glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr)
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
return
}
writeJSON(w, http.StatusOK, result)
return
}
if !isNoSuchTableError(existsErr) {
glog.V(1).Infof("Iceberg: CreateTable existence check failed for %s.%s: %v", flattenNamespacePath(namespace), tableName, existsErr)
writeManagerError(w, existsErr)
return
}
// Stage-create persists metadata in the internal staged area and skips S3Tables registration.
if req.StageCreate {
stagedTablePath := stageCreateStagedTablePath(namespace, tableName, tableUUID)
if err := s.saveMetadataFile(r.Context(), metadataBucket, stagedTablePath, metadataFileName, metadataBytes); err != nil {
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save staged metadata file: "+err.Error())
return
}
stagedMetadataLocation := fmt.Sprintf("s3://%s/%s/metadata/%s", metadataBucket, stagedTablePath, metadataFileName)
if markerErr := s.writeStageCreateMarker(r.Context(), bucketName, namespace, tableName, tableUUID, location, stagedMetadataLocation); markerErr != nil {
glog.V(1).Infof("Iceberg: failed to persist stage-create marker for %s.%s: %v", flattenNamespacePath(namespace), tableName, markerErr)
}
result := LoadTableResult{
MetadataLocation: metadataLocation,
Metadata: metadata,
Config: s.buildFileIOConfig(),
}
writeJSON(w, http.StatusOK, result)
return
}
if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil {
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save metadata file: "+err.Error())
return
}
// Use S3 Tables manager to create table
createReq := &s3tables.CreateTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: tableName,
Format: "ICEBERG",
Metadata: &s3tables.TableMetadata{
Iceberg: &s3tables.IcebergMetadata{
TableUUID: tableUUID.String(),
},
FullMetadata: metadataBytes,
},
MetadataLocation: metadataLocation,
MetadataVersion: 1,
}
var createResp s3tables.CreateTableResponse
err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "CreateTable", createReq, &createResp, identityName)
})
if err != nil {
var tableErr *s3tables.S3TablesError
if errors.As(err, &tableErr) && tableErr.Type == s3tables.ErrCodeTableAlreadyExists {
getReq := &s3tables.GetTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: tableName,
}
var getResp s3tables.GetTableResponse
getErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", getReq, &getResp, identityName)
})
if getErr != nil {
writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error())
return
}
result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName)
if buildErr != nil {
glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr)
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
return
}
writeJSON(w, http.StatusOK, result)
return
}
if strings.Contains(err.Error(), "already exists") {
getReq := &s3tables.GetTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: tableName,
}
var getResp s3tables.GetTableResponse
getErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", getReq, &getResp, identityName)
})
if getErr != nil {
writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error())
return
}
result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName)
if buildErr != nil {
glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr)
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
return
}
writeJSON(w, http.StatusOK, result)
return
}
glog.V(1).Infof("Iceberg: CreateTable error: %v", err)
writeManagerError(w, err)
return
}
// Use returned location if available, otherwise fallback to local one
finalLocation := createResp.MetadataLocation
if finalLocation == "" {
finalLocation = metadataLocation
}
if markerErr := s.deleteStageCreateMarkers(r.Context(), bucketName, namespace, tableName); markerErr != nil {
glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after create: %v", flattenNamespacePath(namespace), tableName, markerErr)
}
result := LoadTableResult{
MetadataLocation: finalLocation,
Metadata: metadata,
Config: s.buildFileIOConfig(),
}
writeJSON(w, http.StatusOK, result)
}
// handleRegisterTable registers an existing metadata.json under a new catalog
// entry without generating new metadata.
func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
namespace := parseNamespace(vars["namespace"])
if len(namespace) == 0 {
writeError(w, http.StatusBadRequest, "BadRequestException", "Namespace is required")
return
}
var req RegisterTableRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body")
return
}
if req.Name == "" {
writeError(w, http.StatusBadRequest, "BadRequestException", errTableNameRequired.Error())
return
}
if req.MetadataLocation == "" {
writeError(w, http.StatusBadRequest, "BadRequestException", "metadata-location is required")
return
}
bucketName := getBucketFromPrefix(r)
bucketARN := buildTableBucketARN(bucketName)
identityName := s3_constants.GetIdentityNameFromContext(r)
// Read the existing metadata object before registering, so a bad location
// is rejected (400) without leaving a dangling catalog entry.
metadataBucket, tablePath, err := parseS3Location(tableLocationFromMetadataLocation(req.MetadataLocation))
if err != nil {
writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid metadata-location: "+err.Error())
return
}
// metadata-location is client-supplied; confine the read to the authorized
// catalog bucket and reject traversal segments so path.Join in
// loadMetadataFile cannot escape into another bucket.
if metadataBucket != bucketName {
writeError(w, http.StatusBadRequest, "BadRequestException", "metadata-location must be within bucket "+bucketName)
return
}
if !isValidTablePath(tablePath) {
writeError(w, http.StatusBadRequest, "BadRequestException", "invalid metadata-location path")
return
}
metadataFileName := path.Base(req.MetadataLocation)
metadataBytes, err := s.loadMetadataFile(r.Context(), metadataBucket, tablePath, metadataFileName)
if err != nil {
glog.V(1).Infof("Iceberg: RegisterTable load metadata at %s: %v", req.MetadataLocation, err)
writeError(w, http.StatusBadRequest, "BadRequestException", "Cannot read metadata at "+req.MetadataLocation)
return
}
registerReq := &s3tables.RegisterTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: req.Name,
MetadataLocation: req.MetadataLocation,
}
var registerResp s3tables.RegisterTableResponse
err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "RegisterTable", registerReq, &registerResp, identityName)
})
if err != nil {
var tableErr *s3tables.S3TablesError
if errors.As(err, &tableErr) {
switch tableErr.Type {
case s3tables.ErrCodeNoSuchNamespace:
writeError(w, http.StatusNotFound, "NoSuchNamespaceException", fmt.Sprintf("Namespace does not exist: %v", namespace))
return
case s3tables.ErrCodeTableAlreadyExists:
writeError(w, http.StatusConflict, "AlreadyExistsException", fmt.Sprintf("Table already exists: %s", req.Name))
return
}
}
glog.V(1).Infof("Iceberg: RegisterTable error: %v", err)
writeManagerError(w, err)
return
}
getResp := s3tables.GetTableResponse{
MetadataLocation: req.MetadataLocation,
Metadata: &s3tables.TableMetadata{FullMetadata: json.RawMessage(metadataBytes)},
}
result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, req.Name)
if buildErr != nil {
glog.Errorf("Iceberg: RegisterTable %s: %v", req.Name, buildErr)
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
return
}
writeJSON(w, http.StatusOK, result)
}
// handleLoadTable loads table metadata.
func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
namespace := parseNamespace(vars["namespace"])
tableName := vars["table"]
if len(namespace) == 0 || tableName == "" {
writeError(w, http.StatusBadRequest, "BadRequestException", "Namespace and table name are required")
return
}
bucketName := getBucketFromPrefix(r)
bucketARN := buildTableBucketARN(bucketName)
// Extract identity from context
identityName := s3_constants.GetIdentityNameFromContext(r)
getReq := &s3tables.GetTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: tableName,
}
var getResp s3tables.GetTableResponse
err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", getReq, &getResp, identityName)
})
if err != nil {
if strings.Contains(err.Error(), "not found") {
writeError(w, http.StatusNotFound, "NoSuchTableException", fmt.Sprintf("Table does not exist: %s", tableName))
return
}
glog.V(1).Infof("Iceberg: LoadTable error: %v", err)
writeManagerError(w, err)
return
}
result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName)
if buildErr != nil {
glog.Errorf("Iceberg: LoadTable %s: %v", tableName, buildErr)
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
return
}
writeJSON(w, http.StatusOK, result)
}
func (s *Server) buildLoadTableResult(getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) (LoadTableResult, error) {
location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
if location == "" {
location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
}
tableUUID := uuid.Nil
if getResp.Metadata != nil && getResp.Metadata.Iceberg != nil && getResp.Metadata.Iceberg.TableUUID != "" {
if parsed, err := uuid.Parse(getResp.Metadata.Iceberg.TableUUID); err == nil {
tableUUID = parsed
}
}
// Use Nil UUID if not found in storage (legacy table)
// Stability is guaranteed by not generating random UUIDs on read
var metadata table.Metadata
var err error
if getResp.Metadata != nil && len(getResp.Metadata.FullMetadata) > 0 {
metadata, err = table.ParseMetadataBytes(getResp.Metadata.FullMetadata)
if err != nil {
glog.Warningf("Iceberg: Failed to parse persisted metadata for %s: %v", tableName, err)
// Attempt to reconstruct from IcebergMetadata if available, otherwise synthetic
// TODO: Extract schema/spec from getResp.Metadata.Iceberg if FullMetadata fails but partial info exists?
// For now, fallback to empty metadata
metadata, err = newTableMetadata(tableUUID, location, nil, nil, nil, nil)
}
} else {
// No full metadata, create synthetic
// TODO: If we had stored schema in IcebergMetadata, we would pass it here
metadata, err = newTableMetadata(tableUUID, location, nil, nil, nil, nil)
}
// A nil metadata would serialize as "metadata":null under HTTP 200, which no
// Iceberg client can parse. Fail the request instead.
if err != nil {
return LoadTableResult{}, fmt.Errorf("build metadata for %s: %w", tableName, err)
}
return LoadTableResult{
MetadataLocation: getResp.MetadataLocation,
Metadata: metadata,
Config: s.buildFileIOConfig(),
}, nil
}
// buildFileIOConfig returns the FileIO properties to advertise to catalog
// clients so they can read the table's data files directly from S3 without
// separately discovering the endpoint. The region defaults to the same
// value baked into table bucket ARNs so clients like DuckDB that require
// a region on attach don't need to be told it out-of-band. See issue #9103.
func (s *Server) buildFileIOConfig() iceberg.Properties {
config := make(iceberg.Properties)
if s.s3Endpoint != "" {
config["s3.endpoint"] = s.s3Endpoint
config["s3.path-style-access"] = "true"
config["s3.region"] = s3tables.DefaultRegion
}
return config
}
// handleTableExists checks if a table exists.
func (s *Server) handleTableExists(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
namespace := parseNamespace(vars["namespace"])
tableName := vars["table"]
if len(namespace) == 0 || tableName == "" {
w.WriteHeader(http.StatusBadRequest)
return
}
bucketName := getBucketFromPrefix(r)
bucketARN := buildTableBucketARN(bucketName)
// Extract identity from context
identityName := s3_constants.GetIdentityNameFromContext(r)
getReq := &s3tables.GetTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: tableName,
}
var getResp s3tables.GetTableResponse
err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", getReq, &getResp, identityName)
})
if err != nil {
w.WriteHeader(http.StatusNotFound)
return
}
w.WriteHeader(http.StatusNoContent)
}
// handleDropTable deletes a table.
func (s *Server) handleDropTable(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
namespace := parseNamespace(vars["namespace"])
tableName := vars["table"]
if len(namespace) == 0 || tableName == "" {
writeError(w, http.StatusBadRequest, "BadRequestException", "Namespace and table name are required")
return
}
bucketName := getBucketFromPrefix(r)
bucketARN := buildTableBucketARN(bucketName)
// Extract identity from context
identityName := s3_constants.GetIdentityNameFromContext(r)
// Resolve the table's storage location from the catalog before the
// delete, so we can purge data files afterwards. The catalog is the
// authoritative owner of this mapping — if the table is not registered,
// there is nothing to clean up and DeleteTable will return NoSuchTable.
var storedMetadataLocation string
lookupReq := &s3tables.GetTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: tableName,
}
var lookupResp s3tables.GetTableResponse
lookupErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", lookupReq, &lookupResp, identityName)
})
if lookupErr == nil {
storedMetadataLocation = lookupResp.MetadataLocation
}
deleteReq := &s3tables.DeleteTableRequest{
TableBucketARN: bucketARN,
Namespace: namespace,
Name: tableName,
}
err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "DeleteTable", deleteReq, nil, identityName)
})
if err != nil {
if strings.Contains(err.Error(), "not found") {
writeError(w, http.StatusNotFound, "NoSuchTableException", fmt.Sprintf("Table does not exist: %s", tableName))
return
}
glog.V(1).Infof("Iceberg: DropTable error: %v", err)
writeManagerError(w, err)
return
}
// Purge data files that lived under the dropped table's location, so a
// subsequent CREATE TABLE at the same name (issue #9074) does not trip
// Trino's "non-empty location" pre-check. We only purge when the catalog
// told us a location: if GetTable above failed, there is no mapping to
// trust and we leave storage alone.
if storedMetadataLocation != "" {
if tableLoc := tableLocationFromMetadataLocation(storedMetadataLocation); tableLoc != "" {
if dataBucket, dataPath, parseErr := parseS3Location(tableLoc); parseErr == nil {
if cleanupErr := s.cleanupStaleTableLocation(r.Context(), dataBucket, dataPath); cleanupErr != nil {
glog.V(1).Infof("Iceberg: failed to purge dropped table location s3://%s/%s: %v", dataBucket, dataPath, cleanupErr)
}
}
}
}
w.WriteHeader(http.StatusNoContent)
}
// handleRenameTable moves a table's catalog pointer to a new namespace/name.
func (s *Server) handleRenameTable(w http.ResponseWriter, r *http.Request) {
var req RenameTableRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body")
return
}
source := parseNamespace(encodeNamespace(req.Source.Namespace))
dest := parseNamespace(encodeNamespace(req.Destination.Namespace))
if len(source) == 0 || req.Source.Name == "" || len(dest) == 0 || req.Destination.Name == "" {
writeError(w, http.StatusBadRequest, "BadRequestException", "source and destination namespace and name are required")
return
}
bucketName := getBucketFromPrefix(r)
bucketARN := buildTableBucketARN(bucketName)
identityName := s3_constants.GetIdentityNameFromContext(r)
renameReq := &s3tables.RenameTableRequest{
TableBucketARN: bucketARN,
SourceNamespace: source,
SourceName: req.Source.Name,
DestNamespace: dest,
DestName: req.Destination.Name,
}
err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
mgrClient := s3tables.NewManagerClient(client)
return s.tablesManager.Execute(r.Context(), mgrClient, "RenameTable", renameReq, nil, identityName)
})
if err != nil {
var tableErr *s3tables.S3TablesError
if errors.As(err, &tableErr) {
switch tableErr.Type {
case s3tables.ErrCodeNoSuchTable:
writeError(w, http.StatusNotFound, "NoSuchTableException", fmt.Sprintf("Table does not exist: %s", req.Source.Name))
return
case s3tables.ErrCodeNoSuchNamespace:
writeError(w, http.StatusNotFound, "NoSuchNamespaceException", fmt.Sprintf("Namespace does not exist: %v", dest))
return
case s3tables.ErrCodeTableAlreadyExists:
writeError(w, http.StatusConflict, "AlreadyExistsException", fmt.Sprintf("Table already exists: %s", req.Destination.Name))
return
case s3tables.ErrCodeInvalidRequest:
writeError(w, http.StatusBadRequest, "BadRequestException", tableErr.Message)
return
}
}
glog.V(1).Infof("Iceberg: RenameTable error: %v", err)
writeManagerError(w, err)
return
}
w.WriteHeader(http.StatusNoContent)
}
// isNoSuchTableError reports whether an error from the S3 Tables manager
// indicates the target table is not registered in the catalog.
func isNoSuchTableError(err error) bool {
if err == nil {
return false
}
var tableErr *s3tables.S3TablesError
if errors.As(err, &tableErr) {
return tableErr.Type == s3tables.ErrCodeNoSuchTable || tableErr.Type == s3tables.ErrCodeNoSuchNamespace
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "not found") || strings.Contains(msg, "no such table")
}
// cleanupStaleTableLocation purges any existing filer entry at the target
// table path inside the regular S3 bucket so that engines which verify an
// empty location (e.g. Trino CTAS) can proceed after a DROP. Callers must
// have confirmed via the catalog that no table is registered at this name —
// live tables must never be touched by this helper. Missing paths are not
// an error.
//
// The tablePath is validated to contain only safe, relative segments before
// being joined with the bucket prefix, so a maliciously crafted location
// (e.g. containing "..") cannot escape the bucket subtree and delete data
// outside of it.
func (s *Server) cleanupStaleTableLocation(ctx context.Context, bucketName, tablePath string) error {
tablePath = strings.Trim(tablePath, "/")
if bucketName == "" || tablePath == "" {
return nil
}
// Reject absolute paths and any traversal segments. path.Clean would
// silently collapse "foo/../bar" into "bar", masking an attempted
// escape, so we check each raw segment explicitly.
for _, segment := range strings.Split(tablePath, "/") {
if segment == "" || segment == "." || segment == ".." || strings.ContainsAny(segment, `\`) {
return fmt.Errorf("cleanupStaleTableLocation: refusing unsafe table path %q", tablePath)
}
}
parentDir := path.Join(s3_constants.DefaultBucketsPath, bucketName, path.Dir(tablePath))
name := path.Base(tablePath)
if name == "" || name == "." || name == ".." || name == "/" {
return fmt.Errorf("cleanupStaleTableLocation: refusing unsafe table path %q", tablePath)
}
return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
err := filer_pb.DoRemove(ctx, client, parentDir, name, true, true, true, false, nil)
if err == nil || errors.Is(err, filer_pb.ErrNotFound) {
return nil
}
return err
})
}
// newTableMetadata creates a new table.Metadata object with the given parameters.
// Uses iceberg-go's MetadataBuilder pattern for proper spec compliance.
func newTableMetadata(
tableUUID uuid.UUID,
location string,
schema *iceberg.Schema,
partitionSpec *iceberg.PartitionSpec,
sortOrder *table.SortOrder,
props iceberg.Properties,
) (table.Metadata, error) {
// Add schema - use provided or create empty schema
var s *iceberg.Schema
if schema != nil {
s = schema
} else {
s = iceberg.NewSchema(0)
}
// Add partition spec
var pSpec *iceberg.PartitionSpec
if partitionSpec != nil {
pSpec = partitionSpec
} else {
unpartitioned := iceberg.NewPartitionSpecID(0)
pSpec = &unpartitioned
}
// Add sort order
var so table.SortOrder
if sortOrder != nil {
so = *sortOrder
} else {
so = table.UnsortedSortOrder
}
// Create properties map if nil
if props == nil {
props = make(iceberg.Properties)
}
// Create metadata directly using the constructor which ensures spec compliance for V2
return table.NewMetadataWithUUID(s, pSpec, so, location, props, tableUUID)
}