From 4f24b756ba5c25ead29c65c31c1c2432da7c136d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 7 Nov 2026 16:37:31 -0800 Subject: [PATCH] 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:////-, 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 * 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 -, so a later rename never leaves its files under a name another table can take. Co-Authored-By: Claude Opus 5.5 * 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 * 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 * test: expect tables created without a location in their own directory Co-Authored-By: Claude Opus 5.5 * 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 * s3tables: remove the table directory when a create backs out Co-Authored-By: Claude Opus 5.5 * iceberg: answer a create that lost the race with the registered table Co-Authored-By: Claude Opus 5.5 * iceberg: answer a view create that lost the race with the registered view Co-Authored-By: Claude Opus 5.5 * iceberg: fail a staged commit whose name another create registered Co-Authored-By: Claude Opus 5.5 * iceberg: remove the directory of a create whose name another table took Co-Authored-By: Claude Opus 5.5 * iceberg: keep the metadata of a staged commit whose name is taken Co-Authored-By: Claude Opus 5.5 * test: write the lakekeeper table's data under its warehouse location Co-Authored-By: Claude Opus 5.5 * iceberg: commit a staged table at the location of its own stage Co-Authored-By: Claude Opus 5.5 * iceberg: never continue another stage when a commit's own marker is gone Co-Authored-By: Claude Opus 5.5 * iceberg: test filer honors exclusive creates and token-guarded deletes Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- .../duckdb_deterministic_location_test.go | 29 +-- .../catalog_olake/olake_catalog_test.go | 21 +- .../spark_deterministic_location_test.go | 41 ++-- test/s3tables/lakekeeper/lakekeeper_test.go | 22 +- weed/s3api/iceberg/handlers_commit.go | 7 +- weed/s3api/iceberg/handlers_table.go | 76 ++---- weed/s3api/iceberg/handlers_transaction.go | 6 +- weed/s3api/iceberg/handlers_view.go | 25 +- .../iceberg_create_table_conflict_test.go | 6 +- .../s3api/iceberg/iceberg_create_view_test.go | 53 ++++- .../iceberg/iceberg_table_location_test.go | 220 ++++++++++++++++++ weed/s3api/iceberg/stage_create.go | 22 +- weed/s3api/iceberg/utils.go | 14 ++ weed/s3api/s3tables/filer_ops.go | 56 +++-- .../handler_bucket_keeps_data_test.go | 2 +- .../s3tables/handler_location_owner_test.go | 42 +++- weed/s3api/s3tables/handler_table.go | 99 ++++---- weed/s3api/s3tables/handler_view.go | 12 +- .../handler_warehouse_location_test.go | 145 ++++++++++++ weed/s3api/s3tables/manager.go | 2 +- weed/s3api/s3tables/types.go | 2 + weed/s3api/s3tables/utils.go | 3 + 22 files changed, 703 insertions(+), 202 deletions(-) create mode 100644 weed/s3api/iceberg/iceberg_table_location_test.go create mode 100644 weed/s3api/s3tables/handler_warehouse_location_test.go diff --git a/test/s3tables/catalog/duckdb_deterministic_location_test.go b/test/s3tables/catalog/duckdb_deterministic_location_test.go index 256efb68c7..fa89fe2669 100644 --- a/test/s3tables/catalog/duckdb_deterministic_location_test.go +++ b/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 -// // 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 +// //-, 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 - 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", tableLoc, wantPrefix) } sqlContent := fmt.Sprintf(` diff --git a/test/s3tables/catalog_olake/olake_catalog_test.go b/test/s3tables/catalog_olake/olake_catalog_test.go index bbed0c2a3d..9eeefd3ca8 100644 --- a/test/s3tables/catalog_olake/olake_catalog_test.go +++ b/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 +// - 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) } diff --git a/test/s3tables/catalog_spark/spark_deterministic_location_test.go b/test/s3tables/catalog_spark/spark_deterministic_location_test.go index f88618b34c..ae1ae443b4 100644 --- a/test/s3tables/catalog_spark/spark_deterministic_location_test.go +++ b/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 / 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, /-, 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 - // // path, with no UUID-suffixed sibling. + // Layout assertion: each table's files land in one - + // 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 - suffix)", k, table) - } + if len(dirs) != 1 { + t.Fatalf("want the files of %q in one %s- directory, found %d; keys: %v", table, table, len(dirs), keys) } } diff --git a/test/s3tables/lakekeeper/lakekeeper_test.go b/test/s3tables/lakekeeper/lakekeeper_test.go index d3a23c76c4..17379a2521 100644 --- a/test/s3tables/lakekeeper/lakekeeper_test.go +++ b/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) diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index 66e5af7515..43a540c7da 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/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()) diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 6299ccbde3..d33a10d6ca 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/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 { diff --git a/weed/s3api/iceberg/handlers_transaction.go b/weed/s3api/iceberg/handlers_transaction.go index 4e090b55bc..9aab8985d4 100644 --- a/weed/s3api/iceberg/handlers_transaction.go +++ b/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()} diff --git a/weed/s3api/iceberg/handlers_view.go b/weed/s3api/iceberg/handlers_view.go index 0eb621b4dc..eae7473aee 100644 --- a/weed/s3api/iceberg/handlers_view.go +++ b/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, }) diff --git a/weed/s3api/iceberg/iceberg_create_table_conflict_test.go b/weed/s3api/iceberg/iceberg_create_table_conflict_test.go index a158a26ef6..53b0477de9 100644 --- a/weed/s3api/iceberg/iceberg_create_table_conflict_test.go +++ b/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) } diff --git a/weed/s3api/iceberg/iceberg_create_view_test.go b/weed/s3api/iceberg/iceberg_create_view_test.go index b08b40c5cc..83eb974724 100644 --- a/weed/s3api/iceberg/iceberg_create_view_test.go +++ b/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() diff --git a/weed/s3api/iceberg/iceberg_table_location_test.go b/weed/s3api/iceberg/iceberg_table_location_test.go new file mode 100644 index 0000000000..a4d18d67c4 --- /dev/null +++ b/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"}, ®istered, "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() + } + } +} diff --git a/weed/s3api/iceberg/stage_create.go b/weed/s3api/iceberg/stage_create.go index ccc53e1661..d94ee72c09 100644 --- a/weed/s3api/iceberg/stage_create.go +++ b/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 "" +} diff --git a/weed/s3api/iceberg/utils.go b/weed/s3api/iceberg/utils.go index 40bab282be..d3cde8c0a2 100644 --- a/weed/s3api/iceberg/utils.go +++ b/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" diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 17252ee6c8..50c59376f3 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/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()) diff --git a/weed/s3api/s3tables/handler_bucket_keeps_data_test.go b/weed/s3api/s3tables/handler_bucket_keeps_data_test.go index dfeef51285..63730e2cff 100644 --- a/weed/s3api/s3tables/handler_bucket_keeps_data_test.go +++ b/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") } diff --git a/weed/s3api/s3tables/handler_location_owner_test.go b/weed/s3api/s3tables/handler_location_owner_test.go index afba1584ff..9979502e3f 100644 --- a/weed/s3api/s3tables/handler_location_owner_test.go +++ b/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) { diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index a19d9cd8f5..c0205bf73f 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/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 { diff --git a/weed/s3api/s3tables/handler_view.go b/weed/s3api/s3tables/handler_view.go index b949bc2192..2bc8ffca0a 100644 --- a/weed/s3api/s3tables/handler_view.go +++ b/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) { diff --git a/weed/s3api/s3tables/handler_warehouse_location_test.go b/weed/s3api/s3tables/handler_warehouse_location_test.go new file mode 100644 index 0000000000..852060223e --- /dev/null +++ b/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") +} diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go index da32f986ce..c89b0349a3 100644 --- a/weed/s3api/s3tables/manager.go +++ b/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 } } diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index ede7cf6a9c..eba88f81e0 100644 --- a/weed/s3api/s3tables/types.go +++ b/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 { diff --git a/weed/s3api/s3tables/utils.go b/weed/s3api/s3tables/utils.go index 411505773c..9ef77bb2ca 100644 --- a/weed/s3api/s3tables/utils.go +++ b/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.