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.