From 07c9e0db851205919c578e9b7c90f7b7fc760efa Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 18 Jul 2026 19:21:35 -0700 Subject: [PATCH] iceberg maintenance: write absolute s3:// locations into table metadata (#10363) The maintenance operations built manifest, manifest-list, data-file and metadata-log paths with a bare path.Join("metadata", ...), so a single maintenance run wrote scheme-less relative paths into the new snapshot, the metadata-log and the table xattr. SeaweedFS reads those back fine because normalizeIcebergPath accepts both forms, but strict readers resolve every location through S3FileIO and fail with "Invalid S3 URI, cannot determine scheme", leaving the whole table unreadable after any maintenance commit. Add absoluteIcebergPath, the inverse of normalizeIcebergPath, and apply it at every site that authors locations: rewrite_manifests, compact, rewrite_position_delete_files, and the shared commit path. The base is derived from the bucket and table path since that is where the worker physically writes, matching the locations the REST catalog generates. --- weed/worker/tasks/iceberg/compact.go | 6 +-- weed/worker/tasks/iceberg/delete_rewrite.go | 6 +-- weed/worker/tasks/iceberg/exec_test.go | 53 +++++++++++++++++-- weed/worker/tasks/iceberg/filer_io.go | 15 +++++- weed/worker/tasks/iceberg/operations.go | 10 ++-- .../worker/tasks/iceberg/where_filter_test.go | 2 +- 6 files changed, 75 insertions(+), 17 deletions(-) diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index acc4adc51..21b6ae75c 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -262,7 +262,7 @@ func (h *Handler) compactDataFiles( } mergedFileName := fmt.Sprintf("compact-%d-%d-%s-%d.parquet", snapshotID, newSnapID, artifactSuffix, binIdx) - mergedFilePath := path.Join("data", mergedFileName) + mergedFilePath := absoluteIcebergPath(bucketName, tablePath, "data", mergedFileName) var mergedData []byte var recordCount int64 @@ -408,7 +408,7 @@ func (h *Handler) compactDataFiles( var manifestBuf bytes.Buffer manifestFileName := fmt.Sprintf("compact-%d-%s-spec%d.avro", newSnapID, artifactSuffix, se.specID) newManifest, err := iceberg.WriteManifest( - path.Join("metadata", manifestFileName), + absoluteIcebergPath(bucketName, tablePath, "metadata", manifestFileName), &manifestBuf, version, ps, @@ -470,7 +470,7 @@ func (h *Handler) compactDataFiles( writtenArtifacts = append(writtenArtifacts, artifact{dir: metaDir, fileName: manifestListFileName}) // Commit: add new snapshot and update main branch ref - manifestListLocation := path.Join("metadata", manifestListFileName) + manifestListLocation := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestListFileName) err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error { // Guard: verify table head hasn't advanced since we planned. cs := currentMeta.CurrentSnapshot() diff --git a/weed/worker/tasks/iceberg/delete_rewrite.go b/weed/worker/tasks/iceberg/delete_rewrite.go index 8e0c4f167..4b5e3fb81 100644 --- a/weed/worker/tasks/iceberg/delete_rewrite.go +++ b/weed/worker/tasks/iceberg/delete_rewrite.go @@ -440,7 +440,7 @@ func (h *Handler) rewritePositionDeleteFiles( dfBuilder, err := iceberg.NewDataFileBuilder( spec, iceberg.EntryContentPosDeletes, - path.Join("data", fileName), + absoluteIcebergPath(bucketName, tablePath, "data", fileName), iceberg.ParquetFile, group.Partition, nil, nil, @@ -512,7 +512,7 @@ func (h *Handler) rewritePositionDeleteFiles( return "", nil, fmt.Errorf("partition spec %d not found", specID) } manifestName := fmt.Sprintf("rewrite-delete-%d-%s-spec%d.avro", newSnapID, artifactSuffix, specID) - manifestPath := path.Join("metadata", manifestName) + manifestPath := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestName) mf, manifestBytes, err := writeManifestWithContent( manifestPath, version, @@ -542,7 +542,7 @@ func (h *Handler) rewritePositionDeleteFiles( } writtenArtifacts = append(writtenArtifacts, artifact{dir: metaDir, fileName: manifestListName}) - manifestListLocation := path.Join("metadata", manifestListName) + manifestListLocation := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestListName) err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error { cs := currentMeta.CurrentSnapshot() if cs == nil || cs.SnapshotID != snapshotID { diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go index 2ac29356e..c728766c5 100644 --- a/weed/worker/tasks/iceberg/exec_test.go +++ b/weed/worker/tasks/iceberg/exec_test.go @@ -766,6 +766,53 @@ func TestRewriteManifestsExecution(t *testing.T) { if updates == 0 { t.Error("expected at least one UpdateEntry call for xattr update") } + + // The spec requires absolute locations — strict readers (Spark/Trino via + // S3FileIO) reject scheme-less paths, so verify every written location. + wantPrefix := "s3://test-bucket/analytics/events/metadata/" + newMeta, _, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("reload metadata: %v", err) + } + newSnap := newMeta.CurrentSnapshot() + if newSnap == nil || !strings.HasPrefix(newSnap.ManifestList, wantPrefix+"snap-") { + t.Fatalf("new snapshot manifest list should be absolute, got %+v", newSnap) + } + foundPreviousEntry := false + for mle := range newMeta.PreviousFiles() { + if mle.MetadataFile == wantPrefix+"v1.metadata.json" { + foundPreviousEntry = true + } + } + if !foundPreviousEntry { + t.Error("metadata-log should record the previous metadata file at its absolute location") + } + mlData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, setup.tablePath(), newSnap.ManifestList) + if err != nil { + t.Fatalf("load new manifest list: %v", err) + } + newManifests, err := iceberg.ReadManifestList(bytes.NewReader(mlData)) + if err != nil { + t.Fatalf("parse new manifest list: %v", err) + } + for _, mf := range newManifests { + if !strings.HasPrefix(mf.FilePath(), wantPrefix+"merged-") { + t.Errorf("merged manifest path should be absolute, got %q", mf.FilePath()) + } + } + tableEntry := fs.getEntry(path.Join(s3tables.TablesPath, setup.BucketName, setup.Namespace), setup.TableName) + if tableEntry == nil { + t.Fatal("table entry missing") + } + var xattrMeta struct { + MetadataLocation string `json:"metadataLocation"` + } + if err := json.Unmarshal(tableEntry.Extended[s3tables.ExtendedKeyMetadata], &xattrMeta); err != nil { + t.Fatalf("unmarshal table xattr: %v", err) + } + if !strings.HasPrefix(xattrMeta.MetadataLocation, wantPrefix+"v2-") { + t.Errorf("xattr metadataLocation should be absolute, got %q", xattrMeta.MetadataLocation) + } } func TestRewriteManifestsBelowThreshold(t *testing.T) { @@ -3318,7 +3365,7 @@ func TestRewritePositionDeleteFilesExecution(t *testing.T) { if len(liveDeletePaths) != 1 { t.Fatalf("expected 1 live rewritten delete file, got %v", liveDeletePaths) } - if !strings.HasPrefix(liveDeletePaths[0], "data/rewrite-delete-") { + if !strings.HasPrefix(liveDeletePaths[0], "s3://tb/ns/tbl/data/rewrite-delete-") { t.Fatalf("expected rewritten delete file path, got %q", liveDeletePaths[0]) } } @@ -3545,7 +3592,7 @@ func TestRewritePositionDeleteFilesPreservesUnsupportedMultiTargetDeletes(t *tes if posPaths[0] != "data/pd3.parquet" && posPaths[1] != "data/pd3.parquet" { t.Fatalf("expected multi-target delete file to be preserved, got %v", posPaths) } - if !strings.HasPrefix(posPaths[0], "data/rewrite-delete-") && !strings.HasPrefix(posPaths[1], "data/rewrite-delete-") { + if !strings.HasPrefix(posPaths[0], "s3://tb/ns/tbl/data/rewrite-delete-") && !strings.HasPrefix(posPaths[1], "s3://tb/ns/tbl/data/rewrite-delete-") { t.Fatalf("expected rewritten delete file to remain live, got %v", posPaths) } } @@ -3614,7 +3661,7 @@ func TestRewritePositionDeleteFilesRebuildsMixedDeleteManifests(t *testing.T) { } posPaths, eqPaths := loadLiveDeleteFilePaths(t, client, setup.BucketName, setup.tablePath()) - if len(posPaths) != 1 || !strings.HasPrefix(posPaths[0], "data/rewrite-delete-") { + if len(posPaths) != 1 || !strings.HasPrefix(posPaths[0], "s3://tb/ns/tbl/data/rewrite-delete-") { t.Fatalf("expected only the rewritten position delete file to remain live, got %v", posPaths) } if len(eqPaths) != 1 || eqPaths[0] != "data/eq1.parquet" { diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index 406ae4560..ba338f13f 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/weed/worker/tasks/iceberg/filer_io.go @@ -251,6 +251,16 @@ func normalizeIcebergPath(icebergPath, bucketName, tablePath string) string { return p } +// absoluteIcebergPath is the inverse of normalizeIcebergPath: it builds the +// absolute s3:// URI for a file under the table root. The Iceberg spec +// requires absolute locations in metadata — strict readers (Spark/Trino via +// S3FileIO) reject paths with no scheme. The base is derived from +// bucketName/tablePath because that is where this package physically writes +// every file, regardless of the location recorded in the table metadata. +func absoluteIcebergPath(bucketName, tablePath string, elem ...string) string { + return "s3://" + path.Join(append([]string{bucketName, tablePath}, elem...)...) +} + // saveFilerFile saves a file to the filer. func saveFilerFile(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, fileName string, content []byte) error { resp, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{ @@ -285,8 +295,9 @@ func deleteFilerFile(ctx context.Context, client filer_pb.SeaweedFilerClient, di // writing and passes the previous metadata xattr back to the filer as a // server-side precondition so concurrent writers fail with a retryable // metadata version conflict. -// newMetadataLocation is the table-relative path to the new metadata file -// (e.g. "metadata/v3.metadata.json"). +// newMetadataLocation is the absolute location of the new metadata file +// (e.g. "s3://bucket/ns/table/metadata/v3.metadata.json"), matching the form +// the S3 Tables catalog stores. func updateTableMetadataXattr(ctx context.Context, client filer_pb.SeaweedFilerClient, tableDir string, expectedVersion int, newFullMetadata []byte, newMetadataLocation string) error { tableName := path.Base(tableDir) parentDir := path.Dir(tableDir) diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go index 8a946a2be..3430d7703 100644 --- a/weed/worker/tasks/iceberg/operations.go +++ b/weed/worker/tasks/iceberg/operations.go @@ -456,7 +456,7 @@ func (h *Handler) rewriteManifests( for _, se := range specMap { totalEntries += len(se.entries) manifestFileName := fmt.Sprintf("merged-%d-%s-spec%d.avro", newSnapshotID, artifactSuffix, se.specID) - manifestPath := path.Join("metadata", manifestFileName) + manifestPath := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestFileName) var manifestBuf bytes.Buffer mergedManifest, err := iceberg.WriteManifest( @@ -500,7 +500,7 @@ func (h *Handler) rewriteManifests( writtenArtifacts = append(writtenArtifacts, artifact{dir: metaDir, fileName: manifestListFileName}) // Create new snapshot with the rewritten manifest list - manifestListLocation := path.Join("metadata", manifestListFileName) + manifestListLocation := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestListFileName) err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error { // Guard: verify table head hasn't advanced since we planned. @@ -590,9 +590,9 @@ func (h *Handler) commitWithRetry( return fmt.Errorf("load metadata (attempt %d): %w", attempt, err) } - // Build new metadata — pass the current metadata file path so the + // Build new metadata — pass the current metadata file location so the // metadata log correctly records where the previous version lives. - currentMetaFilePath := path.Join("metadata", metaFileName) + currentMetaFilePath := absoluteIcebergPath(bucketName, tablePath, "metadata", metaFileName) builder, err := table.MetadataBuilderFromBase(meta, currentMetaFilePath) if err != nil { return fmt.Errorf("create metadata builder (attempt %d): %w", attempt, err) @@ -632,7 +632,7 @@ func (h *Handler) commitWithRetry( // Update the table entry's xattr with new metadata (CAS on version) tableDir := path.Join(s3tables.TablesPath, bucketName, tablePath) - newMetadataLocation := path.Join("metadata", newMetadataFileName) + newMetadataLocation := absoluteIcebergPath(bucketName, tablePath, "metadata", newMetadataFileName) err = updateTableMetadataXattr(ctx, filerClient, tableDir, currentVersion, metadataBytes, newMetadataLocation) if err != nil { // Use a detached context for cleanup so staged files are removed diff --git a/weed/worker/tasks/iceberg/where_filter_test.go b/weed/worker/tasks/iceberg/where_filter_test.go index 46290648b..f62149a94 100644 --- a/weed/worker/tasks/iceberg/where_filter_test.go +++ b/weed/worker/tasks/iceberg/where_filter_test.go @@ -274,7 +274,7 @@ func TestCompactDataFilesWhereFilter(t *testing.T) { var compactedCount int for _, p := range liveDataPaths { switch { - case strings.HasPrefix(p, "data/compact-"): + case strings.HasPrefix(p, "s3://tb/ns/tbl/data/compact-"): compactedCount++ case p == "data/eu-1.parquet", p == "data/eu-2.parquet": default: