test(iceberg): cross-engine regression coverage for deterministic table locations (#9074) (#9253)

* test(trino): regression for unique-table-location=false (#9074)

With #9246's namespace-location property, Trino's REST catalog can
resolve table locations even when the connector is configured with
`iceberg.unique-table-location=false`, and CREATE/CTAS lands at the
deterministic <namespace-location>/<tableName> path with no
UUID-suffixed sibling. Lock that in:

- writeTrinoConfig now parameterizes the unique-table-location flag via
  a withDeterministicTableLocation() option.
- setupTrinoTest forwards config options.
- TestTrinoDeterministicLocationCTAS exercises a fresh CREATE TABLE +
  CTAS with the flag flipped off and asserts the on-disk layout has
  no UUID-suffixed sibling under the namespace dir, proving each table
  occupies a single dir.

Refs #9074

* test(spark): regression for CTAS without explicit location (#9074)

iceberg-spark has no equivalent of Trino's unique-table-location flag —
its REST catalog interactions always produce the deterministic
<namespace-location>/<tableName> path. Without #9246's namespace-location
property, Spark cannot resolve a table location for a CREATE TABLE that
omits an explicit LOCATION clause; with it, the operation succeeds and
the table lands at the expected single-dir-per-table layout.

TestSparkDeterministicLocationCTAS walks the same scenario as the Trino
test: CREATE TABLE without LOCATION, INSERT, CTAS, SELECT count, then
asserts via S3 ListObjectsV2 that no UUID-suffixed sibling directory
appears under the namespace.

Refs #9074

* test(duckdb): read table at deterministic location via REST catalog (#9074)

DuckDB's iceberg extension is a read-only consumer in this flow — there
is no client-side UUID-suffixing toggle to test. The relevant question
post-#9246 is whether DuckDB can ATTACH the REST catalog and resolve a
table at a deterministic <bucket>/<namespace>/<tableName> path produced
by writers that don't suffix UUIDs (iceberg-spark, pyiceberg, Trino with
unique-table-location=false).

TestDuckDBDeterministicLocationRead creates a namespace + minimal table
via direct REST API calls (so no client-side UUID is added), confirms
the catalog returns a deterministic location URL, then runs DuckDB
through ATTACH ... TYPE 'ICEBERG' and DESCRIBE on the table. Asserting
DESCRIBE succeeds proves DuckDB walked the catalog → metadata → schema
chain against the deterministic on-disk path.

The test skips gracefully when the DuckDB image lacks the iceberg
extension or the ATTACH-iceberg syntax, so older base images don't fail
the suite.

Refs #9074

test/s3tables/catalog/duckdb_deterministic_location_test.go

@@ -0,0 +1,211 @@
+package catalog
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+)
+
+// TestDuckDBDeterministicLocationRead validates that DuckDB's iceberg
+// extension can resolve and scan a table that lives at the deterministic
+// <bucket>/<namespace>/<tableName> path produced when no client-side UUID
+// is appended to the table location. This is the layout produced by:
+//
+//   - iceberg-spark (no concept of UUID-suffixed locations)
+//   - pyiceberg (no concept of UUID-suffixed locations)
+//   - Trino with iceberg.unique-table-location=false
+//
+// Post-#9246, REST clients can resolve table locations under a namespace
+// because GetNamespace advertises a default `location` property. This
+// test creates a namespace + table via the REST API directly (so no
+// client-side UUID is added), then asks DuckDB to ATTACH the catalog and
+// describe the table — verifying DuckDB walks the catalog → metadata →
+// schema chain successfully against a deterministic on-disk path.
+//
+// The test does not insert data files — DuckDB scanning an empty table
+// is sufficient to validate the catalog/metadata resolution path. If
+// DuckDB's iceberg extension is unavailable or the build is too old to
+// support the ATTACH syntax, the test skips rather than fails.
+func TestDuckDBDeterministicLocationRead(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+	if !testutil.HasDocker() {
+		t.Skip("Docker not available, skipping DuckDB integration test")
+	}
+
+	env := newOAuthTestEnv(t)
+	defer env.cleanup(t)
+	env.start(t)
+
+	bucketName := "duckdb-detloc-" + randomSuffix()
+	namespace := "detloc_ns_" + randomSuffix()
+	tableName := "detloc_tbl_" + randomSuffix()
+
+	createTableBucketViaShell(t, env, bucketName)
+
+	token := requestOAuthToken(t, env, env.accessKey, env.secretKey)
+	createNamespaceWithToken(t, env, token, bucketName, namespace)
+	createIcebergTableWithToken(t, env, token, bucketName, namespace, tableName)
+
+	// Sanity-check the catalog row exposes a deterministic location.
+	tableLoc := getTableLocation(t, env, token, bucketName, namespace, tableName)
+	wantPrefix := fmt.Sprintf("s3://%s/%s/%s", bucketName, namespace, tableName)
+	if !strings.HasPrefix(tableLoc, wantPrefix) {
+		t.Fatalf("table location %q should start with %q (deterministic, no UUID suffix)", tableLoc, wantPrefix)
+	}
+	if strings.Contains(strings.TrimPrefix(tableLoc, wantPrefix), "-") {
+		t.Fatalf("table location %q should not have a -<uuid> suffix after %q", tableLoc, wantPrefix)
+	}
+
+	sqlContent := fmt.Sprintf(`
+INSTALL iceberg;
+LOAD iceberg;
+
+CREATE SECRET iceberg_secret (
+    TYPE ICEBERG,
+    ENDPOINT 'http://host.docker.internal:%d',
+    CLIENT_ID '%s',
+    CLIENT_SECRET '%s'
+);
+
+CREATE SECRET s3_secret (
+    TYPE S3,
+    KEY_ID '%s',
+    SECRET '%s',
+    ENDPOINT 'host.docker.internal:%d',
+    URL_STYLE 'path',
+    USE_SSL false
+);
+
+ATTACH 's3://%s/' AS detloc_cat (TYPE 'ICEBERG', SECRET iceberg_secret);
+
+DESCRIBE detloc_cat.%s.%s;
+
+SELECT 'duckdb deterministic-location describe ok' AS marker;
+`,
+		env.icebergPort, env.accessKey, env.secretKey,
+		env.accessKey, env.secretKey, env.s3Port,
+		bucketName,
+		namespace, tableName,
+	)
+
+	sqlFile := filepath.Join(env.dataDir, "duckdb_detloc.sql")
+	if err := os.WriteFile(sqlFile, []byte(sqlContent), 0644); err != nil {
+		t.Fatalf("write SQL file: %v", err)
+	}
+
+	cmd := exec.Command("docker", "run", "--rm",
+		"-v", fmt.Sprintf("%s:/test", env.dataDir),
+		"--add-host", "host.docker.internal:host-gateway",
+		"--entrypoint", "duckdb",
+		"duckdb/duckdb:latest",
+		"-init", "/test/duckdb_detloc.sql",
+		"-c", "SELECT 1",
+	)
+	output, err := cmd.CombinedOutput()
+	outputStr := string(output)
+	t.Logf("DuckDB output:\n%s", outputStr)
+
+	if err != nil {
+		// Tolerate environments where the Iceberg extension/version is too
+		// limited to exercise this code path. The intent is to gain
+		// coverage where it's available, not to require it everywhere.
+		if strings.Contains(outputStr, "iceberg extension is not available") ||
+			strings.Contains(outputStr, "Failed to load") ||
+			strings.Contains(outputStr, "syntax error") ||
+			strings.Contains(outputStr, "Unknown extension") {
+			t.Skipf("DuckDB image lacks the required iceberg extension or syntax: %v", err)
+		}
+		t.Fatalf("DuckDB run failed: %v\nOutput:\n%s", err, outputStr)
+	}
+
+	if !strings.Contains(outputStr, "duckdb deterministic-location describe ok") {
+		t.Fatalf("DuckDB DESCRIBE on deterministic-location table did not complete successfully:\n%s", outputStr)
+	}
+}
+
+// createIcebergTableWithToken creates a minimal Iceberg table via the REST
+// catalog using the provided OAuth bearer token. The table is created
+// without an explicit location, so the server picks the deterministic
+// <bucket>/<namespace>/<tableName> path.
+func createIcebergTableWithToken(t *testing.T, env *oauthTestEnv, token, bucketName, namespace, tableName string) {
+	t.Helper()
+
+	body := fmt.Sprintf(`{
+  "name": %q,
+  "schema": {
+    "type": "struct",
+    "schema-id": 0,
+    "fields": [
+      {"id": 1, "name": "id", "required": true, "type": "long"},
+      {"id": 2, "name": "label", "required": false, "type": "string"}
+    ]
+  }
+}`, tableName)
+
+	url := fmt.Sprintf("%s/v1/%s/namespaces/%s/tables", env.icebergURL(), bucketName, namespace)
+	req, err := http.NewRequest(http.MethodPost, url, strings.NewReader(body))
+	if err != nil {
+		t.Fatalf("create table request: %v", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+token)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+	defer resp.Body.Close()
+
+	respBody, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("create table failed: status=%d body=%s", resp.StatusCode, respBody)
+	}
+	t.Logf("Created table %s.%s in bucket %s", namespace, tableName, bucketName)
+}
+
+// getTableLocation fetches the table's metadata via the REST catalog and
+// returns the location field reported in the response.
+func getTableLocation(t *testing.T, env *oauthTestEnv, token, bucketName, namespace, tableName string) string {
+	t.Helper()
+
+	url := fmt.Sprintf("%s/v1/%s/namespaces/%s/tables/%s", env.icebergURL(), bucketName, namespace, tableName)
+	req, err := http.NewRequest(http.MethodGet, url, nil)
+	if err != nil {
+		t.Fatalf("load table request: %v", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+token)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("load table: %v", err)
+	}
+	defer resp.Body.Close()
+
+	respBody, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("load table failed: status=%d body=%s", resp.StatusCode, respBody)
+	}
+	// The metadata.location field is nested. A naive substring match is
+	// sufficient for asserting the expected path shape.
+	body := string(respBody)
+	const key = `"location":"`
+	idx := strings.Index(body, key)
+	if idx < 0 {
+		t.Fatalf("LoadTable response missing location field: %s", body)
+	}
+	rest := body[idx+len(key):]
+	end := strings.Index(rest, `"`)
+	if end < 0 {
+		t.Fatalf("malformed location field: %s", body)
+	}
+	return rest[:end]
+}

test/s3tables/catalog_spark/spark_deterministic_location_test.go

@@ -0,0 +1,135 @@
+package catalog_spark
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+// TestSparkDeterministicLocationCTAS exercises Spark's iceberg-spark
+// connector against the SeaweedFS REST catalog with no explicit `LOCATION`
+// clause and no UUID-suffixing. iceberg-spark does not implement Trino's
+// unique-table-location feature, so every CREATE TABLE here produces a
+// deterministic <namespace-location>/<tableName> path. The test asserts
+// the fresh-CTAS pattern works AND the resulting on-disk layout has no
+// UUID-suffixed sibling — i.e. one directory per table.
+//
+// The fix in #9246 (advertise a default `location` namespace property)
+// is what allows iceberg-spark to resolve the table location at all when
+// the user did not pass one. Without it, a Spark-side CREATE TABLE on a
+// namespace with no `location` property fails to compute a path.
+func TestSparkDeterministicLocationCTAS(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	env, _, tableBucket := setupSparkTestEnv(t)
+
+	namespace := "detloc_" + randomString(6)
+	srcTable := "src_" + randomString(6)
+	ctasTable := "ctas_" + randomString(6)
+
+	sql := fmt.Sprintf(`
+spark.sql("CREATE NAMESPACE iceberg.%s")
+print("namespace created")
+
+spark.sql("""
+CREATE TABLE iceberg.%s.%s (
+    id INT,
+    name STRING
+) USING iceberg
+""")
+print("source table created")
+
+spark.sql("""
+INSERT INTO iceberg.%s.%s VALUES
+    (1, 'alpha'),
+    (2, 'beta'),
+    (3, 'gamma')
+""")
+print("data inserted")
+
+# CTAS without explicit LOCATION — the exact pattern that breaks on Trino
+# without #9246, and that iceberg-spark needs the namespace-location
+# property to resolve.
+spark.sql("""
+CREATE TABLE iceberg.%s.%s
+USING iceberg
+AS SELECT * FROM iceberg.%s.%s
+""")
+print("ctas created")
+
+count = spark.sql("SELECT COUNT(*) AS c FROM iceberg.%s.%s").collect()[0]['c']
+print(f"ctas row count: {count}")
+`,
+		namespace,
+		namespace, srcTable,
+		namespace, srcTable,
+		namespace, ctasTable, namespace, srcTable,
+		namespace, ctasTable,
+	)
+
+	output := runSparkPySQL(t, env.sparkContainer, sql, env.icebergRestPort, env.s3Port)
+
+	for _, want := range []string{"namespace created", "source table created", "data inserted", "ctas created", "ctas row count: 3"} {
+		if !strings.Contains(output, want) {
+			t.Fatalf("expected output to contain %q; full output:\n%s", want, output)
+		}
+	}
+
+	// Layout assertion: each table's data must land at the deterministic
+	// <namespace>/<tableName>/ path, with no UUID-suffixed sibling.
+	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)
+}
+
+func listSparkBucketKeysUnder(t *testing.T, env *TestEnvironment, bucket, prefix string) []string {
+	t.Helper()
+	cfg := aws.Config{
+		Region:       "us-east-1",
+		Credentials:  aws.NewCredentialsCache(credentials.NewStaticCredentialsProvider(env.accessKey, env.secretKey, "")),
+		BaseEndpoint: aws.String(fmt.Sprintf("http://localhost:%d", env.s3Port)),
+	}
+	client := s3.NewFromConfig(cfg, func(o *s3.Options) { o.UsePathStyle = true })
+	out, err := client.ListObjectsV2(context.Background(), &s3.ListObjectsV2Input{
+		Bucket: aws.String(bucket),
+		Prefix: aws.String(prefix),
+	})
+	if err != nil {
+		t.Fatalf("ListObjectsV2 prefix=%s: %v", prefix, err)
+	}
+	keys := make([]string, 0, len(out.Contents))
+	for _, obj := range out.Contents {
+		keys = append(keys, aws.ToString(obj.Key))
+	}
+	return keys
+}
+
+func expectKeyAtSparkLocation(t *testing.T, keys []string, namespace, table string) {
+	t.Helper()
+	want := namespace + "/" + table + "/"
+	for _, k := range keys {
+		if strings.HasPrefix(k, want) {
+			return
+		}
+	}
+	t.Fatalf("expected at least one object under %q; got keys: %v", want, keys)
+}
+
+func rejectUUIDSuffixedSparkSibling(t *testing.T, keys []string, namespace, table string) {
+	t.Helper()
+	prefix := namespace + "/" + table + "-"
+	for _, k := range keys {
+		if strings.HasPrefix(k, prefix) {
+			t.Fatalf("found UUID-suffixed sibling key %q for table %q (Spark should produce no -<uuid> suffix)", k, table)
+		}
+	}
+}

