feat: drop table location mapping support (#8458)
* feat: drop table location mapping support Disable external metadata locations for S3 Tables and remove the table location mapping index entirely. Table metadata must live under the table bucket paths, so lookups no longer use mapping directories. Changes: - Remove mapping lookup and cache from bucket path resolution - Reject metadataLocation in CreateTable and UpdateTable - Remove mapping helpers and tests * compile * refactor * fix: accept metadataLocation in S3 Tables API requests We removed the external table location mapping feature, but still need to accept and store metadataLocation values from clients like Trino. The mapping feature was an internal implementation detail that mapped external buckets to internal table paths. The metadataLocation field itself is part of the S3 Tables API and should be preserved. * fmt * fix: handle MetadataLocation in UpdateTable requests Mirror handleCreateTable behavior by updating metadata.MetadataLocation when req.MetadataLocation is provided in UpdateTable requests. This ensures table metadata location can be updated, not just set during creation.
This commit is contained in:
@@ -49,9 +49,6 @@ type BucketRegistry struct {
|
|||||||
metadataCache map[string]*BucketMetaData
|
metadataCache map[string]*BucketMetaData
|
||||||
metadataCacheLock sync.RWMutex
|
metadataCacheLock sync.RWMutex
|
||||||
|
|
||||||
tableLocationCache map[string]string // Cache for table location mappings (external bucket -> mapped root path)
|
|
||||||
tableLocationLock sync.RWMutex
|
|
||||||
|
|
||||||
notFound map[string]struct{}
|
notFound map[string]struct{}
|
||||||
notFoundLock sync.RWMutex
|
notFoundLock sync.RWMutex
|
||||||
s3a *S3ApiServer
|
s3a *S3ApiServer
|
||||||
@@ -60,7 +57,6 @@ type BucketRegistry struct {
|
|||||||
func NewBucketRegistry(s3a *S3ApiServer) *BucketRegistry {
|
func NewBucketRegistry(s3a *S3ApiServer) *BucketRegistry {
|
||||||
br := &BucketRegistry{
|
br := &BucketRegistry{
|
||||||
metadataCache: make(map[string]*BucketMetaData),
|
metadataCache: make(map[string]*BucketMetaData),
|
||||||
tableLocationCache: make(map[string]string),
|
|
||||||
notFound: make(map[string]struct{}),
|
notFound: make(map[string]struct{}),
|
||||||
s3a: s3a,
|
s3a: s3a,
|
||||||
}
|
}
|
||||||
@@ -164,7 +160,6 @@ func buildBucketMetadata(accountManager AccountManager, entry *filer_pb.Entry) *
|
|||||||
func (r *BucketRegistry) RemoveBucketMetadata(entry *filer_pb.Entry) {
|
func (r *BucketRegistry) RemoveBucketMetadata(entry *filer_pb.Entry) {
|
||||||
r.removeMetadataCache(entry.Name)
|
r.removeMetadataCache(entry.Name)
|
||||||
r.unMarkNotFound(entry.Name)
|
r.unMarkNotFound(entry.Name)
|
||||||
r.removeTableLocationCache(entry.Name)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *BucketRegistry) GetBucketMetadata(bucketName string) (*BucketMetaData, s3err.ErrorCode) {
|
func (r *BucketRegistry) GetBucketMetadata(bucketName string) (*BucketMetaData, s3err.ErrorCode) {
|
||||||
@@ -229,12 +224,6 @@ func (r *BucketRegistry) removeMetadataCache(bucket string) {
|
|||||||
delete(r.metadataCache, bucket)
|
delete(r.metadataCache, bucket)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *BucketRegistry) removeTableLocationCache(bucket string) {
|
|
||||||
r.tableLocationLock.Lock()
|
|
||||||
defer r.tableLocationLock.Unlock()
|
|
||||||
delete(r.tableLocationCache, bucket)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *BucketRegistry) markNotFound(bucket string) {
|
func (r *BucketRegistry) markNotFound(bucket string) {
|
||||||
r.notFoundLock.Lock()
|
r.notFoundLock.Lock()
|
||||||
defer r.notFoundLock.Unlock()
|
defer r.notFoundLock.Unlock()
|
||||||
|
|||||||
@@ -1,9 +1,7 @@
|
|||||||
package s3api
|
package s3api
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
|
||||||
"errors"
|
"errors"
|
||||||
"io"
|
|
||||||
"path"
|
"path"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
@@ -43,136 +41,12 @@ func (s3a *S3ApiServer) isTableBucket(bucket string) bool {
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s3a *S3ApiServer) tableLocationDir(bucket string) (string, bool) {
|
|
||||||
if bucket == "" {
|
|
||||||
return "", false
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check cache first
|
|
||||||
if s3a.bucketRegistry != nil {
|
|
||||||
s3a.bucketRegistry.tableLocationLock.RLock()
|
|
||||||
if tablePath, ok := s3a.bucketRegistry.tableLocationCache[bucket]; ok {
|
|
||||||
s3a.bucketRegistry.tableLocationLock.RUnlock()
|
|
||||||
return tablePath, tablePath != ""
|
|
||||||
}
|
|
||||||
s3a.bucketRegistry.tableLocationLock.RUnlock()
|
|
||||||
}
|
|
||||||
|
|
||||||
tablePath, err := s3a.lookupTableLocationMapping(bucket, s3tables.GetTableLocationMappingDir())
|
|
||||||
|
|
||||||
// Only cache definitive results: successful lookup (tablePath set) or definitive not-found (ErrNotFound)
|
|
||||||
// Don't cache transient errors to avoid treating temporary failures as permanent misses
|
|
||||||
if err == nil || errors.Is(err, filer_pb.ErrNotFound) {
|
|
||||||
if s3a.bucketRegistry != nil {
|
|
||||||
s3a.bucketRegistry.tableLocationLock.Lock()
|
|
||||||
s3a.bucketRegistry.tableLocationCache[bucket] = tablePath
|
|
||||||
s3a.bucketRegistry.tableLocationLock.Unlock()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if tablePath == "" {
|
|
||||||
if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
|
|
||||||
glog.V(1).Infof("table location mapping lookup failed for %s: %v", bucket, err)
|
|
||||||
}
|
|
||||||
return "", false
|
|
||||||
}
|
|
||||||
|
|
||||||
return tablePath, true
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s3a *S3ApiServer) readTableLocationMappingFromDirectory(mappingDir string) (string, error) {
|
|
||||||
var mappedPath string
|
|
||||||
conflict := false
|
|
||||||
|
|
||||||
err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
|
||||||
stream, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
|
|
||||||
Directory: mappingDir,
|
|
||||||
Limit: 4294967295, // math.MaxUint32
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
for {
|
|
||||||
resp, recvErr := stream.Recv()
|
|
||||||
if recvErr == io.EOF {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if recvErr != nil {
|
|
||||||
return recvErr
|
|
||||||
}
|
|
||||||
if resp == nil || resp.Entry == nil || resp.Entry.IsDirectory || len(resp.Entry.Content) == 0 {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
candidate := normalizeTableLocationMappingPath(string(resp.Entry.Content))
|
|
||||||
if candidate == "" {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if mappedPath == "" {
|
|
||||||
mappedPath = candidate
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if mappedPath != candidate {
|
|
||||||
conflict = true
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
|
|
||||||
if conflict {
|
|
||||||
glog.V(1).Infof("table location mapping conflict under %s: multiple mapped roots found", mappingDir)
|
|
||||||
return "", nil
|
|
||||||
}
|
|
||||||
return mappedPath, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s3a *S3ApiServer) lookupTableLocationMapping(bucket, mappingDir string) (string, error) {
|
|
||||||
entry, err := s3a.getEntry(mappingDir, bucket)
|
|
||||||
if err != nil || entry == nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
if entry.IsDirectory {
|
|
||||||
return s3a.readTableLocationMappingFromDirectory(path.Join(mappingDir, bucket))
|
|
||||||
}
|
|
||||||
if len(entry.Content) == 0 {
|
|
||||||
return "", nil
|
|
||||||
}
|
|
||||||
return normalizeTableLocationMappingPath(string(entry.Content)), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func normalizeTableLocationMappingPath(rawPath string) string {
|
|
||||||
rawPath = strings.TrimSpace(rawPath)
|
|
||||||
if rawPath == "" {
|
|
||||||
return ""
|
|
||||||
}
|
|
||||||
|
|
||||||
normalized := path.Clean("/" + strings.TrimPrefix(rawPath, "/"))
|
|
||||||
tablesPrefix := strings.TrimSuffix(s3tables.TablesPath, "/") + "/"
|
|
||||||
if !strings.HasPrefix(normalized, tablesPrefix) {
|
|
||||||
return normalized
|
|
||||||
}
|
|
||||||
|
|
||||||
remaining := strings.TrimPrefix(normalized, tablesPrefix)
|
|
||||||
bucketName, _, _ := strings.Cut(remaining, "/")
|
|
||||||
if bucketName == "" {
|
|
||||||
return ""
|
|
||||||
}
|
|
||||||
return path.Join(s3tables.TablesPath, bucketName)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s3a *S3ApiServer) bucketRoot(bucket string) string {
|
func (s3a *S3ApiServer) bucketRoot(bucket string) string {
|
||||||
// Returns the unified buckets root path for all bucket types
|
// Returns the unified buckets root path for all bucket types
|
||||||
return s3a.option.BucketsPath
|
return s3a.option.BucketsPath
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s3a *S3ApiServer) bucketDir(bucket string) string {
|
func (s3a *S3ApiServer) bucketDir(bucket string) string {
|
||||||
if tablePath, ok := s3a.tableLocationDir(bucket); ok {
|
|
||||||
return tablePath
|
|
||||||
}
|
|
||||||
return path.Join(s3a.bucketRoot(bucket), bucket)
|
return path.Join(s3a.bucketRoot(bucket), bucket)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -221,8 +95,5 @@ func (s3a *S3ApiServer) bucketExists(bucket string) (bool, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s3a *S3ApiServer) getBucketEntry(bucket string) (*filer_pb.Entry, error) {
|
func (s3a *S3ApiServer) getBucketEntry(bucket string) (*filer_pb.Entry, error) {
|
||||||
if tablePath, ok := s3a.tableLocationDir(bucket); ok {
|
|
||||||
return s3a.getEntry(path.Dir(tablePath), path.Base(tablePath))
|
|
||||||
}
|
|
||||||
return s3a.getEntry(s3a.option.BucketsPath, bucket)
|
return s3a.getEntry(s3a.option.BucketsPath, bucket)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,45 +0,0 @@
|
|||||||
package s3api
|
|
||||||
|
|
||||||
import "testing"
|
|
||||||
|
|
||||||
func TestNormalizeTableLocationMappingPath(t *testing.T) {
|
|
||||||
testCases := []struct {
|
|
||||||
name string
|
|
||||||
raw string
|
|
||||||
want string
|
|
||||||
}{
|
|
||||||
{
|
|
||||||
name: "legacy table path maps to bucket root",
|
|
||||||
raw: "/buckets/warehouse/analytics/orders",
|
|
||||||
want: "/buckets/warehouse",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "already bucket root",
|
|
||||||
raw: "/buckets/warehouse",
|
|
||||||
want: "/buckets/warehouse",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "relative buckets path normalized and reduced",
|
|
||||||
raw: "buckets/warehouse/analytics/orders",
|
|
||||||
want: "/buckets/warehouse",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "non buckets path preserved",
|
|
||||||
raw: "/tmp/custom/path",
|
|
||||||
want: "/tmp/custom/path",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "empty path",
|
|
||||||
raw: "",
|
|
||||||
want: "",
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, tc := range testCases {
|
|
||||||
t.Run(tc.name, func(t *testing.T) {
|
|
||||||
if got := normalizeTableLocationMappingPath(tc.raw); got != tc.want {
|
|
||||||
t.Fatalf("normalizeTableLocationMappingPath(%q)=%q, want %q", tc.raw, got, tc.want)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,7 +1,6 @@
|
|||||||
package s3tables
|
package s3tables
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
@@ -10,7 +9,6 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
||||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -242,10 +240,6 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := h.updateTableLocationMapping(r.Context(), client, "", req.MetadataLocation, tablePath); err != nil {
|
|
||||||
glog.V(1).Infof("failed to update table location mapping for %s: %v", req.MetadataLocation, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -943,9 +937,6 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
|
|||||||
if err := h.deleteDirectory(r.Context(), client, tablePath); err != nil {
|
if err := h.deleteDirectory(r.Context(), client, tablePath); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := h.deleteTableLocationMapping(r.Context(), client, metadata.MetadataLocation, tablePath); err != nil {
|
|
||||||
glog.V(1).Infof("failed to delete table location mapping for %s: %v", metadata.MetadataLocation, err)
|
|
||||||
}
|
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -1090,9 +1081,6 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
|
|||||||
return ErrVersionTokenMismatch
|
return ErrVersionTokenMismatch
|
||||||
}
|
}
|
||||||
|
|
||||||
// Capture old metadata location before mutation for stale mapping cleanup
|
|
||||||
oldMetadataLocation := metadata.MetadataLocation
|
|
||||||
|
|
||||||
// Update metadata
|
// Update metadata
|
||||||
if req.Metadata != nil {
|
if req.Metadata != nil {
|
||||||
if metadata.Metadata == nil {
|
if metadata.Metadata == nil {
|
||||||
@@ -1131,9 +1119,6 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
|
|||||||
if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata, metadataBytes); err != nil {
|
if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata, metadataBytes); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := h.updateTableLocationMapping(r.Context(), client, oldMetadataLocation, metadata.MetadataLocation, tablePath); err != nil {
|
|
||||||
glog.V(1).Infof("failed to update table location mapping for %s -> %s: %v", oldMetadataLocation, metadata.MetadataLocation, err)
|
|
||||||
}
|
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -1149,104 +1134,3 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
|
|||||||
})
|
})
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *S3TablesHandler) updateTableLocationMapping(ctx context.Context, client filer_pb.SeaweedFilerClient, oldMetadataLocation, newMetadataLocation, tablePath string) error {
|
|
||||||
newTableLocationBucket, ok := parseTableLocationBucket(newMetadataLocation)
|
|
||||||
if !ok {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
tableBucketPath, ok := tableBucketPathFromTablePath(tablePath)
|
|
||||||
if !ok {
|
|
||||||
return fmt.Errorf("invalid table path for location mapping: %s", tablePath)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := h.ensureDirectory(ctx, client, GetTableLocationMappingDir()); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if err := h.ensureTableLocationMappingBucketDir(ctx, client, newTableLocationBucket); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// If the metadata location changed, remove this table's stale mapping entry from the old bucket.
|
|
||||||
if oldMetadataLocation != "" && oldMetadataLocation != newMetadataLocation {
|
|
||||||
oldTableLocationBucket, ok := parseTableLocationBucket(oldMetadataLocation)
|
|
||||||
if ok && oldTableLocationBucket != newTableLocationBucket {
|
|
||||||
if err := h.removeTableLocationMappingEntry(ctx, client, oldTableLocationBucket, tablePath); err != nil {
|
|
||||||
glog.V(1).Infof("failed to delete stale mapping for %s: %v", oldTableLocationBucket, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return h.upsertFile(ctx, client, GetTableLocationMappingEntryPath(newTableLocationBucket, tablePath), []byte(tableBucketPath))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (h *S3TablesHandler) deleteTableLocationMapping(ctx context.Context, client filer_pb.SeaweedFilerClient, metadataLocation, tablePath string) error {
|
|
||||||
tableLocationBucket, ok := parseTableLocationBucket(metadataLocation)
|
|
||||||
if !ok {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
return h.removeTableLocationMappingEntry(ctx, client, tableLocationBucket, tablePath)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (h *S3TablesHandler) ensureTableLocationMappingBucketDir(ctx context.Context, client filer_pb.SeaweedFilerClient, tableLocationBucket string) error {
|
|
||||||
mappingDir := GetTableLocationMappingDir()
|
|
||||||
bucketMappingPath := GetTableLocationMappingPath(tableLocationBucket)
|
|
||||||
|
|
||||||
resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
|
|
||||||
Directory: mappingDir,
|
|
||||||
Name: tableLocationBucket,
|
|
||||||
})
|
|
||||||
if err == nil {
|
|
||||||
if resp != nil && resp.Entry != nil && resp.Entry.IsDirectory {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if removeErr := h.deleteEntryIfExists(ctx, client, bucketMappingPath); removeErr != nil && !errors.Is(removeErr, filer_pb.ErrNotFound) {
|
|
||||||
return removeErr
|
|
||||||
}
|
|
||||||
} else if !errors.Is(err, filer_pb.ErrNotFound) {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
return h.ensureDirectory(ctx, client, bucketMappingPath)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (h *S3TablesHandler) removeTableLocationMappingEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, tableLocationBucket, tablePath string) error {
|
|
||||||
entryPath := GetTableLocationMappingEntryPath(tableLocationBucket, tablePath)
|
|
||||||
if err := h.deleteEntryIfExists(ctx, client, entryPath); err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return h.removeTableLocationMappingBucketDirIfEmpty(ctx, client, tableLocationBucket)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (h *S3TablesHandler) removeTableLocationMappingBucketDirIfEmpty(ctx context.Context, client filer_pb.SeaweedFilerClient, tableLocationBucket string) error {
|
|
||||||
bucketMappingPath := GetTableLocationMappingPath(tableLocationBucket)
|
|
||||||
|
|
||||||
stream, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{
|
|
||||||
Directory: bucketMappingPath,
|
|
||||||
Limit: 1,
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
for {
|
|
||||||
resp, recvErr := stream.Recv()
|
|
||||||
if recvErr == io.EOF {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
if recvErr != nil {
|
|
||||||
return recvErr
|
|
||||||
}
|
|
||||||
if resp != nil && resp.Entry != nil {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := h.deleteEntryIfExists(ctx, client, bucketMappingPath); err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -1,77 +0,0 @@
|
|||||||
package s3tables
|
|
||||||
|
|
||||||
import (
|
|
||||||
"strings"
|
|
||||||
"testing"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestGetTableLocationMappingEntryPathPerTable(t *testing.T) {
|
|
||||||
tableLocationBucket := "shared-location--table-s3"
|
|
||||||
tablePathA := GetTablePath("warehouse", "analytics", "orders")
|
|
||||||
tablePathB := GetTablePath("warehouse", "analytics", "customers")
|
|
||||||
|
|
||||||
entryPathA := GetTableLocationMappingEntryPath(tableLocationBucket, tablePathA)
|
|
||||||
entryPathARepeat := GetTableLocationMappingEntryPath(tableLocationBucket, tablePathA)
|
|
||||||
entryPathB := GetTableLocationMappingEntryPath(tableLocationBucket, tablePathB)
|
|
||||||
|
|
||||||
if entryPathA != entryPathARepeat {
|
|
||||||
t.Fatalf("mapping entry path should be deterministic: %q != %q", entryPathA, entryPathARepeat)
|
|
||||||
}
|
|
||||||
if entryPathA == entryPathB {
|
|
||||||
t.Fatalf("mapping entry path should differ per table path: %q == %q", entryPathA, entryPathB)
|
|
||||||
}
|
|
||||||
|
|
||||||
expectedPrefix := GetTableLocationMappingPath(tableLocationBucket) + "/"
|
|
||||||
if !strings.HasPrefix(entryPathA, expectedPrefix) {
|
|
||||||
t.Fatalf("mapping entry path %q should start with %q", entryPathA, expectedPrefix)
|
|
||||||
}
|
|
||||||
if strings.TrimPrefix(entryPathA, expectedPrefix) == "" {
|
|
||||||
t.Fatalf("mapping entry name should not be empty: %q", entryPathA)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestTableBucketPathFromTablePath(t *testing.T) {
|
|
||||||
testCases := []struct {
|
|
||||||
name string
|
|
||||||
tablePath string
|
|
||||||
expected string
|
|
||||||
ok bool
|
|
||||||
}{
|
|
||||||
{
|
|
||||||
name: "valid table path",
|
|
||||||
tablePath: GetTablePath("warehouse", "analytics", "orders"),
|
|
||||||
expected: GetTableBucketPath("warehouse"),
|
|
||||||
ok: true,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "valid table bucket root",
|
|
||||||
tablePath: GetTableBucketPath("warehouse"),
|
|
||||||
expected: GetTableBucketPath("warehouse"),
|
|
||||||
ok: true,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "invalid non-tables path",
|
|
||||||
tablePath: "/tmp/warehouse/analytics/orders",
|
|
||||||
expected: "",
|
|
||||||
ok: false,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "invalid empty bucket segment",
|
|
||||||
tablePath: "/buckets/",
|
|
||||||
expected: "",
|
|
||||||
ok: false,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, tc := range testCases {
|
|
||||||
t.Run(tc.name, func(t *testing.T) {
|
|
||||||
actual, ok := tableBucketPathFromTablePath(tc.tablePath)
|
|
||||||
if ok != tc.ok {
|
|
||||||
t.Fatalf("tableBucketPathFromTablePath(%q) ok=%v, want %v", tc.tablePath, ok, tc.ok)
|
|
||||||
}
|
|
||||||
if actual != tc.expected {
|
|
||||||
t.Fatalf("tableBucketPathFromTablePath(%q)=%q, want %q", tc.tablePath, actual, tc.expected)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -2,7 +2,6 @@ package s3tables
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
"crypto/sha1"
|
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/url"
|
"net/url"
|
||||||
@@ -21,7 +20,6 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
tableLocationMappingsDirPath = "/etc/s3tables"
|
|
||||||
tableObjectRootDirName = ".objects"
|
tableObjectRootDirName = ".objects"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -109,43 +107,6 @@ func GetTableObjectBucketPath(bucketName string) string {
|
|||||||
return path.Join(GetTableObjectRootDir(), bucketName)
|
return path.Join(GetTableObjectRootDir(), bucketName)
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetTableLocationMappingDir returns the root path for table location bucket mappings
|
|
||||||
func GetTableLocationMappingDir() string {
|
|
||||||
return tableLocationMappingsDirPath
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetTableLocationMappingPath returns the filer path for a table location bucket mapping
|
|
||||||
func GetTableLocationMappingPath(tableLocationBucket string) string {
|
|
||||||
return path.Join(GetTableLocationMappingDir(), tableLocationBucket)
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetTableLocationMappingEntryPath returns the filer path for a table-specific mapping entry.
|
|
||||||
// Each table gets its own entry so multiple tables can share the same external table-location bucket.
|
|
||||||
func GetTableLocationMappingEntryPath(tableLocationBucket, tablePath string) string {
|
|
||||||
return path.Join(GetTableLocationMappingPath(tableLocationBucket), tableLocationMappingEntryName(tablePath))
|
|
||||||
}
|
|
||||||
|
|
||||||
func tableLocationMappingEntryName(tablePath string) string {
|
|
||||||
normalized := path.Clean("/" + strings.TrimSpace(strings.TrimPrefix(tablePath, "/")))
|
|
||||||
sum := sha1.Sum([]byte(normalized))
|
|
||||||
return hex.EncodeToString(sum[:])
|
|
||||||
}
|
|
||||||
|
|
||||||
func tableBucketPathFromTablePath(tablePath string) (string, bool) {
|
|
||||||
normalized := path.Clean("/" + strings.TrimSpace(strings.TrimPrefix(tablePath, "/")))
|
|
||||||
tablesPrefix := strings.TrimSuffix(TablesPath, "/") + "/"
|
|
||||||
if !strings.HasPrefix(normalized, tablesPrefix) {
|
|
||||||
return "", false
|
|
||||||
}
|
|
||||||
|
|
||||||
remaining := strings.TrimPrefix(normalized, tablesPrefix)
|
|
||||||
bucketName, _, _ := strings.Cut(remaining, "/")
|
|
||||||
if bucketName == "" {
|
|
||||||
return "", false
|
|
||||||
}
|
|
||||||
return path.Join(TablesPath, bucketName), true
|
|
||||||
}
|
|
||||||
|
|
||||||
// Metadata structures
|
// Metadata structures
|
||||||
|
|
||||||
type tableBucketMetadata struct {
|
type tableBucketMetadata struct {
|
||||||
@@ -244,22 +205,6 @@ func ValidateBucketName(name string) error {
|
|||||||
return validateBucketName(name)
|
return validateBucketName(name)
|
||||||
}
|
}
|
||||||
|
|
||||||
func parseTableLocationBucket(metadataLocation string) (string, bool) {
|
|
||||||
if !strings.HasPrefix(metadataLocation, "s3://") {
|
|
||||||
return "", false
|
|
||||||
}
|
|
||||||
trimmed := strings.TrimPrefix(metadataLocation, "s3://")
|
|
||||||
trimmed = strings.TrimSuffix(trimmed, "/")
|
|
||||||
if trimmed == "" {
|
|
||||||
return "", false
|
|
||||||
}
|
|
||||||
bucket, _, _ := strings.Cut(trimmed, "/")
|
|
||||||
if bucket == "" || !strings.HasSuffix(bucket, "--table-s3") {
|
|
||||||
return "", false
|
|
||||||
}
|
|
||||||
return bucket, true
|
|
||||||
}
|
|
||||||
|
|
||||||
// BuildBucketARN builds a bucket ARN with the provided region and account ID.
|
// BuildBucketARN builds a bucket ARN with the provided region and account ID.
|
||||||
// If region is empty, the ARN will omit the region field.
|
// If region is empty, the ARN will omit the region field.
|
||||||
func BuildBucketARN(region, accountID, bucketName string) (string, error) {
|
func BuildBucketARN(region, accountID, bucketName string) (string, error) {
|
||||||
|
|||||||
Reference in New Issue
Block a user