azureblob: list very large containers in parallel with list_parallelism

Add a new "list_parallelism" advanced option (requires use_arrow_list) which
splits a directory's blob-name keyspace into ranges and lists them
concurrently using the Arrow startFrom/endBefore range parameters. Both
recursive (ListR) and single-directory listings benefit.

The keyspace is sharded with a single-case (digit+lowercase) character
ladder. Single case is deliberate: the service validates startFrom /
endBefore case-insensitively (while listing case-sensitively), so a
mixed-case boundary pair is rejected with a 400; its endBefore filtering is
byte-ordered, so the listing is exact.

The service only honours endBefore on the Arrow listing path of accounts with
Blob Listing with Apache Arrow enabled: elsewhere it either rejects the
request with 400 OperationNotSupportedWithFeatureMissing or, if it fell back
to XML, would ignore the shards' upper bounds. Both cases are detected and
the parallel listing falls back to a sequential listing.

Listing 100,000 objects from a well-connected server:

    config                 time    vs XML
    xml                   23.9s     1.0x
    use_arrow_list         6.9s     3.5x
    list_parallelism=2     4.4s     5.5x
    list_parallelism=4     2.8s     8.4x
    list_parallelism=6     2.2s    10.9x
    list_parallelism=10    1.5s    15.4x
    list_parallelism=20    1.4s    17.6x
    list_parallelism=30    1.1s    21.7x

So Apache Arrow listing alone is ~3.5x faster than XML, and parallel listing
takes that to ~20x. The ladder caps at 37 shards, so the benefit plateaus
beyond list_parallelism=30.

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

backend/azureblob/arrow.go

@@ -0,0 +1,169 @@
+//go:build !plan9 && !solaris && !js
+
+// Parallel listing support for azureblob, built on the Apache Arrow
+// listing format.
+//
+// The Arrow listing exposes the server-side startFrom (inclusive) and endBefore
+// (exclusive) name-range parameters, which let a listing be sharded by
+// blob-name range and listed concurrently.
+
+package azureblob
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
+	"github.com/rclone/rclone/backend/azureblob/arrowlist"
+	"github.com/rclone/rclone/fs"
+	"golang.org/x/sync/errgroup"
+)
+
+// listArrowParallel lists (containerName, directory) by splitting the blob name
+// keyspace into shards and listing them concurrently. opts must already request
+// the Arrow format; it is cloned per shard with StartFrom/EndBefore set. It
+// returns the number of raw items seen. directory/prefix are already normalised
+// by the caller.
+func (f *Fs) listArrowParallel(ctx context.Context, containerName, directory, prefix string, addContainer bool, opts *arrowlist.ListBlobsHierarchyOptions, delimiter string, fn listFn) (int, error) {
+	// Parallel listing needs the arrowlist client (server-side endBefore is
+	// only honoured on the Arrow path) - without it fall back to sequential.
+	if _, err := f.arrowCntSVC(containerName); err != nil {
+		fs.Debugf(f, "Not using parallel Arrow listing: %v", err)
+		return f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn)
+	}
+
+	// Split the keyspace by a single-case character ladder (see
+	// arrowLadderBoundaries for why single-case matters).
+	points := arrowLadderBoundaries(directory, arrowShardTarget(f.opt.ListParallelism))
+	if len(points) < 1 {
+		// Couldn't usefully split - list sequentially.
+		return f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn)
+	}
+
+	// Build half-open [startFrom, endBefore) shards from the sorted boundary
+	// points p0..pk: ["", p0), [p0, p1), ..., [pk, "").
+	type shard struct{ startFrom, endBefore string }
+	shards := make([]shard, 0, len(points)+1)
+	prev := ""
+	for _, p := range points {
+		shards = append(shards, shard{prev, p})
+		prev = p
+	}
+	shards = append(shards, shard{prev, ""})
+
+	// fn (and the list.Helper behind it) is not concurrency-safe, so serialise
+	// delivery. Only the cheap callback is locked, not the HTTP/decode work.
+	var fnMu sync.Mutex
+	safeFn := func(remote string, item *container.BlobItem, isDir bool) error {
+		fnMu.Lock()
+		defer fnMu.Unlock()
+		return fn(remote, item, isDir)
+	}
+
+	// Schedule every shard; SetLimit bounds concurrency, so more shards than
+	// workers gives free load balancing as workers pick up the next shard.
+	var total int64
+	g, gCtx := errgroup.WithContext(ctx)
+	g.SetLimit(f.opt.ListParallelism)
+	for _, sh := range shards {
+		if gCtx.Err() != nil {
+			break
+		}
+		g.Go(func() error {
+			// Clone opts per shard so concurrent shards don't share StartFrom/EndBefore.
+			shardOpts := *opts
+			if sh.startFrom != "" {
+				startFrom := sh.startFrom
+				shardOpts.StartFrom = &startFrom
+			} else {
+				shardOpts.StartFrom = nil
+			}
+			if sh.endBefore != "" {
+				endBefore := sh.endBefore
+				shardOpts.EndBefore = &endBefore
+			} else {
+				shardOpts.EndBefore = nil
+			}
+			n, err := f.listBlobsPager(gCtx, containerName, directory, prefix, addContainer, &shardOpts, delimiter, safeFn)
+			atomic.AddInt64(&total, int64(n))
+			return err
+		})
+	}
+	err := g.Wait()
+	found := int(atomic.LoadInt64(&total))
+	if isEndBeforeUnsupported(err) {
+		// The account can't honour the shards' endBefore bounds, so parallel
+		// listing can't be used on it.
+		f.arrowXMLFallback.Store(true)
+		if found == 0 {
+			// No shard delivered anything yet (the normal case - Arrow listing
+			// enablement is per account so every shard fails on its first
+			// page) so it is safe to relist sequentially.
+			fs.Debugf(f, "Parallel Arrow listing not supported by this account (%v) - listing sequentially", err)
+			return f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn)
+		}
+		// Some entries were already delivered so a rerun would duplicate them.
+		return found, fmt.Errorf("parallel listing aborted after partial results (retry, or unset list_parallelism): %w", err)
+	}
+	return found, err
+}
+
+// isEndBeforeUnsupported reports whether err means the service can't honour
+// the endBefore listing bound, so sharded parallel listing must not be used.
+//
+// This happens two ways: the arrowlist pager's ErrEndBeforeXMLFallback (the
+// server answered XML with endBefore set), or the service rejecting the
+// endBefore parameter with 400 OperationNotSupportedWithFeatureMissing "The
+// requested operation is not allowed as EndBefore parameter support is
+// missing on this account" - endBefore is only supported on the Arrow
+// listing path of accounts with Blob Listing with Apache Arrow enabled.
+func isEndBeforeUnsupported(err error) bool {
+	if errors.Is(err, arrowlist.ErrEndBeforeXMLFallback) {
+		return true
+	}
+	var respErr *azcore.ResponseError
+	return errors.As(err, &respErr) && respErr.ErrorCode == "OperationNotSupportedWithFeatureMissing"
+}
+
+// arrowShardTarget is the number of ladder boundaries to use for the given
+// parallelism (a few per worker for load balancing).
+func arrowShardTarget(parallelism int) int {
+	return parallelism * 4
+}
+
+// arrowLadderBoundaries returns up to target sorted ENCODED-space boundaries of
+// the form directory+c, used to split the keyspace into shards.
+//
+// The partition is always correct (no dup/gap) regardless of the alphabet: keys
+// whose next byte is outside the alphabet simply fall into an adjacent shard;
+// the alphabet only affects balance. A single-case (digit+lowercase) alphabet
+// is deliberate: the boundaries are then monotonic under BOTH byte order (the
+// listing order) AND the service's case-insensitive startFrom/endBefore
+// validation, so server-side endBefore is accepted. (A mixed-case ladder gives
+// pairs like [".../Z", ".../a") which are byte-ascending but rejected 400 as
+// "endbefore precedes startfrom".) directory is already encoded; the alphabet
+// is ASCII (encoder-invariant) and ascending, so the result is sorted.
+func arrowLadderBoundaries(directory string, target int) []string {
+	const alphabet = "0123456789abcdefghijklmnopqrstuvwxyz"
+	if target < 1 {
+		return nil
+	}
+	if target > len(alphabet) {
+		target = len(alphabet)
+	}
+	points := make([]string, 0, target)
+	last := -1
+	for i := 0; i < target; i++ {
+		idx := i * len(alphabet) / target
+		if idx == last {
+			continue
+		}
+		last = idx
+		points = append(points, directory+string(alphabet[idx]))
+	}
+	return points
+}

backend/azureblob/arrow_internal_test.go

@@ -4,9 +4,13 @@ package azureblob
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"strings"
 	"testing"
 
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/rclone/rclone/backend/azureblob/arrowlist"
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fstest"
 	"github.com/rclone/rclone/fstest/fstests"
@@ -34,6 +38,14 @@ func (f *Fs) testArrowList(t *testing.T) {
 		fn()
 	}
 
+	// withParallel runs fn with ListParallelism temporarily set.
+	withParallel := func(n int, fn func()) {
+		old := f.opt.ListParallelism
+		f.opt.ListParallelism = n
+		defer func() { f.opt.ListParallelism = 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)
@@ -130,4 +142,80 @@ func (f *Fs) testArrowList(t *testing.T) {
 		}
 		assert.True(t, foundSubDir, "expected %q to be listed as a directory", dir+"/sub")
 	})
