From 761ec7da00012dcff398bbe19ea8624d57a17f95 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 7 Apr 2026 12:21:22 -0700 Subject: [PATCH] fix(iceberg): use dot separator for namespace paths instead of unit separator (#8960) * fix(iceberg): use dot separator for namespace paths instead of unit separator The Iceberg REST Catalog handler was using \x1F (unit separator) to join multi-level namespaces when constructing S3 location and filer paths. The S3 Tables storage layer uses "." (dot) as the namespace separator, causing tables created via the Iceberg REST API to point to different paths than where S3 Tables actually stores them. Fixes #8959 * fix(iceberg): use dot separator in log messages for readable namespace output * fix(iceberg): use path.Join for S3 location path segments Use path.Join to construct the namespace/table path segments in fallback S3 locations for robustness and consistency with handleCreateTable. * test(iceberg): add multi-level namespace integration tests for Spark and Trino Add regression tests for #8959 that create a two-level namespace (e.g. "analytics.daily"), create a table under it, insert data, and query it back. This exercises the dot-separated namespace path construction and verifies that Spark/Trino can actually read the data at the S3 location returned by the Iceberg REST API. * fix(test): enable nested namespace in Trino Iceberg catalog config Trino requires `iceberg.rest-catalog.nested-namespace-enabled=true` to support multi-level namespaces. Without this, CREATE SCHEMA with a dotted name fails with "Nested namespace is not enabled for this catalog". * fix(test): parse Trino COUNT(*) output as integer instead of substring match Avoids false matches from strings.Contains(output, "3") by parsing the actual numeric result with strconv.Atoi and asserting equality. * fix(test): use separate Trino config for nested namespace test The nested-namespace-enabled=true setting in Trino changes how SHOW SCHEMAS works, causing "Internal error" for all tests sharing that catalog config. Move the flag to a dedicated config used only by TestTrinoMultiLevelNamespace. * fix(iceberg): support parent query parameter in ListNamespaces for nested namespaces Add handling for the Iceberg REST spec's `parent` query parameter in handleListNamespaces. When Trino has nested-namespace-enabled=true, it sends `GET /v1/namespaces?parent=` to list child namespaces. The parent value is decoded from the Iceberg unit separator format and converted to a dot-separated prefix for the S3 Tables layer. Also simplify TestTrinoMultiLevelNamespace to focus on namespace operations (create, list, show tables) rather than data operations, since Trino's REST catalog has a non-empty location check that conflicts with server-side metadata creation. * fix(test): expand Trino multi-level namespace test and merge config helpers - Expand TestTrinoMultiLevelNamespace to create a table with explicit location, insert rows, query them back, and verify the S3 file path contains the dot-separated namespace (not \x1F). This ensures the original #8959 bug would be caught by the Trino integration test. - Merge writeTrinoConfig and writeTrinoNestedNamespaceConfig into a single parameterized function using functional options. --- .../catalog_spark/spark_operations_test.go | 77 +++++++++++ .../catalog_trino/trino_catalog_test.go | 125 +++++++++++++++++- weed/s3api/iceberg/commit_helpers.go | 2 +- weed/s3api/iceberg/handlers_commit.go | 4 +- weed/s3api/iceberg/handlers_namespace.go | 11 ++ weed/s3api/iceberg/handlers_table.go | 8 +- weed/s3api/iceberg/utils.go | 10 +- 7 files changed, 225 insertions(+), 12 deletions(-) diff --git a/test/s3tables/catalog_spark/spark_operations_test.go b/test/s3tables/catalog_spark/spark_operations_test.go index 55437c8e7..c2c6c39ef 100644 --- a/test/s3tables/catalog_spark/spark_operations_test.go +++ b/test/s3tables/catalog_spark/spark_operations_test.go @@ -277,3 +277,80 @@ print(f"Count at snapshot: {count}") t.Logf(">>> Time travel test passed") } + +// TestSparkMultiLevelNamespace tests that multi-level namespaces produce correct +// S3 paths (dot-separated) so that Spark can read back the data it writes. +// Regression test for https://github.com/seaweedfs/seaweedfs/issues/8959 +func TestSparkMultiLevelNamespace(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + env, _, _ := setupSparkTestEnv(t) + + // Use a two-level namespace like "analytics.daily" + nsLevel1 := "analytics_" + randomString(4) + nsLevel2 := "daily_" + randomString(4) + multiNs := fmt.Sprintf("%s.%s", nsLevel1, nsLevel2) + tableName := "events_" + randomString(4) + + // Create multi-level namespace + t.Logf(">>> Creating multi-level namespace: %s", multiNs) + createNsSQL := fmt.Sprintf(` +spark.sql("CREATE NAMESPACE iceberg.%s") +print("Namespace created") +`, multiNs) + output := runSparkPySQL(t, env.sparkContainer, createNsSQL, env.icebergRestPort, env.s3Port) + if !strings.Contains(output, "Namespace created") { + t.Fatalf("multi-level namespace creation failed, output: %s", output) + } + + // Create table under multi-level namespace + t.Logf(">>> Creating table under multi-level namespace") + createTableSQL := fmt.Sprintf(` +spark.sql(""" +CREATE TABLE iceberg.%s.%s ( + id INT, + event STRING, + ts TIMESTAMP +) +USING iceberg +""") +print("Table created") +`, multiNs, tableName) + output = runSparkPySQL(t, env.sparkContainer, createTableSQL, env.icebergRestPort, env.s3Port) + if !strings.Contains(output, "Table created") { + t.Fatalf("table creation under multi-level namespace failed, output: %s", output) + } + + // Insert data + t.Logf(">>> Inserting data into multi-level namespace table") + insertSQL := fmt.Sprintf(` +spark.sql(""" +INSERT INTO iceberg.%s.%s VALUES + (1, 'click', TIMESTAMP '2025-01-01 00:00:00'), + (2, 'view', TIMESTAMP '2025-01-01 01:00:00'), + (3, 'click', TIMESTAMP '2025-01-02 00:00:00') +""") +print("Data inserted") +`, multiNs, tableName) + output = runSparkPySQL(t, env.sparkContainer, insertSQL, env.icebergRestPort, env.s3Port) + if !strings.Contains(output, "Data inserted") { + t.Fatalf("data insertion failed, output: %s", output) + } + + // Query data back — this is the key test: if the namespace path separator + // was wrong (\x1F instead of "."), Spark would not find the data files. + t.Logf(">>> Querying data from multi-level namespace table") + querySQL := fmt.Sprintf(` +result = spark.sql("SELECT COUNT(*) as count FROM iceberg.%s.%s") +count = result.collect()[0]['count'] +print(f"Row count: {count}") +`, multiNs, tableName) + output = runSparkPySQL(t, env.sparkContainer, querySQL, env.icebergRestPort, env.s3Port) + if !strings.Contains(output, "Row count: 3") { + t.Errorf("expected row count 3 from multi-level namespace table, got output: %s", output) + } + + t.Logf(">>> Multi-level namespace test passed") +} diff --git a/test/s3tables/catalog_trino/trino_catalog_test.go b/test/s3tables/catalog_trino/trino_catalog_test.go index d7fb47b7d..c32b06cc4 100644 --- a/test/s3tables/catalog_trino/trino_catalog_test.go +++ b/test/s3tables/catalog_trino/trino_catalog_test.go @@ -82,6 +82,101 @@ func TestTrinoIcebergCatalog(t *testing.T) { runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SHOW TABLES FROM iceberg.%s", schemaName)) } +// TestTrinoMultiLevelNamespace tests that multi-level namespaces (dot-separated) +// produce correct S3 paths so Trino can read back data it writes. +// Regression test for https://github.com/seaweedfs/seaweedfs/issues/8959 +func TestTrinoMultiLevelNamespace(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + env := NewTestEnvironment(t) + defer env.Cleanup(t) + + if !env.dockerAvailable { + t.Skip("Docker not available, skipping Trino integration test") + } + + t.Logf(">>> Starting SeaweedFS...") + env.StartSeaweedFS(t) + + tableBucket := "iceberg-tables" + createTableBucket(t, env, tableBucket) + + configDir := env.writeTrinoConfig(t, tableBucket, withNestedNamespace()) + env.startTrinoContainer(t, configDir) + waitForTrino(t, env.trinoContainer, 60*time.Second) + + // Use a two-level namespace: "analytics.daily" + nsLevel1 := "analytics_" + randomString(4) + nsLevel2 := "daily_" + randomString(4) + flatNs := fmt.Sprintf("%s.%s", nsLevel1, nsLevel2) + // Trino uses double-quoted schema names for multi-level namespaces + multiNs := fmt.Sprintf(`"%s"`, flatNs) + tableName := "events_" + randomString(4) + + // Create multi-level namespace (schema) + t.Logf(">>> Creating multi-level schema: %s", flatNs) + runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS iceberg.%s", multiNs)) + + // Verify the schema shows up + output := runTrinoSQL(t, env.trinoContainer, "SHOW SCHEMAS FROM iceberg") + if !strings.Contains(output, flatNs) { + t.Fatalf("Expected schema %s in output:\n%s", flatNs, output) + } + + // Create table with explicit location to avoid non-empty location conflict. + // The location uses the dot-separated namespace — if #8959 regresses + // (unit separator instead of dot), data would be written to the wrong path. + tableLocation := fmt.Sprintf("s3://%s/%s/%s_%s", tableBucket, flatNs, tableName, randomString(6)) + t.Logf(">>> Creating table at location: %s", tableLocation) + createSQL := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS iceberg.%s.%s ( + id INTEGER, + event VARCHAR, + ts TIMESTAMP(6) + ) WITH ( + format = 'PARQUET', + location = '%s' + )`, multiNs, tableName, tableLocation) + runTrinoSQLAllowExists(t, env.trinoContainer, createSQL) + + // Insert data + t.Logf(">>> Inserting data into multi-level namespace table") + runTrinoSQL(t, env.trinoContainer, fmt.Sprintf(` + INSERT INTO iceberg.%s.%s VALUES + (1, 'click', TIMESTAMP '2025-01-01 00:00:00'), + (2, 'view', TIMESTAMP '2025-01-01 01:00:00'), + (3, 'click', TIMESTAMP '2025-01-02 00:00:00') + `, multiNs, tableName)) + + // Query data back — if the namespace path separator were wrong (\x1F + // instead of "."), the metadata location would point to a non-existent + // S3 path and this query would fail. + t.Logf(">>> Querying data from multi-level namespace table") + countOutput := runTrinoSQL(t, env.trinoContainer, fmt.Sprintf( + "SELECT count(*) FROM iceberg.%s.%s", multiNs, tableName)) + rowCount := mustParseCSVInt64(t, countOutput) + if rowCount != 3 { + t.Fatalf("expected row count 3, got %d", rowCount) + } + + // Verify the S3 file path contains the dot-separated namespace, not \x1F. + filesOutput := runTrinoSQL(t, env.trinoContainer, fmt.Sprintf( + `SELECT file_path FROM iceberg.%s."%s$files" LIMIT 1`, multiNs, tableName)) + filePath := strings.TrimSpace(filesOutput) + if filePath == "" { + t.Fatalf("expected at least one data file, got empty output") + } + if !strings.Contains(filePath, flatNs+"/") { + t.Errorf("expected file path to contain dot-separated namespace %q, got: %s", flatNs, filePath) + } + if strings.Contains(filePath, "\x1F") { + t.Errorf("file path contains unit separator (\\x1F), expected dot separator: %s", filePath) + } + + t.Logf(">>> Trino multi-level namespace test passed") +} + func NewTestEnvironment(t *testing.T) *TestEnvironment { t.Helper() @@ -332,18 +427,32 @@ func testIcebergRestAPI(t *testing.T, env *TestEnvironment) { } } -func (env *TestEnvironment) writeTrinoConfig(t *testing.T, warehouseBucket string) string { +func (env *TestEnvironment) writeTrinoConfig(t *testing.T, warehouseBucket string, opts ...func(*trinoConfigOptions)) string { t.Helper() - configDir := filepath.Join(env.dataDir, "trino") + o := trinoConfigOptions{} + for _, fn := range opts { + fn(&o) + } + + dirName := "trino" + if o.nestedNamespace { + dirName = "trino-nested" + } + configDir := filepath.Join(env.dataDir, dirName) if err := os.MkdirAll(configDir, 0755); err != nil { t.Fatalf("Failed to create Trino config dir: %v", err) } + nestedLine := "" + if o.nestedNamespace { + nestedLine = "\niceberg.rest-catalog.nested-namespace-enabled=true" + } + 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 +iceberg.rest-catalog.warehouse=s3://%s%s iceberg.file-format=PARQUET iceberg.unique-table-location=true @@ -358,7 +467,7 @@ s3.region=us-west-2 # REST catalog authentication iceberg.rest-catalog.security=SIGV4 -`, env.icebergPort, warehouseBucket, env.s3Port, env.accessKey, env.secretKey) +`, env.icebergPort, warehouseBucket, nestedLine, 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) @@ -367,6 +476,14 @@ iceberg.rest-catalog.security=SIGV4 return configDir } +type trinoConfigOptions struct { + nestedNamespace bool +} + +func withNestedNamespace() func(*trinoConfigOptions) { + return func(o *trinoConfigOptions) { o.nestedNamespace = true } +} + func (env *TestEnvironment) startTrinoContainer(t *testing.T, configDir string) { t.Helper() diff --git a/weed/s3api/iceberg/commit_helpers.go b/weed/s3api/iceberg/commit_helpers.go index 742f24cae..1e4c912ba 100644 --- a/weed/s3api/iceberg/commit_helpers.go +++ b/weed/s3api/iceberg/commit_helpers.go @@ -200,7 +200,7 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi markerBucket = metadataBucket } if markerErr := s.deleteStageCreateMarkers(ctx, markerBucket, input.namespace, input.tableName); markerErr != nil { - glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after finalize: %v", encodeNamespace(input.namespace), input.tableName, markerErr) + glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after finalize: %v", flattenNamespacePath(input.namespace), input.tableName, markerErr) } return &CommitTableResponse{ diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index d6c264fce..571ad43dd 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/weed/s3api/iceberg/handlers_commit.go @@ -83,7 +83,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { }) if err != nil { if isS3TablesNotFound(err) { - location := fmt.Sprintf("s3://%s/%s/%s", bucketName, encodeNamespace(namespace), tableName) + location := fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName)) tableUUID := generatedLegacyUUID baseMetadataVersion := 0 baseMetadataLocation := "" @@ -195,7 +195,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { location := tableLocationFromMetadataLocation(getResp.MetadataLocation) if location == "" { - location = fmt.Sprintf("s3://%s/%s/%s", bucketName, encodeNamespace(namespace), tableName) + location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName)) } tableUUID := uuid.Nil if getResp.Metadata != nil && getResp.Metadata.Iceberg != nil && getResp.Metadata.Iceberg.TableUUID != "" { diff --git a/weed/s3api/iceberg/handlers_namespace.go b/weed/s3api/iceberg/handlers_namespace.go index 6b4b770bb..6c2246324 100644 --- a/weed/s3api/iceberg/handlers_namespace.go +++ b/weed/s3api/iceberg/handlers_namespace.go @@ -39,10 +39,21 @@ func (s *Server) handleListNamespaces(w http.ResponseWriter, r *http.Request) { return } + // The Iceberg REST spec allows a "parent" query parameter for hierarchical + // namespace listing. Convert it to the dot-separated prefix used by S3 Tables. + var prefix string + if parent := r.URL.Query().Get("parent"); parent != "" { + parentParts := parseNamespace(parent) + if len(parentParts) > 0 { + prefix = flattenNamespacePath(parentParts) + "." + } + } + // Use S3 Tables manager to list namespaces var resp s3tables.ListNamespacesResponse req := &s3tables.ListNamespacesRequest{ TableBucketARN: bucketARN, + Prefix: prefix, ContinuationToken: pageToken, MaxNamespaces: pageSize, } diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 4462459e8..a4987d197 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -120,7 +120,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { // Generate UUID for the new table tableUUID := uuid.New() - tablePath := path.Join(encodeNamespace(namespace), req.Name) + tablePath := path.Join(flattenNamespacePath(namespace), req.Name) location := strings.TrimSuffix(req.Location, "/") if location == "" { if req.Properties != nil { @@ -179,7 +179,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { } stagedMetadataLocation := fmt.Sprintf("s3://%s/%s/metadata/%s", metadataBucket, stagedTablePath, metadataFileName) 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", encodeNamespace(namespace), tableName, markerErr) + glog.V(1).Infof("Iceberg: failed to persist stage-create marker for %s.%s: %v", flattenNamespacePath(namespace), tableName, markerErr) } result := LoadTableResult{ MetadataLocation: metadataLocation, @@ -266,7 +266,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { 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", encodeNamespace(namespace), tableName, markerErr) + glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after create: %v", flattenNamespacePath(namespace), tableName, markerErr) } result := LoadTableResult{ @@ -324,7 +324,7 @@ func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) { func buildLoadTableResult(getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) LoadTableResult { location := tableLocationFromMetadataLocation(getResp.MetadataLocation) if location == "" { - location = fmt.Sprintf("s3://%s/%s/%s", bucketName, encodeNamespace(namespace), tableName) + location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName)) } tableUUID := uuid.Nil if getResp.Metadata != nil && getResp.Metadata.Iceberg != nil && getResp.Metadata.Iceberg.TableUUID != "" { diff --git a/weed/s3api/iceberg/utils.go b/weed/s3api/iceberg/utils.go index a49ff2f68..dc4639d7e 100644 --- a/weed/s3api/iceberg/utils.go +++ b/weed/s3api/iceberg/utils.go @@ -32,11 +32,19 @@ func parseNamespace(encoded string) []string { return result } -// encodeNamespace encodes namespace parts for response. +// encodeNamespace encodes namespace parts using the Iceberg REST protocol's +// unit separator (0x1F) convention. This is only appropriate for protocol-level +// encoding (e.g. URL path parameters), NOT for filesystem/S3 paths. func encodeNamespace(parts []string) string { return strings.Join(parts, "\x1F") } +// flattenNamespacePath joins namespace parts with "." for use in S3 location +// and filer paths, matching the S3 Tables storage layer convention. +func flattenNamespacePath(parts []string) string { + return strings.Join(parts, ".") +} + func parseS3Location(location string) (bucketName, tablePath string, err error) { if !strings.HasPrefix(location, "s3://") { return "", "", fmt.Errorf("unsupported location: %s", location)