test/s3tables/catalog_trino/trino_catalog_test.go

@@ -406,12 +406,19 @@ func (env *TestEnvironment) writeTrinoConfig(t *testing.T, warehouseBucket strin
 		nestedLine = "\niceberg.rest-catalog.nested-namespace-enabled=true"
 	}
 
+	uniqueTableLocation := "true"
+	if o.deterministicTableLocation {
+		// Skip UUID suffixing in TrinoRestCatalog.createLocationForTable so each
+		// table lands at the deterministic <namespace-location>/<table> path.
+		uniqueTableLocation = "false"
+	}
+
 	config := fmt.Sprintf(`connector.name=iceberg
 iceberg.catalog.type=rest
 iceberg.rest-catalog.uri=http://host.docker.internal:%d
 iceberg.rest-catalog.warehouse=s3://%s%s
 iceberg.file-format=PARQUET
-iceberg.unique-table-location=true
+iceberg.unique-table-location=%s
 
 # S3 storage config
 fs.native-s3.enabled=true
@@ -424,7 +431,7 @@ s3.region=us-west-2
 
 # REST catalog authentication
 iceberg.rest-catalog.security=SIGV4
-`, env.icebergPort, warehouseBucket, nestedLine, env.s3Port, env.accessKey, env.secretKey)
+`, env.icebergPort, warehouseBucket, nestedLine, uniqueTableLocation, env.s3Port, env.accessKey, env.secretKey)
 
 	if err := os.WriteFile(filepath.Join(configDir, "iceberg.properties"), []byte(config), 0644); err != nil {
 		t.Fatalf("Failed to write Trino config: %v", err)
@@ -434,13 +441,21 @@ iceberg.rest-catalog.security=SIGV4
 }
 
 type trinoConfigOptions struct {
-	nestedNamespace bool
+	nestedNamespace            bool
+	deterministicTableLocation bool
 }
 
 func withNestedNamespace() func(*trinoConfigOptions) {
 	return func(o *trinoConfigOptions) { o.nestedNamespace = true }
 }
 
+// withDeterministicTableLocation flips Trino's iceberg.unique-table-location
+// flag to false so each table's <location> is the deterministic
+// <namespace-location>/<tableName> path instead of a UUID-suffixed one.
+func withDeterministicTableLocation() func(*trinoConfigOptions) {
+	return func(o *trinoConfigOptions) { o.deterministicTableLocation = true }
+}
+
 func (env *TestEnvironment) startTrinoContainer(t *testing.T, configDir string) {
 	t.Helper()
 

test/s3tables/catalog_trino/trino_crud_operations_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 // setupTrinoTest is a helper function that sets up the common test environment for all Trino CRUD tests
-func setupTrinoTest(t *testing.T) *TestEnvironment {
+func setupTrinoTest(t *testing.T, configOpts ...func(*trinoConfigOptions)) *TestEnvironment {
 	t.Helper()
 
 	if testing.Short() {
@@ -28,7 +28,7 @@ func setupTrinoTest(t *testing.T) *TestEnvironment {
 	catalogBucket := tableBucket
 	createTableBucket(t, env, tableBucket)
 
-	configDir := env.writeTrinoConfig(t, catalogBucket)
+	configDir := env.writeTrinoConfig(t, catalogBucket, configOpts...)
 	env.startTrinoContainer(t, configDir)
 	waitForTrino(t, env.trinoContainer, 60*time.Second)
 

test/s3tables/catalog_trino/trino_deterministic_location_test.go

@@ -0,0 +1,121 @@
+package catalog_trino
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+// TestTrinoDeterministicLocationCTAS exercises the Trino REST catalog with
+// `iceberg.unique-table-location=false`. With #9246 advertising a default
+// namespace `location` property, Trino's defaultTableLocation is non-null,
+// so even with the unique-table-location flag flipped off the CREATE flow
+// takes the deferred-transaction branch and the listFiles emptiness check
+// passes.
+//
+// The benefit users opt into by flipping the flag is a single dir per
+// table on disk (catalog entry and Iceberg <location> share a path, no
+// sibling `<table>-<uuid>/` directory). This test exercises a fresh CTAS
+// in that mode and asserts the on-disk layout has no UUID-suffixed sibling.
+func TestTrinoDeterministicLocationCTAS(t *testing.T) {
+	env := setupTrinoTest(t, withDeterministicTableLocation())
+	defer env.Cleanup(t)
+
+	schemaName := "detloc_" + randomString(6)
+	srcName := "src_" + randomString(6)
+	ctasName := "ctas_" + randomString(6)
+	srcQualified := fmt.Sprintf("iceberg.%s.%s", schemaName, srcName)
+	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))
+	defer runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("DROP TABLE IF EXISTS %s", srcQualified))
+	defer runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("DROP TABLE IF EXISTS %s", ctasQualified))
+
+	t.Logf(">>> CREATE source %s (no explicit location, unique-table-location=false)", srcQualified)
+	runTrinoSQL(t, env.trinoContainer, fmt.Sprintf(
+		"CREATE TABLE %s (id INTEGER, name VARCHAR)", srcQualified))
+
+	runTrinoSQL(t, env.trinoContainer, fmt.Sprintf(
+		"INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c')", srcQualified))
+
+	ctasSQL := fmt.Sprintf("CREATE TABLE %s AS SELECT * FROM %s", ctasQualified, srcQualified)
+	t.Logf(">>> CTAS (no location override): %s", ctasSQL)
+	runTrinoSQL(t, env.trinoContainer, ctasSQL)
+
+	countOutput := runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SELECT count(*) FROM %s", ctasQualified))
+	if got := mustParseCSVInt64(t, countOutput); got != 3 {
+		t.Fatalf("CTAS target: expected 3 rows, got %d", got)
+	}
+
+	out := runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SELECT name FROM %s ORDER BY id", ctasQualified))
+	for _, want := range []string{"a", "b", "c"} {
+		if !strings.Contains(out, want) {
+			t.Fatalf("CTAS target missing row %q; got:\n%s", want, out)
+		}
+	}
+
+	// Layout assertion: with unique-table-location=false, both tables must
+	// land at the deterministic <bucket>/<schema>/<tableName> path without
+	// any UUID-suffixed sibling directory under the namespace.
+	keys := listBucketKeysUnder(t, env, "iceberg-tables", schemaName+"/")
+	expectKeyAtLocation(t, keys, schemaName, srcName)
+	expectKeyAtLocation(t, keys, schemaName, ctasName)
+	rejectUUIDSuffixedSibling(t, keys, schemaName, srcName)
+	rejectUUIDSuffixedSibling(t, keys, schemaName, ctasName)
+}
+
+// listBucketKeysUnder lists every object key under the given bucket prefix
+// using the test S3 endpoint.
+func listBucketKeysUnder(t *testing.T, env *TestEnvironment, bucket, prefix string) []string {
+	t.Helper()
+	cfg := aws.Config{
+		Region:       "us-east-1",
+		Credentials:  aws.NewCredentialsCache(credentials.NewStaticCredentialsProvider(env.accessKey, env.secretKey, "")),
+		BaseEndpoint: aws.String(fmt.Sprintf("http://%s:%d", env.bindIP, env.s3Port)),
+	}
+	client := s3.NewFromConfig(cfg, func(o *s3.Options) { o.UsePathStyle = true })
+	out, err := client.ListObjectsV2(context.Background(), &s3.ListObjectsV2Input{
+		Bucket: aws.String(bucket),
+		Prefix: aws.String(prefix),
+	})
+	if err != nil {
+		t.Fatalf("ListObjectsV2 prefix=%s: %v", prefix, err)
+	}
+	keys := make([]string, 0, len(out.Contents))
+	for _, obj := range out.Contents {
+		keys = append(keys, aws.ToString(obj.Key))
+	}
+	return keys
+}
+
+// expectKeyAtLocation asserts at least one object key sits under
+// <schema>/<table>/ (any depth), proving the table has a deterministic path.
+func expectKeyAtLocation(t *testing.T, keys []string, schema, table string) {
+	t.Helper()
+	want := schema + "/" + table + "/"
+	for _, k := range keys {
+		if strings.HasPrefix(k, want) {
+			return
+		}
+	}
+	t.Fatalf("expected at least one object under %q; got keys: %v", want, keys)
+}
+
+// rejectUUIDSuffixedSibling asserts no object key sits under
+// <schema>/<table>-<hex>/, since a UUID-suffixed sibling would mean Trino
+// did add a UUID despite unique-table-location=false.
+func rejectUUIDSuffixedSibling(t *testing.T, keys []string, schema, table string) {
+	t.Helper()
+	prefix := schema + "/" + table + "-"
+	for _, k := range keys {
+		if strings.HasPrefix(k, prefix) {
+			t.Fatalf("found UUID-suffixed sibling key %q for table %q (unique-table-location=false should produce no -<uuid> suffix)", k, table)
+		}
+	}
+}