diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go
index 1e2349937..f3ff610c0 100644
--- a/weed/s3api/s3tables/handler_table.go
+++ b/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
diff --git a/weed/s3api/s3tables/table_data_dir_test.go b/weed/s3api/s3tables/table_data_dir_test.go
index ad3244958..ffbcc3c79 100644
--- a/weed/s3api/s3tables/table_data_dir_test.go
+++ b/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)
 		}
 	}
 }
diff --git a/weed/s3api/s3tables/utils.go b/weed/s3api/s3tables/utils.go
index a0bd86fb7..f04b31e6a 100644
--- a/weed/s3api/s3tables/utils.go
+++ b/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]
diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go
index 21b6ae75c..d8f10ebec 100644
--- a/weed/worker/tasks/iceberg/compact.go
+++ b/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)
 		}
diff --git a/weed/worker/tasks/iceberg/delete_rewrite.go b/weed/worker/tasks/iceberg/delete_rewrite.go
index 4b5e3fb81..e633cefee 100644
--- a/weed/worker/tasks/iceberg/delete_rewrite.go
+++ b/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
diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go
index d53f1c54b..bbf704e4c 100644
--- a/weed/worker/tasks/iceberg/detection.go
+++ b/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)
 		}
diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go
index c728766c5..a1b4aea65 100644
--- a/weed/worker/tasks/iceberg/exec_test.go
+++ b/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)
 	}
diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go
index ba338f13f..4f03f3337 100644
--- a/weed/worker/tasks/iceberg/filer_io.go
+++ b/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.
diff --git a/weed/worker/tasks/iceberg/handler_test.go b/weed/worker/tasks/iceberg/handler_test.go
index dc1c7cb7b..365949be0 100644
--- a/weed/worker/tasks/iceberg/handler_test.go
+++ b/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(
diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go
index 3430d7703..ae994e46e 100644
--- a/weed/worker/tasks/iceberg/operations.go
+++ b/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
diff --git a/weed/worker/tasks/iceberg/planning_index.go b/weed/worker/tasks/iceberg/planning_index.go
index a2401836a..320cafda2 100644
--- a/weed/worker/tasks/iceberg/planning_index.go
+++ b/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)
diff --git a/weed/worker/tasks/iceberg/where_filter_test.go b/weed/worker/tasks/iceberg/where_filter_test.go
index f62149a94..eef4a904b 100644
--- a/weed/worker/tasks/iceberg/where_filter_test.go
+++ b/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)
 	}