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

@@ -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()

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 {

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" {

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)

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

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: