azureblob: add use_arrow_list flag for experimental Apache Arrow listing

Add a new "use_arrow_list" advanced option which fetches directory listings
using the ListBlobs Apache Arrow response format instead of XML. This can be
substantially 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). If the feature is not enabled the
service returns XML and the listing transparently falls back to it. Connection
string auth is not supported - it also falls back to the normal XML listing.

The option is hidden from the configurator, command line help and docs until the
feature is officially released.

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")
+	})
+}

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, "/")

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