diff --git a/backend/azureblob/arrow_internal_test.go b/backend/azureblob/arrow_internal_test.go new file mode 100644 index 000000000..7bf76941c --- /dev/null +++ b/backend/azureblob/arrow_internal_test.go @@ -0,0 +1,133 @@ +//go:build !plan9 && !solaris && !js + +package azureblob + +import ( + "context" + "strings" + "testing" + + "github.com/rclone/rclone/fs" + "github.com/rclone/rclone/fstest" + "github.com/rclone/rclone/fstest/fstests" + "github.com/rclone/rclone/lib/random" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// testArrowList exercises the experimental Apache Arrow listing. +// +// It is skipped automatically if Arrow listing is not enabled on the test +// account (the server returns XML, or 409 on HNS accounts). +func (f *Fs) testArrowList(t *testing.T) { + ctx := context.Background() + containerName, _ := f.split("probe") + if containerName == "" { + t.Skip("Arrow list test needs a container in the remote root") + } + + // withArrow runs fn with UseArrowList temporarily set. + withArrow := func(on bool, fn func()) { + old := f.opt.UseArrowList + f.opt.UseArrowList = on + defer func() { f.opt.UseArrowList = old }() + fn() + } + + // Create a set of known objects under a unique directory: flat files plus + // one in a subdirectory (so we exercise both the flat and delimited paths). + dir := "arrow-list-" + random.String(8) + // Includes upper-case and digit first-characters: these byte-sort between + // the digit and lowercase ladder boundaries, so they verify the server's + // endBefore filtering is byte-ordered (not case-insensitive, which would + // drop them from a shard and make parallel != sequential). + objects := []string{"a.txt", "b.txt", "c.txt", "d.txt", "e.txt", "sub/x.txt", "Foo.txt", "Zoo.txt", "0num.txt", "9end.txt"} + wantSizes := map[string]int64{} + created := make([]string, 0, len(objects)) + for i, n := range objects { + contents := random.String(10 + i) + remote := dir + "/" + n + item := fstest.NewItem(remote, contents, fstest.Time("2001-05-06T04:05:06.499Z")) + _ = fstests.PutTestContents(ctx, t, f, &item, contents, true) + created = append(created, remote) + wantSizes[remote] = int64(len(contents)) + } + defer func() { + for _, remote := range created { + if o, err := f.NewObject(ctx, remote); err == nil { + _ = o.Remove(ctx) + } + } + }() + + // listing returns remote->size for dir (dirs reported as size -1). + listing := func(recurse bool) (map[string]int64, error) { + got := map[string]int64{} + add := func(entries fs.DirEntries) error { + for _, e := range entries { + if o, ok := e.(fs.Object); ok { + got[e.Remote()] = o.Size() + } else { + got[e.Remote()] = -1 + } + } + return nil + } + if recurse { + return got, f.ListR(ctx, dir, add) + } + entries, err := f.List(ctx, dir) + if err != nil { + return nil, err + } + _ = add(entries) + return got, nil + } + + // Probe: is Arrow available? If not, skip cleanly. + var probeErr error + withArrow(true, func() { _, probeErr = listing(true) }) + if probeErr != nil { + if strings.Contains(probeErr.Error(), "OperationNotSupportedWithFeatureMissing") || strings.Contains(probeErr.Error(), "409") { + t.Skipf("Arrow listing not available on this account: %v", probeErr) + } + require.NoError(t, probeErr) + } + + // Arrow listing must equal XML listing, recursively and per-directory + // (including the "sub" subdirectory). + t.Run("EqualsXML", func(t *testing.T) { + for _, recurse := range []bool{true, false} { + var xml, arrow map[string]int64 + var err error + withArrow(false, func() { xml, err = listing(recurse) }) + require.NoError(t, err) + withArrow(true, func() { arrow, err = listing(recurse) }) + require.NoError(t, err) + assert.Equal(t, xml, arrow, "recurse=%v: Arrow listing differs from XML", recurse) + } + + // Recursive Arrow listing contains every object with the right size. + var arrow map[string]int64 + withArrow(true, func() { arrow, _ = listing(true) }) + for remote, size := range wantSizes { + assert.Equal(t, size, arrow[remote], remote) + } + + // Single-directory Arrow listing reports "sub" as a directory. + var entries fs.DirEntries + withArrow(true, func() { + var err error + entries, err = f.List(ctx, dir) + require.NoError(t, err) + }) + foundSubDir := false + for _, e := range entries { + if e.Remote() == dir+"/sub" { + _, isObj := e.(fs.Object) + foundSubDir = !isObj + } + } + assert.True(t, foundSubDir, "expected %q to be listed as a directory", dir+"/sub") + }) +} diff --git a/backend/azureblob/azureblob.go b/backend/azureblob/azureblob.go index a2e52eecf..fbbed2c98 100644 --- a/backend/azureblob/azureblob.go +++ b/backend/azureblob/azureblob.go @@ -26,12 +26,15 @@ import ( "github.com/Azure/azure-sdk-for-go/sdk/azcore" "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy" + "github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime" + "github.com/Azure/azure-sdk-for-go/sdk/azcore/to" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/sas" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service" + "github.com/rclone/rclone/backend/azureblob/arrowlist" "github.com/rclone/rclone/backend/azureblob/auth" "github.com/rclone/rclone/fs" "github.com/rclone/rclone/fs/chunksize" @@ -235,6 +238,26 @@ minutes per megabyte on average, it will time out ( avoid the time out.`, Default: maxListChunkSize, Advanced: true, + }, { + Name: "use_arrow_list", + Help: `Use the experimental Apache Arrow listing format. + +If set, directory listings are fetched using the ListBlobs Apache +Arrow response format instead of XML. This can be faster for very +large containers. + +This is EXPERIMENTAL and requires the "Blob Listing with Apache Arrow" +preview feature to be enabled on the storage account. It is NOT +supported on accounts with a hierarchical namespace (ADLS Gen2) - +those return a 409 error. If the feature is not enabled the server +returns XML and the listing transparently falls back to the normal XML +path (logged at debug level). + +Not supported with connection_string auth - falls back to normal +listing.`, + Default: false, + Advanced: true, + Hide: fs.OptionHideBoth, }, { Name: "access_tier", Help: `Access tier of blob: hot, cool, cold or archive. @@ -388,6 +411,7 @@ type Options struct { UseCopyBlob bool `config:"use_copy_blob"` UploadConcurrency int `config:"upload_concurrency"` ListChunkSize uint `config:"list_chunk"` + UseArrowList bool `config:"use_arrow_list"` AccessTier string `config:"access_tier"` ArchiveTierDelete bool `config:"archive_tier_delete"` DisableCheckSum bool `config:"disable_checksum"` @@ -407,8 +431,10 @@ type Fs struct { opt Options // parsed config options ci *fs.ConfigInfo // global config features *fs.Features // optional features - cntSVCcacheMu sync.Mutex // mutex to protect cntSVCcache + cntSVCcacheMu sync.Mutex // mutex to protect cntSVCcache and arrowCntSVCcache cntSVCcache map[string]*container.Client // reference to containerClient per container + arrowCntSVCcache map[string]*arrowlist.Client // reference to arrowlist client per container + arrowClientOpts *arrowlist.ClientOptions // client options for the arrowlist clients svc *service.Client // client to access azblob cred azcore.TokenCredential // how to generate tokens (may be nil) usingSharedKeyCred bool // set if using shared key credentials @@ -623,14 +649,15 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e ci := fs.GetConfig(ctx) f := &Fs{ - name: name, - opt: *opt, - ci: ci, - pacer: fs.NewPacer(ctx, pacer.NewS3(pacer.MinSleep(minSleep), pacer.MaxSleep(maxSleep), pacer.DecayConstant(decayConstant))), - uploadToken: pacer.NewTokenDispenser(ci.Transfers), - copyToken: pacer.NewTokenDispenser(opt.CopyTotalConcurrency), - cache: bucket.NewCache(), - cntSVCcache: make(map[string]*container.Client, 1), + name: name, + opt: *opt, + ci: ci, + pacer: fs.NewPacer(ctx, pacer.NewS3(pacer.MinSleep(minSleep), pacer.MaxSleep(maxSleep), pacer.DecayConstant(decayConstant))), + uploadToken: pacer.NewTokenDispenser(ci.Transfers), + copyToken: pacer.NewTokenDispenser(opt.CopyTotalConcurrency), + cache: bucket.NewCache(), + cntSVCcache: make(map[string]*container.Client, 1), + arrowCntSVCcache: make(map[string]*arrowlist.Client, 1), } f.publicAccess = container.PublicAccessType(opt.PublicAccess) f.setRoot(root) @@ -677,6 +704,15 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e f.usingSharedKeyCred = res.UsingSharedKeyCred f.anonymous = res.Anonymous + // Client options for the arrowlist clients, using the same transport and + // gzip policy as the SDK clients built above. + f.arrowClientOpts = &arrowlist.ClientOptions{ + ClientOptions: azcore.ClientOptions{ + Transport: auth.Transporter(ctx), + PerCallPolicies: []policy.Policy{setAcceptEncodingGzip{}}, + }, + } + // if using Container level SAS put the container client into the cache if opt.SASURL != "" && res.Container != "" { _ = f.cntSVC(res.Container) @@ -715,6 +751,44 @@ func (f *Fs) cntSVC(containerName string) (containerClient *container.Client) { return containerClient } +// errArrowAuthUnsupported is returned by arrowCntSVC when the configured +// credentials can't be used for Arrow listing. +var errArrowAuthUnsupported = errors.New("credentials not supported for Arrow listing (connection_string auth is not supported)") + +// return the arrowlist client for the container passed in +// +// Returns errArrowAuthUnsupported if the configured credentials can't be +// reused for the arrowlist client's pipeline. +func (f *Fs) arrowCntSVC(containerName string) (client *arrowlist.Client, err error) { + // The container URL includes any SAS token in its query + url := f.cntSVC(containerName).URL() + f.cntSVCcacheMu.Lock() + defer f.cntSVCcacheMu.Unlock() + if client, ok := f.arrowCntSVCcache[containerName]; ok { + return client, nil + } + switch { + case f.usingSharedKeyCred: + // Covers account+key and the emulator (auth fills in Account/Key) + var cred *arrowlist.SharedKeyCredential + cred, err = arrowlist.NewSharedKeyCredential(f.opt.Account, f.opt.Key) + if err == nil { + client, err = arrowlist.NewClientWithSharedKeyCredential(url, cred, f.arrowClientOpts) + } + case f.cred != nil: + client, err = arrowlist.NewClient(url, f.cred, f.arrowClientOpts) + case f.anonymous || f.opt.SASURL != "": + client, err = arrowlist.NewClientWithNoCredential(url, f.arrowClientOpts) + default: + return nil, errArrowAuthUnsupported + } + if err != nil { + return nil, err + } + f.arrowCntSVCcache[containerName] = client + return client, nil +} + // Return an Object from a path // // If it can't be found it returns the error fs.ErrorObjectNotFound. @@ -1091,19 +1165,44 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, delimiter = "/" } - pager := f.cntSVC(containerName).NewListBlobsHierarchyPager(delimiter, &container.ListBlobsHierarchyOptions{ - // Copy, Metadata, Snapshots, UncommittedBlobs, Deleted, Tags, Versions, LegalHold, ImmutabilityPolicy, DeletedWithVersions bool - Include: container.ListBlobsInclude{ - Copy: false, - Metadata: true, - Snapshots: false, - UncommittedBlobs: false, - Deleted: false, + opts := &arrowlist.ListBlobsHierarchyOptions{ + ListBlobsHierarchyOptions: container.ListBlobsHierarchyOptions{ + // Copy, Metadata, Snapshots, UncommittedBlobs, Deleted, Tags, Versions, LegalHold, ImmutabilityPolicy, DeletedWithVersions bool + Include: container.ListBlobsInclude{ + Copy: false, + Metadata: true, + Snapshots: false, + UncommittedBlobs: false, + Deleted: false, + }, + Prefix: &directory, + MaxResults: &maxResults, }, - Prefix: &directory, - MaxResults: &maxResults, - }) + } + // Experimental: request the Apache Arrow listing format. The arrowlist + // pager requests an Arrow IPC stream and decodes it, falling back to XML + // if the account doesn't have Arrow listing enabled. Skip the + // maxResults==1 probe (isEmpty) which doesn't benefit. + useArrow := f.opt.UseArrowList && maxResults != 1 + if useArrow { + opts.UseArrowFormat = to.Ptr(true) + } + + var pager *runtime.Pager[container.ListBlobsHierarchyResponse] + if useArrow { + arrowSVC, err := f.arrowCntSVC(containerName) + if err != nil { + fs.Debugf(f, "Not using Arrow listing: %v", err) + useArrow = false + } else { + pager = arrowSVC.NewListBlobsHierarchyPager(delimiter, opts) + } + } + if pager == nil { + pager = f.cntSVC(containerName).NewListBlobsHierarchyPager(delimiter, &opts.ListBlobsHierarchyOptions) + } foundItems := 0 + checkedArrow := false for pager.More() { var response container.ListBlobsHierarchyResponse err := f.pacer.Call(func() (bool, error) { @@ -1113,6 +1212,16 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, return f.shouldRetry(ctx, err) }) + // If Arrow was requested but the service answered with XML, Arrow + // listing is not enabled on this account; the listing is still correct + // but not accelerated. + if useArrow && !checkedArrow && err == nil { + checkedArrow = true + if response.ContentType == nil || !strings.HasPrefix(*response.ContentType, arrowlist.ArrowContentType) { + fs.Debugf(f, "Apache Arrow listing requested but server returned XML - Blob Listing with Apache Arrow may not be enabled on this account") + } + } + if err != nil { // Check http error code along with service code, current SDK doesn't populate service code correctly sometimes if storageErr, ok := err.(*azcore.ResponseError); ok && (storageErr.ErrorCode == string(bloberror.ContainerNotFound) || storageErr.StatusCode == http.StatusNotFound) { @@ -1138,7 +1247,12 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, fs.Debugf(f, "Odd name received %q", remote) continue } - isDirectory := isDirectoryMarker(*file.Properties.ContentLength, file.Metadata, remote) + // ContentLength is documented as nullable in the Arrow listing schema + size := int64(-1) + if file.Properties.ContentLength != nil { + size = *file.Properties.ContentLength + } + isDirectory := isDirectoryMarker(size, file.Metadata, remote) if isDirectory { // Don't insert the root directory if remote == f.opt.Enc.ToStandardPath(directory) { @@ -1169,6 +1283,13 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, fs.Debugf(f, "Odd directory name received %q", remote) continue } + // Don't insert the root directory. The Arrow listing returns a + // directory marker blob with the same name as the listed directory + // as a BlobPrefix, whereas XML returns it as a blob (handled + // above). + if remote == f.opt.Enc.ToStandardPath(directory) { + continue + } remote = remote[len(prefix):] // Trim one slash off the remote name remote, _ = strings.CutSuffix(remote, "/") diff --git a/backend/azureblob/azureblob_internal_test.go b/backend/azureblob/azureblob_internal_test.go index 4d52a2fa9..c29450bce 100644 --- a/backend/azureblob/azureblob_internal_test.go +++ b/backend/azureblob/azureblob_internal_test.go @@ -213,6 +213,7 @@ func (f *Fs) InternalTest(t *testing.T) { t.Run("WriteUncommittedBlocks", f.testWriteUncommittedBlocks) t.Run("Metadata", f.testMetadataPaths) t.Run("GzipEncoding", f.testGzipEncoding) + t.Run("ArrowList", f.testArrowList) } // helper to read blob properties for an object