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)