diff --git a/test/s3tables/catalog_trino/trino_recreate_table_test.go b/test/s3tables/catalog_trino/trino_recreate_table_test.go new file mode 100644 index 000000000..91e95400a --- /dev/null +++ b/test/s3tables/catalog_trino/trino_recreate_table_test.go @@ -0,0 +1,120 @@ +package catalog_trino + +import ( + "fmt" + "strings" + "testing" +) + +// TestTrinoCreateDropRecreateTable is a regression test for +// https://github.com/seaweedfs/seaweedfs/issues/9074 +// +// Trino CTAS was failing with "Cannot create a table on a non-empty +// location" because Trino's pre-write check saw leftover data files from +// an earlier table at the same S3 path — DROP TABLE removed the catalog +// entry but left the underlying files behind. The Iceberg REST server now +// treats the S3 Tables catalog as the authority on table existence: when a +// DROP succeeds, it purges the storage under the table's location so that +// a subsequent CREATE at the same path finds it empty. +// +// This test pins an explicit location (so the recreated table lands at the +// exact same path, matching what the bug reporter saw), then walks through: +// 1. CREATE TABLE with explicit location +// 2. INSERT rows (produces data files under that location) +// 3. DROP TABLE (catalog entry removed, data files should also be purged) +// 4. CREATE TABLE again at the same explicit location — previously failed +// with "Cannot create a table on a non-empty location" +// 5. CREATE TABLE AS SELECT on top of the recreated table, which is the +// exact operation reported in the issue. +func TestTrinoCreateDropRecreateTable(t *testing.T) { + env := setupTrinoTest(t) + defer env.Cleanup(t) + + tableBucket := "iceberg-tables" + schemaName := "recreate_" + randomString(6) + tableName := "events_" + randomString(6) + ctasName := "events_ctas_" + randomString(6) + tableLocation := fmt.Sprintf("s3://%s/%s/%s", tableBucket, schemaName, tableName) + ctasLocation := fmt.Sprintf("s3://%s/%s/%s", tableBucket, schemaName, ctasName) + qualified := fmt.Sprintf("iceberg.%s.%s", schemaName, tableName) + ctasQualified := fmt.Sprintf("iceberg.%s.%s", schemaName, ctasName) + + runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS iceberg.%s", schemaName)) + defer runTrinoSQLAllowNamespaceNotEmpty(t, env.trinoContainer, fmt.Sprintf("DROP SCHEMA IF EXISTS iceberg.%s", schemaName)) + // Registered after the DROP SCHEMA defer so LIFO ordering drops the + // tables before the schema. Without this, a failure between the + // CREATE TABLE below and the explicit DROP later in the test would + // leak the table and block the deferred schema drop. + defer runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("DROP TABLE IF EXISTS %s", qualified)) + defer runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("DROP TABLE IF EXISTS %s", ctasQualified)) + + createSQL := fmt.Sprintf(`CREATE TABLE %s ( + id INTEGER, + label VARCHAR + ) WITH ( + format = 'PARQUET', + location = '%s' + )`, qualified, tableLocation) + + t.Logf(">>> CREATE #1: %s (location=%s)", qualified, tableLocation) + runTrinoSQL(t, env.trinoContainer, createSQL) + + t.Logf(">>> INSERT into %s", qualified) + runTrinoSQL(t, env.trinoContainer, fmt.Sprintf( + "INSERT INTO %s VALUES (1, 'alpha'), (2, 'beta')", qualified)) + + countOutput := runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SELECT count(*) FROM %s", qualified)) + if got := mustParseCSVInt64(t, countOutput); got != 2 { + t.Fatalf("after first insert: expected 2 rows, got %d", got) + } + + t.Logf(">>> DROP %s", qualified) + runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("DROP TABLE %s", qualified)) + + // The recreate must succeed at the exact same location; DROP is expected + // to have purged the data files so Trino's pre-write check sees an empty + // path. + t.Logf(">>> CREATE #2 at the same location: %s", qualified) + runTrinoSQL(t, env.trinoContainer, createSQL) + + t.Logf(">>> INSERT into recreated %s", qualified) + runTrinoSQL(t, env.trinoContainer, fmt.Sprintf( + "INSERT INTO %s VALUES (10, 'gamma')", qualified)) + + countOutput = runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SELECT count(*) FROM %s", qualified)) + if got := mustParseCSVInt64(t, countOutput); got != 1 { + t.Fatalf("after recreate+insert: expected 1 row (not %d) — recreated table should not see dropped data", got) + } + + labelOutput := runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SELECT label FROM %s WHERE id = 10", qualified)) + if !strings.Contains(labelOutput, "gamma") { + t.Fatalf("recreated table missing expected row; got:\n%s", labelOutput) + } + + // CTAS on top of the recreated table is the exact operation from #9074. + ctasSQL := fmt.Sprintf(`CREATE TABLE %s +WITH ( + format = 'PARQUET', + location = '%s' +) +AS SELECT * FROM %s`, ctasQualified, ctasLocation, qualified) + + t.Logf(">>> CTAS %s FROM %s", ctasQualified, qualified) + runTrinoSQL(t, env.trinoContainer, ctasSQL) + + countOutput = runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SELECT count(*) FROM %s", ctasQualified)) + if got := mustParseCSVInt64(t, countOutput); got != 1 { + t.Fatalf("CTAS target: expected 1 row, got %d", got) + } + + // Drop the CTAS target and recreate it at the same path to make sure + // the drop-cleanup path is exercised for CTAS too. + t.Logf(">>> DROP then CTAS-recreate %s at the same location", ctasQualified) + runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("DROP TABLE %s", ctasQualified)) + runTrinoSQL(t, env.trinoContainer, ctasSQL) + + countOutput = runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SELECT count(*) FROM %s", ctasQualified)) + if got := mustParseCSVInt64(t, countOutput); got != 1 { + t.Fatalf("CTAS after drop-recreate: expected 1 row, got %d", got) + } +} diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index a4987d197..da36dc8d3 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -1,6 +1,7 @@ package iceberg import ( + "context" "encoding/json" "errors" "fmt" @@ -170,6 +171,35 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { return } + // Authoritative existence check: ask the catalog whether a table is registered + // at this name. If it is, short-circuit with the existing table (idempotent + // CreateTable). Any leftover objects at the target path from a previous + // lifecycle of the same name are purged by handleDropTable on the way out — + // we deliberately do not clean storage here to keep CreateTable free of + // destructive side effects. See issue #9074. + existsReq := &s3tables.GetTableRequest{ + TableBucketARN: bucketARN, + Namespace: namespace, + Name: tableName, + } + var existsResp s3tables.GetTableResponse + existsErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", existsReq, &existsResp, identityName) + }) + if existsErr == nil { + // Table already registered. Return the existing definition so CTAS/IF NOT + // EXISTS flows see a stable response instead of a 409. + result := buildLoadTableResult(existsResp, bucketName, namespace, tableName) + writeJSON(w, http.StatusOK, result) + return + } + if !isNoSuchTableError(existsErr) { + glog.V(1).Infof("Iceberg: CreateTable existence check failed for %s.%s: %v", flattenNamespacePath(namespace), tableName, existsErr) + writeError(w, http.StatusInternalServerError, "InternalServerError", existsErr.Error()) + return + } + // Stage-create persists metadata in the internal staged area and skips S3Tables registration. if req.StageCreate { stagedTablePath := stageCreateStagedTablePath(namespace, tableName, tableUUID) @@ -412,6 +442,25 @@ func (s *Server) handleDropTable(w http.ResponseWriter, r *http.Request) { // Extract identity from context identityName := s3_constants.GetIdentityNameFromContext(r) + // Resolve the table's storage location from the catalog before the + // delete, so we can purge data files afterwards. The catalog is the + // authoritative owner of this mapping — if the table is not registered, + // there is nothing to clean up and DeleteTable will return NoSuchTable. + var storedMetadataLocation string + lookupReq := &s3tables.GetTableRequest{ + TableBucketARN: bucketARN, + Namespace: namespace, + Name: tableName, + } + var lookupResp s3tables.GetTableResponse + lookupErr := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "GetTable", lookupReq, &lookupResp, identityName) + }) + if lookupErr == nil { + storedMetadataLocation = lookupResp.MetadataLocation + } + deleteReq := &s3tables.DeleteTableRequest{ TableBucketARN: bucketARN, Namespace: namespace, @@ -433,9 +482,76 @@ func (s *Server) handleDropTable(w http.ResponseWriter, r *http.Request) { return } + // Purge data files that lived under the dropped table's location, so a + // subsequent CREATE TABLE at the same name (issue #9074) does not trip + // Trino's "non-empty location" pre-check. We only purge when the catalog + // told us a location: if GetTable above failed, there is no mapping to + // trust and we leave storage alone. + if storedMetadataLocation != "" { + if tableLoc := tableLocationFromMetadataLocation(storedMetadataLocation); tableLoc != "" { + if dataBucket, dataPath, parseErr := parseS3Location(tableLoc); parseErr == nil { + if cleanupErr := s.cleanupStaleTableLocation(r.Context(), dataBucket, dataPath); cleanupErr != nil { + glog.V(1).Infof("Iceberg: failed to purge dropped table location s3://%s/%s: %v", dataBucket, dataPath, cleanupErr) + } + } + } + } + w.WriteHeader(http.StatusNoContent) } +// isNoSuchTableError reports whether an error from the S3 Tables manager +// indicates the target table is not registered in the catalog. +func isNoSuchTableError(err error) bool { + if err == nil { + return false + } + var tableErr *s3tables.S3TablesError + if errors.As(err, &tableErr) { + return tableErr.Type == s3tables.ErrCodeNoSuchTable || tableErr.Type == s3tables.ErrCodeNoSuchNamespace + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "not found") || strings.Contains(msg, "no such table") +} + +// cleanupStaleTableLocation purges any existing filer entry at the target +// table path inside the regular S3 bucket so that engines which verify an +// empty location (e.g. Trino CTAS) can proceed after a DROP. Callers must +// have confirmed via the catalog that no table is registered at this name — +// live tables must never be touched by this helper. Missing paths are not +// an error. +// +// The tablePath is validated to contain only safe, relative segments before +// being joined with the bucket prefix, so a maliciously crafted location +// (e.g. containing "..") cannot escape the bucket subtree and delete data +// outside of it. +func (s *Server) cleanupStaleTableLocation(ctx context.Context, bucketName, tablePath string) error { + tablePath = strings.Trim(tablePath, "/") + if bucketName == "" || tablePath == "" { + return nil + } + // Reject absolute paths and any traversal segments. path.Clean would + // silently collapse "foo/../bar" into "bar", masking an attempted + // escape, so we check each raw segment explicitly. + for _, segment := range strings.Split(tablePath, "/") { + if segment == "" || segment == "." || segment == ".." || strings.ContainsAny(segment, `\`) { + return fmt.Errorf("cleanupStaleTableLocation: refusing unsafe table path %q", tablePath) + } + } + parentDir := path.Join(s3_constants.DefaultBucketsPath, bucketName, path.Dir(tablePath)) + name := path.Base(tablePath) + if name == "" || name == "." || name == ".." || name == "/" { + return fmt.Errorf("cleanupStaleTableLocation: refusing unsafe table path %q", tablePath) + } + return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + err := filer_pb.DoRemove(ctx, client, parentDir, name, true, true, true, false, nil) + if err == nil || errors.Is(err, filer_pb.ErrNotFound) { + return nil + } + return err + }) +} + // newTableMetadata creates a new table.Metadata object with the given parameters. // Uses iceberg-go's MetadataBuilder pattern for proper spec compliance. func newTableMetadata(