iceberg maintenance: resolve table files from the recorded location (#10418)

The worker assumed every file of a table sits under its catalog path, so
loadFileByIcebergPath stripped the scheme off a recorded location and joined
the remainder onto /buckets/<bucket>/<ns>/<table>. A table the REST catalog
placed elsewhere in the bucket — which is what a client gets whenever the
catalog path is already occupied — then resolves to a doubled path:

  lookup /buckets/lake/source/t/lake/source/t-0cd81bca-.../metadata/snap-.avro

so the very first manifest list read fails and the job fails again on every
scan interval, indefinitely.

Resolve absolute references (s3:// URIs and /buckets paths) from the bucket
root and keep relative ones under the table's own directory; the
bucket-relative form is now the canonical key everywhere references are
compared. That directory comes from the metadata location the catalog stores,
so reads, writes and deletes all land where the table's other files are
instead of splitting it across two trees. References outside the table's
bucket are rejected rather than silently misresolved.

Rewritten position-delete files now name their data file by absolute URI,
the way the table itself names it, instead of a path relative to the table.

weed/s3api/s3tables/handler_table.go

@@ -1182,7 +1182,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
 
 	// Delete the table
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		dataPath := tableDataDirFromMetadataLocation(metadata.MetadataLocation)
+		dataPath := TableDataDirFromMetadataLocation(metadata.MetadataLocation)
 		if dataPath != "" && dataPath != tablePath && strings.HasPrefix(dataPath+"/", GetTableBucketPath(bucketName)+"/") {
 			// Refuse to purge a data path that is an ancestor of the table's own
 			// name path (e.g. corrupt metadata resolving to the bucket or

weed/s3api/s3tables/table_data_dir_test.go

@@ -16,8 +16,8 @@ func TestTableDataDirFromMetadataLocation(t *testing.T) {
 		{"", ""},
 	}
 	for _, c := range cases {
-		if got := tableDataDirFromMetadataLocation(c.loc); got != c.want {
-			t.Errorf("tableDataDirFromMetadataLocation(%q) = %q, want %q", c.loc, got, c.want)
+		if got := TableDataDirFromMetadataLocation(c.loc); got != c.want {
+			t.Errorf("TableDataDirFromMetadataLocation(%q) = %q, want %q", c.loc, got, c.want)
 		}
 	}
 }

weed/s3api/s3tables/utils.go

@@ -97,11 +97,11 @@ func GetTablePath(bucketName, namespace, tableName string) string {
 	return path.Join(TablesPath, bucketName, namespace, tableName)
 }
 
-// tableDataDirFromMetadataLocation maps a table's s3:// metadata location to the
+// TableDataDirFromMetadataLocation maps a table's s3:// metadata location to the
 // filer directory holding its data. A renamed table is catalog-only, so its data
 // stays at the original location while its catalog entry moves; this lets a drop
 // purge the real data instead of the now-empty catalog path.
-func tableDataDirFromMetadataLocation(metadataLocation string) string {
+func TableDataDirFromMetadataLocation(metadataLocation string) string {
 	loc := strings.TrimSuffix(metadataLocation, "/")
 	if idx := strings.LastIndex(loc, "/metadata/"); idx != -1 {
 		loc = loc[:idx]

weed/worker/tasks/iceberg/compact.go

@@ -44,10 +44,11 @@ func (h *Handler) compactDataFiles(
 	onProgress func(binIdx, totalBins int),
 ) (string, map[string]int64, error) {
 	start := time.Now()
-	meta, metadataFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+	state, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
 	if err != nil {
 		return "", nil, fmt.Errorf("load metadata: %w", err)
 	}
+	meta, dataPath := state.Metadata, state.DataPath
 	predicate, err := parsePartitionPredicate(config.Where, meta)
 	if err != nil {
 		return "", nil, err
@@ -64,7 +65,7 @@ func (h *Handler) compactDataFiles(
 	}
 
 	// Read manifest list
-	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, currentSnap.ManifestList)
+	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, currentSnap.ManifestList)
 	if err != nil {
 		return "", nil, fmt.Errorf("read manifest list: %w", err)
 	}
@@ -93,7 +94,7 @@ func (h *Handler) compactDataFiles(
 	// Collect data file entries from data manifests
 	var allEntries []iceberg.ManifestEntry
 	for _, mf := range dataManifests {
-		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, mf.FilePath())
+		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
 		if err != nil {
 			return "", nil, fmt.Errorf("read manifest %s: %w", mf.FilePath(), err)
 		}
@@ -112,7 +113,7 @@ func (h *Handler) compactDataFiles(
 	if config.ApplyDeletes && len(deleteManifests) > 0 {
 		var allDeleteEntries []iceberg.ManifestEntry
 		for _, mf := range deleteManifests {
-			manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, mf.FilePath())
+			manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
 			if err != nil {
 				return "", nil, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), err)
 			}
@@ -149,14 +150,14 @@ func (h *Handler) compactDataFiles(
 		}
 
 		if len(posDeleteEntries) > 0 {
-			positionDeletes, err = collectPositionDeletes(ctx, filerClient, bucketName, tablePath, posDeleteEntries)
+			positionDeletes, err = collectPositionDeletes(ctx, filerClient, bucketName, dataPath, posDeleteEntries)
 			if err != nil {
 				return "", nil, fmt.Errorf("collect position deletes: %w", err)
 			}
 		}
 
 		if len(eqDeleteEntries) > 0 {
-			eqDeleteGroups, err = collectEqualityDeletes(ctx, filerClient, bucketName, tablePath, eqDeleteEntries, meta.CurrentSchema())
+			eqDeleteGroups, err = collectEqualityDeletes(ctx, filerClient, bucketName, dataPath, eqDeleteEntries, meta.CurrentSchema())
 			if err != nil {
 				return "", nil, fmt.Errorf("collect equality deletes: %w", err)
 			}
@@ -230,8 +231,8 @@ func (h *Handler) compactDataFiles(
 		return entrySeqNum(entry)
 	}
 
-	metaDir := path.Join(s3tables.TablesPath, bucketName, tablePath, "metadata")
-	dataDir := path.Join(s3tables.TablesPath, bucketName, tablePath, "data")
+	metaDir := path.Join(s3tables.TablesPath, bucketName, dataPath, "metadata")
+	dataDir := path.Join(s3tables.TablesPath, bucketName, dataPath, "data")
 
 	// Track written artifacts so we can clean them up if the commit fails.
 	type artifact struct {
@@ -262,14 +263,14 @@ func (h *Handler) compactDataFiles(
 		}
 
 		mergedFileName := fmt.Sprintf("compact-%d-%d-%s-%d.parquet", snapshotID, newSnapID, artifactSuffix, binIdx)
-		mergedFilePath := absoluteIcebergPath(bucketName, tablePath, "data", mergedFileName)
+		mergedFilePath := absoluteIcebergPath(bucketName, dataPath, "data", mergedFileName)
 
 		var mergedData []byte
 		var recordCount int64
 		if rewritePlan != nil && rewritePlan.strategy == "sort" {
-			mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, tablePath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan)
+			mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan)
 		} else {
-			mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, tablePath, bin.Entries, positionDeletes, eqDeleteGroups, schema)
+			mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema)
 		}
 		if err != nil {
 			glog.Warningf("iceberg compact: failed to merge bin %d (%d files): %v", binIdx, len(bin.Entries), err)
@@ -408,7 +409,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(
-			absoluteIcebergPath(bucketName, tablePath, "metadata", manifestFileName),
+			absoluteIcebergPath(bucketName, dataPath, "metadata", manifestFileName),
 			&manifestBuf,
 			version,
 			ps,
@@ -470,8 +471,8 @@ func (h *Handler) compactDataFiles(
 	writtenArtifacts = append(writtenArtifacts, artifact{dir: metaDir, fileName: manifestListFileName})
 
 	// Commit: add new snapshot and update main branch ref
-	manifestListLocation := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestListFileName)
-	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
+	manifestListLocation := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestListFileName)
+	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, state.MetadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
 		// Guard: verify table head hasn't advanced since we planned.
 		cs := currentMeta.CurrentSnapshot()
 		if cs == nil || cs.SnapshotID != snapshotID {
@@ -688,7 +689,7 @@ func partitionKey(partition map[int]any) string {
 func collectPositionDeletes(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	deleteEntries []iceberg.ManifestEntry,
 ) (map[string][]int64, error) {
 	result := make(map[string][]int64)
@@ -696,12 +697,15 @@ func collectPositionDeletes(
 		if entry.DataFile().ContentType() != iceberg.EntryContentPosDeletes {
 			continue
 		}
-		fileDeletes, err := readPositionDeleteFile(ctx, filerClient, bucketName, tablePath, entry.DataFile().FilePath())
+		fileDeletes, err := readPositionDeleteFile(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath())
 		if err != nil {
 			return nil, fmt.Errorf("read position delete file %s: %w", entry.DataFile().FilePath(), err)
 		}
 		for filePath, positions := range fileDeletes {
-			normalized := normalizeIcebergPath(filePath, bucketName, tablePath)
+			normalized, err := normalizeIcebergPath(filePath, bucketName, dataPath)
+			if err != nil {
+				return nil, fmt.Errorf("resolve deleted file %s: %w", filePath, err)
+			}
 			result[normalized] = append(result[normalized], positions...)
 		}
 	}
@@ -720,9 +724,9 @@ func collectPositionDeletes(
 func readPositionDeleteFile(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath, filePath string,
+	bucketName, dataPath, filePath string,
 ) (map[string][]int64, error) {
-	data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, filePath)
+	data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, filePath)
 	if err != nil {
 		return nil, err
 	}
@@ -783,7 +787,7 @@ type equalityDeleteGroup struct {
 func collectEqualityDeletes(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	deleteEntries []iceberg.ManifestEntry,
 	schema *iceberg.Schema,
 ) ([]equalityDeleteGroup, error) {
@@ -809,7 +813,7 @@ func collectEqualityDeletes(
 			groups[groupKey] = gs
 		}
 
-		keys, err := readEqualityDeleteFile(ctx, filerClient, bucketName, tablePath, entry.DataFile().FilePath(), eqFieldIDs, schema)
+		keys, err := readEqualityDeleteFile(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath(), eqFieldIDs, schema)
 		if err != nil {
 			return nil, fmt.Errorf("read equality delete file %s: %w", entry.DataFile().FilePath(), err)
 		}
@@ -831,11 +835,11 @@ func collectEqualityDeletes(
 func readEqualityDeleteFile(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath, filePath string,
+	bucketName, dataPath, filePath string,
 	fieldIDs []int,
 	icebergSchema *iceberg.Schema,
 ) (map[string]struct{}, error) {
-	data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, filePath)
+	data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, filePath)
 	if err != nil {
 		return nil, err
 	}
@@ -922,7 +926,7 @@ func resolveEqualityColIndices(pqSchema *parquet.Schema, fieldIDs []int, iceberg
 func mergeParquetFiles(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	entries []iceberg.ManifestEntry,
 	positionDeletes map[string][]int64,
 	eqDeleteGroups []equalityDeleteGroup,
@@ -933,7 +937,7 @@ func mergeParquetFiles(
 	}
 
 	// Load the first file to obtain the schema for the writer.
-	firstData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, entries[0].DataFile().FilePath())
+	firstData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, entries[0].DataFile().FilePath())
 	if err != nil {
 		return nil, 0, fmt.Errorf("read parquet file %s: %w", entries[0].DataFile().FilePath(), err)
 	}
@@ -954,7 +958,7 @@ func mergeParquetFiles(
 	writer := parquet.NewWriter(&outputBuf, parquetSchema)
 
 	drainReader := func(reader *parquet.Reader, source string) (int64, error) {
-		return visitFilteredParquetRows(ctx, reader, source, bucketName, tablePath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error {
+		return visitFilteredParquetRows(ctx, reader, source, bucketName, dataPath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error {
 			if _, err := writer.WriteRows(filtered); err != nil {
 				return fmt.Errorf("write rows from %s: %w", source, err)
 			}
@@ -980,7 +984,7 @@ func mergeParquetFiles(
 		default:
 		}
 
-		data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, entry.DataFile().FilePath())
+		data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath())
 		if err != nil {
 			writer.Close()
 			return nil, 0, fmt.Errorf("read parquet file %s: %w", entry.DataFile().FilePath(), err)
@@ -1037,7 +1041,7 @@ func resolveEqualityDeleteGroupsForSchema(
 func visitFilteredParquetRows(
 	ctx context.Context,
 	reader *parquet.Reader,
-	source, bucketName, tablePath string,
+	source, bucketName, dataPath string,
 	positionDeletes map[string][]int64,
 	resolvedEqGroups []resolvedEqDeleteGroup,
 	onRows func([]parquet.Row) error,
@@ -1046,7 +1050,10 @@ func visitFilteredParquetRows(
 
 	rows := make([]parquet.Row, 256)
 	filteredRows := make([]parquet.Row, 0, len(rows))
-	normalizedSource := normalizeIcebergPath(source, bucketName, tablePath)
+	normalizedSource, err := normalizeIcebergPath(source, bucketName, dataPath)
+	if err != nil {
+		return 0, err
+	}
 	posDeletes := positionDeletes[normalizedSource]
 	posDeleteIdx := 0
 	var absolutePos int64
@@ -1107,7 +1114,7 @@ func visitFilteredParquetRows(
 func mergeParquetFilesSorted(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	entries []iceberg.ManifestEntry,
 	positionDeletes map[string][]int64,
 	eqDeleteGroups []equalityDeleteGroup,
@@ -1121,7 +1128,7 @@ func mergeParquetFilesSorted(
 		return nil, 0, fmt.Errorf("sorted merge requires sort rewrite plan")
 	}
 
-	firstData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, entries[0].DataFile().FilePath())
+	firstData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, entries[0].DataFile().FilePath())
 	if err != nil {
 		return nil, 0, fmt.Errorf("read parquet file %s: %w", entries[0].DataFile().FilePath(), err)
 	}
@@ -1148,7 +1155,7 @@ func mergeParquetFilesSorted(
 	var allRows []parquet.Row
 
 	collectRows := func(reader *parquet.Reader, source string) (int64, error) {
-		return visitFilteredParquetRows(ctx, reader, source, bucketName, tablePath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error {
+		return visitFilteredParquetRows(ctx, reader, source, bucketName, dataPath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error {
 			for _, row := range filtered {
 				allRows = append(allRows, row.Clone())
 			}
@@ -1168,7 +1175,7 @@ func mergeParquetFilesSorted(
 		default:
 		}
 
-		data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, entry.DataFile().FilePath())
+		data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath())
 		if err != nil {
 			return nil, 0, fmt.Errorf("read parquet file %s: %w", entry.DataFile().FilePath(), err)
 		}

weed/worker/tasks/iceberg/delete_rewrite.go

@@ -40,13 +40,13 @@ type positionDeleteRow struct {
 func hasEligibleDeleteRewrite(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	manifests []iceberg.ManifestFile,
 	config Config,
 	meta table.Metadata,
 	predicate *partitionPredicate,
 ) (bool, error) {
-	groups, _, err := collectDeleteRewriteGroups(ctx, filerClient, bucketName, tablePath, manifests)
+	groups, _, err := collectDeleteRewriteGroups(ctx, filerClient, bucketName, dataPath, manifests)
 	if err != nil {
 		return false, err
 	}
@@ -74,7 +74,7 @@ func hasEligibleDeleteRewrite(
 func collectDeleteRewriteGroups(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	manifests []iceberg.ManifestFile,
 ) (map[string]*deleteRewriteGroup, []iceberg.ManifestEntry, error) {
 	groups := make(map[string]*deleteRewriteGroup)
@@ -85,7 +85,7 @@ func collectDeleteRewriteGroups(
 			continue
 		}
 
-		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, mf.FilePath())
+		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
 		if err != nil {
 			return nil, nil, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), err)
 		}
@@ -101,7 +101,7 @@ func collectDeleteRewriteGroups(
 
 			allPositionEntries = append(allPositionEntries, entry)
 
-			fileDeletes, err := readPositionDeleteFile(ctx, filerClient, bucketName, tablePath, entry.DataFile().FilePath())
+			fileDeletes, err := readPositionDeleteFile(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath())
 			if err != nil {
 				return nil, nil, fmt.Errorf("read position delete file %s: %w", entry.DataFile().FilePath(), err)
 			}
@@ -113,7 +113,13 @@ func collectDeleteRewriteGroups(
 			var referencedPath string
 			var positions []int64
 			for fp, pos := range fileDeletes {
-				referencedPath = normalizeIcebergPath(fp, bucketName, tablePath)
+				normalized, err := normalizeIcebergPath(fp, bucketName, dataPath)
+				if err != nil {
+					return nil, nil, fmt.Errorf("resolve deleted file %s: %w", fp, err)
+				}
+				// The rewritten delete file has to name the data file the way the
+				// table does, so keep the absolute form readers compare against.
+				referencedPath = absoluteIcebergPath(bucketName, normalized)
 				positions = append(positions, pos...)
 			}
 			sort.Slice(positions, func(i, j int) bool { return positions[i] < positions[j] })
@@ -263,17 +269,18 @@ func (h *Handler) rewritePositionDeleteFiles(
 	config Config,
 ) (string, map[string]int64, error) {
 	start := time.Now()
-	meta, metadataFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+	state, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
 	if err != nil {
 		return "", nil, fmt.Errorf("load metadata: %w", err)
 	}
+	meta, dataPath := state.Metadata, state.DataPath
 
 	currentSnap := meta.CurrentSnapshot()
 	if currentSnap == nil || currentSnap.ManifestList == "" {
 		return "no current snapshot", nil, nil
 	}
 
-	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, currentSnap.ManifestList)
+	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, currentSnap.ManifestList)
 	if err != nil {
 		return "", nil, fmt.Errorf("read manifest list: %w", err)
 	}
@@ -289,7 +296,7 @@ func (h *Handler) rewritePositionDeleteFiles(
 		case iceberg.ManifestContentData:
 			dataManifests = append(dataManifests, mf)
 		case iceberg.ManifestContentDeletes:
-			manifestData, readErr := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, mf.FilePath())
+			manifestData, readErr := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
 			if readErr != nil {
 				return "", nil, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), readErr)
 			}
@@ -305,7 +312,7 @@ func (h *Handler) rewritePositionDeleteFiles(
 		}
 	}
 
-	groupMap, allPositionEntries, err := collectDeleteRewriteGroups(ctx, filerClient, bucketName, tablePath, manifests)
+	groupMap, allPositionEntries, err := collectDeleteRewriteGroups(ctx, filerClient, bucketName, dataPath, manifests)
 	if err != nil {
 		return "", nil, err
 	}
@@ -355,8 +362,8 @@ func (h *Handler) rewritePositionDeleteFiles(
 	version := meta.Version()
 	snapshotID := currentSnap.SnapshotID
 	seqNum := currentSnap.SequenceNumber + 1
-	metaDir := path.Join(s3tables.TablesPath, bucketName, tablePath, "metadata")
-	dataDir := path.Join(s3tables.TablesPath, bucketName, tablePath, "data")
+	metaDir := path.Join(s3tables.TablesPath, bucketName, dataPath, "metadata")
+	dataDir := path.Join(s3tables.TablesPath, bucketName, dataPath, "data")
 	artifactSuffix := compactRandomSuffix()
 
 	replacedPaths := make(map[string]struct{})
@@ -440,7 +447,7 @@ func (h *Handler) rewritePositionDeleteFiles(
 			dfBuilder, err := iceberg.NewDataFileBuilder(
 				spec,
 				iceberg.EntryContentPosDeletes,
-				absoluteIcebergPath(bucketName, tablePath, "data", fileName),
+				absoluteIcebergPath(bucketName, dataPath, "data", fileName),
 				iceberg.ParquetFile,
 				group.Partition,
 				nil, nil,
@@ -512,7 +519,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 := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestName)
+		manifestPath := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestName)
 		mf, manifestBytes, err := writeManifestWithContent(
 			manifestPath,
 			version,
@@ -542,8 +549,8 @@ func (h *Handler) rewritePositionDeleteFiles(
 	}
 	writtenArtifacts = append(writtenArtifacts, artifact{dir: metaDir, fileName: manifestListName})
 
-	manifestListLocation := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestListName)
-	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
+	manifestListLocation := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestListName)
+	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, state.MetadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
 		cs := currentMeta.CurrentSnapshot()
 		if cs == nil || cs.SnapshotID != snapshotID {
 			return errStalePlan

weed/worker/tasks/iceberg/detection.go

@@ -120,14 +120,14 @@ func (h *Handler) scanTablesForMaintenance(
 					continue
 				}
 
-				icebergMeta, metadataFileName, planningIndex, err := parseTableMetadataEnvelope(metadataBytes)
+				tablePath := path.Join(nsName, tblName)
+				state, err := parseTableMetadataEnvelope(metadataBytes, bucketName, tablePath)
 				if err != nil {
 					glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot parse iceberg metadata: %v", bucketName, nsName, tblName, err)
 					continue
 				}
 
-				tablePath := path.Join(nsName, tblName)
-				needsWork, err := h.tableNeedsMaintenance(ctx, filerClient, bucketName, tablePath, icebergMeta, metadataFileName, planningIndex, config, ops)
+				needsWork, err := h.tableNeedsMaintenance(ctx, filerClient, bucketName, tablePath, state, config, ops)
 				if err != nil {
 					glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot evaluate maintenance need: %v", bucketName, nsName, tblName, err)
 					continue
@@ -138,8 +138,8 @@ func (h *Handler) scanTablesForMaintenance(
 						Namespace:        nsName,
 						TableName:        tblName,
 						TablePath:        tablePath,
-						MetadataFileName: metadataFileName,
-						Metadata:         icebergMeta,
+						MetadataFileName: state.MetadataFileName,
+						Metadata:         state.Metadata,
 					})
 					if limit > 0 && len(tables) > limit {
 						return tables, nil
@@ -167,13 +167,15 @@ func (h *Handler) tableNeedsMaintenance(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
 	bucketName, tablePath string,
-	meta table.Metadata,
-	metadataFileName string,
-	cachedPlanningIndex *planningIndex,
+	state *tableState,
 	config Config,
 	ops []string,
 ) (bool, error) {
 	config = normalizeDetectionConfig(config)
+	meta := state.Metadata
+	dataPath := state.DataPath
+	metadataFileName := state.MetadataFileName
+	cachedPlanningIndex := state.PlanningIndex
 
 	var predicate *partitionPredicate
 	if strings.TrimSpace(config.Where) != "" {
@@ -209,7 +211,7 @@ func (h *Handler) tableNeedsMaintenance(
 		if manifestsLoaded {
 			return currentManifests, manifestsErr
 		}
-		currentManifests, manifestsErr = loadCurrentManifests(ctx, filerClient, bucketName, tablePath, meta)
+		currentManifests, manifestsErr = loadCurrentManifests(ctx, filerClient, bucketName, dataPath, meta)
 		manifestsLoaded = true
 		return currentManifests, manifestsErr
 	}
@@ -228,7 +230,7 @@ func (h *Handler) tableNeedsMaintenance(
 			planningIndexErrs[op] = err
 			return nil, err
 		}
-		index, err := buildPlanningIndexFromManifests(ctx, filerClient, bucketName, tablePath, meta, config, []string{op}, manifests)
+		index, err := buildPlanningIndexFromManifests(ctx, filerClient, bucketName, dataPath, meta, config, []string{op}, manifests)
 		if err != nil {
 			planningIndexErrs[op] = err
 			return nil, err
@@ -286,7 +288,7 @@ func (h *Handler) tableNeedsMaintenance(
 				opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
 				continue
 			}
-			eligible, err := hasEligibleDeleteRewrite(ctx, filerClient, bucketName, tablePath, manifests, config, meta, predicate)
+			eligible, err := hasEligibleDeleteRewrite(ctx, filerClient, bucketName, dataPath, manifests, config, meta, predicate)
 			if err != nil {
 				opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
 				continue
@@ -308,14 +310,14 @@ func (h *Handler) tableNeedsMaintenance(
 			}
 		case "remove_orphans":
 			if metadataFileName == "" {
-				_, currentMetadataFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+				currentState, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
 				if err != nil {
 					opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
 					continue
 				}
-				metadataFileName = currentMetadataFileName
+				metadataFileName = currentState.MetadataFileName
 			}
-			orphanCandidates, err := collectOrphanCandidates(ctx, filerClient, bucketName, tablePath, meta, metadataFileName, config.OrphanOlderThanHours)
+			orphanCandidates, err := collectOrphanCandidates(ctx, filerClient, bucketName, dataPath, meta, metadataFileName, config.OrphanOlderThanHours)
 			if err != nil {
 				opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
 				continue
@@ -333,11 +335,15 @@ func (h *Handler) tableNeedsMaintenance(
 	return false, nil
 }
 
-func metadataFileNameFromLocation(location, bucketName, tablePath string) string {
+func metadataFileNameFromLocation(location string) string {
 	if location == "" {
 		return ""
 	}
-	return path.Base(normalizeIcebergPath(location, bucketName, tablePath))
+	name := path.Base(location)
+	if name == "." || name == "/" {
+		return ""
+	}
+	return name
 }
 
 func countDataManifests(manifests []iceberg.ManifestFile) int64 {
@@ -353,7 +359,7 @@ func countDataManifests(manifests []iceberg.ManifestFile) int64 {
 func loadCurrentManifests(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	meta table.Metadata,
 ) ([]iceberg.ManifestFile, error) {
 	currentSnap := meta.CurrentSnapshot()
@@ -361,7 +367,7 @@ func loadCurrentManifests(
 		return nil, nil
 	}
 
-	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, currentSnap.ManifestList)
+	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, currentSnap.ManifestList)
 	if err != nil {
 		return nil, fmt.Errorf("read manifest list: %w", err)
 	}
@@ -375,7 +381,7 @@ func loadCurrentManifests(
 func hasEligibleCompaction(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	manifests []iceberg.ManifestFile,
 	config Config,
 	meta table.Metadata,
@@ -408,7 +414,7 @@ func hasEligibleCompaction(
 
 	var allEntries []iceberg.ManifestEntry
 	for _, mf := range dataManifests {
-		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, mf.FilePath())
+		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
 		if err != nil {
 			return false, fmt.Errorf("read manifest %s: %w", mf.FilePath(), err)
 		}

weed/worker/tasks/iceberg/exec_test.go

@@ -259,13 +259,33 @@ type tableSetup struct {
 	BucketName string
 	Namespace  string
 	TableName  string
-	Snapshots  []table.Snapshot
+	// DataPath is the bucket-relative directory holding the table's files when
+	// it differs from the catalog path, as it does for tables an external REST
+	// client created at their own location (e.g. "ns/table-<uuid>"). Such a
+	// table records absolute s3:// URIs for every file it references.
+	DataPath  string
+	Snapshots []table.Snapshot
 }
 
 func (ts tableSetup) tablePath() string {
 	return path.Join(ts.Namespace, ts.TableName)
 }
 
+func (ts tableSetup) dataPath() string {
+	if ts.DataPath != "" {
+		return ts.DataPath
+	}
+	return ts.tablePath()
+}
+
+// fileRef builds the reference a table records for one of its files.
+func (ts tableSetup) fileRef(elem ...string) string {
+	if ts.DataPath == "" {
+		return path.Join(elem...)
+	}
+	return absoluteIcebergPath(ts.BucketName, append([]string{ts.DataPath}, elem...)...)
+}
+
 // populateTable creates the directory hierarchy and metadata entries in the
 // fake filer for a table, writes manifest files referenced by snapshots,
 // and returns the built metadata.
@@ -282,7 +302,7 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 	const metadataVersion = 1
 	internalMeta := map[string]interface{}{
 		"metadataVersion":  metadataVersion,
-		"metadataLocation": path.Join("metadata", fmt.Sprintf("v%d.metadata.json", metadataVersion)),
+		"metadataLocation": setup.fileRef("metadata", fmt.Sprintf("v%d.metadata.json", metadataVersion)),
 		"metadata": map[string]interface{}{
 			"fullMetadata": json.RawMessage(fullMetadataJSON),
 		},
@@ -295,7 +315,7 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 	bucketsPath := s3tables.TablesPath // "/buckets"
 	bucketPath := path.Join(bucketsPath, setup.BucketName)
 	nsPath := path.Join(bucketPath, setup.Namespace)
-	tableFilerPath := path.Join(nsPath, setup.TableName)
+	tableFilerPath := path.Join(bucketPath, setup.dataPath())
 
 	// Register bucket entry (marked as table bucket)
 	fs.putEntry(bucketsPath, setup.BucketName, &filer_pb.Entry{
@@ -340,7 +360,7 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 		dfBuilder, err := iceberg.NewDataFileBuilder(
 			spec,
 			iceberg.EntryContentData,
-			fmt.Sprintf("data/snap-%d-data.parquet", snap.SnapshotID),
+			setup.fileRef("data", fmt.Sprintf("snap-%d-data.parquet", snap.SnapshotID)),
 			iceberg.ParquetFile,
 			map[int]any{},
 			nil, nil,
@@ -360,7 +380,7 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 
 		// Write manifest
 		manifestFileName := fmt.Sprintf("manifest-%d.avro", snap.SnapshotID)
-		manifestPath := path.Join("metadata", manifestFileName)
+		manifestPath := setup.fileRef("metadata", manifestFileName)
 		var manifestBuf bytes.Buffer
 		mf, err := iceberg.WriteManifest(manifestPath, &manifestBuf, version, spec, schema, snap.SnapshotID, []iceberg.ManifestEntry{entry})
 		if err != nil {
@@ -570,6 +590,165 @@ func TestExpireSnapshotsNothingToExpire(t *testing.T) {
 	}
 }
 
+// A table created through the Iceberg REST catalog can live outside its
+// catalog path, and every location it records is then an absolute s3:// URI
+// under that other directory. Maintenance must follow the recorded location
+// instead of joining the URI onto the catalog path.
+func TestMaintenanceOnExternalTableLocation(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	const dataPath = "source/events-0cd81bca-b389-4dda-8dae-3502ce680501"
+	snapAt := func(id int64) string {
+		return fmt.Sprintf("s3://lake/%s/metadata/snap-%d.avro", dataPath, id)
+	}
+	now := time.Now().Add(-10 * time.Second).UnixMilli()
+	setup := tableSetup{
+		BucketName: "lake",
+		Namespace:  "source",
+		TableName:  "events",
+		DataPath:   dataPath,
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: snapAt(1)},
+			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: snapAt(2)},
+			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: snapAt(3)},
+		},
+	}
+	populateTable(t, fs, setup)
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionHours: 0,
+		MaxSnapshotsToKeep:     1,
+		MaxCommitRetries:       3,
+		Operations:             "expire_snapshots",
+	}
+
+	result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
+	if err != nil {
+		t.Fatalf("expireSnapshots: %v", err)
+	}
+	if !strings.Contains(result, "expired 2 snapshot(s)") {
+		t.Errorf("expected two snapshots expired, got %q", result)
+	}
+
+	// The expired snapshots' files are gone from the table's own directory.
+	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, dataPath, "metadata")
+	for _, name := range []string{"snap-1.avro", "snap-2.avro"} {
+		if fs.getEntry(metaDir, name) != nil {
+			t.Errorf("expected %s/%s to be deleted", metaDir, name)
+		}
+	}
+	if fs.getEntry(metaDir, "snap-3.avro") == nil {
+		t.Errorf("expected %s/snap-3.avro to be kept", metaDir)
+	}
+
+	// The new metadata file is written next to the table's files, and the
+	// catalog now points at it there.
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("loadCurrentMetadata: %v", err)
+	}
+	if fs.getEntry(metaDir, state.MetadataFileName) == nil {
+		t.Errorf("expected new metadata file %s in %s", state.MetadataFileName, metaDir)
+	}
+	if got := len(state.Metadata.Snapshots()); got != 1 {
+		t.Errorf("expected 1 snapshot left, got %d", got)
+	}
+}
+
+// Compaction of such a table must keep its output with the rest of the table
+// instead of splitting it across the catalog path and the table's location.
+func TestCompactDataFilesOnExternalTableLocation(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	setup := tableSetup{
+		BucketName: "lake",
+		Namespace:  "source",
+		TableName:  "events",
+		DataPath:   "source/events-0cd81bca-b389-4dda-8dae-3502ce680501",
+	}
+	populateTableWithDeleteFiles(t, fs, setup,
+		[]struct {
+			Name string
+			Rows []struct {
+				ID   int64
+				Name string
+			}
+		}{
+			{"d1.parquet", []struct {
+				ID   int64
+				Name string
+			}{{1, "a"}, {2, "b"}}},
+			{"d2.parquet", []struct {
+				ID   int64
+				Name string
+			}{{3, "c"}}},
+		},
+		nil, nil,
+	)
+
+	handler := NewHandler(nil)
+	config := Config{
+		TargetFileSizeBytes: 256 * 1024 * 1024,
+		MinInputFiles:       2,
+		MaxCommitRetries:    3,
+		ApplyDeletes:        true,
+	}
+
+	result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
+	if err != nil {
+		t.Fatalf("compactDataFiles: %v", err)
+	}
+	if !strings.Contains(result, "compacted 2 files into 1") {
+		t.Fatalf("unexpected result: %q", result)
+	}
+
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("loadCurrentMetadata: %v", err)
+	}
+	if state.DataPath != setup.DataPath {
+		t.Fatalf("data path = %q, want %q", state.DataPath, setup.DataPath)
+	}
+
+	// Everything the new snapshot names lives under the table's own location.
+	wantPrefix := "s3://" + setup.BucketName + "/" + setup.DataPath + "/"
+	newSnap := state.Metadata.CurrentSnapshot()
+	if newSnap == nil || !strings.HasPrefix(newSnap.ManifestList, wantPrefix+"metadata/snap-") {
+		t.Fatalf("manifest list %q should be under %q", newSnap.ManifestList, wantPrefix)
+	}
+	manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata)
+	if err != nil {
+		t.Fatalf("loadCurrentManifests: %v", err)
+	}
+	for _, mf := range manifests {
+		if !strings.HasPrefix(mf.FilePath(), wantPrefix+"metadata/") {
+			t.Errorf("manifest %q should be under %q", mf.FilePath(), wantPrefix)
+		}
+		manifestData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, state.DataPath, mf.FilePath())
+		if err != nil {
+			t.Fatalf("load manifest: %v", err)
+		}
+		entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true)
+		if err != nil {
+			t.Fatalf("read manifest: %v", err)
+		}
+		for _, entry := range entries {
+			if !strings.HasPrefix(entry.DataFile().FilePath(), wantPrefix+"data/") {
+				t.Errorf("data file %q should be under %q", entry.DataFile().FilePath(), wantPrefix)
+			}
+		}
+	}
+
+	// Nothing was written to the catalog path.
+	catalogDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath())
+	for _, subdir := range []string{"metadata", "data"} {
+		if entries := fs.listDir(path.Join(catalogDir, subdir)); len(entries) > 0 {
+			t.Errorf("expected no files under %s/%s, got %d", catalogDir, subdir, len(entries))
+		}
+	}
+}
+
 func TestRemoveOrphansExecution(t *testing.T) {
 	fs, client := startFakeFiler(t)
 
@@ -770,16 +949,16 @@ func TestRewriteManifestsExecution(t *testing.T) {
 	// 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())
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
 	if err != nil {
 		t.Fatalf("reload metadata: %v", err)
 	}
-	newSnap := newMeta.CurrentSnapshot()
+	newSnap := state.Metadata.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() {
+	for mle := range state.Metadata.PreviousFiles() {
 		if mle.MetadataFile == wantPrefix+"v1.metadata.json" {
 			foundPreviousEntry = true
 		}
@@ -1532,7 +1711,7 @@ func TestTableNeedsMaintenanceCachesPlanningIndexBuildError(t *testing.T) {
 		t.Fatalf("parseOperations: %v", err)
 	}
 
-	needsWork, err := handler.tableNeedsMaintenance(context.Background(), client, setup.BucketName, setup.tablePath(), meta, "v1.metadata.json", nil, config, ops)
+	needsWork, err := handler.tableNeedsMaintenance(context.Background(), client, setup.BucketName, setup.tablePath(), &tableState{Metadata: meta, MetadataFileName: "v1.metadata.json", DataPath: setup.dataPath()}, config, ops)
 	if err == nil {
 		t.Fatal("expected planning-index build error")
 	}
@@ -1588,7 +1767,7 @@ func TestTableNeedsMaintenanceScopesPlanningIndexBuildErrorsPerOperation(t *test
 		t.Fatalf("parseOperations: %v", err)
 	}
 
-	needsWork, err := handler.tableNeedsMaintenance(context.Background(), client, setup.BucketName, setup.tablePath(), meta, "v1.metadata.json", nil, config, ops)
+	needsWork, err := handler.tableNeedsMaintenance(context.Background(), client, setup.BucketName, setup.tablePath(), &tableState{Metadata: meta, MetadataFileName: "v1.metadata.json", DataPath: setup.dataPath()}, config, ops)
 	if err != nil {
 		t.Fatalf("expected rewrite_manifests planning to survive compaction planning error, got %v", err)
 	}
@@ -2111,7 +2290,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 	schema := newTestSchema()
 	spec := *iceberg.UnpartitionedSpec
 
-	meta, err := table.NewMetadata(schema, &spec, sortOrder, "s3://"+setup.BucketName+"/"+setup.tablePath(), nil)
+	meta, err := table.NewMetadata(schema, &spec, sortOrder, "s3://"+setup.BucketName+"/"+setup.dataPath(), nil)
 	if err != nil {
 		t.Fatalf("create metadata: %v", err)
 	}
@@ -2119,7 +2298,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 	bucketsPath := s3tables.TablesPath
 	bucketPath := path.Join(bucketsPath, setup.BucketName)
 	nsPath := path.Join(bucketPath, setup.Namespace)
-	tableFilerPath := path.Join(nsPath, setup.TableName)
+	tableFilerPath := path.Join(bucketPath, setup.dataPath())
 	metaDir := path.Join(tableFilerPath, "metadata")
 	dataDir := path.Join(tableFilerPath, "data")
 
@@ -2129,7 +2308,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 	var dataManifestEntries []iceberg.ManifestEntry
 	for _, df := range dataFiles {
 		data := writeTestParquetFile(t, fs, dataDir, df.Name, df.Rows)
-		dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, "data/"+df.Name, iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(df.Rows)), int64(len(data)))
+		dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, setup.fileRef("data", df.Name), iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(df.Rows)), int64(len(data)))
 		if err != nil {
 			t.Fatalf("build data file %s: %v", df.Name, err)
 		}
@@ -2140,7 +2319,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 	// Write data manifest
 	var dataManifestBuf bytes.Buffer
 	dataManifestName := "data-manifest-1.avro"
-	dataMf, err := iceberg.WriteManifest(path.Join("metadata", dataManifestName), &dataManifestBuf, version, spec, schema, 1, dataManifestEntries)
+	dataMf, err := iceberg.WriteManifest(setup.fileRef("metadata", dataManifestName), &dataManifestBuf, version, spec, schema, 1, dataManifestEntries)
 	if err != nil {
 		t.Fatalf("write data manifest: %v", err)
 	}
@@ -2173,7 +2352,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 				Name: pdf.Name, Content: buf.Bytes(),
 				Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(buf.Len())},
 			})
-			dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes, "data/"+pdf.Name, iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(pdf.Rows)), int64(buf.Len()))
+			dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes, setup.fileRef("data", pdf.Name), iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(pdf.Rows)), int64(buf.Len()))
 			if err != nil {
 				t.Fatalf("build pos delete file: %v", err)
 			}
@@ -2185,7 +2364,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 		// metadata to "deletes" and build a ManifestFile with the right content type.
 		var posManifestBuf bytes.Buffer
 		posManifestName := "pos-delete-manifest-1.avro"
-		posManifestPath := path.Join("metadata", posManifestName)
+		posManifestPath := setup.fileRef("metadata", posManifestName)
 		_, err := iceberg.WriteManifest(posManifestPath, &posManifestBuf, version, spec, schema, 1, posDeleteEntries)
 		if err != nil {
 			t.Fatalf("write pos delete manifest: %v", err)
@@ -2225,7 +2404,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 				Name: edf.Name, Content: buf.Bytes(),
 				Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(buf.Len())},
 			})
-			dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentEqDeletes, "data/"+edf.Name, iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(edf.Rows)), int64(buf.Len()))
+			dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentEqDeletes, setup.fileRef("data", edf.Name), iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(edf.Rows)), int64(buf.Len()))
 			if err != nil {
 				t.Fatalf("build eq delete file: %v", err)
 			}
@@ -2236,7 +2415,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 
 		var eqManifestBuf bytes.Buffer
 		eqManifestName := "eq-delete-manifest-1.avro"
-		eqManifestPath := path.Join("metadata", eqManifestName)
+		eqManifestPath := setup.fileRef("metadata", eqManifestName)
 		_, err := iceberg.WriteManifest(eqManifestPath, &eqManifestBuf, version, spec, schema, 1, eqDeleteEntries)
 		if err != nil {
 			t.Fatalf("write eq delete manifest: %v", err)
@@ -2267,8 +2446,8 @@ func populateTableWithDeleteFilesAndSortOrder(
 
 	// Build final metadata with snapshot
 	now := time.Now().UnixMilli()
-	snap := table.Snapshot{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"}
-	builder, err := table.MetadataBuilderFromBase(meta, "s3://"+setup.BucketName+"/"+setup.tablePath())
+	snap := table.Snapshot{SnapshotID: 1, TimestampMs: now, ManifestList: setup.fileRef("metadata", "snap-1.avro")}
+	builder, err := table.MetadataBuilderFromBase(meta, "s3://"+setup.BucketName+"/"+setup.dataPath())
 	if err != nil {
 		t.Fatalf("create metadata builder: %v", err)
 	}
@@ -2286,8 +2465,9 @@ func populateTableWithDeleteFilesAndSortOrder(
 	// Register table structure
 	fullMetadataJSON, _ := json.Marshal(meta)
 	internalMeta := map[string]interface{}{
-		"metadataVersion": 1,
-		"metadata":        map[string]interface{}{"fullMetadata": json.RawMessage(fullMetadataJSON)},
+		"metadataVersion":  1,
+		"metadataLocation": setup.fileRef("metadata", "v1.metadata.json"),
+		"metadata":         map[string]interface{}{"fullMetadata": json.RawMessage(fullMetadataJSON)},
 	}
 	xattr, _ := json.Marshal(internalMeta)
 
@@ -2311,11 +2491,11 @@ func loadLiveDeleteFilePaths(
 ) (posPaths, eqPaths []string) {
 	t.Helper()
 
-	meta, _, err := loadCurrentMetadata(context.Background(), client, bucketName, tablePath)
+	state, err := loadCurrentMetadata(context.Background(), client, bucketName, tablePath)
 	if err != nil {
 		t.Fatalf("loadCurrentMetadata: %v", err)
 	}
-	manifests, err := loadCurrentManifests(context.Background(), client, bucketName, tablePath, meta)
+	manifests, err := loadCurrentManifests(context.Background(), client, bucketName, state.DataPath, state.Metadata)
 	if err != nil {
 		t.Fatalf("loadCurrentManifests: %v", err)
 	}
@@ -2355,11 +2535,11 @@ func rewriteDeleteManifestsAsMixed(
 ) {
 	t.Helper()
 
-	meta, _, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
 	if err != nil {
 		t.Fatalf("loadCurrentMetadata: %v", err)
 	}
-	manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, setup.tablePath(), meta)
+	manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata)
 	if err != nil {
 		t.Fatalf("loadCurrentManifests: %v", err)
 	}
@@ -2385,13 +2565,13 @@ func rewriteDeleteManifestsAsMixed(
 	}
 
 	spec := *iceberg.UnpartitionedSpec
-	version := meta.Version()
+	version := state.Metadata.Version()
 	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
 	manifestName := "mixed-delete-manifest-1.avro"
 	manifestPath := path.Join("metadata", manifestName)
 
 	var manifestBuf bytes.Buffer
-	_, err = iceberg.WriteManifest(manifestPath, &manifestBuf, version, spec, meta.CurrentSchema(), 1, deleteEntries)
+	_, err = iceberg.WriteManifest(manifestPath, &manifestBuf, version, spec, state.Metadata.CurrentSchema(), 1, deleteEntries)
 	if err != nil {
 		t.Fatalf("write mixed delete manifest: %v", err)
 	}
@@ -3204,11 +3384,11 @@ func TestDetectSplitsSortCompactionBinsByCap(t *testing.T) {
 		sortOrder,
 	)
 
-	meta, _, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
 	if err != nil {
 		t.Fatalf("loadCurrentMetadata: %v", err)
 	}
-	manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, setup.tablePath(), meta)
+	manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata)
 	if err != nil {
 		t.Fatalf("loadCurrentManifests: %v", err)
 	}

weed/worker/tasks/iceberg/filer_io.go

@@ -13,7 +13,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/apache/iceberg-go/table"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -124,7 +123,7 @@ func walkFilerEntries(ctx context.Context, client filer_pb.SeaweedFilerClient, d
 }
 
 // loadCurrentMetadata loads and parses the current Iceberg metadata from the table entry's xattr.
-func loadCurrentMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, tablePath string) (table.Metadata, string, error) {
+func loadCurrentMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, tablePath string) (*tableState, error) {
 	dir := path.Join(s3tables.TablesPath, bucketName, path.Dir(tablePath))
 	name := path.Base(tablePath)
 
@@ -133,63 +132,28 @@ func loadCurrentMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient
 		Name:      name,
 	})
 	if err != nil {
-		return nil, "", fmt.Errorf("lookup table entry %s/%s: %w", dir, name, err)
+		return nil, fmt.Errorf("lookup table entry %s/%s: %w", dir, name, err)
 	}
 	if resp == nil || resp.Entry == nil {
-		return nil, "", fmt.Errorf("table entry not found: %s/%s", dir, name)
+		return nil, fmt.Errorf("table entry not found: %s/%s", dir, name)
 	}
 
 	metadataBytes, ok := resp.Entry.Extended[s3tables.ExtendedKeyMetadata]
 	if !ok || len(metadataBytes) == 0 {
-		return nil, "", fmt.Errorf("no metadata xattr on table entry %s/%s", dir, name)
+		return nil, fmt.Errorf("no metadata xattr on table entry %s/%s", dir, name)
 	}
 
-	// Parse internal metadata to extract FullMetadata
-	var internalMeta struct {
-		MetadataVersion  int    `json:"metadataVersion"`
-		MetadataLocation string `json:"metadataLocation,omitempty"`
-		Metadata         *struct {
-			FullMetadata json.RawMessage `json:"fullMetadata,omitempty"`
-		} `json:"metadata,omitempty"`
-	}
-	if err := json.Unmarshal(metadataBytes, &internalMeta); err != nil {
-		return nil, "", fmt.Errorf("unmarshal internal metadata: %w", err)
-	}
-	if internalMeta.Metadata == nil || len(internalMeta.Metadata.FullMetadata) == 0 {
-		return nil, "", fmt.Errorf("no fullMetadata in table xattr")
-	}
-
-	meta, err := table.ParseMetadataBytes(internalMeta.Metadata.FullMetadata)
-	if err != nil {
-		return nil, "", fmt.Errorf("parse iceberg metadata: %w", err)
-	}
-
-	// Use metadataLocation from xattr if available (includes nonce suffix),
-	// otherwise fall back to the canonical name derived from metadataVersion.
-	metadataFileName := path.Base(internalMeta.MetadataLocation)
-	if metadataFileName == "" || metadataFileName == "." {
-		metadataFileName = fmt.Sprintf("v%d.metadata.json", internalMeta.MetadataVersion)
-	}
-	return meta, metadataFileName, nil
+	return parseTableMetadataEnvelope(metadataBytes, bucketName, tablePath)
 }
 
 // loadFileByIcebergPath loads a file from the filer given an Iceberg-style path.
-// Paths may be absolute filer paths, relative (metadata/..., data/...), or
-// location-based (s3://bucket/ns/table/metadata/...).
-//
-// The function normalises the path to a relative form under the table root
-// (e.g. "metadata/snap-1.avro" or "data/region=us/file.parquet") and splits
-// it into the correct filer directory + entry name, so nested sub-directories
-// are resolved properly.
-func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, tablePath, icebergPath string) ([]byte, error) {
-	relPath := path.Clean(normalizeIcebergPath(icebergPath, bucketName, tablePath))
-	relPath = strings.TrimPrefix(relPath, "/")
-	if relPath == "." || relPath == "" || strings.HasPrefix(relPath, "../") {
-		return nil, fmt.Errorf("invalid iceberg path %q", icebergPath)
+// dataPath is the bucket-relative directory the table's files live in.
+func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) ([]byte, error) {
+	fullPath, err := icebergFilerPath(bucketName, dataPath, icebergPath)
+	if err != nil {
+		return nil, err
 	}
-
-	dir := path.Join(s3tables.TablesPath, bucketName, tablePath, path.Dir(relPath))
-	fileName := path.Base(relPath)
+	dir, fileName := path.Dir(fullPath), path.Base(fullPath)
 
 	resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
 		Directory: dir,
@@ -218,47 +182,74 @@ func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClie
 	return data, nil
 }
 
-// normalizeIcebergPath converts an Iceberg path (which may be an S3 URL, an
-// absolute filer path, or a plain relative path) into a relative path under the
-// table root, e.g. "metadata/snap-1.avro" or "data/region=us/file.parquet".
-func normalizeIcebergPath(icebergPath, bucketName, tablePath string) string {
+// normalizeIcebergPath converts an Iceberg path (an S3 URL, an absolute filer
+// path, or a path relative to the table root) into a path relative to the
+// bucket root, e.g. "ns/table/metadata/snap-1.avro". Absolute references are
+// resolved from the bucket root rather than from dataPath: an Iceberg table
+// records where its files actually are, and a client writing through the REST
+// catalog may place them outside the catalog path. The bucket-relative form is
+// the canonical key for comparing references that mix both spellings.
+func normalizeIcebergPath(icebergPath, bucketName, dataPath string) (string, error) {
 	p := icebergPath
-
-	// Strip scheme (e.g. "s3://bucket/ns/table/metadata/file" → "bucket/ns/table/metadata/file")
+	absolute := false
 	if idx := strings.Index(p, "://"); idx >= 0 {
 		p = p[idx+3:]
+		absolute = true
+	} else if strings.HasPrefix(p, "/") {
+		absolute = true
 	}
-
-	// Strip any leading slash
 	p = strings.TrimPrefix(p, "/")
 
-	// Strip bucket+tablePath prefix if present
-	// e.g. "mybucket/ns/table/metadata/file" → "metadata/file"
-	tablePrefix := path.Join(bucketName, tablePath) + "/"
-	if strings.HasPrefix(p, tablePrefix) {
-		return p[len(tablePrefix):]
+	if absolute {
+		rest, ok := cutPathPrefix(p, bucketName)
+		if !ok {
+			// Filer-style references carry the /buckets prefix ahead of the bucket.
+			if withoutTablesPath, hasTablesPath := cutPathPrefix(p, strings.TrimPrefix(s3tables.TablesPath, "/")); hasTablesPath {
+				rest, ok = cutPathPrefix(withoutTablesPath, bucketName)
+			}
+		}
+		if !ok {
+			return "", fmt.Errorf("iceberg path %q is outside bucket %q", icebergPath, bucketName)
+		}
+		p = rest
+	} else {
+		if clean := path.Clean(p); clean == ".." || strings.HasPrefix(clean, "../") {
+			return "", fmt.Errorf("invalid iceberg path %q", icebergPath)
+		}
+		p = path.Join(dataPath, p)
 	}
 
-	// Strip filer TablesPath prefix if present
-	// e.g. "buckets/mybucket/ns/table/metadata/file" → "metadata/file"
-	filerPrefix := strings.TrimPrefix(s3tables.TablesPath, "/")
-	fullPrefix := path.Join(filerPrefix, bucketName, tablePath) + "/"
-	if strings.HasPrefix(p, fullPrefix) {
-		return p[len(fullPrefix):]
+	p = path.Clean(p)
+	if p == "." || p == ".." || strings.HasPrefix(p, "../") {
+		return "", fmt.Errorf("invalid iceberg path %q", icebergPath)
 	}
+	return p, nil
+}
 
-	// Already relative (e.g. "metadata/snap-1.avro")
-	return p
+// cutPathPrefix removes a leading path prefix at a path boundary.
+func cutPathPrefix(p, segment string) (string, bool) {
+	if segment == "" {
+		return p, false
+	}
+	rest, ok := strings.CutPrefix(p, segment+"/")
+	return rest, ok
+}
+
+// icebergFilerPath resolves an Iceberg path to the filer path holding the file.
+func icebergFilerPath(bucketName, dataPath, icebergPath string) (string, error) {
+	relPath, err := normalizeIcebergPath(icebergPath, bucketName, dataPath)
+	if err != nil {
+		return "", err
+	}
+	return path.Join(s3tables.TablesPath, bucketName, relPath), nil
 }
 
 // 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...)...)
+// absolute s3:// URI for a bucket-relative path. The Iceberg spec requires
+// absolute locations in metadata — strict readers (Spark/Trino via S3FileIO)
+// reject paths with no scheme.
+func absoluteIcebergPath(bucketName string, elem ...string) string {
+	return "s3://" + path.Join(append([]string{bucketName}, elem...)...)
 }
 
 // saveFilerFile saves a file to the filer.

weed/worker/tasks/iceberg/handler_test.go

@@ -445,17 +445,24 @@ func TestManifestRewriteNestedPathConsistency(t *testing.T) {
 				t.Errorf("FilePath() = %q, want %q", mf.FilePath(), tc.manifestPath)
 			}
 
-			// normalizeIcebergPath should return the path unchanged when already relative
-			normalized := normalizeIcebergPath(tc.manifestPath, "bucket", "ns/table")
-			if normalized != tc.manifestPath {
-				t.Errorf("normalizeIcebergPath(%q) = %q, want %q", tc.manifestPath, normalized, tc.manifestPath)
+			// A relative path resolves under the table's own directory
+			want := "ns/table/" + tc.manifestPath
+			normalized, err := normalizeIcebergPath(tc.manifestPath, "bucket", "ns/table")
+			if err != nil {
+				t.Fatalf("normalizeIcebergPath(%q): %v", tc.manifestPath, err)
+			}
+			if normalized != want {
+				t.Errorf("normalizeIcebergPath(%q) = %q, want %q", tc.manifestPath, normalized, want)
 			}
 
-			// Verify normalization strips S3 scheme prefix correctly
+			// The S3 URL for the same file resolves to the same place
 			s3Path := "s3://bucket/ns/table/" + tc.manifestPath
-			normalized = normalizeIcebergPath(s3Path, "bucket", "ns/table")
-			if normalized != tc.manifestPath {
-				t.Errorf("normalizeIcebergPath(%q) = %q, want %q", s3Path, normalized, tc.manifestPath)
+			normalized, err = normalizeIcebergPath(s3Path, "bucket", "ns/table")
+			if err != nil {
+				t.Fatalf("normalizeIcebergPath(%q): %v", s3Path, err)
+			}
+			if normalized != want {
+				t.Errorf("normalizeIcebergPath(%q) = %q, want %q", s3Path, normalized, want)
 			}
 		})
 	}
@@ -466,53 +473,133 @@ func TestNormalizeIcebergPath(t *testing.T) {
 		name        string
 		icebergPath string
 		bucket      string
-		tablePath   string
+		dataPath    string
 		expected    string
+		wantErr     bool
 	}{
 		{
-			"relative metadata path",
-			"metadata/snap-1.avro",
-			"mybucket", "ns/table",
-			"metadata/snap-1.avro",
+			name:        "relative metadata path",
+			icebergPath: "metadata/snap-1.avro",
+			bucket:      "mybucket", dataPath: "ns/table",
+			expected: "ns/table/metadata/snap-1.avro",
 		},
 		{
-			"relative data path",
-			"data/file.parquet",
-			"mybucket", "ns/table",
-			"data/file.parquet",
+			name:        "relative data path",
+			icebergPath: "data/file.parquet",
+			bucket:      "mybucket", dataPath: "ns/table",
+			expected: "ns/table/data/file.parquet",
 		},
 		{
-			"S3 URL",
-			"s3://mybucket/ns/table/metadata/snap-1.avro",
-			"mybucket", "ns/table",
-			"metadata/snap-1.avro",
+			name:        "S3 URL",
+			icebergPath: "s3://mybucket/ns/table/metadata/snap-1.avro",
+			bucket:      "mybucket", dataPath: "ns/table",
+			expected: "ns/table/metadata/snap-1.avro",
 		},
 		{
-			"absolute filer path",
-			"/buckets/mybucket/ns/table/data/file.parquet",
-			"mybucket", "ns/table",
-			"data/file.parquet",
+			name:        "absolute filer path",
+			icebergPath: "/buckets/mybucket/ns/table/data/file.parquet",
+			bucket:      "mybucket", dataPath: "ns/table",
+			expected: "ns/table/data/file.parquet",
 		},
 		{
-			"nested data path",
-			"data/region=us/city=sf/file.parquet",
-			"mybucket", "ns/table",
-			"data/region=us/city=sf/file.parquet",
+			name:        "nested data path",
+			icebergPath: "data/region=us/city=sf/file.parquet",
+			bucket:      "mybucket", dataPath: "ns/table",
+			expected: "ns/table/data/region=us/city=sf/file.parquet",
 		},
 		{
-			"S3 URL nested",
-			"s3://mybucket/ns/table/data/region=us/file.parquet",
-			"mybucket", "ns/table",
-			"data/region=us/file.parquet",
+			name:        "S3 URL nested",
+			icebergPath: "s3://mybucket/ns/table/data/region=us/file.parquet",
+			bucket:      "mybucket", dataPath: "ns/table",
+			expected: "ns/table/data/region=us/file.parquet",
+		},
+		{
+			// Written by a client that created the table at its own location.
+			name:        "S3 URL outside the catalog path",
+			icebergPath: "s3://mybucket/ns/table-9f1c/metadata/snap-1.avro",
+			bucket:      "mybucket", dataPath: "ns/table-9f1c",
+			expected: "ns/table-9f1c/metadata/snap-1.avro",
+		},
+		{
+			// Same file, referenced while the worker believes the table sits at
+			// its catalog path: the URI still wins.
+			name:        "S3 URL wins over a stale data path",
+			icebergPath: "s3://mybucket/ns/table-9f1c/metadata/snap-1.avro",
+			bucket:      "mybucket", dataPath: "ns/table",
+			expected: "ns/table-9f1c/metadata/snap-1.avro",
+		},
+		{
+			name:        "other bucket",
+			icebergPath: "s3://otherbucket/ns/table/metadata/snap-1.avro",
+			bucket:      "mybucket", dataPath: "ns/table",
+			wantErr: true,
+		},
+		{
+			name:        "traversal out of the bucket",
+			icebergPath: "../../etc/passwd",
+			bucket:      "mybucket", dataPath: "ns/table",
+			wantErr: true,
 		},
 	}
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			result := normalizeIcebergPath(tc.icebergPath, tc.bucket, tc.tablePath)
+			result, err := normalizeIcebergPath(tc.icebergPath, tc.bucket, tc.dataPath)
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("normalizeIcebergPath(%q, %q, %q) = %q, want error",
+						tc.icebergPath, tc.bucket, tc.dataPath, result)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("normalizeIcebergPath(%q, %q, %q): %v", tc.icebergPath, tc.bucket, tc.dataPath, err)
+			}
 			if result != tc.expected {
 				t.Errorf("normalizeIcebergPath(%q, %q, %q) = %q, want %q",
-					tc.icebergPath, tc.bucket, tc.tablePath, result, tc.expected)
+					tc.icebergPath, tc.bucket, tc.dataPath, result, tc.expected)
+			}
+		})
+	}
+}
+
+func TestTableDataPath(t *testing.T) {
+	tests := []struct {
+		name             string
+		metadataLocation string
+		expected         string
+	}{
+		{
+			name:             "catalog path",
+			metadataLocation: "s3://mybucket/ns/table/metadata/v1.metadata.json",
+			expected:         "ns/table",
+		},
+		{
+			name:             "location outside the catalog path",
+			metadataLocation: "s3://mybucket/ns/table-9f1c/metadata/v3-171.metadata.json",
+			expected:         "ns/table-9f1c",
+		},
+		{
+			name:             "relative location",
+			metadataLocation: "metadata/v1.metadata.json",
+			expected:         "ns/table",
+		},
+		{
+			name:             "missing location",
+			metadataLocation: "",
+			expected:         "ns/table",
+		},
+		{
+			name:             "location in another bucket",
+			metadataLocation: "s3://otherbucket/ns/table/metadata/v1.metadata.json",
+			expected:         "ns/table",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := tableDataPath("mybucket", "ns/table", tc.metadataLocation); got != tc.expected {
+				t.Errorf("tableDataPath(%q) = %q, want %q", tc.metadataLocation, got, tc.expected)
 			}
 		})
 	}
@@ -972,16 +1059,16 @@ func TestCollectPositionDeletes(t *testing.T) {
 		t.Fatalf("collectPositionDeletes: %v", err)
 	}
 
-	// Verify results
-	if len(result["data/file1.parquet"]) != 3 {
-		t.Errorf("expected 3 positions for file1, got %d", len(result["data/file1.parquet"]))
+	// Deleted files are keyed by their bucket-relative path
+	if len(result["ns/tbl/data/file1.parquet"]) != 3 {
+		t.Errorf("expected 3 positions for file1, got %d", len(result["ns/tbl/data/file1.parquet"]))
 	}
-	if len(result["data/file2.parquet"]) != 1 {
-		t.Errorf("expected 1 position for file2, got %d", len(result["data/file2.parquet"]))
+	if len(result["ns/tbl/data/file2.parquet"]) != 1 {
+		t.Errorf("expected 1 position for file2, got %d", len(result["ns/tbl/data/file2.parquet"]))
 	}
 
 	// Verify sorted
-	positions := result["data/file1.parquet"]
+	positions := result["ns/tbl/data/file1.parquet"]
 	for i := 1; i < len(positions); i++ {
 		if positions[i] <= positions[i-1] {
 			t.Errorf("positions not sorted: %v", positions)
@@ -1107,7 +1194,7 @@ func TestMergeParquetFilesWithPositionDeletes(t *testing.T) {
 
 	// Delete rows 1 (bob) and 3 (dave) from file1
 	posDeletes := map[string][]int64{
-		"data/file1.parquet": {1, 3},
+		"ns/tbl/data/file1.parquet": {1, 3},
 	}
 
 	merged, count, err := mergeParquetFiles(

weed/worker/tasks/iceberg/operations.go

@@ -9,7 +9,6 @@ import (
 	"math/rand/v2"
 	"path"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/apache/iceberg-go"
@@ -41,10 +40,11 @@ func (h *Handler) expireSnapshots(
 ) (string, map[string]int64, error) {
 	start := time.Now()
 	// Load current metadata
-	meta, metadataFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+	state, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
 	if err != nil {
 		return "", nil, fmt.Errorf("load metadata: %w", err)
 	}
+	meta, dataPath := state.Metadata, state.DataPath
 
 	snapshots := meta.Snapshots()
 	if len(snapshots) == 0 {
@@ -112,23 +112,25 @@ func (h *Handler) expireSnapshots(
 
 	// Collect all files referenced by each set before modifying metadata.
 	// This lets us determine which files become unreferenced.
-	expiredFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, expiredSnaps)
+	expiredFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, dataPath, expiredSnaps)
 	if err != nil {
 		return "", nil, fmt.Errorf("collect expired snapshot files: %w", err)
 	}
-	keptFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, keptSnaps)
+	keptFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, dataPath, keptSnaps)
 	if err != nil {
 		return "", nil, fmt.Errorf("collect kept snapshot files: %w", err)
 	}
 
-	// Normalize kept file paths for consistent comparison
-	normalizedKept := make(map[string]struct{}, len(keptFiles))
+	// Resolve kept file paths for consistent comparison
+	keptFilerPaths := make(map[string]struct{}, len(keptFiles))
 	for f := range keptFiles {
-		normalizedKept[normalizeIcebergPath(f, bucketName, tablePath)] = struct{}{}
+		if filerPath, err := icebergFilerPath(bucketName, dataPath, f); err == nil {
+			keptFilerPaths[filerPath] = struct{}{}
+		}
 	}
 
 	// Use MetadataBuilder to remove snapshots and create new metadata
-	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
+	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, state.MetadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
 		// Guard: verify table head hasn't changed since we planned
 		cs := currentMeta.CurrentSnapshot()
 		if (cs == nil) != (currentSnapID == 0) || (cs != nil && cs.SnapshotID != currentSnapID) {
@@ -141,16 +143,17 @@ func (h *Handler) expireSnapshots(
 	}
 
 	// Delete files exclusively referenced by expired snapshots (best-effort)
-	tableBasePath := path.Join(s3tables.TablesPath, bucketName, tablePath)
 	deletedCount := 0
 	for filePath := range expiredFiles {
-		normalized := normalizeIcebergPath(filePath, bucketName, tablePath)
-		if _, stillReferenced := normalizedKept[normalized]; stillReferenced {
+		resolved, err := icebergFilerPath(bucketName, dataPath, filePath)
+		if err != nil {
+			glog.Warningf("iceberg maintenance: cannot resolve expired file %s: %v", filePath, err)
 			continue
 		}
-		dir := path.Join(tableBasePath, path.Dir(normalized))
-		fileName := path.Base(normalized)
-		if delErr := deleteFilerFile(ctx, filerClient, dir, fileName); delErr != nil {
+		if _, stillReferenced := keptFilerPaths[resolved]; stillReferenced {
+			continue
+		}
+		if delErr := deleteFilerFile(ctx, filerClient, path.Dir(resolved), path.Base(resolved)); delErr != nil {
 			glog.Warningf("iceberg maintenance: failed to delete unreferenced file %s: %v", filePath, delErr)
 		} else {
 			deletedCount++
@@ -172,7 +175,7 @@ func (h *Handler) expireSnapshots(
 func collectSnapshotFiles(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	snapshots []table.Snapshot,
 ) (map[string]struct{}, error) {
 	files := make(map[string]struct{})
@@ -182,7 +185,7 @@ func collectSnapshotFiles(
 		}
 		files[snap.ManifestList] = struct{}{}
 
-		manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, snap.ManifestList)
+		manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, snap.ManifestList)
 		if err != nil {
 			return nil, fmt.Errorf("read manifest list %s: %w", snap.ManifestList, err)
 		}
@@ -194,7 +197,7 @@ func collectSnapshotFiles(
 		for _, mf := range manifests {
 			files[mf.FilePath()] = struct{}{}
 
-			manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, mf.FilePath())
+			manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
 			if err != nil {
 				return nil, fmt.Errorf("read manifest %s: %w", mf.FilePath(), err)
 			}
@@ -224,12 +227,12 @@ func (h *Handler) removeOrphans(
 ) (string, map[string]int64, error) {
 	start := time.Now()
 	// Load current metadata
-	meta, metadataFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+	state, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
 	if err != nil {
 		return "", nil, fmt.Errorf("load metadata: %w", err)
 	}
 
-	orphanCandidates, err := collectOrphanCandidates(ctx, filerClient, bucketName, tablePath, meta, metadataFileName, config.OrphanOlderThanHours)
+	orphanCandidates, err := collectOrphanCandidates(ctx, filerClient, bucketName, state.DataPath, state.Metadata, state.MetadataFileName, config.OrphanOlderThanHours)
 	if err != nil {
 		return "", nil, fmt.Errorf("collect orphan candidates: %w", err)
 	}
@@ -253,12 +256,12 @@ func (h *Handler) removeOrphans(
 func collectOrphanCandidates(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	meta table.Metadata,
 	metadataFileName string,
 	orphanOlderThanHours int64,
 ) ([]filerFileEntry, error) {
-	referencedFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, meta.Snapshots())
+	referencedFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, dataPath, meta.Snapshots())
 	if err != nil {
 		return nil, fmt.Errorf("collect referenced files: %w", err)
 	}
@@ -268,13 +271,14 @@ func collectOrphanCandidates(
 		referencedFiles[mle.MetadataFile] = struct{}{}
 	}
 
-	normalizedRefs := make(map[string]struct{}, len(referencedFiles))
+	referencedFilerPaths := make(map[string]struct{}, len(referencedFiles))
 	for ref := range referencedFiles {
-		normalizedRefs[ref] = struct{}{}
-		normalizedRefs[normalizeIcebergPath(ref, bucketName, tablePath)] = struct{}{}
+		if filerPath, err := icebergFilerPath(bucketName, dataPath, ref); err == nil {
+			referencedFilerPaths[filerPath] = struct{}{}
+		}
 	}
 
-	tableBasePath := path.Join(s3tables.TablesPath, bucketName, tablePath)
+	tableBasePath := path.Join(s3tables.TablesPath, bucketName, dataPath)
 	safetyThreshold := time.Now().Add(-time.Duration(orphanOlderThanHours) * time.Hour)
 	var candidates []filerFileEntry
 
@@ -288,9 +292,7 @@ func collectOrphanCandidates(
 
 		for _, fe := range fileEntries {
 			entry := fe.Entry
-			fullPath := path.Join(fe.Dir, entry.Name)
-			relPath := strings.TrimPrefix(fullPath, tableBasePath+"/")
-			if _, isReferenced := normalizedRefs[relPath]; isReferenced {
+			if _, isReferenced := referencedFilerPaths[path.Join(fe.Dir, entry.Name)]; isReferenced {
 				continue
 			}
 			if entry.Attributes == nil {
@@ -319,10 +321,11 @@ func (h *Handler) rewriteManifests(
 ) (string, map[string]int64, error) {
 	start := time.Now()
 	// Load current metadata
-	meta, metadataFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+	state, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
 	if err != nil {
 		return "", nil, fmt.Errorf("load metadata: %w", err)
 	}
+	meta, dataPath := state.Metadata, state.DataPath
 	predicate, err := parsePartitionPredicate(config.Where, meta)
 	if err != nil {
 		return "", nil, err
@@ -334,7 +337,7 @@ func (h *Handler) rewriteManifests(
 	}
 
 	// Read manifest list
-	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, currentSnap.ManifestList)
+	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, currentSnap.ManifestList)
 	if err != nil {
 		return "", nil, fmt.Errorf("read manifest list: %w", err)
 	}
@@ -368,7 +371,7 @@ func (h *Handler) rewriteManifests(
 	var manifestsRewritten int64
 
 	for _, mf := range dataManifests {
-		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, tablePath, mf.FilePath())
+		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
 		if err != nil {
 			return "", nil, fmt.Errorf("read manifest %s: %w", mf.FilePath(), err)
 		}
@@ -427,7 +430,7 @@ func (h *Handler) rewriteManifests(
 	newSnapshotID := time.Now().UnixMilli()
 	newSeqNum := currentSnap.SequenceNumber + 1
 	artifactSuffix := compactRandomSuffix()
-	metaDir := path.Join(s3tables.TablesPath, bucketName, tablePath, "metadata")
+	metaDir := path.Join(s3tables.TablesPath, bucketName, dataPath, "metadata")
 
 	// Track written artifacts so we can clean them up if the commit fails.
 	type artifact struct {
@@ -456,7 +459,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 := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestFileName)
+		manifestPath := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestFileName)
 
 		var manifestBuf bytes.Buffer
 		mergedManifest, err := iceberg.WriteManifest(
@@ -500,9 +503,9 @@ func (h *Handler) rewriteManifests(
 	writtenArtifacts = append(writtenArtifacts, artifact{dir: metaDir, fileName: manifestListFileName})
 
 	// Create new snapshot with the rewritten manifest list
-	manifestListLocation := absoluteIcebergPath(bucketName, tablePath, "metadata", manifestListFileName)
+	manifestListLocation := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestListFileName)
 
-	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
+	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, state.MetadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
 		// Guard: verify table head hasn't advanced since we planned.
 		// The merged manifest and manifest list were built against snapshotID;
 		// if the head moved, they reference stale state.
@@ -585,14 +588,15 @@ func (h *Handler) commitWithRetry(
 		}
 
 		// Load current metadata
-		meta, metaFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+		state, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
 		if err != nil {
 			return fmt.Errorf("load metadata (attempt %d): %w", attempt, err)
 		}
+		meta, metaFileName, dataPath := state.Metadata, state.MetadataFileName, state.DataPath
 
 		// Build new metadata — pass the current metadata file location so the
 		// metadata log correctly records where the previous version lives.
-		currentMetaFilePath := absoluteIcebergPath(bucketName, tablePath, "metadata", metaFileName)
+		currentMetaFilePath := absoluteIcebergPath(bucketName, dataPath, "metadata", metaFileName)
 		builder, err := table.MetadataBuilderFromBase(meta, currentMetaFilePath)
 		if err != nil {
 			return fmt.Errorf("create metadata builder (attempt %d): %w", attempt, err)
@@ -625,14 +629,14 @@ func (h *Handler) commitWithRetry(
 		newMetadataFileName := fmt.Sprintf("v%d-%d.metadata.json", newVersion, time.Now().UnixNano())
 
 		// Save new metadata file
-		metaDir := path.Join(s3tables.TablesPath, bucketName, tablePath, "metadata")
+		metaDir := path.Join(s3tables.TablesPath, bucketName, dataPath, "metadata")
 		if err := saveFilerFile(ctx, filerClient, metaDir, newMetadataFileName, metadataBytes); err != nil {
 			return fmt.Errorf("save metadata file (attempt %d): %w", attempt, err)
 		}
 
 		// Update the table entry's xattr with new metadata (CAS on version)
 		tableDir := path.Join(s3tables.TablesPath, bucketName, tablePath)
-		newMetadataLocation := absoluteIcebergPath(bucketName, tablePath, "metadata", newMetadataFileName)
+		newMetadataLocation := absoluteIcebergPath(bucketName, dataPath, "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/planning_index.go

@@ -44,18 +44,28 @@ type tableMetadataEnvelope struct {
 	PlanningIndex json.RawMessage `json:"planningIndex,omitempty"`
 }
 
-func parseTableMetadataEnvelope(metadataBytes []byte) (table.Metadata, string, *planningIndex, error) {
+// tableState is the catalog's view of one table: its Iceberg metadata, the
+// metadata file backing it, where its files live, and the cached planning index.
+type tableState struct {
+	Metadata         table.Metadata
+	MetadataFileName string
+	// DataPath is the bucket-relative directory holding metadata/ and data/.
+	DataPath      string
+	PlanningIndex *planningIndex
+}
+
+func parseTableMetadataEnvelope(metadataBytes []byte, bucketName, tablePath string) (*tableState, error) {
 	var envelope tableMetadataEnvelope
 	if err := json.Unmarshal(metadataBytes, &envelope); err != nil {
-		return nil, "", nil, fmt.Errorf("parse metadata xattr: %w", err)
+		return nil, fmt.Errorf("parse metadata xattr: %w", err)
 	}
 	if envelope.Metadata == nil || len(envelope.Metadata.FullMetadata) == 0 {
-		return nil, "", nil, fmt.Errorf("no fullMetadata in table xattr")
+		return nil, fmt.Errorf("no fullMetadata in table xattr")
 	}
 
 	meta, err := table.ParseMetadataBytes(envelope.Metadata.FullMetadata)
 	if err != nil {
-		return nil, "", nil, fmt.Errorf("parse iceberg metadata: %w", err)
+		return nil, fmt.Errorf("parse iceberg metadata: %w", err)
 	}
 
 	var index *planningIndex
@@ -66,11 +76,30 @@ func parseTableMetadataEnvelope(metadataBytes []byte) (table.Metadata, string, *
 		}
 	}
 
-	metadataFileName := metadataFileNameFromLocation(envelope.MetadataLocation, "", "")
+	metadataFileName := metadataFileNameFromLocation(envelope.MetadataLocation)
 	if metadataFileName == "" {
 		metadataFileName = fmt.Sprintf("v%d.metadata.json", envelope.MetadataVersion)
 	}
-	return meta, metadataFileName, index, nil
+	return &tableState{
+		Metadata:         meta,
+		MetadataFileName: metadataFileName,
+		DataPath:         tableDataPath(bucketName, tablePath, envelope.MetadataLocation),
+		PlanningIndex:    index,
+	}, nil
+}
+
+// tableDataPath resolves the bucket-relative directory holding a table's files.
+// The catalog records the table's real location in the metadata location, which
+// is the catalog path only for tables the catalog placed there itself: a client
+// creating a table through the REST catalog can be handed a location elsewhere
+// in the bucket, e.g. "ns/table-<uuid>" when the catalog path was occupied.
+// Locations outside the table's own bucket fall back to the catalog path.
+func tableDataPath(bucketName, tablePath, metadataLocation string) string {
+	dataDir := s3tables.TableDataDirFromMetadataLocation(metadataLocation)
+	if rel, ok := cutPathPrefix(dataDir, path.Join(s3tables.TablesPath, bucketName)); ok && rel != "" {
+		return rel
+	}
+	return tablePath
 }
 
 func (idx *planningIndex) matchesSnapshot(meta table.Metadata) bool {
@@ -140,7 +169,7 @@ func mergePlanningIndexSections(index, existing *planningIndex) *planningIndex {
 func buildPlanningIndexFromManifests(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
-	bucketName, tablePath string,
+	bucketName, dataPath string,
 	meta table.Metadata,
 	config Config,
 	ops []string,
@@ -159,7 +188,7 @@ func buildPlanningIndexFromManifests(
 	}
 
 	if operationRequested(ops, "compact") {
-		eligible, err := hasEligibleCompaction(ctx, filerClient, bucketName, tablePath, manifests, config, meta, nil)
+		eligible, err := hasEligibleCompaction(ctx, filerClient, bucketName, dataPath, manifests, config, meta, nil)
 		if err != nil {
 			return nil, err
 		}
@@ -213,8 +242,8 @@ func persistPlanningIndex(
 	if err := json.Unmarshal(existingXattr, &internalMeta); err != nil {
 		return fmt.Errorf("unmarshal metadata xattr: %w", err)
 	}
-	if _, _, existingIndex, err := parseTableMetadataEnvelope(existingXattr); err == nil {
-		index = mergePlanningIndexSections(index, existingIndex)
+	if existingState, err := parseTableMetadataEnvelope(existingXattr, bucketName, tablePath); err == nil {
+		index = mergePlanningIndexSections(index, existingState.PlanningIndex)
 	}
 
 	indexJSON, err := json.Marshal(index)

weed/worker/tasks/iceberg/where_filter_test.go

@@ -241,11 +241,11 @@ func TestCompactDataFilesWhereFilter(t *testing.T) {
 		t.Fatalf("unexpected result: %q", result)
 	}
 
-	meta, _, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
 	if err != nil {
 		t.Fatalf("loadCurrentMetadata: %v", err)
 	}
-	manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, setup.tablePath(), meta)
+	manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata)
 	if err != nil {
 		t.Fatalf("loadCurrentManifests: %v", err)
 	}