fix(iceberg): clean stale data before creating a table (#9074) (#9077)

* fix(iceberg): clean stale data before creating a table

CREATE TABLE AS from Trino fails against the SeaweedFS Iceberg REST
catalog with "Cannot create a table on a non-empty location". The
catalog assigns every new table the deterministic <ns>/<table> path,
and Trino's pre-write check rejects the CTAS whenever leftover objects
live there — typically files from a prior DROP that did not purge the
data, or an earlier aborted CTAS.

Make the catalog the authority for table existence: before writing any
metadata, look up the table in the S3 Tables catalog. If it is already
registered, return the existing definition (idempotent create). If it
is not registered, any objects still sitting at the target location
are stale, so purge them before proceeding. Live tables are never
touched — the cleanup path is guarded by the catalog lookup.

Fixes #9074

* test(trino): regression for create/drop/recreate table (#9074)

Exercises the exact sequence from the reported bug: CREATE TABLE without
an explicit location, INSERT, DROP, then CREATE again with the same name,
followed by a CTAS on top. Previously the recreate failed with
"Cannot create a table on a non-empty location" because stale data files
from the dropped table lingered at the deterministic <schema>/<table>
path. The test also asserts the recreated table does not see the dropped
data and that a drop/recreate CTAS cycle works.

* fix(iceberg): purge dropped table location on DROP TABLE

Recreate-at-same-location was failing with "Cannot create a table on a
non-empty location" because DROP TABLE only removed the catalog entry
and left data files behind. Trino's pre-write emptiness check then
rejected the subsequent CREATE.

Look up the table's storage location before deleting the catalog entry,
and after a successful DeleteTable, purge the filer subtree at that
location. The lookup uses the catalog — the authoritative owner of the
name→location mapping — so cleanup is gated on a live table having
existed; failed lookups skip cleanup and leave storage alone.

Also pin the regression test to an explicit, fixed location so the
recreate genuinely targets the same path the drop was supposed to free.

* refactor(iceberg): use errors.As in isNoSuchTableError

* fix(iceberg): drop create-preflight cleanup; harden cleanup path

- Remove the destructive cleanupStaleTableLocation call from the
  CreateTable preflight. Storage cleanup now lives exclusively on the
  DROP path, so CreateTable has no side effects on storage beyond the
  new table's own metadata write.
- Validate tablePath in cleanupStaleTableLocation by rejecting empty,
  ".", "..", or backslash-bearing segments before joining with the
  bucket prefix, so a crafted location cannot escape the bucket
  subtree. path.Clean would silently collapse traversal, so segments
  are checked raw.

* test(trino): defer table drops in recreate test for failure-safe cleanup

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)
+	}
+}

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(