+
+	// Parallel listing must equal sequential listing, recursively and
+	// per-directory, including the "sub" subdirectory.
+	t.Run("ParallelEqualsSequential", func(t *testing.T) {
+		for _, recurse := range []bool{true, false} {
+			var seq, par map[string]int64
+			var err error
+			withArrow(true, func() {
+				withParallel(0, func() { seq, err = listing(recurse) })
+				require.NoError(t, err)
+				withParallel(8, func() { par, err = listing(recurse) })
+				require.NoError(t, err)
+			})
+			assert.Equal(t, seq, par, "recurse=%v: parallel listing differs from sequential", recurse)
+		}
+	})
+
+	// A tiny list_chunk forces each shard to span multiple Arrow pages, so this
+	// exercises pagination (startFrom+endBefore+marker resent together) inside a
+	// shard. The "sub/" boundary puts a.txt..e.txt in one shard => >1 page at
+	// chunk size 2.
+	t.Run("ParallelMultiPage", func(t *testing.T) {
+		oldChunk := f.opt.ListChunkSize
+		f.opt.ListChunkSize = 2
+		defer func() { f.opt.ListChunkSize = oldChunk }()
+		var seq, par map[string]int64
+		var err error
+		withArrow(true, func() {
+			withParallel(0, func() { seq, err = listing(true) })
+			require.NoError(t, err)
+			withParallel(8, func() { par, err = listing(true) })
+			require.NoError(t, err)
+		})
+		assert.Equal(t, seq, par)
+		for remote, size := range wantSizes {
+			assert.Equal(t, size, par[remote], remote)
+		}
+	})
+}
+
+func TestArrowLadderBoundaries(t *testing.T) {
+	for _, tc := range []struct {
+		directory string
+		target    int
+	}{
+		{"", 8},
+		{"dir/", 16},
+		{"dir/", 100}, // capped at alphabet size
+		{"dir/", 1},
+	} {
+		points := arrowLadderBoundaries(tc.directory, tc.target)
+		require.NotEmpty(t, points)
+		assert.LessOrEqual(t, len(points), 62)
+		for i, p := range points {
+			assert.True(t, strings.HasPrefix(p, tc.directory), "boundary %q lacks prefix %q", p, tc.directory)
+			if i > 0 {
+				assert.Less(t, points[i-1], p, "boundaries not strictly ascending")
+			}
+		}
+	}
+	assert.Nil(t, arrowLadderBoundaries("dir/", 0))
+	// alphabet is single-case (digit+lowercase) so boundaries are monotonic
+	// under both byte order and case-insensitive order
+	pts := arrowLadderBoundaries("dir/", 36)
+	for _, p := range pts {
+		assert.Equal(t, p, strings.ToLower(p), "boundary must be lowercase: %q", p)
+	}
+}
+
+func TestIsEndBeforeUnsupported(t *testing.T) {
+	assert.True(t, isEndBeforeUnsupported(arrowlist.ErrEndBeforeXMLFallback))
+	assert.True(t, isEndBeforeUnsupported(fmt.Errorf("wrapped: %w", arrowlist.ErrEndBeforeXMLFallback)))
+	assert.True(t, isEndBeforeUnsupported(&azcore.ResponseError{ErrorCode: "OperationNotSupportedWithFeatureMissing"}))
+	assert.False(t, isEndBeforeUnsupported(&azcore.ResponseError{ErrorCode: "ContainerNotFound"}))
+	assert.False(t, isEndBeforeUnsupported(errors.New("some other error")))
+	assert.False(t, isEndBeforeUnsupported(nil))
 }

backend/azureblob/azureblob.go

@@ -22,6 +22,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
@@ -258,6 +259,22 @@ listing.`,
 			Default:  false,
 			Advanced: true,
 			Hide:     fs.OptionHideBoth,
+		}, {
+			Name: "list_parallelism",
+			Help: `Number of parallel shards to list a directory with.
+
+EXPERIMENTAL. If set greater than 1, the blob name keyspace of each
+directory is split into this many ranges which are listed concurrently
+using the Arrow startFrom/endBefore range parameters. This can
+dramatically speed up listing containers with millions of objects, for
+both recursive (ListR) and single directory listings.
+
+This has no effect unless "use_arrow_list" is also set, as Arrow is the
+only listing path that supports server-side name ranges. The default of
+0 (or 1) lists sequentially.`,
+			Default:  0,
+			Advanced: true,
+			Hide:     fs.OptionHideBoth,
 		}, {
 			Name: "access_tier",
 			Help: `Access tier of blob: hot, cool, cold or archive.
