mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-11 22:38:53 +01:00
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=<ns>` 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.
This commit is contained in:
1 parent
d4548376a1
commit
761ec7da00
7 files changed
+225
-12
No files matched your search
@@ -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{
|
||||
|
||||
@@ -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 != "" {
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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 != "" {
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user