diff --git a/test/s3tables/catalog_spark/spark_register_table_test.go b/test/s3tables/catalog_spark/spark_register_table_test.go new file mode 100644 index 0000000000..1c186c5b61 --- /dev/null +++ b/test/s3tables/catalog_spark/spark_register_table_test.go @@ -0,0 +1,117 @@ +package catalog_spark + +import ( + "bytes" + "context" + "fmt" + "io" + "path" + "regexp" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" +) + +// TestSparkRegisterTableServesItsData registers a table under a new name from +// the current metadata file of another and checks that Spark reads the +// registered rows and can write on top of them. One file is the one the +// catalog wrote; the other is a copy uploaded through the S3 gateway, the way +// the metadata of a table migrated in arrives, which the filer stores in +// chunks rather than inline. A registered table used to load as an empty +// placeholder: SELECT found no rows and INSERT failed its table UUID check. +func TestSparkRegisterTableServesItsData(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + env, _, tableBucket := setupSparkTestEnv(t) + + for _, tc := range []struct { + name string + upload bool + }{ + {"metadata file written by the catalog", false}, + {"metadata file uploaded through S3", true}, + } { + t.Run(tc.name, func(t *testing.T) { + namespace := "register_" + randomString(6) + source := fmt.Sprintf("iceberg.%s.src_%s", namespace, randomString(6)) + table := "reg_" + randomString(6) + ident := fmt.Sprintf("iceberg.%s.%s", namespace, table) + + output := runSparkPySQL(t, env.sparkContainer, fmt.Sprintf(` +spark.sql("CREATE NAMESPACE iceberg.%[1]s") +spark.sql("CREATE TABLE %[2]s (id INT, name STRING) USING iceberg") +spark.sql("INSERT INTO %[2]s VALUES (1, 'alpha'), (2, 'beta'), (3, 'gamma')") +meta = spark.sql("SELECT file FROM %[2]s.metadata_log_entries ORDER BY timestamp DESC LIMIT 1").collect()[0]['file'] +print(f"metadata file: {meta}") +`, namespace, source), env.icebergRestPort, env.s3Port) + match := regexp.MustCompile(`metadata file: (\S+)`).FindStringSubmatch(output) + if match == nil { + t.Fatalf("create/insert of %s failed, output:\n%s", source, output) + } + metadataFile := match[1] + if tc.upload { + metadataFile = uploadMetadataCopy(t, env, tableBucket, metadataFile) + } + + output = runSparkPySQL(t, env.sparkContainer, fmt.Sprintf(` +spark.sql("CALL iceberg.system.register_table('%[1]s.%[2]s', '%[3]s')") +registered = spark.sql("SELECT COUNT(*) AS c FROM %[4]s").collect()[0]['c'] +print(f"registered row count: {registered}") +spark.sql("INSERT INTO %[4]s VALUES (4, 'delta')") +inserted = spark.sql("SELECT COUNT(*) AS c FROM %[4]s").collect()[0]['c'] +print(f"row count after insert: {inserted}") +`, namespace, table, metadataFile, ident), env.icebergRestPort, env.s3Port) + if !strings.Contains(output, "registered row count: 3") { + t.Fatalf("the table registered from %s does not serve its rows, output:\n%s", metadataFile, output) + } + if !strings.Contains(output, "row count after insert: 4") { + t.Fatalf("INSERT into the table registered from %s failed, output:\n%s", metadataFile, output) + } + }) + } +} + +// uploadMetadataCopy puts a copy of metadataFile next to it through the S3 +// gateway and returns the copy's location. +func uploadMetadataCopy(t *testing.T, env *TestEnvironment, bucket, metadataFile string) string { + t.Helper() + key, ok := strings.CutPrefix(metadataFile, "s3://"+bucket+"/") + if !ok { + t.Fatalf("metadata file %s is not in bucket %s", metadataFile, bucket) + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + cfg := aws.Config{ + Region: "us-east-1", + Credentials: aws.NewCredentialsCache(credentials.NewStaticCredentialsProvider(env.accessKey, env.secretKey, "")), + BaseEndpoint: aws.String(fmt.Sprintf("http://localhost:%d", env.s3Port)), + } + client := s3.NewFromConfig(cfg, func(o *s3.Options) { o.UsePathStyle = true }) + + obj, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)}) + if err != nil { + t.Fatalf("GetObject %s: %v", key, err) + } + content, err := io.ReadAll(obj.Body) + obj.Body.Close() + if err != nil { + t.Fatalf("read %s: %v", key, err) + } + + copyKey := path.Join(path.Dir(key), "00003-uploaded.metadata.json") + if _, err := client.PutObject(ctx, &s3.PutObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String(copyKey), + Body: bytes.NewReader(content), + }); err != nil { + t.Fatalf("PutObject %s: %v", copyKey, err) + } + return "s3://" + bucket + "/" + copyKey +} diff --git a/weed/admin/dash/iceberg_data_preview.go b/weed/admin/dash/iceberg_data_preview.go index 1c9e90809d..ccf560e3f7 100644 --- a/weed/admin/dash/iceberg_data_preview.go +++ b/weed/admin/dash/iceberg_data_preview.go @@ -16,6 +16,7 @@ import ( "github.com/parquet-go/parquet-go" "github.com/seaweedfs/seaweedfs/weed/admin/plugin" "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" ) @@ -106,6 +107,7 @@ func (s *AdminServer) GetIcebergTableDataPreview(ctx context.Context, catalogNam s.applyWorkerPreview(ctx, &data, bucketArn, namespaceParts, tableName, rowLimit) return data, nil } + s.fillTableMetadata(ctx, &resp, bucketArn, namespace, tableName) if resp.Metadata == nil || len(resp.Metadata.FullMetadata) == 0 { data.PreviewError = "Table has no Iceberg metadata." return data, nil @@ -271,6 +273,33 @@ func (s *AdminServer) listIcebergDataFiles(ctx context.Context, bucketName, name return files, hasDeletes, notes, nil } +// fillTableMetadata reads the metadata file a location-only record points to: +// a commit that outgrew the request-size limit stores no inline copy, and +// without the file the page reports a healthy table as having no metadata. +func (s *AdminServer) fillTableMetadata(ctx context.Context, resp *s3tables.GetTableResponse, bucketArn, namespaceDir, tableName string) { + if resp.MetadataLocation == "" || (resp.Metadata != nil && len(resp.Metadata.FullMetadata) > 0) { + return + } + bucketName, err := s3tables.ParseBucketNameFromARN(bucketArn) + if err != nil { + bucketName = "" + } + filerPath, err := icebergLocationToFilerPath(resp.MetadataLocation, bucketName, namespaceDir, tableName) + if err != nil { + glog.V(1).Infof("iceberg: cannot resolve metadata location %s: %v", resp.MetadataLocation, err) + return + } + content, err := s.readFilerFileContent(ctx, filerPath) + if err != nil { + glog.V(1).Infof("iceberg: cannot read metadata at %s: %v", resp.MetadataLocation, err) + return + } + if resp.Metadata == nil { + resp.Metadata = &s3tables.TableMetadata{} + } + resp.Metadata.FullMetadata = content +} + // icebergLocationToFilerPath maps an Iceberg file location (s3://bucket/..., // an absolute filer path, or a table-relative path like data/x.parquet) to a // filer path, which must stay under the table-buckets root. diff --git a/weed/admin/dash/s3tables_management.go b/weed/admin/dash/s3tables_management.go index 0db9e81d05..19a997d552 100644 --- a/weed/admin/dash/s3tables_management.go +++ b/weed/admin/dash/s3tables_management.go @@ -388,6 +388,7 @@ func (s *AdminServer) GetIcebergTableDetailsData(ctx context.Context, catalogNam MetadataLocation: resp.MetadataLocation, } + s.fillTableMetadata(ctx, &resp, bucketArn, strings.Join(namespaceParts, "/"), tableName) applyIcebergMetadata(resp.Metadata, &details) s.applyWorkerObservation(&details, bucketArn, namespaceParts, resp.Name) return details, nil diff --git a/weed/s3api/iceberg/commit_helpers.go b/weed/s3api/iceberg/commit_helpers.go index 116ed85862..bd725d019d 100644 --- a/weed/s3api/iceberg/commit_helpers.go +++ b/weed/s3api/iceberg/commit_helpers.go @@ -253,6 +253,21 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi }, nil } +// Manager requests are size-capped like registrations, so a commit whose new +// metadata is too large to ride along flips the pointer by location alone — +// the staged file already holds it. The record's stored copy must be cleared +// with it, or loads would keep serving the pre-commit bytes. +func fitUpdateTableRequest(req *s3tables.UpdateTableRequest) { + if req.Metadata == nil || len(req.Metadata.FullMetadata) == 0 { + return + } + if body, err := json.Marshal(req); err == nil && len(body) <= s3tables.MaxRequestBodySize { + return + } + req.Metadata = &s3tables.TableMetadata{Iceberg: req.Metadata.Iceberg} + req.ClearFullMetadata = true +} + // authorizeCommit asks the S3 Tables layer, before anything is written, whether // the caller may update the table. Both commit paths stage the new metadata // file in the table's directory before UpdateTable authorizes the pointer diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index c31a6d5d5d..1f2aed1ae1 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/weed/s3api/iceberg/handlers_commit.go @@ -266,19 +266,31 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { tableUUID = parsed } } - if tableUUID == uuid.Nil { - tableUUID = generatedLegacyUUID - } + currentBytes, err := s.currentMetadataBytes(r.Context(), getResp, bucketName) + if err != nil { + glog.Errorf("Iceberg: CommitTable load current metadata for %s: %v", tableName, err) + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to load current metadata") + return + } var currentMetadata table.Metadata - if getResp.Metadata != nil && len(getResp.Metadata.FullMetadata) > 0 { - currentMetadata, err = table.ParseMetadataBytes(getResp.Metadata.FullMetadata) + if len(currentBytes) > 0 { + currentMetadata, err = table.ParseMetadataBytes(currentBytes) if err != nil { glog.Errorf("Iceberg: Failed to parse current metadata for %s: %v", tableName, err) writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to parse current metadata") return } - } else { + // A record that kept only a metadata location carries no UUID of + // its own; the one in its metadata file is the table's. + if tableUUID == uuid.Nil { + tableUUID = currentMetadata.TableUUID() + } + } + if tableUUID == uuid.Nil { + tableUUID = generatedLegacyUUID + } + if currentMetadata == nil { currentMetadata, err = newTableMetadata(tableUUID, location, nil, nil, nil, nil) if err != nil { glog.Errorf("Iceberg: CommitTable placeholder metadata for %s: %v", tableName, err) @@ -371,6 +383,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { MetadataVersion: metadataVersion, MetadataLocation: newMetadataLocation, } + fitUpdateTableRequest(updateReq) err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { mgrClient := s3tables.NewManagerClient(client) diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index e83dac4982..d99ba5a708 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -483,18 +483,36 @@ func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) { return } metadataFileName := path.Base(req.MetadataLocation) - metadataBytes, err := s.loadMetadataFile(r.Context(), metadataBucket, tablePath, metadataFileName) + // A table migrated in has its metadata uploaded through the S3 gateway, + // which stores it in chunks rather than inline, so follow the chunks. + metadataBytes, err := s.loadTableMetadataBlob(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 } + registered, err := table.ParseMetadataBytes(metadataBytes) + if err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", fmt.Sprintf("Invalid metadata at %s: %v", req.MetadataLocation, err)) + return + } + registeredMetadata := &s3tables.TableMetadata{ + Iceberg: &s3tables.IcebergMetadata{TableUUID: registered.TableUUID().String()}, + FullMetadata: metadataBytes, + } registerReq := &s3tables.RegisterTableRequest{ TableBucketARN: bucketARN, Namespace: namespace, Name: req.Name, MetadataLocation: req.MetadataLocation, + Metadata: registeredMetadata, + } + // The manager refuses a request larger than s3tables.MaxRequestBodySize. + // A file too large to travel with the registration is registered by its + // location and UUID alone; loads and commits read it from that location. + if body, marshalErr := json.Marshal(registerReq); marshalErr == nil && len(body) > s3tables.MaxRequestBodySize { + registerReq.Metadata = &s3tables.TableMetadata{Iceberg: registeredMetadata.Iceberg} } var registerResp s3tables.RegisterTableResponse err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { @@ -520,7 +538,7 @@ func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) { getResp := s3tables.GetTableResponse{ MetadataLocation: req.MetadataLocation, - Metadata: &s3tables.TableMetadata{FullMetadata: json.RawMessage(metadataBytes)}, + Metadata: registeredMetadata, } result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, req.Name) if buildErr != nil { @@ -599,10 +617,13 @@ func (s *Server) buildLoadTableResult(r *http.Request, getResp s3tables.GetTable // Use Nil UUID if not found in storage (legacy table) // Stability is guaranteed by not generating random UUIDs on read + metadataBytes, err := s.currentMetadataBytes(r.Context(), getResp, bucketName) + if err != nil { + return LoadTableResult{}, fmt.Errorf("load metadata for %s: %w", tableName, err) + } var metadata table.Metadata - var err error - if getResp.Metadata != nil && len(getResp.Metadata.FullMetadata) > 0 { - metadata, err = table.ParseMetadataBytes(getResp.Metadata.FullMetadata) + if len(metadataBytes) > 0 { + metadata, err = table.ParseMetadataBytes(metadataBytes) 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 diff --git a/weed/s3api/iceberg/handlers_transaction.go b/weed/s3api/iceberg/handlers_transaction.go index 6279cc4cc7..4e090b55bc 100644 --- a/weed/s3api/iceberg/handlers_transaction.go +++ b/weed/s3api/iceberg/handlers_transaction.go @@ -148,23 +148,39 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN, if location == "" { location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName)) } + locBucket, locPath, locErr := parseS3Location(location) + if locErr != nil { + return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Invalid table location: " + locErr.Error()} + } + if err := confineMetadataLocation(locBucket, locPath, bucketName); err != nil { + return nil, &icebergRequestError{http.StatusBadRequest, "BadRequestException", err.Error()} + } tableUUID := uuid.Nil if getResp.Metadata != nil && getResp.Metadata.Iceberg != nil && getResp.Metadata.Iceberg.TableUUID != "" { if parsed, parseErr := uuid.Parse(getResp.Metadata.Iceberg.TableUUID); parseErr == nil { tableUUID = parsed } } - if tableUUID == uuid.Nil { - tableUUID = uuid.New() - } + currentBytes, err := s.currentMetadataBytes(ctx, getResp, bucketName) + if err != nil { + glog.Errorf("Iceberg: CommitTransaction load current metadata for %s: %v", tableName, err) + return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Failed to load current metadata"} + } var currentMetadata table.Metadata - if getResp.Metadata != nil && len(getResp.Metadata.FullMetadata) > 0 { - currentMetadata, err = table.ParseMetadataBytes(getResp.Metadata.FullMetadata) + if len(currentBytes) > 0 { + currentMetadata, err = table.ParseMetadataBytes(currentBytes) if err != nil { return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Failed to parse current metadata"} } - } else { + if tableUUID == uuid.Nil { + tableUUID = currentMetadata.TableUUID() + } + } + if tableUUID == uuid.Nil { + tableUUID = uuid.New() + } + if currentMetadata == nil { currentMetadata, err = newTableMetadata(tableUUID, location, nil, nil, nil, nil) if err != nil { glog.Errorf("Iceberg: CommitTransaction placeholder metadata for %s: %v", tableName, err) @@ -204,6 +220,7 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN, if err != nil { return nil, &icebergRequestError{http.StatusBadRequest, "BadRequestException", "Failed to apply statistics updates: " + err.Error()} } + metadataBytes = refreshDefaultNameMapping(metadataBytes, newMetadata) metadataBytes = ensureMetadataSpecCompliance(metadataBytes) metadataBucket, metadataPath, err := parseS3Location(location) @@ -214,6 +231,17 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN, return nil, &icebergRequestError{http.StatusBadRequest, "BadRequestException", err.Error()} } + prevMetadata := cloneTableMetadata(getResp.Metadata) + if len(currentBytes) > 0 && (getResp.Metadata == nil || len(getResp.Metadata.FullMetadata) == 0) { + // UpdateTable keeps the FullMetadata a flip stored when a restore sends + // none, so a record that held only a location gets back what loads + // served from its file rather than the transaction it is rolling back. + prevMetadata = cloneTableMetadata(&s3tables.TableMetadata{ + Iceberg: &s3tables.IcebergMetadata{TableUUID: tableUUID.String()}, + FullMetadata: currentBytes, + }) + } + return &preparedTableCommit{ namespace: namespace, tableName: tableName, @@ -228,7 +256,7 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN, newMetadataLoc: newMetadataLocation, prevMetadataLoc: getResp.MetadataLocation, prevMetadataVersion: getResp.MetadataVersion, - prevMetadata: cloneTableMetadata(getResp.Metadata), + prevMetadata: prevMetadata, }, nil } @@ -285,6 +313,7 @@ func (s *Server) flipTablePointer(ctx context.Context, bucketARN, identityName s MetadataVersion: pc.metadataVersion, MetadataLocation: pc.newMetadataLoc, } + fitUpdateTableRequest(updateReq) return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { mgrClient := s3tables.NewManagerClient(client) return s.tablesManager.Execute(ctx, mgrClient, "UpdateTable", updateReq, nil, identityName) @@ -302,6 +331,7 @@ func (s *Server) rollbackTablePointers(ctx context.Context, bucketARN, identityN for i := range flipped { pc := &flipped[i] updateReq := buildTableRestoreRequest(bucketARN, pc) + fitUpdateTableRequest(updateReq) err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { mgrClient := s3tables.NewManagerClient(client) return s.tablesManager.Execute(ctx, mgrClient, "UpdateTable", updateReq, nil, identityName) diff --git a/weed/s3api/iceberg/iceberg_access_delegation_test.go b/weed/s3api/iceberg/iceberg_access_delegation_test.go index 8e004a375a..5b3ee9ba57 100644 --- a/weed/s3api/iceberg/iceberg_access_delegation_test.go +++ b/weed/s3api/iceberg/iceberg_access_delegation_test.go @@ -64,7 +64,7 @@ func TestWriteLoadResultVariesOnTheDelegationHeader(t *testing.T) { } func TestLoadTableResultOmitsConfigForCredentialVendingClients(t *testing.T) { - s := &Server{s3Endpoint: "http://seaweed.example:8333"} + s := &Server{filerClient: newMemFiler(), s3Endpoint: "http://seaweed.example:8333"} r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil) r.Header.Set(accessDelegationHeader, "vended-credentials") diff --git a/weed/s3api/iceberg/iceberg_create_table_test.go b/weed/s3api/iceberg/iceberg_create_table_test.go index 96f63d0a97..e566f476b9 100644 --- a/weed/s3api/iceberg/iceberg_create_table_test.go +++ b/weed/s3api/iceberg/iceberg_create_table_test.go @@ -102,7 +102,9 @@ func TestBuildLoadTableResultNeverReturnsNilMetadata(t *testing.T) { for name, getResp := range cases { t.Run(name, func(t *testing.T) { r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil) - result, err := (&Server{}).buildLoadTableResult(r, getResp, "bkt", []string{"ns"}, "t") + // The filer is empty, so a record holding only a location has no + // file behind it to serve. + result, err := (&Server{filerClient: newMemFiler()}).buildLoadTableResult(r, getResp, "bkt", []string{"ns"}, "t") if err != nil { t.Fatalf("buildLoadTableResult() error = %v, want nil", err) } diff --git a/weed/s3api/iceberg/iceberg_register_table_test.go b/weed/s3api/iceberg/iceberg_register_table_test.go index da847bd09c..d65e11ab84 100644 --- a/weed/s3api/iceberg/iceberg_register_table_test.go +++ b/weed/s3api/iceberg/iceberg_register_table_test.go @@ -1,13 +1,20 @@ package iceberg import ( + "context" "encoding/json" "net/http" "net/http/httptest" + "path" "strings" "testing" + "github.com/apache/iceberg-go/table" + "github.com/google/uuid" "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" ) // TestRegisterTableRequestWireShape pins the field names to the iceberg-go REST @@ -59,3 +66,299 @@ func TestRegisterTableRejectsOutOfBoundsLocation(t *testing.T) { }) } } + +const registerTestBucket = "warehouse" + +// seedMetadataFile writes content as the file name in the metadata directory +// of finance/ and returns its location. +func seedMetadataFile(fc *memFiler, tableName, name string, content []byte) string { + tablePath := path.Join("finance", tableName) + fc.seed(path.Join(metadataDirPath(registerTestBucket, tablePath), name), &filer_pb.Entry{Name: name, Content: content}) + return "s3://" + registerTestBucket + "/" + tablePath + "/metadata/" + name +} + +// seedRegistrableMetadata writes the metadata of a table with two snapshots, +// main on the second, the way the file a table is registered from holds them. +func seedRegistrableMetadata(t *testing.T, fc *memFiler, tableName string) (string, table.Metadata) { + t.Helper() + metadata := metadataWithSnapshots(t, []int64{1, 2}, 0) + content, err := json.Marshal(metadata) + if err != nil { + t.Fatalf("marshal metadata: %v", err) + } + return seedMetadataFile(fc, tableName, "00001-7c1e2f4a.metadata.json", content), metadata +} + +func registerTableRequest(name, metadataLocation string) *http.Request { + body := `{"name":"` + name + `","metadata-location":"` + metadataLocation + `"}` + r := httptest.NewRequest(http.MethodPost, "/v1/"+registerTestBucket+"/namespaces/finance/register", strings.NewReader(body)) + r = mux.SetURLVars(r, map[string]string{"prefix": registerTestBucket, "namespace": "finance"}) + return r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice")) +} + +func registeredTableRequest(method, name, body string) *http.Request { + r := httptest.NewRequest(method, "/v1/"+registerTestBucket+"/namespaces/finance/tables/"+name, strings.NewReader(body)) + r = mux.SetURLVars(r, map[string]string{"prefix": registerTestBucket, "namespace": "finance", "table": name}) + return r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice")) +} + +// commitWithUUIDAssertion is the commit Spark sends for an INSERT: it asserts +// the table UUID it loaded before changing anything. +func commitWithUUIDAssertion(tableUUID uuid.UUID) string { + return `{"requirements":[{"type":"assert-table-uuid","uuid":"` + tableUUID.String() + `"}],` + + `"updates":[{"action":"set-properties","updates":{"committed":"true"}}]}` +} + +// servedMetadata decodes the table metadata from a load, register, or commit +// response. +func servedMetadata(t *testing.T, what string, w *httptest.ResponseRecorder) table.Metadata { + t.Helper() + if w.Code != http.StatusOK { + t.Fatalf("%s: status = %d, want %d (body: %s)", what, w.Code, http.StatusOK, w.Body.String()) + } + var result struct { + Metadata json.RawMessage `json:"metadata"` + } + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("%s: decode response: %v", what, err) + } + metadata, err := table.ParseMetadataBytes(result.Metadata) + if err != nil { + t.Fatalf("%s: parse response metadata: %v", what, err) + } + return metadata +} + +// checkServesRegistered fails unless got is the registered table, not a +// placeholder: same UUID, same snapshots, same current snapshot. +func checkServesRegistered(t *testing.T, what string, got, want table.Metadata) { + t.Helper() + if got.TableUUID() != want.TableUUID() { + t.Errorf("%s: table UUID = %s, want %s", what, got.TableUUID(), want.TableUUID()) + } + if len(got.Snapshots()) != len(want.Snapshots()) { + t.Errorf("%s: %d snapshots, want %d", what, len(got.Snapshots()), len(want.Snapshots())) + } + if current := got.CurrentSnapshot(); current == nil || current.SnapshotID != want.CurrentSnapshot().SnapshotID { + t.Errorf("%s: current snapshot = %+v, want %d", what, current, want.CurrentSnapshot().SnapshotID) + } +} + +// storedMetadata returns the metadata the catalog record of finance. +// holds. +func storedMetadata(t *testing.T, s *Server, fc *memFiler, name string) *s3tables.TableMetadata { + t.Helper() + var getResp s3tables.GetTableResponse + err := fc.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s.tablesManager.Execute(context.Background(), s3tables.NewManagerClient(client), "GetTable", &s3tables.GetTableRequest{ + TableBucketARN: buildTableBucketARN(registerTestBucket), + Namespace: []string{"finance"}, + Name: name, + }, &getResp, "alice") + }) + if err != nil { + t.Fatalf("GetTable %s: %v", name, err) + } + return getResp.Metadata +} + +// loadAndCommit loads finance. twice, then commits to it with the UUID +// assertion Spark sends, checking each response serves the registered table. +func loadAndCommit(t *testing.T, s *Server, name string, registered table.Metadata) { + t.Helper() + for _, what := range []string{"first load", "second load"} { + w := httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, name, "")) + checkServesRegistered(t, what, servedMetadata(t, what, w), registered) + } + + w := httptest.NewRecorder() + s.handleUpdateTable(w, registeredTableRequest(http.MethodPost, name, commitWithUUIDAssertion(registered.TableUUID()))) + committed := servedMetadata(t, "commit", w) + checkServesRegistered(t, "commit", committed, registered) + if got := committed.Properties()["committed"]; got != "true" { + t.Errorf("commit: property committed = %q, want true", got) + } + + w = httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, name, "")) + loaded := servedMetadata(t, "load after commit", w) + checkServesRegistered(t, "load after commit", loaded, registered) + if got := loaded.Properties()["committed"]; got != "true" { + t.Errorf("load after commit: property committed = %q, want true", got) + } +} + +// A registered table serves the metadata it was registered from on every +// load, and a commit that asserts the table UUID lands on it. Loads used to +// serve an empty placeholder with a nil UUID, which hid the table's data and +// failed every commit on the UUID assertion. +func TestRegisterTableServesRegisteredMetadata(t *testing.T) { + fc := newMemFiler() + seedNamespace(fc, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + location, registered := seedRegistrableMetadata(t, fc, "events") + + w := httptest.NewRecorder() + s.handleRegisterTable(w, registerTableRequest("events", location)) + checkServesRegistered(t, "register", servedMetadata(t, "register", w), registered) + + // The record keeps what was registered, as a created table's does, rather + // than only pointing at the file. + stored := storedMetadata(t, s, fc, "events") + if stored == nil || len(stored.FullMetadata) == 0 { + t.Fatal("the catalog record does not keep the registered metadata") + } + if stored.Iceberg == nil || stored.Iceberg.TableUUID != registered.TableUUID().String() { + t.Errorf("catalog record UUID = %+v, want %s", stored.Iceberg, registered.TableUUID()) + } + + loadAndCommit(t, s, "events", registered) +} + +// A table registered before the record kept its metadata has only the +// metadata location in its record. Loading it reads the file there, and the +// first commit starts from that file rather than from a placeholder. +func TestLoadTableReadsMetadataFileOfOlderRegistration(t *testing.T) { + fc := newMemFiler() + seedNamespace(fc, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + location, registered := seedRegistrableMetadata(t, fc, "events") + + // The record an older RegisterTable left: the location and nothing else. + err := fc.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s.tablesManager.Execute(context.Background(), s3tables.NewManagerClient(client), "RegisterTable", &s3tables.RegisterTableRequest{ + TableBucketARN: buildTableBucketARN(registerTestBucket), + Namespace: []string{"finance"}, + Name: "events", + MetadataLocation: location, + }, nil, "alice") + }) + if err != nil { + t.Fatalf("seed older registration: %v", err) + } + + loadAndCommit(t, s, "events", registered) + + // The first commit stores the metadata in the record, under the UUID of the + // file it read rather than a fresh one. + stored := storedMetadata(t, s, fc, "events") + if stored == nil || stored.Iceberg == nil { + t.Fatal("the first commit stored no metadata in the record") + } + if stored.Iceberg.TableUUID != registered.TableUUID().String() { + t.Errorf("record table UUID after first commit = %s, want the file's %s", stored.Iceberg.TableUUID, registered.TableUUID()) + } +} + +// seedOversizedMetadata writes a metadata file larger than the manager's +// request limit, so it can only be registered by location and UUID. +func seedOversizedMetadata(t *testing.T, fc *memFiler, tableName string) (string, table.Metadata) { + t.Helper() + builder, err := table.MetadataBuilderFromBase(metadataWithSnapshots(t, []int64{1, 2}, 0), "") + if err != nil { + t.Fatalf("MetadataBuilderFromBase() error = %v", err) + } + if err := builder.SetProperties(map[string]string{"padding": strings.Repeat("x", s3tables.MaxRequestBodySize)}); err != nil { + t.Fatalf("SetProperties() error = %v", err) + } + registered, err := builder.Build() + if err != nil { + t.Fatalf("Build() error = %v", err) + } + content, err := json.Marshal(registered) + if err != nil { + t.Fatalf("marshal metadata: %v", err) + } + return seedMetadataFile(fc, tableName, "00001-7c1e2f4a.metadata.json", content), registered +} + +// A metadata file too large to travel in the manager request is registered by +// its location and UUID alone, and loads read it from there. Sending it whole +// failed every such registration on the manager's request size limit. +func TestRegisterTableKeepsOversizedMetadataByLocation(t *testing.T) { + fc := newMemFiler() + seedNamespace(fc, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + + location, registered := seedOversizedMetadata(t, fc, "events") + + w := httptest.NewRecorder() + s.handleRegisterTable(w, registerTableRequest("events", location)) + checkServesRegistered(t, "register", servedMetadata(t, "register", w), registered) + + stored := storedMetadata(t, s, fc, "events") + if stored == nil || stored.Iceberg == nil || stored.Iceberg.TableUUID != registered.TableUUID().String() { + t.Errorf("catalog record metadata = %+v, want the file's UUID %s", stored, registered.TableUUID()) + } + if stored != nil && len(stored.FullMetadata) != 0 { + t.Errorf("catalog record holds %d bytes of metadata, over the manager's request limit", len(stored.FullMetadata)) + } + + w = httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, "events", "")) + checkServesRegistered(t, "load", servedMetadata(t, "load", w), registered) +} + +// A commit to a location-only record cannot ship the new metadata back either: +// the pointer flips by location alone, and the record must not keep a stored +// copy that would shadow the freshly staged file. +func TestCommitTableOnOversizedRegistration(t *testing.T) { + fc := newMemFiler() + seedNamespace(fc, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + + location, registered := seedOversizedMetadata(t, fc, "events") + w := httptest.NewRecorder() + s.handleRegisterTable(w, registerTableRequest("events", location)) + checkServesRegistered(t, "register", servedMetadata(t, "register", w), registered) + + w = httptest.NewRecorder() + s.handleUpdateTable(w, registeredTableRequest(http.MethodPost, "events", commitWithUUIDAssertion(registered.TableUUID()))) + committed := servedMetadata(t, "commit", w) + checkServesRegistered(t, "commit", committed, registered) + if got := committed.Properties()["committed"]; got != "true" { + t.Errorf("commit: property committed = %q, want true", got) + } + + if stored := storedMetadata(t, s, fc, "events"); stored != nil && len(stored.FullMetadata) != 0 { + t.Errorf("catalog record holds %d bytes of metadata, over the manager's request limit", len(stored.FullMetadata)) + } + + w = httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, "events", "")) + if got := servedMetadata(t, "load", w).Properties()["committed"]; got != "true" { + t.Errorf("load: property committed = %q, want true", got) + } +} + +// A file that is not Iceberg metadata is the client's error, and no catalog +// record is left behind for it. +func TestRegisterTableRejectsInvalidMetadata(t *testing.T) { + for name, content := range map[string][]byte{ + "truncated": []byte(`{"format-version":2,"table-uuid":`), + "empty file": nil, + } { + t.Run(name, func(t *testing.T) { + fc := newMemFiler() + seedNamespace(fc, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + location := seedMetadataFile(fc, "events", "00001-7c1e2f4a.metadata.json", content) + + w := httptest.NewRecorder() + s.handleRegisterTable(w, registerTableRequest("events", location)) + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusBadRequest, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "BadRequestException") { + t.Errorf("body = %s, want a BadRequestException", w.Body.String()) + } + + w = httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, "events", "")) + if w.Code != http.StatusNotFound { + t.Errorf("load after rejected register = %d, want %d", w.Code, http.StatusNotFound) + } + }) + } +} diff --git a/weed/s3api/iceberg/iceberg_transaction_test.go b/weed/s3api/iceberg/iceberg_transaction_test.go index e1a776a820..4d43748ced 100644 --- a/weed/s3api/iceberg/iceberg_transaction_test.go +++ b/weed/s3api/iceberg/iceberg_transaction_test.go @@ -2,11 +2,23 @@ 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/google/uuid" + "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" ) func TestDecodeCommitTransactionRequest(t *testing.T) { @@ -252,3 +264,241 @@ func (e *tableEntry) applyUpdate(req *s3tables.UpdateTableRequest) { e.metadataVersion = req.MetadataVersion } } + +// A transaction against a registration that kept only its metadata location +// reads that file: the UUID assertion the client took from its load passes, +// and the commit lands on the registered snapshots instead of a placeholder. +func TestCommitTransactionReadsMetadataFileOfOlderRegistration(t *testing.T) { + fc := newMemFiler() + seedNamespace(fc, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + location, registered := seedRegistrableMetadata(t, fc, "events") + + err := fc.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s.tablesManager.Execute(context.Background(), s3tables.NewManagerClient(client), "RegisterTable", &s3tables.RegisterTableRequest{ + TableBucketARN: buildTableBucketARN(registerTestBucket), + Namespace: []string{"finance"}, + Name: "events", + MetadataLocation: location, + }, nil, "alice") + }) + if err != nil { + t.Fatalf("seed older registration: %v", err) + } + + commitTransaction := func(requirements, updates string) *httptest.ResponseRecorder { + body := `{"table-changes":[{"identifier":{"namespace":["finance"],"name":"events"},` + + `"requirements":[` + requirements + `],"updates":[` + updates + `]}]}` + r := httptest.NewRequest(http.MethodPost, "/v1/"+registerTestBucket+"/transactions/commit", strings.NewReader(body)) + r = mux.SetURLVars(r, map[string]string{"prefix": registerTestBucket}) + r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice")) + w := httptest.NewRecorder() + s.handleCommitTransaction(w, r) + return w + } + + w := commitTransaction( + `{"type":"assert-table-uuid","uuid":"`+registered.TableUUID().String()+`"}`, + `{"action":"set-properties","updates":{"committed":"true"}}`) + if w.Code != http.StatusNoContent { + t.Fatalf("uuid-asserting transaction: status = %d, want %d (body: %s)", w.Code, http.StatusNoContent, w.Body.String()) + } + + w = commitTransaction("", `{"action":"set-properties","updates":{"again":"true"}}`) + if w.Code != http.StatusNoContent { + t.Fatalf("requirement-free transaction: status = %d, want %d (body: %s)", w.Code, http.StatusNoContent, w.Body.String()) + } + + w = httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, "events", "")) + loaded := servedMetadata(t, "load after transactions", w) + checkServesRegistered(t, "load after transactions", loaded, registered) + if got := loaded.Properties()["committed"]; got != "true" { + t.Errorf("property committed = %q, want true", got) + } + if got := loaded.Properties()["again"]; got != "true" { + t.Errorf("property again = %q, want true", got) + } +} + +// A transaction re-stamps the default name mapping the way a single-table +// commit does, so readers of field-id-less files keep resolving columns even +// when the table's record started without one. +func TestCommitTransactionRefreshesNameMapping(t *testing.T) { + fc := newMemFiler() + seedNamespace(fc, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "label", Type: iceberg.PrimitiveTypes.String}, + ) + metadata, err := newTableMetadata(uuid.New(), "s3://"+registerTestBucket+"/finance/events", schema, nil, nil, nil) + if err != nil { + t.Fatalf("newTableMetadata: %v", err) + } + raw, err := json.Marshal(metadata) + if err != nil { + t.Fatalf("marshal metadata: %v", err) + } + var doc map[string]any + if err := json.Unmarshal(raw, &doc); err != nil { + t.Fatalf("decode metadata: %v", err) + } + if props, ok := doc["properties"].(map[string]any); ok { + delete(props, table.DefaultNameMappingKey) + } + if raw, err = json.Marshal(doc); err != nil { + t.Fatalf("re-marshal metadata: %v", err) + } + location := seedMetadataFile(fc, "events", "00001-7c1e2f4a.metadata.json", raw) + + err = fc.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s.tablesManager.Execute(context.Background(), s3tables.NewManagerClient(client), "RegisterTable", &s3tables.RegisterTableRequest{ + TableBucketARN: buildTableBucketARN(registerTestBucket), + Namespace: []string{"finance"}, + Name: "events", + MetadataLocation: location, + }, nil, "alice") + }) + if err != nil { + t.Fatalf("seed older registration: %v", err) + } + + body := `{"table-changes":[{"identifier":{"namespace":["finance"],"name":"events"},` + + `"updates":[{"action":"set-properties","updates":{"committed":"true"}}]}]}` + r := httptest.NewRequest(http.MethodPost, "/v1/"+registerTestBucket+"/transactions/commit", strings.NewReader(body)) + r = mux.SetURLVars(r, map[string]string{"prefix": registerTestBucket}) + r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice")) + w := httptest.NewRecorder() + s.handleCommitTransaction(w, r) + if w.Code != http.StatusNoContent { + t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusNoContent, w.Body.String()) + } + + w = httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, "events", "")) + loaded := servedMetadata(t, "load after transaction", w) + mappingJSON, ok := loaded.Properties()[table.DefaultNameMappingKey] + if !ok || mappingJSON == "" { + t.Fatalf("committed metadata lost %s: %v", table.DefaultNameMappingKey, loaded.Properties()) + } +} + +// failUpdateFiler fails entry updates on one path, so a transaction's flip of +// that table fails after the tables before it in the request have flipped. +type failUpdateFiler struct { + *memFiler + failPath string +} + +func (f *failUpdateFiler) WithFilerClient(_ bool, fn func(client filer_pb.SeaweedFilerClient) error) error { + return fn(f) +} + +func (f *failUpdateFiler) UpdateEntry(ctx context.Context, in *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + if path.Join(in.Directory, in.Entry.Name) == f.failPath { + return nil, errors.New("injected update failure") + } + return f.memFiler.UpdateEntry(ctx, in, opts...) +} + +// A transaction that fails after flipping a table rolls it back. UpdateTable +// cannot clear the FullMetadata a flip stored, so for a registration that kept +// only its metadata location the restore has to write back the file's +// metadata: otherwise loads go on serving the change the client was told +// failed, and a client retrying the transaction applies it twice. +func TestCommitTransactionRollbackRestoresOlderRegistration(t *testing.T) { + fc := &failUpdateFiler{memFiler: newMemFiler()} + seedNamespace(fc.memFiler, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + location, registered := seedRegistrableMetadata(t, fc.memFiler, "events") + + err := fc.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s.tablesManager.Execute(context.Background(), s3tables.NewManagerClient(client), "RegisterTable", &s3tables.RegisterTableRequest{ + TableBucketARN: buildTableBucketARN(registerTestBucket), + Namespace: []string{"finance"}, + Name: "events", + MetadataLocation: location, + }, nil, "alice") + }) + if err != nil { + t.Fatalf("seed older registration: %v", err) + } + otherLocation, _ := seedRegistrableMetadata(t, fc.memFiler, "other") + w := httptest.NewRecorder() + s.handleRegisterTable(w, registerTableRequest("other", otherLocation)) + if w.Code != http.StatusOK { + t.Fatalf("register other: status = %d (body: %s)", w.Code, w.Body.String()) + } + + // events flips first; the flip of other then fails and rolls events back. + fc.failPath = s3tables.GetTablePath(registerTestBucket, "finance", "other") + change := func(name string) string { + return `{"identifier":{"namespace":["finance"],"name":"` + name + `"},` + + `"updates":[{"action":"set-properties","updates":{"committed":"true"}}]}` + } + body := `{"table-changes":[` + change("events") + `,` + change("other") + `]}` + r := httptest.NewRequest(http.MethodPost, "/v1/"+registerTestBucket+"/transactions/commit", strings.NewReader(body)) + r = mux.SetURLVars(r, map[string]string{"prefix": registerTestBucket}) + r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice")) + w = httptest.NewRecorder() + s.handleCommitTransaction(w, r) + if w.Code == http.StatusNoContent { + t.Fatal("transaction succeeded although the flip of other failed") + } + + w = httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, "events", "")) + loaded := servedMetadata(t, "load after rolled-back transaction", w) + checkServesRegistered(t, "load after rolled-back transaction", loaded, registered) + if got, ok := loaded.Properties()["committed"]; ok { + t.Errorf("load after rolled-back transaction serves its property committed = %q", got) + } +} + +// A transaction on a registration kept by location alone — one whose metadata +// is too large to ship — flips the pointer by location too, and the record +// holds no stored copy that would shadow the staged file. +func TestCommitTransactionOnOversizedRegistration(t *testing.T) { + fc := newMemFiler() + seedNamespace(fc, registerTestBucket, "finance", "alice") + s := NewServer(fc, nil) + location, registered := seedOversizedMetadata(t, fc, "events") + + err := fc.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s.tablesManager.Execute(context.Background(), s3tables.NewManagerClient(client), "RegisterTable", &s3tables.RegisterTableRequest{ + TableBucketARN: buildTableBucketARN(registerTestBucket), + Namespace: []string{"finance"}, + Name: "events", + MetadataLocation: location, + }, nil, "alice") + }) + if err != nil { + t.Fatalf("seed location-only registration: %v", err) + } + + body := `{"table-changes":[{"identifier":{"namespace":["finance"],"name":"events"},` + + `"requirements":[{"type":"assert-table-uuid","uuid":"` + registered.TableUUID().String() + `"}],` + + `"updates":[{"action":"set-properties","updates":{"committed":"true"}}]}]}` + r := httptest.NewRequest(http.MethodPost, "/v1/"+registerTestBucket+"/transactions/commit", strings.NewReader(body)) + r = mux.SetURLVars(r, map[string]string{"prefix": registerTestBucket}) + r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice")) + w := httptest.NewRecorder() + s.handleCommitTransaction(w, r) + if w.Code != http.StatusNoContent { + t.Fatalf("oversized transaction: status = %d, want %d (body: %s)", w.Code, http.StatusNoContent, w.Body.String()) + } + + if stored := storedMetadata(t, s, fc, "events"); stored != nil && len(stored.FullMetadata) != 0 { + t.Errorf("catalog record holds %d bytes of metadata, over the manager's request limit", len(stored.FullMetadata)) + } + + w = httptest.NewRecorder() + s.handleLoadTable(w, registeredTableRequest(http.MethodGet, "events", "")) + loaded := servedMetadata(t, "load after transaction", w) + checkServesRegistered(t, "load after transaction", loaded, registered) + if got := loaded.Properties()["committed"]; got != "true" { + t.Errorf("load after transaction: property committed = %q, want true", got) + } +} diff --git a/weed/s3api/iceberg/metadata_files.go b/weed/s3api/iceberg/metadata_files.go index 7d9021dc45..64e4a21484 100644 --- a/weed/s3api/iceberg/metadata_files.go +++ b/weed/s3api/iceberg/metadata_files.go @@ -2,6 +2,7 @@ package iceberg import ( "context" + "errors" "fmt" "os" "path" @@ -169,3 +170,31 @@ func (s *Server) loadMetadataFile(ctx context.Context, bucketName, tablePath, me } return content, nil } + +// currentMetadataBytes returns the Iceberg metadata a table serves: the copy +// kept in its catalog record or, for a record that holds only a metadata +// location, the file at that location. Tables registered before the record +// kept their metadata look like that, and serving an empty placeholder for +// them hides every snapshot and fails every commit on a table UUID mismatch. +// It returns nil, for the caller's placeholder, when the record has neither +// or the file is gone. +func (s *Server) currentMetadataBytes(ctx context.Context, getResp s3tables.GetTableResponse, bucketName string) ([]byte, error) { + if getResp.Metadata != nil && len(getResp.Metadata.FullMetadata) > 0 { + return getResp.Metadata.FullMetadata, nil + } + if getResp.MetadataLocation == "" { + return nil, nil + } + metadataBucket, tablePath, err := parseS3Location(tableLocationFromMetadataLocation(getResp.MetadataLocation)) + if err != nil { + return nil, err + } + if err := confineMetadataLocation(metadataBucket, tablePath, bucketName); err != nil { + return nil, err + } + content, err := s.loadTableMetadataBlob(ctx, metadataBucket, tablePath, path.Base(getResp.MetadataLocation)) + if errors.Is(err, filer_pb.ErrNotFound) { + return nil, nil + } + return content, err +} diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index eda5e9ae6c..521fcb868a 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -42,8 +42,9 @@ const ( EntryTypeTable = "table" EntryTypeView = "view" - // Maximum request body size (10MB) - maxRequestBodySize = 10 * 1024 * 1024 + // MaxRequestBodySize is the largest request body the handler reads (10MB). + // In-process callers that send a table's full metadata check against it. + MaxRequestBodySize = 10 * 1024 * 1024 ) var ( @@ -438,15 +439,15 @@ func (h *S3TablesHandler) readRequestBody(r *http.Request, v interface{}) error defer r.Body.Close() // Limit request body size to prevent unbounded reads - limitedReader := io.LimitReader(r.Body, maxRequestBodySize+1) + limitedReader := io.LimitReader(r.Body, MaxRequestBodySize+1) body, err := io.ReadAll(limitedReader) if err != nil { return fmt.Errorf("failed to read request body: %w", err) } // Check if body exceeds size limit - if len(body) > maxRequestBodySize { - return fmt.Errorf("request body too large: exceeds maximum size of %d bytes", maxRequestBodySize) + if len(body) > MaxRequestBodySize { + return fmt.Errorf("request body too large: exceeds maximum size of %d bytes", MaxRequestBodySize) } if len(body) == 0 { diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 79c41cf674..46687901b9 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -269,6 +269,7 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req VersionToken: versionToken, MetadataVersion: metadataVersionFromLocation(req.MetadataLocation), MetadataLocation: req.MetadataLocation, + Metadata: req.Metadata, } metadataBytes, err := json.Marshal(metadata) @@ -1758,6 +1759,9 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque metadata.Metadata.FullMetadata = req.Metadata.FullMetadata } } + if req.ClearFullMetadata && metadata.Metadata != nil { + metadata.Metadata.FullMetadata = nil + } if req.MetadataLocation != "" { metadata.MetadataLocation = req.MetadataLocation } diff --git a/weed/s3api/s3tables/handler_table_test.go b/weed/s3api/s3tables/handler_table_test.go index aacd11f80c..9ab2bd7379 100644 --- a/weed/s3api/s3tables/handler_table_test.go +++ b/weed/s3api/s3tables/handler_table_test.go @@ -1,6 +1,13 @@ package s3tables -import "testing" +import ( + "encoding/json" + "net/http" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) func TestMetadataVersionFromLocation(t *testing.T) { cases := []struct { @@ -23,3 +30,76 @@ func TestMetadataVersionFromLocation(t *testing.T) { } } } + +// A registration keeps the metadata it was given in the record, the way +// CreateTable does, so GetTable hands it back alongside the location. +func TestRegisterTableStoresMetadata(t *testing.T) { + fs, m := startRenameManager(t) + + const tableUUID = "6f1d0c5e-7a4b-4d2e-9c3f-1a2b3c4d5e6f" + location := "s3://" + renameTestBucket + "/ns/reg/metadata/00004-6f1d0c5e.metadata.json" + metadata := &TableMetadata{ + Iceberg: &IcebergMetadata{TableUUID: tableUUID}, + FullMetadata: json.RawMessage(`{"format-version":2,"table-uuid":"` + tableUUID + `"}`), + } + + status, resp := runHandler(t, m, fs.Client, "RegisterTable", &RegisterTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "reg", + MetadataLocation: location, + Metadata: metadata, + }) + require.Equal(t, http.StatusOK, status, "%+v", resp) + + got, err := runGetTable(t, m, fs, "ns", "reg") + require.NoError(t, err) + assert.Equal(t, location, got.MetadataLocation) + assert.Equal(t, 4, got.MetadataVersion) + require.NotNil(t, got.Metadata, "the registered metadata must be kept with the record") + require.NotNil(t, got.Metadata.Iceberg) + assert.Equal(t, tableUUID, got.Metadata.Iceberg.TableUUID) + assert.JSONEq(t, string(metadata.FullMetadata), string(got.Metadata.FullMetadata)) +} + +// UpdateTable drops the record's stored metadata copy when asked, leaving the +// table tracked by its location — a commit whose metadata outgrew the request +// limit must not leave the stale copy shadowing the new file. +func TestUpdateTableClearsFullMetadata(t *testing.T) { + fs, m := startRenameManager(t) + + const tableUUID = "6f1d0c5e-7a4b-4d2e-9c3f-1a2b3c4d5e6f" + location := "s3://" + renameTestBucket + "/ns/reg/metadata/00004-6f1d0c5e.metadata.json" + status, resp := runHandler(t, m, fs.Client, "RegisterTable", &RegisterTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "reg", + MetadataLocation: location, + Metadata: &TableMetadata{ + Iceberg: &IcebergMetadata{TableUUID: tableUUID}, + FullMetadata: json.RawMessage(`{"format-version":2,"table-uuid":"` + tableUUID + `"}`), + }, + }) + require.Equal(t, http.StatusOK, status, "%+v", resp) + + newLocation := "s3://" + renameTestBucket + "/ns/reg/metadata/00005-6f1d0c5e.metadata.json" + status, resp = runHandler(t, m, fs.Client, "UpdateTable", &UpdateTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "reg", + Metadata: &TableMetadata{Iceberg: &IcebergMetadata{TableUUID: tableUUID}}, + MetadataVersion: 5, + MetadataLocation: newLocation, + ClearFullMetadata: true, + }) + require.Equal(t, http.StatusOK, status, "%+v", resp) + + got, err := runGetTable(t, m, fs, "ns", "reg") + require.NoError(t, err) + assert.Equal(t, newLocation, got.MetadataLocation) + assert.Equal(t, 5, got.MetadataVersion) + require.NotNil(t, got.Metadata) + require.NotNil(t, got.Metadata.Iceberg) + assert.Equal(t, tableUUID, got.Metadata.Iceberg.TableUUID) + assert.Empty(t, got.Metadata.FullMetadata, "the stored copy must be dropped, not shadow the location") +} diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index 261c724be0..9681b14b19 100644 --- a/weed/s3api/s3tables/types.go +++ b/weed/s3api/s3tables/types.go @@ -208,10 +208,11 @@ type CreateTableResponse struct { } type RegisterTableRequest struct { - TableBucketARN string `json:"tableBucketARN"` - Namespace []string `json:"namespace"` - Name string `json:"name"` - MetadataLocation string `json:"metadataLocation"` + TableBucketARN string `json:"tableBucketARN"` + Namespace []string `json:"namespace"` + Name string `json:"name"` + MetadataLocation string `json:"metadataLocation"` + Metadata *TableMetadata `json:"metadata,omitempty"` } type RegisterTableResponse struct { @@ -281,6 +282,9 @@ type UpdateTableRequest struct { Metadata *TableMetadata `json:"metadata,omitempty"` MetadataVersion int `json:"metadataVersion,omitempty"` MetadataLocation string `json:"metadataLocation,omitempty"` + // ClearFullMetadata drops the record's stored metadata copy; a table whose + // metadata no longer fits in a request is tracked by location alone. + ClearFullMetadata bool `json:"clearFullMetadata,omitempty"` } type UpdateTableResponse struct { diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go index 74288613e5..96b04bb326 100644 --- a/weed/worker/tasks/iceberg/detection.go +++ b/weed/worker/tasks/iceberg/detection.go @@ -135,6 +135,10 @@ func (h *Handler) scanTablesForMaintenance( } continue } + if err := state.loadMetadata(ctx, filerClient, bucketName); err != nil { + glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot load iceberg metadata: %v", bucketName, nsName, tblName, err) + continue + } if !isIcebergTableEntry(tableEntry.Extended, state.Metadata) { glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: not an iceberg table", bucketName, nsName, tblName) continue diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index 9c05e010d2..440514c790 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/weed/worker/tasks/iceberg/filer_io.go @@ -16,6 +16,7 @@ import ( "sync" "time" + "github.com/apache/iceberg-go/table" "github.com/parquet-go/parquet-go" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -154,6 +155,9 @@ func loadCurrentMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient if err != nil { return nil, err } + if err := state.loadMetadata(ctx, client, bucketName); err != nil { + return nil, err + } // Refuse to rewrite an entry that only looks like an Iceberg table. Every // operation here rewrites or deletes files under the table location, and a // foreign format's files are unreferenced by Iceberg metadata by definition. @@ -163,6 +167,41 @@ func loadCurrentMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient return state, nil } +// metadataFileMaxBytes bounds the chunked read a location-only record may +// trigger: the location comes from the catalog record and can name any file +// in the bucket, not only ones a commit wrote. Inline content is exempt — +// the REST load path accepts it unbounded and the lookup already holds it in +// memory, so refusing it would only strand a table that loads fine. +var metadataFileMaxBytes int64 = 64 << 20 + +// loadMetadata fills in Metadata for a record that keeps no inline copy: a +// commit that outgrew the request-size limit stores only the metadata file's +// location, so the bytes come from the file it points to. +func (s *tableState) loadMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName string) error { + if s.Metadata != nil { + return nil + } + entry, err := lookupFileByIcebergPath(ctx, client, bucketName, s.DataPath, s.MetadataLocation) + if err != nil { + return fmt.Errorf("load metadata file %s: %w", s.MetadataLocation, err) + } + if len(entry.Content) == 0 && len(entry.Chunks) > 0 { + if size := int64(filer.FileSize(entry)); size > metadataFileMaxBytes { + return fmt.Errorf("metadata file %s too large: %d bytes", s.MetadataLocation, size) + } + } + content, err := readEntryContent(ctx, client, entry, s.MetadataLocation) + if err != nil { + return fmt.Errorf("load metadata file %s: %w", s.MetadataLocation, err) + } + meta, err := table.ParseMetadataBytes(content) + if err != nil { + return fmt.Errorf("parse iceberg metadata at %s: %w", s.MetadataLocation, err) + } + s.Metadata = meta + return nil +} + func isFilerNotFound(err error) bool { return errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound } @@ -231,7 +270,12 @@ func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClie if err != nil { return nil, err } + return readEntryContent(ctx, client, entry, icebergPath) +} +// readEntryContent returns an entry's bytes: its inline content, or a chunked +// file read through the volume servers. +func readEntryContent(ctx context.Context, client filer_pb.SeaweedFilerClient, entry *filer_pb.Entry, name string) ([]byte, error) { if len(entry.Content) > 0 || len(entry.Chunks) == 0 { return entry.Content, nil } @@ -243,7 +287,7 @@ func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClie } data, err := io.ReadAll(reader) if err != nil { - return nil, fmt.Errorf("read chunked file %s: %w", icebergPath, err) + return nil, fmt.Errorf("read chunked file %s: %w", name, err) } return data, nil } diff --git a/weed/worker/tasks/iceberg/filer_io_load_test.go b/weed/worker/tasks/iceberg/filer_io_load_test.go new file mode 100644 index 0000000000..b0d1f4e6b7 --- /dev/null +++ b/weed/worker/tasks/iceberg/filer_io_load_test.go @@ -0,0 +1,133 @@ +package iceberg + +import ( + "context" + "encoding/json" + "fmt" + "path" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// A commit that outgrew the request-size limit stores no inline metadata: the +// record keeps only metadataLocation. Maintenance must follow it or the table +// is skipped on every scan. +func TestLoadCurrentMetadataFollowsMetadataLocation(t *testing.T) { + fs, client := startFakeFiler(t) + ctx := context.Background() + + setup := tableSetup{BucketName: "b", Namespace: "ns", TableName: "t"} + meta := buildTestMetadata(t, nil, nil, 0, nil, nil, nil) + metaJSON, err := json.Marshal(meta) + if err != nil { + t.Fatalf("marshal metadata: %v", err) + } + + bucketPath := path.Join(s3tables.TablesPath, setup.BucketName) + nsPath := path.Join(bucketPath, setup.Namespace) + tableDir := path.Join(bucketPath, setup.tablePath()) + metadataFileName := "v1.metadata.json" + metadataLocation := fmt.Sprintf("s3://%s/%s/metadata/%s", setup.BucketName, setup.tablePath(), metadataFileName) + + fs.putEntry(s3tables.TablesPath, setup.BucketName, &filer_pb.Entry{ + Name: setup.BucketName, + IsDirectory: true, + Extended: map[string][]byte{s3tables.ExtendedKeyTableBucket: []byte("true")}, + }) + fs.putEntry(bucketPath, setup.Namespace, &filer_pb.Entry{ + Name: setup.Namespace, + IsDirectory: true, + }) + fs.putEntry(path.Join(tableDir, "metadata"), metadataFileName, &filer_pb.Entry{ + Name: metadataFileName, + Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(metaJSON))}, + Content: metaJSON, + }) + + xattr, err := json.Marshal(map[string]interface{}{ + "metadataVersion": 7, + "metadataLocation": metadataLocation, + }) + if err != nil { + t.Fatalf("marshal xattr: %v", err) + } + fs.putEntry(nsPath, setup.TableName, &filer_pb.Entry{ + Name: setup.TableName, + IsDirectory: true, + Extended: map[string][]byte{ + s3tables.ExtendedKeyMetadata: xattr, + s3tables.ExtendedKeyMetadataVersion: metadataVersionXattr(7), + }, + }) + + state, err := loadCurrentMetadata(ctx, client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("loadCurrentMetadata: %v", err) + } + if state.Metadata == nil { + t.Fatal("loadCurrentMetadata returned no metadata for a location-only record") + } + if got := state.Metadata.Location(); got != "s3://test-bucket/test-table" { + t.Fatalf("metadata location = %q, want the file's own location", got) + } + if state.MetadataFileName != metadataFileName { + t.Fatalf("MetadataFileName = %q, want %q", state.MetadataFileName, metadataFileName) + } + if state.DataPath != setup.tablePath() { + t.Fatalf("DataPath = %q, want %q", state.DataPath, setup.tablePath()) + } +} + +// A location-only record whose metadata file sits inline loads at any size: +// the REST load path accepts inline content unbounded, so a bound here would +// strand a table the catalog itself wrote. Only a chunked file — where the +// location can name any object in the bucket — is capped. +func TestLoadMetadataBoundsOnlyChunkedFiles(t *testing.T) { + fs, client := startFakeFiler(t) + ctx := context.Background() + + original := metadataFileMaxBytes + metadataFileMaxBytes = 8 + defer func() { metadataFileMaxBytes = original }() + + setup := tableSetup{BucketName: "b", Namespace: "ns", TableName: "t"} + meta := buildTestMetadata(t, nil, nil, 0, nil, nil, nil) + metaJSON, err := json.Marshal(meta) + if err != nil { + t.Fatalf("marshal metadata: %v", err) + } + if int64(len(metaJSON)) <= metadataFileMaxBytes { + t.Fatal("test needs the inline metadata file to exceed the bound") + } + metadataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata") + + inlineLocation := fmt.Sprintf("s3://%s/%s/metadata/inline.metadata.json", setup.BucketName, setup.tablePath()) + fs.putEntry(metadataDir, "inline.metadata.json", &filer_pb.Entry{ + Name: "inline.metadata.json", + Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(metaJSON))}, + Content: metaJSON, + }) + + state := &tableState{DataPath: setup.tablePath(), MetadataLocation: inlineLocation} + if err := state.loadMetadata(ctx, client, setup.BucketName); err != nil { + t.Fatalf("loadMetadata on inline content: %v", err) + } + if state.Metadata == nil { + t.Fatal("inline metadata file over the bound was not loaded") + } + + chunkedLocation := fmt.Sprintf("s3://%s/%s/metadata/chunked.metadata.json", setup.BucketName, setup.tablePath()) + fs.putEntry(metadataDir, "chunked.metadata.json", &filer_pb.Entry{ + Name: "chunked.metadata.json", + Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(metadataFileMaxBytes + 1)}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,0123456789ab"}}, + }) + state = &tableState{DataPath: setup.tablePath(), MetadataLocation: chunkedLocation} + if err := state.loadMetadata(ctx, client, setup.BucketName); err == nil || !strings.Contains(err.Error(), "too large") { + t.Fatalf("chunked file over the bound: err = %v, want too-large rejection", err) + } +} diff --git a/weed/worker/tasks/iceberg/planning_index.go b/weed/worker/tasks/iceberg/planning_index.go index 885e2817e8..bf3b794112 100644 --- a/weed/worker/tasks/iceberg/planning_index.go +++ b/weed/worker/tasks/iceberg/planning_index.go @@ -55,6 +55,9 @@ type tableMetadataEnvelope struct { type tableState struct { Metadata table.Metadata MetadataFileName string + // MetadataLocation is the record's metadata file pointer, and the only + // copy for a commit that outgrew the request-size limit. + MetadataLocation string // DataPath is the bucket-relative directory holding metadata/ and data/. DataPath string PlanningIndex *planningIndex @@ -91,15 +94,17 @@ func parseTableMetadataEnvelope(metadataBytes []byte, bucketName, tablePath stri if envelope.Format != "" && envelope.Format != s3tables.FormatIceberg { return nil, fmt.Errorf("%w: %s", errForeignFormat, envelope.Format) } - if envelope.Metadata == nil || len(envelope.Metadata.FullMetadata) == 0 { + var meta table.Metadata + if envelope.Metadata != nil && len(envelope.Metadata.FullMetadata) > 0 { + parsed, err := table.ParseMetadataBytes(envelope.Metadata.FullMetadata) + if err != nil { + return nil, fmt.Errorf("parse iceberg metadata: %w", err) + } + meta = parsed + } else if envelope.MetadataLocation == "" { return nil, fmt.Errorf("no fullMetadata in table xattr") } - meta, err := table.ParseMetadataBytes(envelope.Metadata.FullMetadata) - if err != nil { - return nil, fmt.Errorf("parse iceberg metadata: %w", err) - } - var index *planningIndex if len(envelope.PlanningIndex) > 0 { if err := json.Unmarshal(envelope.PlanningIndex, &index); err != nil { @@ -115,6 +120,7 @@ func parseTableMetadataEnvelope(metadataBytes []byte, bucketName, tablePath stri return &tableState{ Metadata: meta, MetadataFileName: metadataFileName, + MetadataLocation: envelope.MetadataLocation, DataPath: tableDataPath(bucketName, tablePath, envelope.MetadataLocation), PlanningIndex: index, }, nil