@@ -412,6 +429,7 @@ type Options struct {
 	UploadConcurrency    int                  `config:"upload_concurrency"`
 	ListChunkSize        uint                 `config:"list_chunk"`
 	UseArrowList         bool                 `config:"use_arrow_list"`
+	ListParallelism      int                  `config:"list_parallelism"`
 	AccessTier           string               `config:"access_tier"`
 	ArchiveTierDelete    bool                 `config:"archive_tier_delete"`
 	DisableCheckSum      bool                 `config:"disable_checksum"`
@@ -435,6 +453,7 @@ type Fs struct {
 	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
+	arrowXMLFallback   atomic.Bool                  // set once the server answers XML so parallel listing is not retried
 	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
@@ -634,6 +653,9 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e
 	if opt.ListChunkSize > maxListChunkSize {
 		return nil, fmt.Errorf("blob list size can't be greater than %v - was %v", maxListChunkSize, opt.ListChunkSize)
 	}
+	if opt.ListParallelism > 1 && !opt.UseArrowList {
+		fs.Logf(nil, "azureblob: list_parallelism has no effect without use_arrow_list - listing sequentially")
+	}
 
 	if opt.AccessTier == "" {
 		opt.AccessTier = string(defaultAccessTier)
@@ -1188,6 +1210,40 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string,
 		opts.UseArrowFormat = to.Ptr(true)
 	}
 
+	var foundItems int
+	var err error
+	if useArrow && f.opt.ListParallelism > 1 && !f.arrowXMLFallback.Load() {
+		// Arrow exposes server-side startFrom/endBefore name ranges, so the
+		// keyspace can be sharded and listed concurrently.
+		foundItems, err = f.listArrowParallel(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn)
+	} else {
+		foundItems, err = f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn)
+	}
+	if err != nil {
+		return err
+	}
+	if f.opt.DirectoryMarkers && foundItems == 0 && directory != "" {
+		// Determine whether the directory exists or not by whether it has a marker
+		_, err := f.readMetaData(ctx, containerName, directory)
+		if err != nil {
+			if err == fs.ErrorObjectNotFound {
+				return fs.ErrorDirNotFound
+			}
+			return err
+		}
+	}
+	return nil
+}
+
+// listBlobsPager runs the hierarchy listing described by opts, calling fn for
+// each blob and subdirectory, and returns the number of raw items seen.
+// delimiter selects flat (recurse) vs hierarchical listing. If opts requests
+// the Apache Arrow format the listing goes through the arrowlist pager
+// (falling back to the SDK pager if the credentials don't support it); if the
+// service then answers with XML (Arrow listing not enabled) a debug message
+// is logged.
+func (f *Fs) listBlobsPager(ctx context.Context, containerName, directory, prefix string, addContainer bool, opts *arrowlist.ListBlobsHierarchyOptions, delimiter string, fn listFn) (foundItems int, err error) {
+	useArrow := opts.UseArrowFormat != nil && *opts.UseArrowFormat
 	var pager *runtime.Pager[container.ListBlobsHierarchyResponse]
 	if useArrow {
 		arrowSVC, err := f.arrowCntSVC(containerName)
@@ -1201,7 +1257,6 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string,
 	if pager == nil {
 		pager = f.cntSVC(containerName).NewListBlobsHierarchyPager(delimiter, &opts.ListBlobsHierarchyOptions)
 	}
-	foundItems := 0
 	checkedArrow := false
 	for pager.More() {
 		var response container.ListBlobsHierarchyResponse
@@ -1225,9 +1280,9 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string,
 		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) {
-				return fs.ErrorDirNotFound
+				return foundItems, fs.ErrorDirNotFound
 			}
-			return err
+			return foundItems, err
 		}
 		// Advance marker to next
 		// marker = response.NextMarker
@@ -1268,7 +1323,7 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string,
 			// Send object
 			err = fn(remote, file, isDirectory)
 			if err != nil {
-				return err
+				return foundItems, err
 			}
 		}
 		// Send the subdirectories
@@ -1302,21 +1357,11 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string,
 			// Send object
 			err = fn(remote, nil, true)
 			if err != nil {
-				return err
+				return foundItems, err
 			}
 		}
 	}
-	if f.opt.DirectoryMarkers && foundItems == 0 && directory != "" {
-		// Determine whether the directory exists or not by whether it has a marker
-		_, err := f.readMetaData(ctx, containerName, directory)
-		if err != nil {
-			if err == fs.ErrorObjectNotFound {
-				return fs.ErrorDirNotFound
-			}
-			return err
-		}
-	}
-	return nil
+	return foundItems, nil
 }
 
 // Convert a list item into a DirEntry