diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index 02b50ee2b..78bd26d33 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/weed/s3api/iceberg/handlers_commit.go @@ -160,8 +160,9 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { } if baseMetadata == nil { - baseMetadata = newTableMetadata(tableUUID, location, nil, nil, nil, nil) - if baseMetadata == nil { + var buildErr error + if baseMetadata, buildErr = newTableMetadata(tableUUID, location, nil, nil, nil, nil); buildErr != nil { + glog.Errorf("Iceberg: CommitTable placeholder metadata for %s: %v", tableName, buildErr) writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build current metadata") return } @@ -216,11 +217,12 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { return } } else { - currentMetadata = newTableMetadata(tableUUID, location, nil, nil, nil, nil) - } - if currentMetadata == nil { - writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build current metadata") - return + currentMetadata, err = newTableMetadata(tableUUID, location, nil, nil, nil, nil) + if err != nil { + glog.Errorf("Iceberg: CommitTable placeholder metadata for %s: %v", tableName, err) + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build current metadata") + return + } } for _, requirement := range req.Requirements { diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 5901d17fa..309547c4a 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -188,9 +188,10 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { } // Build proper Iceberg table metadata using iceberg-go types - metadata := newTableMetadata(tableUUID, location, req.Schema, req.PartitionSpec, req.WriteOrder, req.Properties) - if metadata == nil { - writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") + 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 } @@ -245,7 +246,12 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { 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 := s.buildLoadTableResult(existsResp, bucketName, namespace, tableName) + 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 } @@ -318,7 +324,12 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error()) return } - result := s.buildLoadTableResult(getResp, bucketName, namespace, tableName) + 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 } @@ -337,7 +348,12 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error()) return } - result := s.buildLoadTableResult(getResp, bucketName, namespace, tableName) + 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 } @@ -449,7 +465,12 @@ func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) { MetadataLocation: req.MetadataLocation, Metadata: &s3tables.TableMetadata{FullMetadata: json.RawMessage(metadataBytes)}, } - result := s.buildLoadTableResult(getResp, bucketName, namespace, req.Name) + 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) } @@ -493,11 +514,16 @@ func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) { return } - result := s.buildLoadTableResult(getResp, bucketName, namespace, tableName) + 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 { +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)) @@ -512,27 +538,32 @@ func (s *Server) buildLoadTableResult(getResp s3tables.GetTableResponse, bucketN // 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 { - var err error 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 = newTableMetadata(tableUUID, location, nil, nil, nil, nil) + 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 = newTableMetadata(tableUUID, location, nil, nil, nil, nil) + 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 @@ -780,7 +811,7 @@ func newTableMetadata( partitionSpec *iceberg.PartitionSpec, sortOrder *table.SortOrder, props iceberg.Properties, -) table.Metadata { +) (table.Metadata, error) { // Add schema - use provided or create empty schema var s *iceberg.Schema if schema != nil { @@ -812,11 +843,5 @@ func newTableMetadata( } // Create metadata directly using the constructor which ensures spec compliance for V2 - metadata, err := table.NewMetadataWithUUID(s, pSpec, so, location, props, tableUUID) - if err != nil { - glog.Errorf("Failed to create metadata: %v", err) - return nil - } - - return metadata + return table.NewMetadataWithUUID(s, pSpec, so, location, props, tableUUID) } diff --git a/weed/s3api/iceberg/handlers_transaction.go b/weed/s3api/iceberg/handlers_transaction.go index 8e2d24c87..f46bdfa37 100644 --- a/weed/s3api/iceberg/handlers_transaction.go +++ b/weed/s3api/iceberg/handlers_transaction.go @@ -155,10 +155,11 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN, return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Failed to parse current metadata"} } } else { - currentMetadata = newTableMetadata(tableUUID, location, nil, nil, nil, nil) - } - if currentMetadata == nil { - return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Failed to build current metadata"} + currentMetadata, err = newTableMetadata(tableUUID, location, nil, nil, nil, nil) + if err != nil { + glog.Errorf("Iceberg: CommitTransaction placeholder metadata for %s: %v", tableName, err) + return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Failed to build current metadata"} + } } for _, requirement := range requirements { diff --git a/weed/s3api/iceberg/iceberg_create_table_test.go b/weed/s3api/iceberg/iceberg_create_table_test.go index 78e1234e1..721cb463e 100644 --- a/weed/s3api/iceberg/iceberg_create_table_test.go +++ b/weed/s3api/iceberg/iceberg_create_table_test.go @@ -1,8 +1,14 @@ package iceberg import ( + "encoding/json" "errors" + "strings" "testing" + + "github.com/apache/iceberg-go" + "github.com/google/uuid" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" ) func TestValidateCreateTableRequestRequiresName(t *testing.T) { @@ -35,3 +41,68 @@ func TestIsStageCreateEnabledFalseValues(t *testing.T) { } } } + +func mustParseSchema(t *testing.T, raw string) *iceberg.Schema { + t.Helper() + var schema iceberg.Schema + if err := json.Unmarshal([]byte(raw), &schema); err != nil { + t.Fatalf("parse schema: %v", err) + } + return &schema +} + +const variantSchema = `{"type":"struct","schema-id":0,"fields":[ + {"id":1,"name":"id","required":true,"type":"long"}, + {"id":2,"name":"payload","required":false,"type":"variant"}]}` + +// A v3-only column type without format-version 3 is the client's mistake, and +// the reason has to travel back to them rather than only into the server log. +func TestNewTableMetadataRejectsV3TypeBelowV3(t *testing.T) { + _, err := newTableMetadata(uuid.New(), "s3://bkt/ns/t", mustParseSchema(t, variantSchema), nil, nil, nil) + if err == nil { + t.Fatal("newTableMetadata() error = nil, want invalid schema") + } + if !errors.Is(err, iceberg.ErrInvalidSchema) { + t.Fatalf("newTableMetadata() error = %v, want ErrInvalidSchema", err) + } + + // writeManagerError turns this into a 400; see TestWriteManagerError. + if !strings.Contains(err.Error(), "variant is not supported until v3") { + t.Errorf("error = %q, want the underlying reason", err.Error()) + } +} + +func TestNewTableMetadataAcceptsV3TypeAtV3(t *testing.T) { + metadata, err := newTableMetadata(uuid.New(), "s3://bkt/ns/t", mustParseSchema(t, variantSchema), nil, nil, + iceberg.Properties{"format-version": "3"}) + if err != nil { + t.Fatalf("newTableMetadata() error = %v, want nil", err) + } + if got := metadata.Version(); got != 3 { + t.Fatalf("metadata.Version() = %d, want 3", got) + } + if _, found := metadata.CurrentSchema().FindFieldByName("payload"); !found { + t.Error("variant field missing from stored schema") + } +} + +// A LoadTable response must never carry nil metadata: it serializes as +// "metadata":null under HTTP 200, which no Iceberg client can parse. +func TestBuildLoadTableResultNeverReturnsNilMetadata(t *testing.T) { + cases := map[string]s3tables.GetTableResponse{ + "no stored metadata": {MetadataLocation: "s3://bkt/ns/t/metadata/v1.metadata.json"}, + "empty full metadata": {Metadata: &s3tables.TableMetadata{}}, + "unparseable metadata": {Metadata: &s3tables.TableMetadata{FullMetadata: json.RawMessage(`{"nope":`)}}, + } + for name, getResp := range cases { + t.Run(name, func(t *testing.T) { + result, err := (&Server{}).buildLoadTableResult(getResp, "bkt", []string{"ns"}, "t") + if err != nil { + t.Fatalf("buildLoadTableResult() error = %v, want nil", err) + } + if result.Metadata == nil { + t.Fatal("buildLoadTableResult() returned nil metadata with no error") + } + }) + } +} diff --git a/weed/s3api/iceberg/iceberg_invalid_name_test.go b/weed/s3api/iceberg/iceberg_invalid_name_test.go index af7c5e415..bab683af8 100644 --- a/weed/s3api/iceberg/iceberg_invalid_name_test.go +++ b/weed/s3api/iceberg/iceberg_invalid_name_test.go @@ -6,6 +6,8 @@ import ( "net/http" "net/http/httptest" "testing" + + "github.com/apache/iceberg-go" ) func TestNameValidationError(t *testing.T) { @@ -41,6 +43,8 @@ func TestWriteManagerError(t *testing.T) { wantType string }{ {"invalid name is a client error", fmt.Errorf("invalid namespace name: only 'a-z', '0-9', and '_' are allowed"), http.StatusBadRequest, "BadRequestException"}, + {"rejected schema is a client error", fmt.Errorf("%w: for v2: variant is not supported until v3", iceberg.ErrInvalidSchema), http.StatusBadRequest, "BadRequestException"}, + {"bad format version is a client error", fmt.Errorf("%w: 4", iceberg.ErrInvalidFormatVersion), http.StatusBadRequest, "BadRequestException"}, {"everything else is a server fault", fmt.Errorf("all filers failed, last error: connection refused"), http.StatusInternalServerError, "InternalServerError"}, } for _, c := range cases { diff --git a/weed/s3api/iceberg/utils.go b/weed/s3api/iceberg/utils.go index f01130e17..94277a2f5 100644 --- a/weed/s3api/iceberg/utils.go +++ b/weed/s3api/iceberg/utils.go @@ -2,12 +2,14 @@ package iceberg import ( "encoding/json" + "errors" "fmt" "net/http" "os" "strconv" "strings" + "github.com/apache/iceberg-go" "github.com/gorilla/mux" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -123,11 +125,19 @@ func nameValidationError(err error) bool { return false } -// writeManagerError maps a residual s3tables manager error to a response: -// name validation failures are client errors (400); anything else is a server -// fault (500). +// writeManagerError maps a residual error to a response: name validation +// failures and rejected schemas come from the request body, so they are client +// errors (400) and must carry the reason -- a variant field without +// format-version 3 otherwise reads as a bare 500 with "variant is not supported +// until v3" only in the log. Anything else is a server fault (500). func writeManagerError(w http.ResponseWriter, err error) { - if nameValidationError(err) { + switch { + case nameValidationError(err), + errors.Is(err, iceberg.ErrInvalidSchema), + errors.Is(err, iceberg.ErrInvalidPartitionSpec), + errors.Is(err, iceberg.ErrInvalidTypeString), + errors.Is(err, iceberg.ErrInvalidTransform), + errors.Is(err, iceberg.ErrInvalidArgument): writeError(w, http.StatusBadRequest, "BadRequestException", err.Error()) return }