s3tables, iceberg: give tables created without a location their own directory (#12475)

* s3tables: give a table created without a location its own directory

An Iceberg table created without a location kept its files under its name
path, so a rename left them where the next table of that name would land.
Such a table now gets s3://<bucket>/<namespace>/<name>-<uuid>, stored as the
record's warehouseLocation and returned by GetTable; its first commit writes
there without moving the root. Records without one keep resolving to their
name path, so existing tables need no rewrite. Lance datasets stay at their
name path, where Lance clients and the vector engine look for them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: create tables and views in a directory of their own

A table or view created without a location took its name path, or a
unique one only when that path was already taken. It now always gets
<name>-<uuid>, so a later rename never leaves its files under a name
another table can take.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: load and commit a table at its warehouse location

A table created through S3 Tables without a location has no metadata until
its first commit; loading or committing it now uses the directory recorded
for it, and only a record without one falls back to its name path.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: fail a stage-create whose marker is not saved

The marker is how the commit finds the location the stage-create handed
out. Without it the commit fell back to the name path while the client's
files sat under the returned location.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: expect tables created without a location in their own directory

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: rename a table that keeps its files elsewhere before reusing its name

Enterprise keeps the storage marker on a renamed table's files, so the name
of a table whose files sit under its name path stays taken there.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3tables: remove the table directory when a create backs out

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: answer a create that lost the race with the registered table

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: answer a view create that lost the race with the registered view

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: fail a staged commit whose name another create registered

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: remove the directory of a create whose name another table took

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: keep the metadata of a staged commit whose name is taken

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: write the lakekeeper table's data under its warehouse location

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: commit a staged table at the location of its own stage

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: never continue another stage when a commit's own marker is gone

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: test filer honors exclusive creates and token-guarded deletes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

test/s3tables/catalog/duckdb_deterministic_location_test.go

@@ -14,20 +14,10 @@ import (
 )
 
 // TestDuckDBDeterministicLocationRead validates that DuckDB's iceberg
-// extension can resolve and scan a table that lives at the deterministic
-// <bucket>/<namespace>/<tableName> path produced when no client-side UUID
-// is appended to the table location. This is the layout produced by:
-//
-//   - iceberg-spark (no concept of UUID-suffixed locations)
-//   - pyiceberg (no concept of UUID-suffixed locations)
-//   - Trino with iceberg.unique-table-location=false
-//
-// Post-#9246, REST clients can resolve table locations under a namespace
-// because GetNamespace advertises a default `location` property. This
-// test creates a namespace + table via the REST API directly (so no
-// client-side UUID is added), then asks DuckDB to ATTACH the catalog and
-// describe the table — verifying DuckDB walks the catalog → metadata →
-// schema chain successfully against a deterministic on-disk path.
+// extension can resolve and scan a table created through the REST API with no
+// location. The catalog gives such a table the directory
+// <bucket>/<namespace>/<tableName>-<uuid>, and DuckDB has to reach its
+// metadata and schema through the catalog rather than by the table's name.
 //
 // The test does not insert data files — DuckDB scanning an empty table
 // is sufficient to validate the catalog/metadata resolution path. If
@@ -55,14 +45,11 @@ func TestDuckDBDeterministicLocationRead(t *testing.T) {
 	createNamespaceWithToken(t, env, token, bucketName, namespace)
 	createIcebergTableWithToken(t, env, token, bucketName, namespace, tableName)
 
-	// Sanity-check the catalog row exposes a deterministic location.
+	// The table gets a directory of its own, named after it and its UUID.
 	tableLoc := getTableLocation(t, env, token, bucketName, namespace, tableName)
-	wantPrefix := fmt.Sprintf("s3://%s/%s/%s", bucketName, namespace, tableName)
-	if !strings.HasPrefix(tableLoc, wantPrefix) {
-		t.Fatalf("table location %q should start with %q (deterministic, no UUID suffix)", tableLoc, wantPrefix)
-	}
-	if strings.Contains(strings.TrimPrefix(tableLoc, wantPrefix), "-") {
-		t.Fatalf("table location %q should not have a -<uuid> suffix after %q", tableLoc, wantPrefix)
+	wantPrefix := fmt.Sprintf("s3://%s/%s/%s-", bucketName, namespace, tableName)
+	if !strings.HasPrefix(tableLoc, wantPrefix) || strings.Contains(strings.TrimPrefix(tableLoc, wantPrefix), "/") {
+		t.Fatalf("table location %q should be a directory %s<uuid>", tableLoc, wantPrefix)
 	}
 
 	sqlContent := fmt.Sprintf(`

test/s3tables/catalog_olake/olake_catalog_test.go

@@ -476,12 +476,24 @@ func (env *TestEnvironment) readTable(t *testing.T, mode string) string {
 
 // listTableMetadata lists the table's metadata directory through the filer so
 // the repair assertion looks at real objects rather than at what the catalog
-// reports about itself.
+// reports about itself. A table created without a location sits in a
+// <table>-<uuid> directory, so every directory of the table's name is read.
 func (env *TestEnvironment) listTableMetadata(t *testing.T) []string {
 	t.Helper()
 
-	url := fmt.Sprintf("http://%s:%d/buckets/%s/%s/%s/metadata/?limit=200",
-		env.bindIP, env.filerPort, tableBucketName, expectedNamespace, expectedTable)
+	var names []string
+	for _, dir := range env.listFilerDir(t, fmt.Sprintf("/buckets/%s/%s/", tableBucketName, expectedNamespace)) {
+		if dir == expectedTable || strings.HasPrefix(dir, expectedTable+"-") {
+			names = append(names, env.listFilerDir(t, fmt.Sprintf("/buckets/%s/%s/%s/metadata/", tableBucketName, expectedNamespace, dir))...)
+		}
+	}
+	return names
+}
+
+func (env *TestEnvironment) listFilerDir(t *testing.T, dir string) []string {
+	t.Helper()
+
+	url := fmt.Sprintf("http://%s:%d%s?limit=200", env.bindIP, env.filerPort, dir)
 	req, err := http.NewRequest(http.MethodGet, url, nil)
 	if err != nil {
 		t.Fatalf("Failed to build filer request: %v", err)
@@ -494,6 +506,9 @@ func (env *TestEnvironment) listTableMetadata(t *testing.T) []string {
 	}
 	defer resp.Body.Close()
 	body, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode == http.StatusNotFound {
+		return nil
+	}
 	if resp.StatusCode != http.StatusOK {
 		t.Fatalf("filer listing returned %d: %s", resp.StatusCode, body)
 	}

test/s3tables/catalog_spark/spark_deterministic_location_test.go

@@ -13,11 +13,9 @@ import (
 
 // TestSparkDeterministicLocationCTAS exercises Spark's iceberg-spark
 // connector against the SeaweedFS REST catalog with no explicit `LOCATION`
-// clause and no UUID-suffixing. iceberg-spark does not implement Trino's
-// unique-table-location feature, so every CREATE TABLE here produces a
-// deterministic <namespace-location>/<tableName> path. The test asserts
-// the fresh-CTAS pattern works AND the resulting on-disk layout has no
-// UUID-suffixed sibling — i.e. one directory per table.
+// clause. The catalog picks the location, <namespace>/<tableName>-<uuid>, and
+// the test asserts the fresh-CTAS pattern works and each table's files sit in
+// that one directory, none under the bare table name.
 //
 // The fix in #9246 (advertise a default `location` namespace property)
 // is what allows iceberg-spark to resolve the table location at all when
@@ -82,13 +80,11 @@ print(f"ctas row count: {count}")
 		}
 	}
 
-	// Layout assertion: each table's data must land at the deterministic
-	// <namespace>/<tableName>/ path, with no UUID-suffixed sibling.
+	// Layout assertion: each table's files land in one <tableName>-<uuid>
+	// directory, and nothing sits under the bare table name.
 	keys := listSparkBucketKeysUnder(t, env, tableBucket, namespace+"/")
-	expectKeyAtSparkLocation(t, keys, namespace, srcTable)
-	expectKeyAtSparkLocation(t, keys, namespace, ctasTable)
-	rejectUUIDSuffixedSparkSibling(t, keys, namespace, srcTable)
-	rejectUUIDSuffixedSparkSibling(t, keys, namespace, ctasTable)
+	expectKeysInOneSparkLocation(t, keys, namespace, srcTable)
+	expectKeysInOneSparkLocation(t, keys, namespace, ctasTable)
 }
 
 func listSparkBucketKeysUnder(t *testing.T, env *TestEnvironment, bucket, prefix string) []string {
@@ -113,23 +109,18 @@ func listSparkBucketKeysUnder(t *testing.T, env *TestEnvironment, bucket, prefix
 	return keys
 }
 
-func expectKeyAtSparkLocation(t *testing.T, keys []string, namespace, table string) {
+func expectKeysInOneSparkLocation(t *testing.T, keys []string, namespace, table string) {
 	t.Helper()
-	want := namespace + "/" + table + "/"
+	dirs := map[string]bool{}
 	for _, k := range keys {
-		if strings.HasPrefix(k, want) {
-			return
+		if strings.HasPrefix(k, namespace+"/"+table+"/") {
+			t.Fatalf("found key %q under the bare table name %q", k, table)
+		}
+		if rest, ok := strings.CutPrefix(k, namespace+"/"+table+"-"); ok {
+			dirs[strings.SplitN(rest, "/", 2)[0]] = true
 		}
 	}
-	t.Fatalf("expected at least one object under %q; got keys: %v", want, keys)
-}
-
-func rejectUUIDSuffixedSparkSibling(t *testing.T, keys []string, namespace, table string) {
-	t.Helper()
-	prefix := namespace + "/" + table + "-"
-	for _, k := range keys {
-		if strings.HasPrefix(k, prefix) {
-			t.Fatalf("found UUID-suffixed sibling key %q for table %q (Spark should produce no -<uuid> suffix)", k, table)
-		}
+	if len(dirs) != 1 {
+		t.Fatalf("want the files of %q in one %s-<uuid> directory, found %d; keys: %v", table, table, len(dirs), keys)
 	}
 }

test/s3tables/lakekeeper/lakekeeper_test.go

@@ -384,7 +384,15 @@ func runLakekeeperTableBucketRepro(t *testing.T, env *TestEnvironment) {
 		t.Fatalf("Create S3 client failed: %v", err)
 	}
 
-	objectKey := fmt.Sprintf("%s/%s/data/part-%d.parquet", namespaceName, tableName, time.Now().UnixNano())
+	table, err := client.GetTable(ctx, bucketARN, namespaceName, tableName)
+	if err != nil {
+		t.Fatalf("GetTable failed: %v", err)
+	}
+	tableDir := strings.TrimPrefix(table.WarehouseLocation, "s3://"+bucketName+"/")
+	if tableDir == "" || tableDir == table.WarehouseLocation {
+		t.Fatalf("unexpected warehouse location %q", table.WarehouseLocation)
+	}
+	objectKey := fmt.Sprintf("%s/data/part-%d.parquet", tableDir, time.Now().UnixNano())
 	createResp, err := s3Client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{
 		Bucket: aws.String(bucketName),
 		Key:    aws.String(objectKey),
@@ -556,6 +564,18 @@ func (c *s3TablesClient) CreateTable(ctx context.Context, bucketARN, namespace,
 	return c.doRequest(ctx, "CreateTable", http.MethodPut, path, req, nil)
 }
 
+func (c *s3TablesClient) GetTable(ctx context.Context, bucketARN, namespace, name string) (*s3tables.GetTableResponse, error) {
+	query := url.Values{}
+	query.Set("tableBucketARN", bucketARN)
+	query.Set("namespace", namespace)
+	query.Set("name", name)
+	var resp s3tables.GetTableResponse
+	if err := c.doRequest(ctx, "GetTable", http.MethodGet, "/get-table?"+query.Encode(), nil, &resp); err != nil {
+		return nil, err
+	}
+	return &resp, nil
+}
+
 func (c *s3TablesClient) DeleteTable(ctx context.Context, bucketARN, namespace, name string) error {
 	path := "/tables/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(namespace) + "/" + url.PathEscape(name)
 	return c.doRequest(ctx, "DeleteTable", http.MethodDelete, path, nil, nil)

weed/s3api/iceberg/handlers_commit.go

@@ -119,7 +119,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 				var latestMarker *stageCreateMarker
 				if stageCreateEnabled {
 					var markerErr error
-					latestMarker, markerErr = s.loadLatestStageCreateMarker(r.Context(), bucketName, namespace, tableName)
+					latestMarker, markerErr = s.loadLatestStageCreateMarker(r.Context(), bucketName, namespace, tableName, assignedTableUUID(raw.Updates))
 					if markerErr != nil {
 						writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to load stage-create marker: "+markerErr.Error())
 						return
@@ -247,10 +247,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
-		if location == "" {
-			location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
-		}
+		location := tableLocation(getResp, bucketName, namespace, tableName)
 		locBucket, locPath, locErr := parseS3Location(location)
 		if locErr != nil {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid table location: "+locErr.Error())

weed/s3api/iceberg/handlers_table.go

@@ -110,34 +110,6 @@ func (s *Server) handleListTables(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, result)
 }
 
-// tablePathOccupied reports whether a filer entry already exists at the catalog
-// name path. A leftover directory there (e.g. data kept when another table was
-// renamed to this name) means a table created at the default location would
-// overwrite it, so the caller routes the new table to a unique location. A
-// lookup failure other than not-found is returned so the caller can fail the
-// create rather than fall back to the default path on a transient filer error.
-func (s *Server) tablePathOccupied(ctx context.Context, bucketName, tablePath string) (bool, error) {
-	full := path.Join(s3tables.TablesPath, bucketName, tablePath)
-	dir, name := path.Split(full)
-	dir = strings.TrimSuffix(dir, "/")
-	occupied := false
-	err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		_, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
-			Directory: dir,
-			Name:      name,
-		})
-		if errors.Is(err, filer_pb.ErrNotFound) {
-			return nil
-		}
-		if err != nil {
-			return err
-		}
-		occupied = true
-		return nil
-	})
-	return occupied, err
-}
-
 // authorizeCreateTable checks the caller may create the table at location, for
 // the create paths that touch the bucket before the catalog registers it.
 func (s *Server) authorizeCreateTable(ctx context.Context, bucketARN string, namespace []string, tableName, identityName, location string) error {
@@ -183,20 +155,11 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 
 	// Generate UUID for the new table
 	tableUUID := uuid.New()
-	tablePath := path.Join(flattenNamespacePath(namespace), req.Name)
+	// A location the server picks is unique to the table, so its files never
+	// sit under a name a rename can hand to another table.
+	tablePath := path.Join(flattenNamespacePath(namespace), req.Name+"-"+tableUUID.String())
 	location := strings.TrimSuffix(req.Location, "/")
 	if location == "" {
-		// If a leftover directory already occupies the default location (data
-		// kept when another table was renamed to this name), route this table
-		// to a unique location so it cannot overwrite that table's files.
-		occupied, err := s.tablePathOccupied(r.Context(), bucketName, tablePath)
-		if err != nil {
-			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to check table location: "+err.Error())
-			return
-		}
-		if occupied {
-			tablePath = path.Join(flattenNamespacePath(namespace), req.Name+"-"+tableUUID.String())
-		}
 		if req.Properties != nil {
 			if warehouse := strings.TrimSuffix(req.Properties["warehouse"], "/"); warehouse != "" {
 				location = fmt.Sprintf("%s/%s", warehouse, tablePath)
@@ -319,8 +282,11 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		stagedMetadataLocation := fmt.Sprintf("s3://%s/%s/metadata/%s", metadataBucket, stagedTablePath, metadataFileName)
+		// The marker is how the commit finds the location handed out here.
 		if markerErr := s.writeStageCreateMarker(r.Context(), bucketName, namespace, tableName, tableUUID, location, stagedMetadataLocation); markerErr != nil {
 			glog.V(1).Infof("Iceberg: failed to persist stage-create marker for %s.%s: %v", flattenNamespacePath(namespace), tableName, markerErr)
+			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save stage-create marker")
+			return
 		}
 		// The staged table is not registered yet, so there is nothing to
 		// authorize an update against; the create gate above is the write
@@ -366,7 +332,10 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		mgrClient := s3tables.NewManagerClient(client)
 		return s.tablesManager.Execute(r.Context(), mgrClient, "CreateTable", createReq, &createResp, identityName)
 	})
-
+	if err == nil && createResp.MetadataLocation != metadataLocation {
+		// Another create registered the name first.
+		err = &s3tables.S3TablesError{Type: s3tables.ErrCodeTableAlreadyExists, Message: fmt.Sprintf("table %s already exists", tableName)}
+	}
 	if err != nil {
 		// The file is named by this request alone, so it can go once the
 		// catalog refused the registration before a record naming it became
@@ -378,6 +347,13 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 			s.discardStagedMetadata(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataLocation, stageToken)
 		}
 		if isS3TablesAlreadyExists(err) {
+			if strings.HasSuffix(metadataPath, tablePath) {
+				// The name went to another table before this request wrote a
+				// record, so nothing names the directory the server picked.
+				if cleanupErr := s.cleanupStaleTableLocation(r.Context(), metadataBucket, metadataPath); cleanupErr != nil {
+					glog.V(1).Infof("Iceberg: CreateTable cleanup %s: %v", metadataPath, cleanupErr)
+				}
+			}
 			writeTableAlreadyExists(w, namespace, tableName)
 			return
 		}
@@ -386,17 +362,6 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// The catalog answers a create for a name that already holds a table with
-	// that table's own metadata location rather than an error. The table was
-	// registered by a request that won the race, not by this one. A winner
-	// registered without a location comes back empty, which is no match either:
-	// this request always sends one, and a fresh create echoes it.
-	if createResp.MetadataLocation != metadataLocation {
-		s.discardStagedMetadata(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataLocation, stageToken)
-		writeTableAlreadyExists(w, namespace, tableName)
-		return
-	}
-	finalLocation := metadataLocation
 	if markerErr := s.deleteStageCreateMarkers(r.Context(), bucketName, namespace, tableName); markerErr != nil {
 		glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after create: %v", flattenNamespacePath(namespace), tableName, markerErr)
 	}
@@ -405,7 +370,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		return s.mayUpdateTable(r, bucketName, namespace, req.Name)
 	})
 	result := LoadTableResult{
-		MetadataLocation:   finalLocation,
+		MetadataLocation:   metadataLocation,
 		Metadata:           metadata,
 		Config:             config,
 		StorageCredentials: storageCredentials,
@@ -586,10 +551,7 @@ func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) buildLoadTableResult(r *http.Request, getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) (LoadTableResult, error) {
-	location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
-	if location == "" {
-		location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
-	}
+	location := tableLocation(getResp, bucketName, namespace, tableName)
 	tableUUID := uuid.Nil
 	if getResp.Metadata != nil && getResp.Metadata.Iceberg != nil && getResp.Metadata.Iceberg.TableUUID != "" {
 		if parsed, err := uuid.Parse(getResp.Metadata.Iceberg.TableUUID); err == nil {

weed/s3api/iceberg/handlers_transaction.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
-	"path"
 	"strings"
 
 	"github.com/apache/iceberg-go/table"
@@ -144,10 +143,7 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN,
 		return nil, reqErr
 	}
 
-	location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
-	if location == "" {
-		location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
-	}
+	location := tableLocation(getResp, bucketName, namespace, tableName)
 	locBucket, locPath, locErr := parseS3Location(location)
 	if locErr != nil {
 		return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Invalid table location: " + locErr.Error()}

weed/s3api/iceberg/handlers_view.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/view"
+	"github.com/google/uuid"
 	"github.com/gorilla/mux"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -99,7 +100,7 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) {
 	bucketARN := buildTableBucketARN(bucketName)
 	identityName := s3_constants.GetIdentityNameFromContext(r)
 
-	viewPath := path.Join(flattenNamespacePath(namespace), req.Name)
+	viewPath := path.Join(flattenNamespacePath(namespace), req.Name+"-"+uuid.NewString())
 	location := strings.TrimSuffix(req.Location, "/")
 	if location == "" {
 		if req.Properties != nil {
@@ -202,6 +203,22 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if createResp.MetadataLocation != metadataLocation {
+		// Another create registered the name first; answer with its view.
+		existsResp, existsErr := s.getView(r, namespace, req.Name)
+		if existsErr != nil {
+			writeError(w, http.StatusConflict, "AlreadyExistsException", fmt.Sprintf("View already exists: %s", req.Name))
+			return
+		}
+		result, buildErr := s.buildViewResponse(r, existsResp, bucketName, namespace, req.Name)
+		if buildErr != nil {
+			writeError(w, http.StatusInternalServerError, "InternalServerError", buildErr.Error())
+			return
+		}
+		writeLoadResult(w, http.StatusOK, result)
+		return
+	}
+
 	// Persist the metadata file only after the catalog registers the view, so a
 	// missing namespace or name collision fails before any bytes hit storage.
 	if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes, false); err != nil {
@@ -214,15 +231,11 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	finalLocation := createResp.MetadataLocation
-	if finalLocation == "" {
-		finalLocation = metadataLocation
-	}
 	config, _ := s.buildFileIOConfig(r, location, func() bool {
 		return s.mayUpdateView(r, bucketName, namespace, req.Name)
 	})
 	writeLoadResult(w, http.StatusOK, ViewResponse{
-		MetadataLocation: finalLocation,
+		MetadataLocation: metadataLocation,
 		Metadata:         metadata,
 		Config:           config,
 	})

weed/s3api/iceberg/iceberg_create_table_conflict_test.go

@@ -189,7 +189,11 @@ func (e *conflictEnv) metadataFiles() []string {
 // straight from the filer the way a client reading S3 directly would.
 func (e *conflictEnv) metadataFile(location string) table.Metadata {
 	e.t.Helper()
-	entry := e.filer.Get(e.metadataDir(), path.Base(location))
+	bucket, tablePath, err := parseS3Location(tableLocationFromMetadataLocation(location))
+	if err != nil {
+		e.t.Fatalf("parse %s: %v", location, err)
+	}
+	entry := e.filer.Get(metadataDirPath(bucket, tablePath), path.Base(location))
 	if entry == nil {
 		e.t.Fatalf("no metadata file behind %s", location)
 	}

weed/s3api/iceberg/iceberg_create_view_test.go

@@ -21,6 +21,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 // memFiler is a minimal in-memory SeaweedFilerClient backing the view handlers
@@ -33,6 +35,8 @@ type memFiler struct {
 	// failFileCreate, when set, makes CreateEntry fail for non-directory entries,
 	// simulating a metadata-file write error.
 	failFileCreate error
+	// onLookup, when set, runs at the start of every LookupDirectoryEntry.
+	onLookup func(p string)
 }
 
 func newMemFiler() *memFiler {
@@ -48,6 +52,9 @@ func (m *memFiler) seed(p string, entry *filer_pb.Entry) {
 }
 
 func (m *memFiler) LookupDirectoryEntry(_ context.Context, in *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	if hook := m.onLookup; hook != nil {
+		hook(path.Join(in.Directory, in.Name))
+	}
 	entry, ok := m.entries[path.Join(in.Directory, in.Name)]
 	if !ok {
 		return nil, filer_pb.ErrNotFound
@@ -59,12 +66,22 @@ func (m *memFiler) CreateEntry(_ context.Context, in *filer_pb.CreateEntryReques
 	if m.failFileCreate != nil && !in.Entry.IsDirectory {
 		return nil, m.failFileCreate
 	}
+	if _, taken := m.entries[path.Join(in.Directory, in.Entry.Name)]; taken && in.OExcl {
+		return &filer_pb.CreateEntryResponse{ErrorCode: filer_pb.FilerError_ENTRY_ALREADY_EXISTS}, nil
+	}
 	m.entries[path.Join(in.Directory, in.Entry.Name)] = in.Entry
 	return &filer_pb.CreateEntryResponse{}, nil
 }
 
 func (m *memFiler) DeleteEntry(_ context.Context, in *filer_pb.DeleteEntryRequest, _ ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) {
 	prefix := path.Join(in.Directory, in.Name)
+	if entry, ok := m.entries[prefix]; ok {
+		for key, want := range in.ExpectedExtended {
+			if string(entry.Extended[key]) != string(want) {
+				return nil, status.Errorf(codes.FailedPrecondition, "%s: %s changed", prefix, key)
+			}
+		}
+	}
 	for p := range m.entries {
 		if p == prefix || strings.HasPrefix(p, prefix+"/") {
 			delete(m.entries, p)
@@ -200,7 +217,16 @@ func TestCreateViewDuplicateDoesNotClobberMetadata(t *testing.T) {
 		t.Fatalf("first create status = %d, want 200 (body: %s)", w.Code, w.Body.String())
 	}
 
-	metadataKey := path.Join(s3tables.GetTablePath(bucket, "ns", "v"), "metadata", "v1.metadata.json")
+	var created struct {
+		MetadataLocation string `json:"metadata-location"`
+	}
+	if err := json.Unmarshal(w.Body.Bytes(), &created); err != nil {
+		t.Fatalf("decode create response: %v", err)
+	}
+	if !strings.HasPrefix(created.MetadataLocation, "s3://"+bucket+"/ns/v-") {
+		t.Fatalf("metadata-location = %q, want one under a directory of the view's own", created.MetadataLocation)
+	}
+	metadataKey := path.Join(s3tables.TablesPath, strings.TrimPrefix(created.MetadataLocation, "s3://"))
 	first, ok := fc.entries[metadataKey]
 	if !ok {
 		t.Fatalf("metadata file not written at %s", metadataKey)
@@ -219,6 +245,31 @@ func TestCreateViewDuplicateDoesNotClobberMetadata(t *testing.T) {
 	}
 }
 
+// A view create that loses the race for its name answers with the view the
+// catalog registered and writes no metadata of its own.
+func TestCreateViewThatLosesTheRaceReturnsTheRegisteredView(t *testing.T) {
+	const bucket = "warehouse"
+	fc := newMemFiler()
+	seedNamespace(fc, bucket, "ns", s3_constants.AccountAdminId)
+	s := NewServer(fc, nil)
+	first := httptest.NewRecorder()
+	raceAtSecondLookup(fc, s3tables.GetTablePath(bucket, "ns", "v"), func() {
+		s.handleCreateView(first, newCreateViewRequest(t, "ns", "v", "SELECT 1"))
+	})
+	second := httptest.NewRecorder()
+	s.handleCreateView(second, newCreateViewRequest(t, "ns", "v", "SELECT 2"))
+
+	won, lost := decodeCreated(t, first), decodeCreated(t, second)
+	if lost != won {
+		t.Errorf("losing create answered %+v, want the registered view %+v", lost, won)
+	}
+	for p, entry := range fc.entries {
+		if !entry.IsDirectory && strings.HasSuffix(p, ".metadata.json") && p != path.Join(s3tables.TablesPath, strings.TrimPrefix(won.MetadataLocation, "s3://")) {
+			t.Errorf("losing create wrote %s", p)
+		}
+	}
+}
+
 func TestCreateViewRollsBackEntryWhenMetadataWriteFails(t *testing.T) {
 	const bucket = "warehouse"
 	fc := newMemFiler()

weed/s3api/iceberg/iceberg_table_location_test.go

@@ -0,0 +1,220 @@
+package iceberg
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"path"
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+// A table created without a location gets a directory named after it and its
+// UUID, so it never lands on files another table left under its name.
+func TestCreateTableWithoutLocationUsesItsOwnDirectory(t *testing.T) {
+	const bucket = "warehouse"
+	for _, body := range []string{`{"name":"orders"}`, `{"name":"orders","location":"s3://warehouse"}`} {
+		fc := newMemFiler()
+		seedNamespace(fc, bucket, "finance", "alice")
+		s := NewServer(fc, nil)
+
+		w := httptest.NewRecorder()
+		s.handleCreateTable(w, newCreateTableRequest(t, bucket, "finance", body, "alice"))
+		if w.Code != http.StatusOK {
+			t.Fatalf("%s: status = %d (body: %s)", body, w.Code, w.Body.String())
+		}
+		var created struct {
+			MetadataLocation string `json:"metadata-location"`
+			Metadata         struct {
+				Location  string `json:"location"`
+				TableUUID string `json:"table-uuid"`
+			} `json:"metadata"`
+		}
+		if err := json.Unmarshal(w.Body.Bytes(), &created); err != nil {
+			t.Fatalf("decode: %v", err)
+		}
+		want := "s3://warehouse/finance/orders-" + created.Metadata.TableUUID
+		if created.Metadata.TableUUID == "" || created.Metadata.Location != want {
+			t.Errorf("%s: location = %q, want %q", body, created.Metadata.Location, want)
+		}
+		if !strings.HasPrefix(created.MetadataLocation, want+"/metadata/") {
+			t.Errorf("%s: metadata-location = %q, want one under %s", body, created.MetadataLocation, want)
+		}
+	}
+}
+
+func TestTableLocation(t *testing.T) {
+	for name, tc := range map[string]struct {
+		resp s3tables.GetTableResponse
+		want string
+	}{
+		"metadata":  {s3tables.GetTableResponse{MetadataLocation: "s3://b/ns/t-1/metadata/v2.metadata.json", WarehouseLocation: "s3://b/ns/t-0"}, "s3://b/ns/t-1"},
+		"warehouse": {s3tables.GetTableResponse{WarehouseLocation: "s3://b/ns/t-0/"}, "s3://b/ns/t-0"},
+		"legacy":    {s3tables.GetTableResponse{}, "s3://b/ns/t"},
+	} {
+		if got := tableLocation(tc.resp, "b", []string{"ns"}, "t"); got != tc.want {
+			t.Errorf("%s: tableLocation = %q, want %q", name, got, tc.want)
+		}
+	}
+}
+
+// Two creates can both find a name free. The one the catalog registers second
+// is a conflict and leaves nothing in the directory the server picked for it.
+func TestCreateTableThatLosesTheRaceIsConflict(t *testing.T) {
+	const bucket = "warehouse"
+	fc := newMemFiler()
+	seedNamespace(fc, bucket, "finance", "alice")
+	s := NewServer(fc, nil)
+	first := httptest.NewRecorder()
+	raceAtSecondLookup(fc, s3tables.GetTablePath(bucket, "finance", "orders"), func() {
+		s.handleCreateTable(first, newCreateTableRequest(t, bucket, "finance", `{"name":"orders"}`, "alice"))
+	})
+	second := httptest.NewRecorder()
+	s.handleCreateTable(second, newCreateTableRequest(t, bucket, "finance", `{"name":"orders"}`, "alice"))
+
+	won := decodeCreated(t, first)
+	if second.Code != http.StatusConflict {
+		t.Fatalf("losing create = %d (%s), want 409", second.Code, second.Body.String())
+	}
+	wonDir := path.Join(s3tables.TablesPath, strings.TrimPrefix(won.Metadata.Location, "s3://"))
+	for p := range fc.entries {
+		if strings.HasPrefix(p, s3tables.GetTablePath(bucket, "finance", "orders-")) && p != wonDir && !strings.HasPrefix(p, wonDir+"/") {
+			t.Errorf("losing create left %s", p)
+		}
+	}
+}
+
+// Two commits of one staged table pick the same metadata file name. The one
+// that finds the table registered, and moved on to newer metadata, fails
+// without removing the file the registered table's history still names.
+func TestStagedCommitThatLosesTheRaceFails(t *testing.T) {
+	const bucket = "warehouse"
+	fc := newMemFiler()
+	seedNamespace(fc, bucket, "finance", "alice")
+	s := NewServer(fc, nil)
+
+	tableUUID := uuid.New()
+	location := "s3://" + bucket + "/finance/orders-" + tableUUID.String()
+	base, err := newTableMetadata(tableUUID, location, nil, nil, nil, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	input := createOnCommitInput{
+		bucketARN:    buildTableBucketARN(bucket),
+		markerBucket: bucket,
+		namespace:    []string{"finance"},
+		tableName:    "orders",
+		identityName: "alice",
+		location:     location,
+		tableUUID:    tableUUID,
+		baseMetadata: base,
+	}
+	if _, reqErr := s.finalizeCreateOnCommit(context.Background(), input); reqErr != nil {
+		t.Fatalf("first commit: %+v", reqErr)
+	}
+	ctx := context.Background()
+	mgr := s3tables.NewManagerClient(fc)
+	var registered s3tables.GetTableResponse
+	if err := s.tablesManager.Execute(ctx, mgr, "GetTable", &s3tables.GetTableRequest{TableBucketARN: input.bucketARN, Namespace: input.namespace, Name: "orders"}, &registered, "alice"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.tablesManager.Execute(ctx, mgr, "UpdateTable", &s3tables.UpdateTableRequest{
+		TableBucketARN: input.bucketARN, Namespace: input.namespace, Name: "orders", VersionToken: registered.VersionToken,
+		MetadataVersion: 2, MetadataLocation: location + "/metadata/v2.metadata.json",
+	}, nil, "alice"); err != nil {
+		t.Fatal(err)
+	}
+
+	_, reqErr := s.finalizeCreateOnCommit(ctx, input)
+	if reqErr == nil || reqErr.status != http.StatusConflict {
+		t.Fatalf("second commit error = %+v, want a conflict", reqErr)
+	}
+	published := path.Join(s3tables.TablesPath, strings.TrimPrefix(location, "s3://"), "metadata", "v1.metadata.json")
+	if _, ok := fc.entries[published]; !ok {
+		t.Errorf("losing commit removed %s", published)
+	}
+}
+
+// Two clients can stage the same name, each at a location of its own. A commit
+// lands at the location of its own stage, never at the location of another
+// stage, even once its own marker is gone.
+func TestStagedCommitKeepsItsOwnLocation(t *testing.T) {
+	const bucket = "warehouse"
+	for _, ownMarkerGone := range []bool{false, true} {
+		fc := newMemFiler()
+		seedNamespace(fc, bucket, "finance", "alice")
+		s := NewServer(fc, nil)
+		stage := func() createdResult {
+			w := httptest.NewRecorder()
+			s.handleCreateTable(w, newCreateTableRequest(t, bucket, "finance", `{"name":"orders","stage-create":true}`, "alice"))
+			return decodeCreated(t, w)
+		}
+		first, second := stage(), stage()
+		if ownMarkerGone {
+			markerDir := stageCreateMarkerDir(bucket, []string{"finance"}, "orders")
+			for p, entry := range fc.entries {
+				if path.Dir(p) == markerDir && strings.Contains(string(entry.Content), first.Metadata.TableUUID) {
+					delete(fc.entries, p)
+				}
+			}
+		}
+
+		body := `{"requirements":[{"type":"assert-create"}],"updates":[{"action":"assign-uuid","uuid":"` + first.Metadata.TableUUID + `"}]}`
+		r := httptest.NewRequest(http.MethodPost, "/v1/"+bucket+"/namespaces/finance/tables/orders", strings.NewReader(body))
+		r = mux.SetURLVars(r, map[string]string{"prefix": bucket, "namespace": "finance", "table": "orders"})
+		r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice"))
+		w := httptest.NewRecorder()
+		s.handleUpdateTable(w, r)
+
+		committed := decodeCreated(t, w)
+		if strings.HasPrefix(committed.MetadataLocation, second.Metadata.Location+"/") {
+			t.Errorf("own marker gone %v: commit landed at the other stage's %s", ownMarkerGone, committed.MetadataLocation)
+		}
+		if !ownMarkerGone && (!strings.HasPrefix(committed.MetadataLocation, first.Metadata.Location+"/metadata/") || committed.Metadata.Location != first.Metadata.Location) {
+			t.Errorf("commit landed at %s (location %s), want under %s", committed.MetadataLocation, committed.Metadata.Location, first.Metadata.Location)
+		}
+	}
+}
+
+type createdResult struct {
+	MetadataLocation string `json:"metadata-location"`
+	Metadata         struct {
+		Location  string `json:"location"`
+		TableUUID string `json:"table-uuid"`
+		ViewUUID  string `json:"view-uuid"`
+	} `json:"metadata"`
+}
+
+func decodeCreated(t *testing.T, w *httptest.ResponseRecorder) createdResult {
+	t.Helper()
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d (body: %s)", w.Code, w.Body.String())
+	}
+	var created createdResult
+	if err := json.Unmarshal(w.Body.Bytes(), &created); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	return created
+}
+
+// raceAtSecondLookup runs create at the second lookup of record: after the
+// handler found the name free, as the catalog looks at it again.
+func raceAtSecondLookup(fc *memFiler, record string, create func()) {
+	seen := 0
+	fc.onLookup = func(p string) {
+		if p != record {
+			return
+		}
+		seen++
+		if seen == 2 {
+			fc.onLookup = nil
+			create()
+		}
+	}
+}

weed/s3api/iceberg/stage_create.go

@@ -90,7 +90,10 @@ func (s *Server) pruneExpiredStageCreateMarkers(ctx context.Context, bucketName
 	})
 }
 
-func (s *Server) loadLatestStageCreateMarker(ctx context.Context, bucketName string, namespace []string, tableName string) (*stageCreateMarker, error) {
+// loadLatestStageCreateMarker returns the newest live marker for the table.
+// Each stage has a location of its own, so given the table UUID a commit
+// assigns, only the marker staged with that UUID is returned.
+func (s *Server) loadLatestStageCreateMarker(ctx context.Context, bucketName string, namespace []string, tableName string, tableUUID string) (*stageCreateMarker, error) {
 	opCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
 
@@ -137,6 +140,9 @@ func (s *Server) loadLatestStageCreateMarker(ctx context.Context, bucketName str
 			if err != nil {
 				createdAt = time.Time{}
 			}
+			if tableUUID != "" && !strings.EqualFold(marker.TableUUID, tableUUID) {
+				continue
+			}
 			if latestMarker == nil || createdAt.After(latestCreatedAt) {
 				candidate := marker
 				latestMarker = &candidate
@@ -274,3 +280,17 @@ func isStageCreateEnabled() bool {
 		return true
 	}
 }
+
+// assignedTableUUID returns the UUID a commit's assign-uuid update names.
+func assignedTableUUID(updates []json.RawMessage) string {
+	for _, raw := range updates {
+		var update struct {
+			Action string `json:"action"`
+			UUID   string `json:"uuid"`
+		}
+		if json.Unmarshal(raw, &update) == nil && update.Action == "assign-uuid" {
+			return update.UUID
+		}
+	}
+	return ""
+}

weed/s3api/iceberg/utils.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"path"
 	"strconv"
 	"strings"
 
@@ -75,6 +76,19 @@ func tableLocationFromMetadataLocation(metadataLocation string) string {
 	return trimmed
 }
 
+// tableLocation is where the table GetTable answered for keeps its files: the
+// directory of its metadata, else the one assigned when it was created
+// without a location, else its name path.
+func tableLocation(getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) string {
+	if location := tableLocationFromMetadataLocation(getResp.MetadataLocation); location != "" {
+		return location
+	}
+	if location := strings.TrimSuffix(getResp.WarehouseLocation, "/"); location != "" {
+		return location
+	}
+	return fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
+}
+
 // accessDelegationHeader is how a client asks the catalog to hand back storage
 // credentials along with the table metadata.
 const accessDelegationHeader = "X-Iceberg-Access-Delegation"

weed/s3api/s3tables/filer_ops.go

@@ -999,15 +999,32 @@ func eachDirEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, dir s
 }
 
 // CatalogRecordRoot returns the directory the catalog record at recordPath
-// keeps its files under: the location it names inside the bucket, or its own
-// path.
-func CatalogRecordRoot(bucketName, recordPath, metadataLocation string) string {
-	if dir := TableDataDirFromMetadataLocation(metadataLocation); strings.HasPrefix(dir, GetTableBucketPath(bucketName)+"/") {
+// keeps its files under: the one its metadata location names inside the
+// bucket, else the warehouse location assigned at create, else its own path.
+func CatalogRecordRoot(bucketName, recordPath, metadataLocation, warehouseLocation string) string {
+	bucketPath := GetTableBucketPath(bucketName) + "/"
+	if dir := TableDataDirFromMetadataLocation(metadataLocation); strings.HasPrefix(dir, bucketPath) {
+		return dir
+	}
+	if dir := warehouseDir(warehouseLocation); strings.HasPrefix(dir, bucketPath) {
 		return dir
 	}
 	return recordPath
 }
 
+func warehouseDir(warehouseLocation string) string {
+	loc := strings.TrimPrefix(strings.TrimSuffix(warehouseLocation, "/"), "s3://")
+	if loc == "" {
+		return ""
+	}
+	return path.Join(TablesPath, loc)
+}
+
+// root returns the directory the record at recordPath keeps its files under.
+func (m *tableMetadataInternal) root(bucketName, recordPath string) string {
+	return CatalogRecordRoot(bucketName, recordPath, m.MetadataLocation, m.WarehouseLocation)
+}
+
 // CatalogRecordRoots maps every table and view record of the bucket, in any
 // namespace, to the directories it keeps its files under: the one its metadata
 // gives it and, while an update moves it, the one it moves to. A renamed record
@@ -1036,10 +1053,11 @@ func CatalogRecordRoots(ctx context.Context, client filer_pb.SeaweedFilerClient,
 					continue
 				}
 				var record struct {
-					MetadataLocation string `json:"metadataLocation"`
+					MetadataLocation  string `json:"metadataLocation"`
+					WarehouseLocation string `json:"warehouseLocation"`
 				}
 				_ = json.Unmarshal(data, &record)
-				roots[recordPath] = append(roots[recordPath], CatalogRecordRoot(bucketName, recordPath, record.MetadataLocation))
+				roots[recordPath] = append(roots[recordPath], CatalogRecordRoot(bucketName, recordPath, record.MetadataLocation, record.WarehouseLocation))
 			}
 			return true
 		}); err != nil {
@@ -1073,27 +1091,27 @@ func RootClaimed(roots map[string][]string, selfPath, root string) bool {
 	return false
 }
 
-// checkLocationUnclaimed returns ErrLocationInUse when the root the record at
-// recordPath would have with metadataLocation is claimed by another record.
-func checkLocationUnclaimed(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, recordPath, metadataLocation string) error {
+// checkLocationUnclaimed returns ErrLocationInUse when root, the one the record
+// at recordPath would have, is claimed by another record.
+func checkLocationUnclaimed(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, recordPath, root string) error {
 	roots, err := CatalogRecordRoots(ctx, client, bucketName)
 	if err != nil {
 		return err
 	}
-	if RootClaimed(roots, recordPath, CatalogRecordRoot(bucketName, recordPath, metadataLocation)) {
+	if RootClaimed(roots, recordPath, root) {
 		return ErrLocationInUse
 	}
 	return nil
 }
 
-// claimCatalogEntryAt claims path like claimCatalogEntry, then checks the root
-// metadataLocation gives it again with the record written: of two creates
-// racing for one root, at least one sees the other and backs out.
-func (h *S3TablesHandler) claimCatalogEntryAt(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, path, metadataLocation string, attrs map[string][]byte, emptyDirs ...string) error {
+// claimCatalogEntryAt claims path like claimCatalogEntry, then checks its root
+// again with the record written: of two creates racing for one root, at least
+// one sees the other and backs out.
+func (h *S3TablesHandler) claimCatalogEntryAt(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, path, root string, attrs map[string][]byte, emptyDirs ...string) error {
 	if err := h.claimCatalogEntry(ctx, client, path, attrs); err != nil {
 		return err
 	}
-	if err := checkLocationUnclaimed(ctx, client, bucketName, path, metadataLocation); err != nil {
+	if err := checkLocationUnclaimed(ctx, client, bucketName, path, root); err != nil {
 		// The claim still carries its mark, so no reader has seen the record.
 		return h.backOutChild(ctx, client, path, attrs, false, false, emptyDirs, err)
 	}
@@ -1106,7 +1124,7 @@ func (h *S3TablesHandler) claimCatalogEntryAt(ctx context.Context, client filer_
 // one is checked again. Of two writers racing for it at least one sees the
 // other, and the old root stays held until the move lands. A move that backs
 // out drops the reservation and leaves the metadata as it was.
-func (h *S3TablesHandler) moveCatalogRecord(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, path, metadataLocation string, stored, moving []byte, write func(extended map[string][]byte) error) error {
+func (h *S3TablesHandler) moveCatalogRecord(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, path, root string, stored, moving []byte, write func(extended map[string][]byte) error) error {
 	if err := h.mutateCatalogRecord(ctx, client, path, func(extended map[string][]byte) error {
 		if !bytes.Equal(extended[ExtendedKeyMetadata], stored) {
 			return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata)
@@ -1119,7 +1137,7 @@ func (h *S3TablesHandler) moveCatalogRecord(ctx context.Context, client filer_pb
 	}); err != nil {
 		return err
 	}
-	err := checkLocationUnclaimed(ctx, client, bucketName, path, metadataLocation)
+	err := checkLocationUnclaimed(ctx, client, bucketName, path, root)
 	if err == nil {
 		err = h.mutateCatalogRecord(ctx, client, path, func(extended map[string][]byte) error {
 			if !bytes.Equal(extended[ExtendedKeyMoving], moving) {
@@ -1144,9 +1162,9 @@ func (h *S3TablesHandler) moveCatalogRecord(ctx context.Context, client filer_pb
 
 // refuseClaimedLocation answers a request whose record would share its files
 // with another table or view of the bucket.
-func (h *S3TablesHandler) refuseClaimedLocation(w http.ResponseWriter, r *http.Request, filerClient FilerClient, bucketName, recordPath, metadataLocation string) error {
+func (h *S3TablesHandler) refuseClaimedLocation(w http.ResponseWriter, r *http.Request, filerClient FilerClient, bucketName, recordPath, root string) error {
 	err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return checkLocationUnclaimed(r.Context(), client, bucketName, recordPath, metadataLocation)
+		return checkLocationUnclaimed(r.Context(), client, bucketName, recordPath, root)
 	})
 	if errors.Is(err, ErrLocationInUse) {
 		h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error())

weed/s3api/s3tables/handler_bucket_keeps_data_test.go

@@ -485,7 +485,7 @@ func TestDeleteTableBucketKeepsALeftoverTakenBackAfterItsClaimIsLifted(t *testin
 	assert.Contains(t, fs.Get(bucketPath, "ns").Extended, ExtendedKeyMetadata)
 	require.NotNil(t, fs.Get(nsPath, "t"), "the table the create answered for must survive")
 	assert.Contains(t, fs.Get(nsPath, "t").Extended, ExtendedKeyMetadata)
-	assert.NotNil(t, fs.Get(path.Join(nsPath, "t"), "data"), "and so must its data directory")
+	assert.NotNil(t, fs.Get(recordRoot(t, fs, nsPath, "t"), "data"), "and so must its data directory")
 	assert.NotNil(t, fs.Get(path.Join(bucketPath, "old"), keptDataFile), "the file under the other leftover must survive")
 	assert.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive")
 }

weed/s3api/s3tables/handler_location_owner_test.go

@@ -64,10 +64,7 @@ func TestLocationOwnedByOneRecord(t *testing.T) {
 		DestNamespace:   []string{"ns2"},
 		DestName:        "t",
 	})
-	t.Run("renamed table", func(t *testing.T) {
-		checkRefusals(t)
-		refused(t, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "t", Format: FormatIceberg})
-	})
+	t.Run("renamed table", checkRefusals)
 
 	t.Run("free locations", func(t *testing.T) {
 		requireOK(t, m, fs, "UpdateTable", &UpdateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "own", MetadataVersion: 2, MetadataLocation: renameLocation("ns/own/metadata/v2.metadata.json")})
@@ -214,7 +211,7 @@ func TestLocationRaceBacksOut(t *testing.T) {
 			var oldRootErr error
 			client := &rootRaceClient{SeaweedFilerClient: fs.Client, f: fs, dir: nsPath, name: "a", location: location}
 			client.then = func() {
-				oldRootErr = checkLocationUnclaimed(context.Background(), fs.Client, renameTestBucket, path.Join(nsPath, "b"), oldLocation)
+				oldRootErr = checkLocationUnclaimed(context.Background(), fs.Client, renameTestBucket, path.Join(nsPath, "b"), CatalogRecordRoot(renameTestBucket, path.Join(nsPath, "b"), oldLocation, ""))
 			}
 			status, resp := runHandler(t, m, client, operation, update)
 			assert.Equal(t, http.StatusConflict, status, "%+v", resp)
@@ -259,6 +256,7 @@ func TestLocationRaceBacksOut(t *testing.T) {
 	t.Run("rename", func(t *testing.T) {
 		fs, m := startRenameManager(t)
 		requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "c", Format: FormatIceberg})
+		dropWarehouseLocation(t, fs, nsPath, "c")
 		client := &rootRaceClient{SeaweedFilerClient: fs.Client, f: fs, dir: nsPath, name: "dest", location: renameLocation("ns/dest/metadata/v1.metadata.json")}
 		status, resp := runHandler(t, m, client, "RenameTable", &RenameTableRequest{
 			TableBucketARN: mustBucketARN(t), SourceNamespace: []string{"ns"}, SourceName: "c", DestNamespace: []string{"ns"}, DestName: "dest",
@@ -297,6 +295,30 @@ func TestMoveReservationLeftBehind(t *testing.T) {
 	assert.NotContains(t, fs.Get(nsPath, "t").Extended, ExtendedKeyMoving)
 }
 
+// recordRoot returns the directory the record at dir/name keeps its files under.
+func recordRoot(t *testing.T, fs *s3tablestest.MemFiler, dir, name string) string {
+	t.Helper()
+	entry := fs.Get(dir, name)
+	require.NotNil(t, entry)
+	var metadata tableMetadataInternal
+	require.NoError(t, json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata))
+	return metadata.root(renameTestBucket, path.Join(dir, name))
+}
+
+// dropWarehouseLocation turns the record at dir/name into one written before
+// tables got a directory of their own, which keeps its files under its path.
+func dropWarehouseLocation(t *testing.T, fs *s3tablestest.MemFiler, dir, name string) {
+	t.Helper()
+	entry := fs.Get(dir, name)
+	require.NotNil(t, entry)
+	var metadata tableMetadataInternal
+	require.NoError(t, json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata))
+	metadata.WarehouseLocation = ""
+	data, err := json.Marshal(&metadata)
+	require.NoError(t, err)
+	entry.Extended[ExtendedKeyMetadata] = data
+}
+
 func mustRoots(t *testing.T, fs *s3tablestest.MemFiler) map[string][]string {
 	t.Helper()
 	roots, err := CatalogRecordRoots(context.Background(), fs.Client, renameTestBucket)
@@ -340,12 +362,16 @@ func TestRecordGoneDropsItsMoveReservation(t *testing.T) {
 
 	t.Run("rename", func(t *testing.T) {
 		fs, m := startRenameManager(t)
-		reserve(t, fs, "t")
+		requireOK(t, m, fs, "CreateTable", &CreateTableRequest{
+			TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "r", Format: FormatIceberg,
+			MetadataLocation: renameLocation("ns/r-files/metadata/v1.metadata.json"),
+		})
+		reserve(t, fs, "r")
 		requireOK(t, m, fs, "RenameTable", &RenameTableRequest{
-			TableBucketARN: mustBucketARN(t), SourceNamespace: []string{"ns"}, SourceName: "t", DestNamespace: []string{"ns"}, DestName: "u",
+			TableBucketARN: mustBucketARN(t), SourceNamespace: []string{"ns"}, SourceName: "r", DestNamespace: []string{"ns"}, DestName: "u",
 		})
 		assert.NotContains(t, fs.Get(nsPath, "u").Extended, ExtendedKeyMoving)
-		reuse(t, m, fs, "t")
+		reuse(t, m, fs, "r")
 	})
 
 	t.Run("delete", func(t *testing.T) {

weed/s3api/s3tables/handler_table.go

@@ -14,6 +14,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/google/uuid"
+
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 )
@@ -112,7 +114,16 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
 		return err
 	}
-	if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, tablePath, req.MetadataLocation); err != nil {
+	// An Iceberg table created without a location gets a directory of its own,
+	// so its files never sit under a name a rename can hand to another table.
+	// A Lance dataset stays at its name path, where Lance clients and the
+	// vector engine look for it.
+	metadataLocation, warehouseLocation := req.MetadataLocation, ""
+	if metadataLocation == "" && req.Format == FormatIceberg {
+		warehouseLocation = fmt.Sprintf("s3://%s/%s/%s-%s", bucketName, namespaceName, tableName, uuid.NewString())
+	}
+	root := CatalogRecordRoot(bucketName, tablePath, metadataLocation, warehouseLocation)
+	if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, tablePath, root); err != nil {
 		return err
 	}
 
@@ -121,16 +132,17 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 	versionToken := generateVersionToken()
 
 	metadata := &tableMetadataInternal{
-		Name:             tableName,
-		Namespace:        namespaceName,
-		Format:           req.Format,
-		CreatedAt:        now,
-		ModifiedAt:       now,
-		OwnerAccountID:   target.ownerAccountID, // Inherit namespace owner for consistency
-		VersionToken:     versionToken,
-		MetadataVersion:  max(req.MetadataVersion, 1),
-		MetadataLocation: req.MetadataLocation,
-		Metadata:         req.Metadata,
+		Name:              tableName,
+		Namespace:         namespaceName,
+		Format:            req.Format,
+		CreatedAt:         now,
+		ModifiedAt:        now,
+		OwnerAccountID:    target.ownerAccountID, // Inherit namespace owner for consistency
+		VersionToken:      versionToken,
+		MetadataVersion:   max(req.MetadataVersion, 1),
+		MetadataLocation:  metadataLocation,
+		Metadata:          req.Metadata,
+		WarehouseLocation: warehouseLocation,
 	}
 
 	metadataBytes, err := json.Marshal(metadata)
@@ -154,7 +166,11 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		}
 
 		// Create data subdirectory for Iceberg files
-		dataPath := tablePath + "/data"
+		dataParent := tablePath
+		if warehouseLocation != "" {
+			dataParent = root
+		}
+		dataPath := dataParent + "/data"
 		if err := h.ensureDirectory(r.Context(), client, dataPath); err != nil {
 			return err
 		}
@@ -167,10 +183,10 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		if len(tagsBytes) > 0 {
 			attrs[ExtendedKeyTags] = tagsBytes
 		}
-		if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, tablePath, req.MetadataLocation, attrs, dataPath, tablePath); err != nil {
+		if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, tablePath, root, attrs, dataPath, dataParent, tablePath); err != nil {
 			return err
 		}
-		return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceCreatedAt(target.namespaceCreatedAt), tablePath, attrs, dataPath, tablePath)
+		return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceCreatedAt(target.namespaceCreatedAt), tablePath, attrs, dataPath, dataParent, tablePath)
 	})
 
 	if err != nil {
@@ -292,7 +308,8 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req
 		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check table: %v", err))
 		return err
 	}
-	if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, tablePath, req.MetadataLocation); err != nil {
+	root := CatalogRecordRoot(bucketName, tablePath, req.MetadataLocation, "")
+	if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, tablePath, root); err != nil {
 		return err
 	}
 
@@ -325,7 +342,7 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req
 		if err := h.ensureDirectory(r.Context(), client, tablePath); err != nil {
 			return err
 		}
-		if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, tablePath, req.MetadataLocation, attrs, tablePath); err != nil {
+		if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, tablePath, root, attrs, tablePath); err != nil {
 			return err
 		}
 		return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceCreatedAt(target.namespaceCreatedAt), tablePath, attrs, tablePath)
@@ -500,17 +517,18 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request,
 	}
 
 	resp := &GetTableResponse{
-		Name:             metadata.Name,
-		TableARN:         tableARN,
-		Namespace:        expandNamespace(metadata.Namespace),
-		Format:           metadata.Format,
-		CreatedAt:        metadata.CreatedAt,
-		ModifiedAt:       metadata.ModifiedAt,
-		OwnerAccountID:   metadata.OwnerAccountID,
-		MetadataLocation: metadata.MetadataLocation,
-		MetadataVersion:  metadata.MetadataVersion,
-		VersionToken:     metadata.VersionToken,
-		Metadata:         metadata.Metadata,
+		Name:              metadata.Name,
+		TableARN:          tableARN,
+		Namespace:         expandNamespace(metadata.Namespace),
+		Format:            metadata.Format,
+		CreatedAt:         metadata.CreatedAt,
+		ModifiedAt:        metadata.ModifiedAt,
+		OwnerAccountID:    metadata.OwnerAccountID,
+		MetadataLocation:  metadata.MetadataLocation,
+		MetadataVersion:   metadata.MetadataVersion,
+		VersionToken:      metadata.VersionToken,
+		Metadata:          metadata.Metadata,
+		WarehouseLocation: metadata.WarehouseLocation,
 	}
 
 	h.writeJSON(w, http.StatusOK, resp)
@@ -1065,11 +1083,11 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
 
 	// Delete the table
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		if err := checkLocationUnclaimed(r.Context(), client, bucketName, tablePath, metadata.MetadataLocation); err != nil {
+		dataPath := metadata.root(bucketName, tablePath)
+		if err := checkLocationUnclaimed(r.Context(), client, bucketName, tablePath, dataPath); err != nil {
 			return err
 		}
-		dataPath := TableDataDirFromMetadataLocation(metadata.MetadataLocation)
-		if dataPath != "" && dataPath != tablePath && strings.HasPrefix(dataPath+"/", GetTableBucketPath(bucketName)+"/") {
+		if dataPath != tablePath {
 			// 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
 			// namespace root): the bucket-scope check above still admits the
@@ -1381,11 +1399,7 @@ func tablePurgeDir(entry *filer_pb.Entry, tablePath, bucketName string) (string,
 	if err := json.Unmarshal(data, &metadata); err != nil {
 		return "", fmt.Errorf("failed to unmarshal table metadata: %w", err)
 	}
-	dataPath := TableDataDirFromMetadataLocation(metadata.MetadataLocation)
-	if dataPath != "" && dataPath != tablePath && strings.HasPrefix(dataPath+"/", GetTableBucketPath(bucketName)+"/") {
-		return dataPath, nil
-	}
-	return tablePath, nil
+	return metadata.root(bucketName, tablePath), nil
 }
 
 // canReadEntry reports whether the principal may read the catalog entry a
@@ -1715,8 +1729,8 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ
 	if err := h.refuseClaimedNamespace(w, r, filerClient, destNamespacePath, destNamespace, isDestNamespace); err != nil {
 		return err
 	}
-	if metadata.MetadataLocation == "" {
-		if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, destPath, ""); err != nil {
+	if metadata.root(bucketName, srcPath) == srcPath {
+		if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, destPath, destPath); err != nil {
 			return err
 		}
 	}
@@ -1799,8 +1813,8 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ
 			h.removeLeftoverDirectory(ctx, client, destPath)
 			return releaseClaim(err)
 		}
-		if metadata.MetadataLocation == "" {
-			if err := checkLocationUnclaimed(r.Context(), client, bucketName, destPath, ""); err != nil {
+		if metadata.root(bucketName, srcPath) == srcPath {
+			if err := checkLocationUnclaimed(r.Context(), client, bucketName, destPath, destPath); err != nil {
 				return backOutDest(err)
 			}
 		}
@@ -2145,9 +2159,10 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 	if req.ClearFullMetadata && metadata.Metadata != nil {
 		metadata.Metadata.FullMetadata = nil
 	}
-	movesRoot := req.MetadataLocation != "" && CatalogRecordRoot(target.bucketName, target.tablePath, req.MetadataLocation) != CatalogRecordRoot(target.bucketName, target.tablePath, metadata.MetadataLocation)
+	newRoot := CatalogRecordRoot(target.bucketName, target.tablePath, req.MetadataLocation, metadata.WarehouseLocation)
+	movesRoot := req.MetadataLocation != "" && newRoot != metadata.root(target.bucketName, target.tablePath)
 	if movesRoot {
-		if err := h.refuseClaimedLocation(w, r, filerClient, target.bucketName, target.tablePath, req.MetadataLocation); err != nil {
+		if err := h.refuseClaimedLocation(w, r, filerClient, target.bucketName, target.tablePath, newRoot); err != nil {
 			return err
 		}
 	}
@@ -2192,7 +2207,7 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		if !movesRoot {
 			return h.mutateCatalogRecord(r.Context(), client, target.tablePath, write)
 		}
-		return h.moveCatalogRecord(r.Context(), client, target.bucketName, target.tablePath, metadata.MetadataLocation, target.storedMetadata, metadataBytes, write)
+		return h.moveCatalogRecord(r.Context(), client, target.bucketName, target.tablePath, newRoot, target.storedMetadata, metadataBytes, write)
 	})
 
 	if err != nil {

weed/s3api/s3tables/handler_view.go

@@ -138,7 +138,8 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
 		return err
 	}
-	if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, viewPath, req.MetadataLocation); err != nil {
+	root := CatalogRecordRoot(bucketName, viewPath, req.MetadataLocation, "")
+	if err := h.refuseClaimedLocation(w, r, filerClient, bucketName, viewPath, root); err != nil {
 		return err
 	}
 
@@ -172,7 +173,7 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques
 			ExtendedKeyEntryType: []byte(EntryTypeView),
 			ExtendedKeyDeleting:  []byte(createClaim),
 		}
-		if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, viewPath, req.MetadataLocation, attrs, viewPath); err != nil {
+		if err := h.claimCatalogEntryAt(r.Context(), client, bucketName, viewPath, root, attrs, viewPath); err != nil {
 			return err
 		}
 		return h.confirmParent(r.Context(), client, namespacePath, isNamespace, viewPath, attrs, viewPath)
@@ -558,9 +559,10 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques
 			metadata.Metadata.FullMetadata = req.Metadata.FullMetadata
 		}
 	}
-	movesRoot := req.MetadataLocation != "" && CatalogRecordRoot(target.bucketName, viewPath, req.MetadataLocation) != CatalogRecordRoot(target.bucketName, viewPath, metadata.MetadataLocation)
+	newRoot := CatalogRecordRoot(target.bucketName, viewPath, req.MetadataLocation, metadata.WarehouseLocation)
+	movesRoot := req.MetadataLocation != "" && newRoot != metadata.root(target.bucketName, viewPath)
 	if movesRoot {
-		if err := h.refuseClaimedLocation(w, r, filerClient, target.bucketName, viewPath, req.MetadataLocation); err != nil {
+		if err := h.refuseClaimedLocation(w, r, filerClient, target.bucketName, viewPath, newRoot); err != nil {
 			return err
 		}
 	}
@@ -600,7 +602,7 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques
 		if !movesRoot {
 			return h.mutateCatalogRecord(r.Context(), client, viewPath, write)
 		}
-		return h.moveCatalogRecord(r.Context(), client, target.bucketName, viewPath, metadata.MetadataLocation, target.storedMetadata, metadataBytes, write)
+		return h.moveCatalogRecord(r.Context(), client, target.bucketName, viewPath, newRoot, target.storedMetadata, metadataBytes, write)
 	})
 	if err != nil {
 		if errors.Is(err, ErrLocationInUse) {

weed/s3api/s3tables/handler_warehouse_location_test.go

@@ -0,0 +1,145 @@
+package s3tables
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"path"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+// A table created without a location keeps its files in a directory of its
+// own, so a rename leaves nothing under the old name and the name can be used
+// again without either table reaching the other's files.
+func TestCreateTableWithoutLocationGetsItsOwnDirectory(t *testing.T) {
+	fs, m := startRenameManager(t)
+	bucketARN := mustBucketARN(t)
+	nsPath := GetNamespacePath(renameTestBucket, "ns")
+
+	requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "w", Format: FormatIceberg})
+	var table GetTableResponse
+	getTable(t, m, fs, bucketARN, "w", &table)
+	assert.Empty(t, table.MetadataLocation)
+	assert.True(t, strings.HasPrefix(table.WarehouseLocation, renameLocation("ns/w-")), table.WarehouseLocation)
+
+	firstRoot := recordRoot(t, fs, nsPath, "w")
+	assert.Equal(t, path.Join(TablesPath, strings.TrimPrefix(table.WarehouseLocation, "s3://")), firstRoot)
+	require.NotNil(t, fs.Get(firstRoot, "data"))
+	fs.PutFile(path.Join(firstRoot, "data"), "part-0.parquet", time.Now())
+	assert.Nil(t, fs.Get(path.Join(nsPath, "w"), "data"), "the catalog entry holds no files")
+
+	requireOK(t, m, fs, "RenameTable", &RenameTableRequest{TableBucketARN: bucketARN, SourceNamespace: []string{"ns"}, SourceName: "w", DestNamespace: []string{"ns"}, DestName: "moved"})
+	assert.Equal(t, firstRoot, recordRoot(t, fs, nsPath, "moved"), "a rename keeps the directory")
+
+	requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "w", Format: FormatIceberg})
+	secondRoot := recordRoot(t, fs, nsPath, "w")
+	assert.NotEqual(t, firstRoot, secondRoot)
+	assert.False(t, PathsOverlap(firstRoot, secondRoot))
+
+	requireOK(t, m, fs, "DeleteTable", &DeleteTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "w"})
+	assert.NotNil(t, fs.Get(path.Join(firstRoot, "data"), "part-0.parquet"), "deleting the new table leaves the renamed one's files")
+	requireOK(t, m, fs, "DeleteTable", &DeleteTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "moved"})
+	assert.Nil(t, fs.Get(path.Dir(firstRoot), path.Base(firstRoot)), "deleting the renamed table purges its directory")
+}
+
+// A Lance dataset stays at its name path, where Lance clients and the vector
+// engine look for it.
+func TestCreateLanceTableWithoutLocationKeepsItsPath(t *testing.T) {
+	fs, m := startRenameManager(t)
+	bucketARN := mustBucketARN(t)
+	requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "l", Format: FormatLance})
+	var table GetTableResponse
+	getTable(t, m, fs, bucketARN, "l", &table)
+	assert.Empty(t, table.MetadataLocation)
+	assert.Empty(t, table.WarehouseLocation)
+	assert.Equal(t, GetTablePath(renameTestBucket, "ns", "l"), recordRoot(t, fs, GetNamespacePath(renameTestBucket, "ns"), "l"))
+}
+
+// Records written before tables got their own directory carry no warehouse
+// location and keep resolving to their name path, so nothing has to be
+// rewritten when a gateway is upgraded.
+func TestRecordWithoutWarehouseLocationKeepsItsPath(t *testing.T) {
+	fs, m := startRenameManager(t)
+	bucketARN := mustBucketARN(t)
+	nsPath := GetNamespacePath(renameTestBucket, "ns")
+	requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "old", Format: FormatIceberg})
+	dropWarehouseLocation(t, fs, nsPath, "old")
+	assert.Equal(t, path.Join(nsPath, "old"), recordRoot(t, fs, nsPath, "old"))
+	assert.Equal(t, path.Join(nsPath, "t"), recordRoot(t, fs, nsPath, "t"), "a metadata location still decides the root")
+
+	status, resp := runHandler(t, m, fs.Client, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "alias", Format: FormatIceberg, MetadataLocation: renameLocation("ns/old/metadata/v1.metadata.json")})
+	assert.Equal(t, http.StatusConflict, status, "%+v", resp)
+}
+
+// The first commit of a table writes its metadata into the assigned directory,
+// which is the root the record already holds, so it scans no other record.
+func TestFirstCommitIntoWarehouseLocationKeepsTheRoot(t *testing.T) {
+	fs, m := startRenameManager(t)
+	bucketARN := mustBucketARN(t)
+	nsPath := GetNamespacePath(renameTestBucket, "ns")
+	requireOK(t, m, fs, "CreateTable", &CreateTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "w", Format: FormatIceberg})
+	var table GetTableResponse
+	getTable(t, m, fs, bucketARN, "w", &table)
+	root := recordRoot(t, fs, nsPath, "w")
+
+	status, resp := runHandler(t, m, &noListClient{SeaweedFilerClient: fs.Client}, "UpdateTable", &UpdateTableRequest{
+		TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: "w", MetadataVersion: 1,
+		MetadataLocation: table.WarehouseLocation + "/metadata/v1.metadata.json",
+	})
+	require.Equal(t, http.StatusOK, status, "%+v", resp)
+	assert.Equal(t, root, recordRoot(t, fs, nsPath, "w"))
+	assert.NotContains(t, fs.Get(nsPath, "w").Extended, ExtendedKeyMoving)
+}
+
+// A create that backs out after making the table's directory removes it, so
+// failed creates leave nothing behind under the namespace.
+func TestBackedOutCreateRemovesItsDirectory(t *testing.T) {
+	fs, m := startEmptyBucket(t)
+	requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t))
+	bucketPath := GetTableBucketPath(renameTestBucket)
+	nsPath := GetNamespacePath(renameTestBucket, "ns")
+	recreated := recreatedNamespace(t, "ns")
+	swapped := false
+	fs.OnLookup = func(dir, name string) {
+		switch {
+		case dir == nsPath && name == "t" && !swapped:
+			swapped = true
+			fs.Put(bucketPath, "ns", recreated)
+		case path.Dir(dir) == nsPath && name == "data":
+			// The filer makes the parents of an entry it creates.
+			fs.Put(nsPath, path.Base(dir), nil)
+		}
+	}
+
+	code, resp := runHandler(t, m, fs.Client, "CreateTable", createTableRequest(t, "ns", "t"))
+	require.Equal(t, http.StatusNotFound, code, "%+v", resp)
+	fs.OnLookup = nil
+
+	var left []string
+	require.NoError(t, eachDirEntry(context.Background(), fs.Client, nsPath, func(entry *filer_pb.Entry) bool {
+		left = append(left, entry.Name)
+		return true
+	}))
+	assert.Empty(t, left)
+}
+
+func getTable(t *testing.T, m *Manager, fs *s3tablestest.MemFiler, bucketARN, name string, out *GetTableResponse) {
+	t.Helper()
+	require.NoError(t, m.Execute(context.Background(), NewManagerClient(fs.Client), "GetTable", &GetTableRequest{TableBucketARN: bucketARN, Namespace: []string{"ns"}, Name: name}, out, ""))
+}
+
+type noListClient struct {
+	filer_pb.SeaweedFilerClient
+}
+
+func (c *noListClient) ListEntries(context.Context, *filer_pb.ListEntriesRequest, ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	return nil, errors.New("listing unavailable")
+}

weed/s3api/s3tables/manager.go

@@ -137,7 +137,7 @@ func (m *Manager) AuthorizeCreateTable(ctx context.Context, filerClient FilerCli
 		tablePath := GetTablePath(target.bucketName, target.namespaceName, target.tableName)
 		if authErr = ValidateMetadataLocation(req.MetadataLocation, target.bucketName, target.namespaceName); authErr != nil {
 			m.handler.writeError(recorder, http.StatusBadRequest, ErrCodeInvalidRequest, authErr.Error())
-		} else if authErr = m.handler.refuseClaimedLocation(recorder, httpReq, filerClient, target.bucketName, tablePath, req.MetadataLocation); authErr == nil {
+		} else if authErr = m.handler.refuseClaimedLocation(recorder, httpReq, filerClient, target.bucketName, tablePath, CatalogRecordRoot(target.bucketName, tablePath, req.MetadataLocation, "")); authErr == nil {
 			return nil
 		}
 	}

weed/s3api/s3tables/types.go

@@ -243,6 +243,8 @@ type GetTableResponse struct {
 	VersionToken     string         `json:"versionToken"`
 	MetadataVersion  int            `json:"metadataVersion"`
 	Metadata         *TableMetadata `json:"metadata,omitempty"`
+	// WarehouseLocation is set on tables created without a location.
+	WarehouseLocation string `json:"warehouseLocation,omitempty"`
 }
 
 type ListTablesRequest struct {

weed/s3api/s3tables/utils.go

@@ -280,6 +280,9 @@ type tableMetadataInternal struct {
 	MetadataVersion  int            `json:"metadataVersion"`
 	MetadataLocation string         `json:"metadataLocation,omitempty"`
 	Metadata         *TableMetadata `json:"metadata,omitempty"`
+	// WarehouseLocation is where a table created without a location keeps
+	// its files. Records written before it existed keep them under their path.
+	WarehouseLocation string `json:"warehouseLocation,omitempty"`
 }
 
 // IsTableBucketEntry returns true when the entry is marked as a table bucket.