From b62bbb99fc9d9d85107b81452b211701e921b434 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nguy=E1=BB=85n=20L=E1=BB=99c=20Ph=C3=BAc?= <112325531+omninguyen@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:15:42 +0700 Subject: [PATCH] fix(iceberg): serve a registered table's metadata on load and commit (#12159) * iceberg: serve a registered table's metadata on load and commit RegisterTable saved only the metadata location, and loads and commits never read the file there. They served a placeholder with no schema, no snapshots and a new random table UUID on every load, so a registered table read as empty, an INSERT failed against the empty schema, and every other commit failed its assert-table-uuid requirement. Metadata uploaded through the S3 gateway, as a migrated table's is, is stored in chunks that registration did not read, so it registered an empty table. Read the file following its chunks, reject one that is not Iceberg metadata with a 400, and keep its content and table UUID in the record the way CreateTable does. Records registered before this change hold only the location. Loads and commits now read the file there, and the first commit stores it in the record under the file's table UUID. * iceberg: transactions read a location-only registration's metadata file * iceberg: cover transactions on a location-only registration * iceberg: refresh the default name mapping on transaction commits * iceberg: cover name-mapping refresh in transactions * iceberg: roll a transaction back to a registration's metadata file A transaction that failed after flipping a table registered before the record kept its metadata could not undo the flip. The rollback restored the record's prior metadata, which for such a table is only a location, and UpdateTable keeps the full metadata a flip stored when the restore sends none. Loads went on serving the change the client was told had failed, and a client retrying the transaction could apply it twice. Restore the metadata loads served from the file instead, under the file's table UUID. * iceberg: register oversized metadata by its location Registration sends the metadata file's content to the S3 Tables manager, which refuses a request over its 10MB body limit, so a table whose metadata is larger could no longer be registered at all. Register such a file by its location and table UUID alone; loads read it from that location, as they do for a record that never kept its metadata. Export the limit as s3tables.MaxRequestBodySize so the catalog can check against it. A commit to such a table still sends the whole metadata to UpdateTable and stays over the limit. * iceberg: commit oversized metadata by its location * iceberg: let maintenance follow a location-only table record * admin: read a table's metadata file when the record keeps none * iceberg: bound the metadata file a location-only record reads * iceberg: bound only chunked metadata reads in the worker Inline content is what the catalog itself wrote and the REST load path accepts it at any size, so rejecting it in loadMetadata stranded tables over 64 MiB from every maintenance scan. Cap only the chunked read, where the recorded location can name any object in the bucket. --------- Co-authored-by: chrislusf --- .../spark_register_table_test.go | 117 +++++++ weed/admin/dash/iceberg_data_preview.go | 29 ++ weed/admin/dash/s3tables_management.go | 1 + weed/s3api/iceberg/commit_helpers.go | 15 + weed/s3api/iceberg/handlers_commit.go | 25 +- weed/s3api/iceberg/handlers_table.go | 31 +- weed/s3api/iceberg/handlers_transaction.go | 44 ++- .../iceberg/iceberg_access_delegation_test.go | 2 +- .../iceberg/iceberg_create_table_test.go | 4 +- .../iceberg/iceberg_register_table_test.go | 303 ++++++++++++++++++ .../s3api/iceberg/iceberg_transaction_test.go | 250 +++++++++++++++ weed/s3api/iceberg/metadata_files.go | 29 ++ weed/s3api/s3tables/handler.go | 11 +- weed/s3api/s3tables/handler_table.go | 4 + weed/s3api/s3tables/handler_table_test.go | 82 ++++- weed/s3api/s3tables/types.go | 12 +- weed/worker/tasks/iceberg/detection.go | 4 + weed/worker/tasks/iceberg/filer_io.go | 46 ++- .../tasks/iceberg/filer_io_load_test.go | 133 ++++++++ weed/worker/tasks/iceberg/planning_index.go | 18 +- 20 files changed, 1123 insertions(+), 37 deletions(-) create mode 100644 test/s3tables/catalog_spark/spark_register_table_test.go create mode 100644 weed/worker/tasks/iceberg/filer_io_load_test.go 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