diff --git a/test/s3tables/catalog/duckdb_deterministic_location_test.go b/test/s3tables/catalog/duckdb_deterministic_location_test.go new file mode 100644 index 000000000..8cccdddad --- /dev/null +++ b/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 +// // 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 - 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 +// // 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] +} diff --git a/test/s3tables/catalog_spark/spark_deterministic_location_test.go b/test/s3tables/catalog_spark/spark_deterministic_location_test.go new file mode 100644 index 000000000..f88618b34 --- /dev/null +++ b/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 / 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 + // // 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 - suffix)", k, table) + } + } +} diff --git a/test/s3tables/catalog_trino/trino_catalog_test.go b/test/s3tables/catalog_trino/trino_catalog_test.go index e937f66a5..7645dd2a8 100644 --- a/test/s3tables/catalog_trino/trino_catalog_test.go +++ b/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 / 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 is the deterministic +// / 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() diff --git a/test/s3tables/catalog_trino/trino_crud_operations_test.go b/test/s3tables/catalog_trino/trino_crud_operations_test.go index e33d03e1c..4b2eecdf0 100644 --- a/test/s3tables/catalog_trino/trino_crud_operations_test.go +++ b/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) diff --git a/test/s3tables/catalog_trino/trino_deterministic_location_test.go b/test/s3tables/catalog_trino/trino_deterministic_location_test.go new file mode 100644 index 000000000..7cb2e943b --- /dev/null +++ b/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 share a path, no +// sibling `
-/` 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 // 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 +// /
/ (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 +// /
-/, 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 - suffix)", k, table) + } + } +}