azureblob: add arrowlist package for Apache Arrow blob listing

This adds a temporary subpackage implementing the experimental "Blob
Listing with Apache Arrow" feature on top of the released Azure azblob
SDK.

The Azure SDK for Go supports Arrow listing on its unreleased
feature/storage/bifrost branch (commit c6fa341ca22b) but that can't be
used in a release. This package exposes the same options and pager
interface as the experimental SDK, built on the released SDK's public
API plus verbatim copies of its Arrow decoder and of the two internal
auth policies (shared key signing and storage bearer challenge) which
the released SDK does not export. Copied files carry MIT attribution
headers pinned to their upstream source.

When Arrow listing ships in a released azblob the whole package will be
deleted and callers pointed back at the SDK - the interface is
deliberately source compatible to make that a mechanical change.

Also exports auth.Transporter so the new package's pipeline can use
rclone's fshttp transport, and adds github.com/apache/arrow-go/v18 as a
direct dependency for decoding the Arrow IPC stream.

backend/azureblob/arrowlist/arrow.go

@@ -0,0 +1,421 @@
+//go:build !plan9 && !solaris && !js
+
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+//
+// Adapted from internal/arrow/arrow.go in the Azure SDK for Go
+// github.com/Azure/azure-sdk-for-go/sdk/storage/azblob at commit
+// c6fa341ca22b (branch feature/storage/bifrost), retyped onto the SDK's
+// public type aliases so it compiles against the released azblob module.
+
+package arrowlist
+
+import (
+	"encoding/base64"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/lease"
+	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/ipc"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+)
+
+const (
+	// ArrowContentType is the Content-Type for Apache Arrow IPC stream responses.
+	ArrowContentType = "application/vnd.apache.arrow.stream"
+
+	// ArrowAcceptHeader is the Accept header value to request Arrow format.
+	ArrowAcceptHeader = ArrowContentType
+
+	// resourceTypeBlobPrefix is the ResourceType value that identifies a virtual directory prefix
+	// in hierarchy listing responses. In XML, prefixes are separate <BlobPrefix> elements;
+	// in Arrow, all rows share the same schema and are distinguished by this field value.
+	resourceTypeBlobPrefix = "blobprefix"
+)
+
+// HandleFlatListResponse parses an Arrow IPC stream response into a ContainerClientListBlobFlatSegmentResponse.
+func HandleFlatListResponse(resp *http.Response) (container.ListBlobsFlatResponse, error) {
+	result := container.ListBlobsFlatResponse{}
+
+	// Extract response headers
+	extractResponseHeaders(resp, &result.ClientRequestID, &result.ContentType, &result.Date, &result.RequestID, &result.Version)
+
+	// Parse Arrow IPC stream
+	items, nextMarker, err := parseArrowStream(resp.Body)
+	if err != nil {
+		return result, fmt.Errorf("failed to parse Arrow IPC response: %w", err)
+	}
+
+	result.ListBlobsFlatSegmentResponse = container.ListBlobsFlatSegmentResponse{
+		Segment: &container.BlobFlatListSegment{
+			BlobItems: items,
+		},
+		NextMarker: nextMarker,
+	}
+
+	return result, nil
+}
+
+// HandleHierarchyListResponse parses an Arrow IPC stream response into a ContainerClientListBlobHierarchySegmentResponse.
+func HandleHierarchyListResponse(resp *http.Response) (container.ListBlobsHierarchyResponse, error) {
+	result := container.ListBlobsHierarchyResponse{}
+
+	extractResponseHeaders(resp, &result.ClientRequestID, &result.ContentType, &result.Date, &result.RequestID, &result.Version)
+
+	items, nextMarker, err := parseArrowStream(resp.Body)
+	if err != nil {
+		return result, fmt.Errorf("failed to parse Arrow IPC response: %w", err)
+	}
+
+	// Separate blob items from blob prefixes based on ResourceType field.
+	var blobItems []*container.BlobItem
+	var blobPrefixes []*container.BlobPrefix
+	for _, item := range items {
+		if item.Properties != nil && item.Properties.ResourceType != nil && *item.Properties.ResourceType == resourceTypeBlobPrefix {
+			blobPrefixes = append(blobPrefixes, &container.BlobPrefix{
+				Name:       item.Name,
+				Properties: item.Properties,
+			})
+		} else {
+			blobItems = append(blobItems, item)
+		}
+	}
+
+	result.ListBlobsHierarchySegmentResponse = container.ListBlobsHierarchySegmentResponse{
+		Segment: &container.BlobHierarchyListSegment{
+			BlobItems:    blobItems,
+			BlobPrefixes: blobPrefixes,
+		},
+		NextMarker: nextMarker,
+	}
+
+	return result, nil
+}
+
+func extractResponseHeaders(resp *http.Response, clientRequestID, contentType **string, date **time.Time, requestID, version **string) {
+	if val := resp.Header.Get("x-ms-client-request-id"); val != "" {
+		*clientRequestID = &val
+	}
+	if val := resp.Header.Get("Content-Type"); val != "" {
+		*contentType = &val
+	}
+	if val := resp.Header.Get("Date"); val != "" {
+		if t, err := time.Parse(time.RFC1123, val); err == nil {
+			*date = &t
+		}
+	}
+	if val := resp.Header.Get("x-ms-request-id"); val != "" {
+		*requestID = &val
+	}
+	if val := resp.Header.Get("x-ms-version"); val != "" {
+		*version = &val
+	}
+}
+
+// parseArrowStream reads an Arrow IPC stream and converts it to BlobItems.
+func parseArrowStream(body io.Reader) ([]*container.BlobItem, *string, error) {
+	reader, err := ipc.NewReader(body, ipc.WithAllocator(memory.DefaultAllocator))
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create Arrow IPC reader: %w", err)
+	}
+	defer reader.Release()
+
+	// Extract NextMarker from schema metadata
+	var nextMarker *string
+	var numRecords int
+	if md := reader.Schema().Metadata(); md.Len() > 0 {
+		if idx := md.FindKey("NextMarker"); idx >= 0 {
+			val := md.Values()[idx]
+			if val != "" {
+				nextMarker = &val
+			}
+		}
+		if idx := md.FindKey("NumberOfRecords"); idx >= 0 {
+			if n, err := strconv.Atoi(md.Values()[idx]); err == nil {
+				numRecords = n
+			}
+		}
+	}
+
+	// Build column index for the schema
+	schema := reader.Schema()
+	colIndex := buildColumnIndex(schema)
+
+	// Pre-allocate if we know the count
+	items := make([]*container.BlobItem, 0, numRecords)
+
+	// Iterate over record batches
+	for reader.Next() {
+		rec := reader.RecordBatch()
+		rows := int(rec.NumRows())
+
+		for row := 0; row < rows; row++ {
+			item := &container.BlobItem{
+				Properties: &container.BlobProperties{},
+			}
+			populateBlobItem(item, rec, row, colIndex)
+			items = append(items, item)
+		}
+	}
+
+	if err := reader.Err(); err != nil {
+		return nil, nil, fmt.Errorf("error reading Arrow record batches: %w", err)
+	}
+
+	return items, nextMarker, nil
+}
+
+// columnIndex maps column names to their indices in the record batch for O(1) lookup.
+type columnIndex map[string]int
+
+func buildColumnIndex(schema *arrow.Schema) columnIndex {
+	idx := make(columnIndex, len(schema.Fields()))
+	for i, f := range schema.Fields() {
+		idx[f.Name] = i
+	}
+	return idx
+}
+
+// populateBlobItem fills a BlobItem from a single row across all columns.
+func populateBlobItem(item *container.BlobItem, rec arrow.RecordBatch, row int, colIdx columnIndex) {
+	// Top-level BlobItem fields
+	item.Name = getStringField(rec, row, colIdx, "Name")
+	item.Deleted = getBoolField(rec, row, colIdx, "Deleted")
+	item.Snapshot = getStringField(rec, row, colIdx, "Snapshot")
+	item.VersionID = getStringField(rec, row, colIdx, "VersionId")
+	item.IsCurrentVersion = getBoolField(rec, row, colIdx, "IsCurrentVersion")
+	item.HasVersionsOnly = getBoolField(rec, row, colIdx, "HasVersionsOnly")
+
+	// Metadata maps
+	item.Metadata = getMapField(rec, row, colIdx, "Metadata")
+	item.OrMetadata = getMapField(rec, row, colIdx, "OrMetadata")
+
+	// Tags
+	if tags := getMapField(rec, row, colIdx, "Tags"); tags != nil {
+		var blobTags []*container.BlobTag
+		for k, v := range tags {
+			key := k
+			blobTags = append(blobTags, &container.BlobTag{Key: &key, Value: v})
+		}
+		item.BlobTags = &container.BlobTags{BlobTagSet: blobTags}
+	}
+
+	// BlobProperties fields
+	p := item.Properties
+	p.CreationTime = getTimestampField(rec, row, colIdx, "Creation-Time")
+	p.LastModified = getTimestampField(rec, row, colIdx, "Last-Modified")
+	p.ETag = getETagField(rec, row, colIdx, "Etag")
+	p.ContentLength = getInt64Field(rec, row, colIdx, "Content-Length")
+	p.ContentType = getStringField(rec, row, colIdx, "Content-Type")
+	p.ContentEncoding = getStringField(rec, row, colIdx, "Content-Encoding")
+	p.ContentLanguage = getStringField(rec, row, colIdx, "Content-Language")
+	p.ContentDisposition = getStringField(rec, row, colIdx, "Content-Disposition")
+	p.CacheControl = getStringField(rec, row, colIdx, "Cache-Control")
+	p.ContentMD5 = getBase64BytesField(rec, row, colIdx, "Content-MD5")
+	p.BlobType = getEnumField[container.BlobType](rec, row, colIdx, "BlobType")
+	p.AccessTier = getEnumField[container.AccessTier](rec, row, colIdx, "AccessTier")
+	p.AccessTierInferred = getBoolField(rec, row, colIdx, "AccessTierInferred")
+	p.AccessTierChangeTime = getTimestampField(rec, row, colIdx, "AccessTierChangeTime")
+	p.LeaseState = getEnumField[lease.StateType](rec, row, colIdx, "LeaseState")
+	p.LeaseStatus = getEnumField[lease.StatusType](rec, row, colIdx, "LeaseStatus")
+	p.LeaseDuration = getEnumField[lease.DurationType](rec, row, colIdx, "LeaseDuration")
+	p.ServerEncrypted = getBoolField(rec, row, colIdx, "ServerEncrypted")
+	p.CustomerProvidedKeySHA256 = getStringField(rec, row, colIdx, "CustomerProvidedKeySha256")
+	p.EncryptionScope = getStringField(rec, row, colIdx, "EncryptionScope")
+	p.IncrementalCopy = getBoolField(rec, row, colIdx, "IncrementalCopy")
+	p.IsSealed = getBoolField(rec, row, colIdx, "Sealed")
+	p.ArchiveStatus = getEnumField[container.ArchiveStatus](rec, row, colIdx, "ArchiveStatus")
+	p.RehydratePriority = getEnumField[blob.RehydratePriority](rec, row, colIdx, "RehydratePriority")
+	p.CopyID = getStringField(rec, row, colIdx, "CopyId")
+	p.CopyStatus = getEnumField[blob.CopyStatusType](rec, row, colIdx, "CopyStatus")
+	p.CopySource = getStringField(rec, row, colIdx, "CopySource")
+	p.CopyProgress = getStringField(rec, row, colIdx, "CopyProgress")
+	p.CopyCompletionTime = getTimestampField(rec, row, colIdx, "CopyCompletionTime")
+	p.CopyStatusDescription = getStringField(rec, row, colIdx, "CopyStatusDescription")
+	p.DestinationSnapshot = getStringField(rec, row, colIdx, "CopyDestinationSnapshot")
+	p.ImmutabilityPolicyExpiresOn = getTimestampField(rec, row, colIdx, "ImmutabilityPolicyUntilDate")
+	p.ImmutabilityPolicyMode = getEnumField[container.ImmutabilityPolicyMode](rec, row, colIdx, "ImmutabilityPolicyMode")
+	p.LegalHold = getBoolField(rec, row, colIdx, "LegalHold")
+	p.DeletedTime = getTimestampField(rec, row, colIdx, "DeletedTime")
+	p.RemainingRetentionDays = getInt32Field(rec, row, colIdx, "RemainingRetentionDays")
+	p.LastAccessedOn = getTimestampField(rec, row, colIdx, "LastAccessTime")
+	p.TagCount = getInt32Field(rec, row, colIdx, "TagCount")
+	p.BlobSequenceNumber = getInt64Field(rec, row, colIdx, "x-ms-blob-sequence-number")
+	p.ResourceType = getStringField(rec, row, colIdx, "ResourceType")
+
+	// Content-CRC64 is a string field in Arrow but not directly mapped to BlobProperties.
+	// The field exists in the Arrow schema but has no corresponding field in the Go SDK BlobProperties.
+	// It is silently ignored for forward compatibility.
+}
+
+// Field accessor helpers that safely handle missing columns and null values.
+
+func getStringField(rec arrow.RecordBatch, row int, colIdx columnIndex, name string) *string {
+	idx, ok := colIdx[name]
+	if !ok {
+		return nil
+	}
+	col := rec.Column(idx)
+	if col.IsNull(row) {
+		return nil
+	}
+	if arr, ok := col.(*array.String); ok {
+		val := arr.Value(row)
+		return &val
+	}
+	return nil
+}
+
+func getBoolField(rec arrow.RecordBatch, row int, colIdx columnIndex, name string) *bool {
+	idx, ok := colIdx[name]
+	if !ok {
+		return nil
+	}
+	col := rec.Column(idx)
+	if col.IsNull(row) {
+		return nil
+	}
+	if arr, ok := col.(*array.Boolean); ok {
+		val := arr.Value(row)
+		return &val
+	}
+	return nil
+}
+
+func getTimestampField(rec arrow.RecordBatch, row int, colIdx columnIndex, name string) *time.Time {
+	idx, ok := colIdx[name]
+	if !ok {
+		return nil
+	}
+	col := rec.Column(idx)
+	if col.IsNull(row) {
+		return nil
+	}
+	if arr, ok := col.(*array.Timestamp); ok {
+		val := arr.Value(row)
+		// Arrow timestamps are stored as int64 with a unit; convert to time.Time
+		dt := val.ToTime(arr.DataType().(*arrow.TimestampType).Unit)
+		return &dt
+	}
+	return nil
+}
+
+func getInt64Field(rec arrow.RecordBatch, row int, colIdx columnIndex, name string) *int64 {
+	idx, ok := colIdx[name]
+	if !ok {
+		return nil
+	}
+	col := rec.Column(idx)
+	if col.IsNull(row) {
+		return nil
+	}
+	if arr, ok := col.(*array.Uint64); ok {
+		val := int64(arr.Value(row))
+		return &val
+	}
+	if arr, ok := col.(*array.Int64); ok {
+		val := arr.Value(row)
+		return &val
+	}
+	return nil
+}
+
+func getInt32Field(rec arrow.RecordBatch, row int, colIdx columnIndex, name string) *int32 {
+	idx, ok := colIdx[name]
+	if !ok {
+		return nil
+	}
+	col := rec.Column(idx)
+	if col.IsNull(row) {
+		return nil
+	}
+	if arr, ok := col.(*array.Uint64); ok {
+		val := int32(arr.Value(row))
+		return &val
+	}
+	if arr, ok := col.(*array.Int32); ok {
+		val := arr.Value(row)
+		return &val
+	}
+	return nil
+}
+
+func getETagField(rec arrow.RecordBatch, row int, colIdx columnIndex, name string) *azcore.ETag {
+	s := getStringField(rec, row, colIdx, name)
+	if s == nil {
+		return nil
+	}
+	etag := azcore.ETag(*s)
+	return &etag
+}
+
+func getBase64BytesField(rec arrow.RecordBatch, row int, colIdx columnIndex, name string) []byte {
+	s := getStringField(rec, row, colIdx, name)
+	if s == nil {
+		return nil
+	}
+	decoded, err := base64.StdEncoding.DecodeString(*s)
+	if err != nil {
+		return nil
+	}
+	return decoded
+}
+
+// getEnumField returns a pointer to a string-typed enum value.
+type stringEnum interface {
+	~string
+}
+
+func getEnumField[T stringEnum](rec arrow.RecordBatch, row int, colIdx columnIndex, name string) *T {
+	s := getStringField(rec, row, colIdx, name)
+	if s == nil {
+		return nil
+	}
+	val := T(*s)
+	return &val
+}
+
+func getMapField(rec arrow.RecordBatch, row int, colIdx columnIndex, name string) map[string]*string {
+	idx, ok := colIdx[name]
+	if !ok {
+		return nil
+	}
+	col := rec.Column(idx)
+	if col.IsNull(row) {
+		return nil
+	}
+	mapArr, ok := col.(*array.Map)
+	if !ok {
+		return nil
+	}
+
+	// Get the offsets for this row's map entries
+	start := int(mapArr.Offsets()[row])
+	end := int(mapArr.Offsets()[row+1])
+	if start == end {
+		return nil
+	}
+
+	keys := mapArr.Keys().(*array.String)
+	values := mapArr.Items().(*array.String)
+
+	result := make(map[string]*string, end-start)
+	for i := start; i < end; i++ {
+		k := keys.Value(i)
+		if values.IsNull(i) {
+			result[k] = nil
+		} else {
+			v := values.Value(i)
+			result[k] = &v
+		}
+	}
+	return result
+}

backend/azureblob/arrowlist/arrow_test.go

@@ -0,0 +1,374 @@
+//go:build !plan9 && !solaris && !js
+
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+//
+// Adapted from internal/arrow/arrow_test.go in the Azure SDK for Go
+// github.com/Azure/azure-sdk-for-go/sdk/storage/azblob at commit
+// c6fa341ca22b (branch feature/storage/bifrost).
+
+package arrowlist
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"testing"
+
+	"github.com/apache/arrow-go/v18/arrow"
+	arrowArray "github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/ipc"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/stretchr/testify/require"
+)
+
+// coreSchema returns a basic Arrow schema for testing with common blob fields.
+func coreSchema(md *arrow.Metadata) *arrow.Schema {
+	fields := []arrow.Field{
+		{Name: "Name", Type: arrow.BinaryTypes.String, Nullable: false},
+		{Name: "Creation-Time", Type: &arrow.TimestampType{Unit: arrow.Microsecond}, Nullable: true},
+		{Name: "Last-Modified", Type: &arrow.TimestampType{Unit: arrow.Microsecond}, Nullable: true},
+		{Name: "BlobType", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "Etag", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "Content-Length", Type: arrow.PrimitiveTypes.Uint64, Nullable: true},
+		{Name: "Content-Type", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "ServerEncrypted", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+		{Name: "AccessTier", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "LeaseState", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "LeaseStatus", Type: arrow.BinaryTypes.String, Nullable: true},
+	}
+	return arrow.NewSchema(fields, md)
+}
+
+// buildArrowStream constructs a synthetic Arrow IPC stream from a schema and a populate function.
+func buildArrowStream(t *testing.T, schema *arrow.Schema, populate func(builder *arrowArray.RecordBuilder)) []byte {
+	t.Helper()
+	alloc := memory.DefaultAllocator
+
+	var buf bytes.Buffer
+	w := ipc.NewWriter(&buf, ipc.WithSchema(schema), ipc.WithAllocator(alloc))
+
+	builder := arrowArray.NewRecordBuilder(alloc, schema)
+	defer builder.Release()
+
+	populate(builder)
+
+	rec := builder.NewRecordBatch()
+	defer rec.Release()
+
+	require.NoError(t, w.Write(rec))
+	require.NoError(t, w.Close())
+
+	return buf.Bytes()
+}
+
+func makeHTTPResponse(data []byte, headers map[string]string) *http.Response {
+	h := http.Header{}
+	h.Set("Content-Type", ArrowContentType)
+	for k, v := range headers {
+		h.Set(k, v)
+	}
+	return &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     h,
+		Body:       io.NopCloser(bytes.NewReader(data)),
+	}
+}
+
+func TestHandleFlatListResponse_BasicParsing(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker", "NumberOfRecords"}, []string{"marker123", "2"})
+	schema := coreSchema(&md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		b.Field(0).(*arrowArray.StringBuilder).Append("blob1.txt")
+		b.Field(1).(*arrowArray.TimestampBuilder).Append(arrow.Timestamp(1700000000000000))
+		b.Field(2).(*arrowArray.TimestampBuilder).Append(arrow.Timestamp(1700000000000000))
+		b.Field(3).(*arrowArray.StringBuilder).Append("BlockBlob")
+		b.Field(4).(*arrowArray.StringBuilder).Append("0x1234")
+		b.Field(5).(*arrowArray.Uint64Builder).Append(1024)
+		b.Field(6).(*arrowArray.StringBuilder).Append("application/octet-stream")
+		b.Field(7).(*arrowArray.BooleanBuilder).Append(true)
+		b.Field(8).(*arrowArray.StringBuilder).Append("Hot")
+		b.Field(9).(*arrowArray.StringBuilder).Append("available")
+		b.Field(10).(*arrowArray.StringBuilder).Append("unlocked")
+
+		b.Field(0).(*arrowArray.StringBuilder).Append("blob2.txt")
+		b.Field(1).(*arrowArray.TimestampBuilder).Append(arrow.Timestamp(1700000001000000))
+		b.Field(2).(*arrowArray.TimestampBuilder).Append(arrow.Timestamp(1700000001000000))
+		b.Field(3).(*arrowArray.StringBuilder).Append("BlockBlob")
+		b.Field(4).(*arrowArray.StringBuilder).Append("0x5678")
+		b.Field(5).(*arrowArray.Uint64Builder).Append(2048)
+		b.Field(6).(*arrowArray.StringBuilder).Append("text/plain")
+		b.Field(7).(*arrowArray.BooleanBuilder).Append(false)
+		b.Field(8).(*arrowArray.StringBuilder).Append("Cool")
+		b.Field(9).(*arrowArray.StringBuilder).Append("available")
+		b.Field(10).(*arrowArray.StringBuilder).Append("unlocked")
+	})
+
+	resp := makeHTTPResponse(data, nil)
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+
+	require.NotNil(t, result.NextMarker)
+	require.Equal(t, "marker123", *result.NextMarker)
+	require.Len(t, result.Segment.BlobItems, 2)
+
+	item := result.Segment.BlobItems[0]
+	require.Equal(t, "blob1.txt", *item.Name)
+	require.Equal(t, "BlockBlob", string(*item.Properties.BlobType))
+	require.Equal(t, int64(1024), *item.Properties.ContentLength)
+	require.Equal(t, "Hot", string(*item.Properties.AccessTier))
+}
+
+func TestHandleFlatListResponse_EmptyRecordBatch(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker", "NumberOfRecords"}, []string{"", "0"})
+	schema := coreSchema(&md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		// No rows added
+	})
+
+	resp := makeHTTPResponse(data, nil)
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+	require.Nil(t, result.NextMarker)
+	require.Empty(t, result.Segment.BlobItems)
+}
+
+func TestHandleFlatListResponse_NullFields(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker"}, []string{""})
+	schema := coreSchema(&md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		b.Field(0).(*arrowArray.StringBuilder).Append("blob.txt")
+		b.Field(1).(*arrowArray.TimestampBuilder).AppendNull()
+		b.Field(2).(*arrowArray.TimestampBuilder).AppendNull()
+		b.Field(3).(*arrowArray.StringBuilder).AppendNull()
+		b.Field(4).(*arrowArray.StringBuilder).AppendNull()
+		b.Field(5).(*arrowArray.Uint64Builder).AppendNull()
+		b.Field(6).(*arrowArray.StringBuilder).AppendNull()
+		b.Field(7).(*arrowArray.BooleanBuilder).AppendNull()
+		b.Field(8).(*arrowArray.StringBuilder).AppendNull()
+		b.Field(9).(*arrowArray.StringBuilder).AppendNull()
+		b.Field(10).(*arrowArray.StringBuilder).AppendNull()
+	})
+
+	resp := makeHTTPResponse(data, nil)
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+	require.Len(t, result.Segment.BlobItems, 1)
+
+	item := result.Segment.BlobItems[0]
+	require.Equal(t, "blob.txt", *item.Name)
+	require.Nil(t, item.Properties.CreationTime)
+	require.Nil(t, item.Properties.BlobType)
+	require.Nil(t, item.Properties.ETag)
+	require.Nil(t, item.Properties.ContentLength)
+	require.Nil(t, item.Properties.ContentType)
+	require.Nil(t, item.Properties.ServerEncrypted)
+}
+
+func TestHandleFlatListResponse_UnknownColumnsIgnored(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker"}, []string{""})
+	fields := []arrow.Field{
+		{Name: "Name", Type: arrow.BinaryTypes.String, Nullable: false},
+		{Name: "FutureField", Type: arrow.BinaryTypes.String, Nullable: true},
+	}
+	schema := arrow.NewSchema(fields, &md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		b.Field(0).(*arrowArray.StringBuilder).Append("blob.txt")
+		b.Field(1).(*arrowArray.StringBuilder).Append("some-future-value")
+	})
+
+	resp := makeHTTPResponse(data, nil)
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+	require.Len(t, result.Segment.BlobItems, 1)
+	require.Equal(t, "blob.txt", *result.Segment.BlobItems[0].Name)
+}
+
+func TestHandleFlatListResponse_MultiRecordBatch(t *testing.T) {
+	alloc := memory.DefaultAllocator
+	md := arrow.NewMetadata([]string{"NextMarker", "NumberOfRecords"}, []string{"", "3"})
+	schema := coreSchema(&md)
+
+	var buf bytes.Buffer
+	w := ipc.NewWriter(&buf, ipc.WithSchema(schema), ipc.WithAllocator(alloc))
+
+	// First batch: 2 rows
+	builder := arrowArray.NewRecordBuilder(alloc, schema)
+	for i := 0; i < 2; i++ {
+		builder.Field(0).(*arrowArray.StringBuilder).Append("batch1_blob")
+		for j := 1; j < 11; j++ {
+			builder.Field(j).AppendNull()
+		}
+	}
+	rec := builder.NewRecordBatch()
+	require.NoError(t, w.Write(rec))
+	rec.Release()
+	builder.Release()
+
+	// Second batch: 1 row
+	builder = arrowArray.NewRecordBuilder(alloc, schema)
+	builder.Field(0).(*arrowArray.StringBuilder).Append("batch2_blob")
+	for j := 1; j < 11; j++ {
+		builder.Field(j).AppendNull()
+	}
+	rec = builder.NewRecordBatch()
+	require.NoError(t, w.Write(rec))
+	rec.Release()
+	builder.Release()
+
+	require.NoError(t, w.Close())
+
+	resp := makeHTTPResponse(buf.Bytes(), nil)
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+	require.Len(t, result.Segment.BlobItems, 3)
+}
+
+func TestHandleHierarchyListResponse_SeparatesBlobsAndPrefixes(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker"}, []string{""})
+	fields := []arrow.Field{
+		{Name: "Name", Type: arrow.BinaryTypes.String, Nullable: false},
+		{Name: "ResourceType", Type: arrow.BinaryTypes.String, Nullable: true},
+	}
+	schema := arrow.NewSchema(fields, &md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		// A regular blob
+		b.Field(0).(*arrowArray.StringBuilder).Append("folder/file.txt")
+		b.Field(1).(*arrowArray.StringBuilder).AppendNull()
+
+		// A blob prefix (virtual directory)
+		b.Field(0).(*arrowArray.StringBuilder).Append("folder/subfolder/")
+		b.Field(1).(*arrowArray.StringBuilder).Append("blobprefix")
+	})
+
+	resp := makeHTTPResponse(data, nil)
+	result, err := HandleHierarchyListResponse(resp)
+	require.NoError(t, err)
+
+	require.Len(t, result.Segment.BlobItems, 1)
+	require.Equal(t, "folder/file.txt", *result.Segment.BlobItems[0].Name)
+
+	require.Len(t, result.Segment.BlobPrefixes, 1)
+	require.Equal(t, "folder/subfolder/", *result.Segment.BlobPrefixes[0].Name)
+}
+
+func TestHandleFlatListResponse_ResponseHeaders(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker"}, []string{""})
+	fields := []arrow.Field{
+		{Name: "Name", Type: arrow.BinaryTypes.String, Nullable: false},
+	}
+	schema := arrow.NewSchema(fields, &md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		b.Field(0).(*arrowArray.StringBuilder).Append("blob.txt")
+	})
+
+	resp := makeHTTPResponse(data, map[string]string{
+		"x-ms-request-id":        "req-123",
+		"x-ms-version":           "2026-10-06",
+		"x-ms-client-request-id": "client-req-456",
+		"Date":                   "Mon, 01 Jan 2024 00:00:00 GMT",
+	})
+
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+	require.Equal(t, "req-123", *result.RequestID)
+	require.Equal(t, "2026-10-06", *result.Version)
+	require.Equal(t, "client-req-456", *result.ClientRequestID)
+	require.NotNil(t, result.Date)
+}
+
+func TestHandleFlatListResponse_MapFields(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker"}, []string{""})
+	fields := []arrow.Field{
+		{Name: "Name", Type: arrow.BinaryTypes.String, Nullable: false},
+		{Name: "Metadata", Type: arrow.MapOf(arrow.BinaryTypes.String, arrow.BinaryTypes.String), Nullable: true},
+		{Name: "Tags", Type: arrow.MapOf(arrow.BinaryTypes.String, arrow.BinaryTypes.String), Nullable: true},
+	}
+	schema := arrow.NewSchema(fields, &md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		b.Field(0).(*arrowArray.StringBuilder).Append("blob.txt")
+
+		metaBuilder := b.Field(1).(*arrowArray.MapBuilder)
+		metaBuilder.Append(true)
+		metaBuilder.KeyBuilder().(*arrowArray.StringBuilder).Append("key1")
+		metaBuilder.ItemBuilder().(*arrowArray.StringBuilder).Append("value1")
+
+		tagBuilder := b.Field(2).(*arrowArray.MapBuilder)
+		tagBuilder.Append(true)
+		tagBuilder.KeyBuilder().(*arrowArray.StringBuilder).Append("env")
+		tagBuilder.ItemBuilder().(*arrowArray.StringBuilder).Append("prod")
+	})
+
+	resp := makeHTTPResponse(data, nil)
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+	require.Len(t, result.Segment.BlobItems, 1)
+
+	item := result.Segment.BlobItems[0]
+	require.NotNil(t, item.Metadata)
+	require.Equal(t, "value1", *item.Metadata["key1"])
+
+	require.NotNil(t, item.BlobTags)
+	require.Len(t, item.BlobTags.BlobTagSet, 1)
+}
+
+func TestHandleFlatListResponse_BooleanFields(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker"}, []string{""})
+	fields := []arrow.Field{
+		{Name: "Name", Type: arrow.BinaryTypes.String, Nullable: false},
+		{Name: "Deleted", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+		{Name: "IsCurrentVersion", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+		{Name: "ServerEncrypted", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+	}
+	schema := arrow.NewSchema(fields, &md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		b.Field(0).(*arrowArray.StringBuilder).Append("blob.txt")
+		b.Field(1).(*arrowArray.BooleanBuilder).Append(true)
+		b.Field(2).(*arrowArray.BooleanBuilder).Append(false)
+		b.Field(3).(*arrowArray.BooleanBuilder).Append(true)
+	})
+
+	resp := makeHTTPResponse(data, nil)
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+
+	item := result.Segment.BlobItems[0]
+	require.True(t, *item.Deleted)
+	require.False(t, *item.IsCurrentVersion)
+	require.True(t, *item.Properties.ServerEncrypted)
+}
+
+func TestHandleFlatListResponse_VersionAndSnapshot(t *testing.T) {
+	md := arrow.NewMetadata([]string{"NextMarker"}, []string{""})
+	fields := []arrow.Field{
+		{Name: "Name", Type: arrow.BinaryTypes.String, Nullable: false},
+		{Name: "VersionId", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "Snapshot", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "IsCurrentVersion", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+	}
+	schema := arrow.NewSchema(fields, &md)
+
+	data := buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		b.Field(0).(*arrowArray.StringBuilder).Append("versioned-blob.txt")
+		b.Field(1).(*arrowArray.StringBuilder).Append("2024-01-01T00:00:00.0000000Z")
+		b.Field(2).(*arrowArray.StringBuilder).Append("snap-123")
+		b.Field(3).(*arrowArray.BooleanBuilder).Append(true)
+	})
+
+	resp := makeHTTPResponse(data, nil)
+	result, err := HandleFlatListResponse(resp)
+	require.NoError(t, err)
+
+	item := result.Segment.BlobItems[0]
+	require.Equal(t, "2024-01-01T00:00:00.0000000Z", *item.VersionID)
+	require.Equal(t, "snap-123", *item.Snapshot)
+	require.True(t, *item.IsCurrentVersion)
+}

backend/azureblob/arrowlist/arrowlist.go

@@ -0,0 +1,259 @@
+//go:build !plan9 && !solaris && !js
+
+// Package arrowlist implements the experimental "Blob Listing with Apache
+// Arrow" feature on top of the released Azure azblob SDK.
+//
+// The Azure SDK for Go has support for Arrow listing on the unreleased
+// feature/storage/bifrost branch (commit c6fa341ca22b) where the listing
+// pagers decode Arrow IPC stream responses into the ordinary ListBlobs
+// response types. Until that ships in a tagged azblob release this package
+// fills the gap: it exposes the same options and pager interface as the
+// experimental SDK, built only on the public API of the released SDK plus
+// copies of two small internal auth policies (see sharedkey.go and
+// challenge_policy.go).
+//
+// This package is temporary. When Arrow listing ships in a released azblob
+// the whole package should be deleted and its callers pointed back at
+// container.Client.NewListBlobsHierarchyPager - the option and response
+// types here are deliberately source compatible with the SDK's to make that
+// a mechanical change.
+package arrowlist
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"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/storage/azblob/container"
+)
+
+const (
+	// moduleName and moduleVersion identify this package in the azcore
+	// telemetry policy's User-Agent fragment.
+	moduleName    = "github.com/rclone/rclone/backend/azureblob/arrowlist"
+	moduleVersion = "v0.0.1"
+
+	// serviceVersion is the x-ms-version sent with listing requests. Arrow
+	// responses and the startFrom/endBefore parameters need at least
+	// 2026-06-06.
+	serviceVersion = "2026-06-06"
+
+	// tokenScope is the OAuth scope for Azure Storage, as used by the SDK's
+	// own clients.
+	tokenScope = "https://storage.azure.com/.default"
+)
+
+// ErrEndBeforeXMLFallback is returned by the pager when the server answered
+// with XML instead of Arrow while EndBefore was set. The service only honours
+// endBefore on the Arrow listing path, so an XML page could silently contain
+// blobs outside the requested range and cannot be used.
+var ErrEndBeforeXMLFallback = errors.New("arrow listing fell back to XML with endBefore set")
+
+// ListBlobsHierarchyOptions provides the configuration for
+// Client.NewListBlobsHierarchyPager.
+//
+// It mirrors the experimental SDK's container.ListBlobsHierarchyOptions:
+// the released SDK's options are embedded (providing Include, Marker,
+// MaxResults, Prefix and StartFrom by promotion) with the two fields the
+// released SDK lacks added alongside.
+type ListBlobsHierarchyOptions struct {
+	container.ListBlobsHierarchyOptions
+	// EndBefore limits listing to blobs whose full container path is
+	// lexically before it (exclusive). Only honoured on Arrow responses.
+	EndBefore *string
+	// UseArrowFormat requests the response as an Apache Arrow IPC stream.
+	UseArrowFormat *bool
+}
+
+// ClientOptions contains the optional parameters for client creation.
+type ClientOptions struct {
+	azcore.ClientOptions
+}
+
+// Client issues Arrow listing requests to a single container.
+type Client struct {
+	endpoint string
+	azClient *azcore.Client
+}
+
+// newClient creates a Client for the container at containerURL sending
+// requests through a pipeline with the given auth policy (which may be nil
+// for anonymous or SAS access).
+func newClient(containerURL string, authPolicy policy.Policy, options *ClientOptions) (*Client, error) {
+	if options == nil {
+		options = &ClientOptions{}
+	}
+	plOpts := runtime.PipelineOptions{}
+	if authPolicy != nil {
+		plOpts.PerRetry = []policy.Policy{authPolicy}
+	}
+	azClient, err := azcore.NewClient(moduleName, moduleVersion, plOpts, &options.ClientOptions)
+	if err != nil {
+		return nil, err
+	}
+	return &Client{
+		endpoint: containerURL,
+		azClient: azClient,
+	}, nil
+}
+
+// NewClient creates a Client authenticating with a token credential.
+func NewClient(containerURL string, cred azcore.TokenCredential, options *ClientOptions) (*Client, error) {
+	return newClient(containerURL, NewStorageChallengePolicy(cred, tokenScope, false), options)
+}
+
+// NewClientWithNoCredential creates a Client with no authentication, for
+// anonymous access or a containerURL carrying a SAS token in its query.
+func NewClientWithNoCredential(containerURL string, options *ClientOptions) (*Client, error) {
+	return newClient(containerURL, nil, options)
+}
+
+// NewClientWithSharedKeyCredential creates a Client signing requests with the
+// account's shared key.
+func NewClientWithSharedKeyCredential(containerURL string, cred *SharedKeyCredential, options *ClientOptions) (*Client, error) {
+	return newClient(containerURL, NewSharedKeyCredPolicy(cred), options)
+}
+
+// NewListBlobsHierarchyPager returns a pager over the container's blobs,
+// requesting Arrow responses when o.UseArrowFormat is set and returning the
+// same response type as container.Client.NewListBlobsHierarchyPager.
+//
+// If the service does not support Arrow listing (the feature not enabled on
+// the account, or an HNS account) it answers with XML, which is decoded
+// transparently - unless EndBefore is set, in which case the fetcher returns
+// ErrEndBeforeXMLFallback (see there).
+func (c *Client) NewListBlobsHierarchyPager(delimiter string, o *ListBlobsHierarchyOptions) *runtime.Pager[container.ListBlobsHierarchyResponse] {
+	// Copy the options so advancing the Marker doesn't mutate the caller's struct.
+	opts := ListBlobsHierarchyOptions{}
+	if o != nil {
+		opts = *o
+	}
+	useArrow := opts.UseArrowFormat != nil && *opts.UseArrowFormat
+	return runtime.NewPager(runtime.PagingHandler[container.ListBlobsHierarchyResponse]{
+		More: func(page container.ListBlobsHierarchyResponse) bool {
+			return page.NextMarker != nil && len(*page.NextMarker) > 0
+		},
+		Fetcher: func(ctx context.Context, page *container.ListBlobsHierarchyResponse) (container.ListBlobsHierarchyResponse, error) {
+			if page != nil {
+				opts.Marker = page.NextMarker
+			}
+			req, err := c.listCreateRequest(ctx, delimiter, &opts, useArrow)
+			if err != nil {
+				return container.ListBlobsHierarchyResponse{}, err
+			}
+			resp, err := c.azClient.Pipeline().Do(req)
+			if err != nil {
+				return container.ListBlobsHierarchyResponse{}, err
+			}
+			if !runtime.HasStatusCode(resp, http.StatusOK) {
+				return container.ListBlobsHierarchyResponse{}, runtime.NewResponseError(resp)
+			}
+			if useArrow && resp.Header.Get("Content-Type") == ArrowContentType {
+				return HandleHierarchyListResponse(resp)
+			}
+			if opts.EndBefore != nil {
+				return container.ListBlobsHierarchyResponse{}, ErrEndBeforeXMLFallback
+			}
+			return handleXMLResponse(resp)
+		},
+	})
+}
+
+// listCreateRequest builds the ListBlobs request, mirroring the generated
+// request builders in the SDK.
+func (c *Client) listCreateRequest(ctx context.Context, delimiter string, opts *ListBlobsHierarchyOptions, useArrow bool) (*policy.Request, error) {
+	req, err := runtime.NewRequest(ctx, http.MethodGet, c.endpoint)
+	if err != nil {
+		return nil, err
+	}
+	// Reading the existing query first preserves any SAS token in the endpoint.
+	reqQP := req.Raw().URL.Query()
+	reqQP.Set("comp", "list")
+	reqQP.Set("restype", "container")
+	reqQP.Set("delimiter", delimiter)
+	if include := formatInclude(opts.Include); include != "" {
+		reqQP.Set("include", include)
+	}
+	if opts.Marker != nil {
+		reqQP.Set("marker", *opts.Marker)
+	}
+	if opts.MaxResults != nil {
+		reqQP.Set("maxresults", strconv.FormatInt(int64(*opts.MaxResults), 10))
+	}
+	if opts.Prefix != nil {
+		reqQP.Set("prefix", *opts.Prefix)
+	}
+	if opts.StartFrom != nil {
+		reqQP.Set("startFrom", *opts.StartFrom)
+	}
+	if useArrow && opts.EndBefore != nil {
+		reqQP.Set("endBefore", *opts.EndBefore)
+	}
+	req.Raw().URL.RawQuery = strings.ReplaceAll(reqQP.Encode(), "+", "%20")
+	if useArrow {
+		// Stream the Arrow body rather than buffering it in the pipeline.
+		runtime.SkipBodyDownload(req)
+		req.Raw().Header["Accept"] = []string{ArrowAcceptHeader}
+	} else {
+		req.Raw().Header["Accept"] = []string{"application/xml"}
+	}
+	req.Raw().Header["x-ms-version"] = []string{serviceVersion}
+	return req, nil
+}
+
+// handleXMLResponse decodes an XML ListBlobs response, mirroring the SDK's
+// generated ListBlobHierarchySegmentHandleResponse.
+func handleXMLResponse(resp *http.Response) (container.ListBlobsHierarchyResponse, error) {
+	result := container.ListBlobsHierarchyResponse{}
+	extractResponseHeaders(resp, &result.ClientRequestID, &result.ContentType, &result.Date, &result.RequestID, &result.Version)
+	if err := runtime.UnmarshalAsXML(resp, &result.ListBlobsHierarchySegmentResponse); err != nil {
+		return container.ListBlobsHierarchyResponse{}, err
+	}
+	return result, nil
+}
+
+// formatInclude renders the include datasets as the comma separated query
+// parameter value, in the same order as the SDK's unexported
+// ListBlobsInclude.format.
+func formatInclude(l container.ListBlobsInclude) string {
+	var include []string
+	if l.Copy {
+		include = append(include, "copy")
+	}
+	if l.Deleted {
+		include = append(include, "deleted")
+	}
+	if l.DeletedWithVersions {
+		include = append(include, "deletedwithversions")
+	}
+	if l.ImmutabilityPolicy {
+		include = append(include, "immutabilitypolicy")
+	}
+	if l.LegalHold {
+		include = append(include, "legalhold")
+	}
+	if l.Metadata {
+		include = append(include, "metadata")
+	}
+	if l.Snapshots {
+		include = append(include, "snapshots")
+	}
+	if l.Tags {
+		include = append(include, "tags")
+	}
+	if l.UncommittedBlobs {
+		include = append(include, "uncommittedblobs")
+	}
+	if l.Versions {
+		include = append(include, "versions")
+	}
+	if l.Permissions {
+		include = append(include, "permissions")
+	}
+	return strings.Join(include, ",")
+}

backend/azureblob/arrowlist/arrowlist_test.go

@@ -0,0 +1,199 @@
+//go:build !plan9 && !solaris && !js
+
+package arrowlist
+
+import (
+	"context"
+	"encoding/base64"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
+	"github.com/apache/arrow-go/v18/arrow"
+	arrowArray "github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// arrowPage builds a one row Arrow IPC listing page with the given blob name
+// and NextMarker.
+func arrowPage(t *testing.T, name, nextMarker string) []byte {
+	md := arrow.NewMetadata([]string{"NextMarker", "NumberOfRecords"}, []string{nextMarker, "1"})
+	fields := []arrow.Field{
+		{Name: "Name", Type: arrow.BinaryTypes.String, Nullable: false},
+		{Name: "Content-Length", Type: arrow.PrimitiveTypes.Uint64, Nullable: true},
+	}
+	schema := arrow.NewSchema(fields, &md)
+	return buildArrowStream(t, schema, func(b *arrowArray.RecordBuilder) {
+		b.Field(0).(*arrowArray.StringBuilder).Append(name)
+		b.Field(1).(*arrowArray.Uint64Builder).Append(42)
+	})
+}
+
+func TestPagerArrow(t *testing.T) {
+	var requests []*http.Request
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests = append(requests, r.Clone(context.Background()))
+		w.Header().Set("Content-Type", ArrowContentType)
+		if r.URL.Query().Get("marker") == "" {
+			_, _ = w.Write(arrowPage(t, "blob1.txt", "marker1"))
+		} else {
+			_, _ = w.Write(arrowPage(t, "blob2.txt", ""))
+		}
+	}))
+	defer srv.Close()
+
+	// SAS style query on the endpoint must be preserved in requests.
+	client, err := NewClientWithNoCredential(srv.URL+"/testcontainer?sv=fakesas", nil)
+	require.NoError(t, err)
+
+	opts := &ListBlobsHierarchyOptions{
+		ListBlobsHierarchyOptions: container.ListBlobsHierarchyOptions{
+			Include:    container.ListBlobsInclude{Metadata: true, Tags: true},
+			Prefix:     to.Ptr("dir/"),
+			MaxResults: to.Ptr(int32(1000)),
+			StartFrom:  to.Ptr("testcontainer/dir/a"),
+		},
+		EndBefore:      to.Ptr("testcontainer/dir/n"),
+		UseArrowFormat: to.Ptr(true),
+	}
+	pager := client.NewListBlobsHierarchyPager("/", opts)
+
+	var names []string
+	for pager.More() {
+		page, err := pager.NextPage(context.Background())
+		require.NoError(t, err)
+		assert.Equal(t, ArrowContentType, *page.ContentType)
+		for _, item := range page.Segment.BlobItems {
+			names = append(names, *item.Name)
+			assert.Equal(t, int64(42), *item.Properties.ContentLength)
+		}
+	}
+	assert.Equal(t, []string{"blob1.txt", "blob2.txt"}, names)
+
+	require.Len(t, requests, 2)
+	q := requests[0].URL.Query()
+	assert.Equal(t, "list", q.Get("comp"))
+	assert.Equal(t, "container", q.Get("restype"))
+	assert.Equal(t, "/", q.Get("delimiter"))
+	assert.Equal(t, "metadata,tags", q.Get("include"))
+	assert.Equal(t, "dir/", q.Get("prefix"))
+	assert.Equal(t, "1000", q.Get("maxresults"))
+	assert.Equal(t, "testcontainer/dir/a", q.Get("startFrom"))
+	assert.Equal(t, "testcontainer/dir/n", q.Get("endBefore"))
+	assert.Equal(t, "fakesas", q.Get("sv"))
+	assert.Empty(t, q.Get("marker"))
+	assert.Equal(t, ArrowAcceptHeader, requests[0].Header.Get("Accept"))
+	assert.Equal(t, serviceVersion, requests[0].Header.Get("x-ms-version"))
+	assert.Equal(t, "marker1", requests[1].URL.Query().Get("marker"))
+
+	// The caller's options must not have been mutated by paging.
+	assert.Nil(t, opts.Marker)
+}
+
+const xmlListResponse = `<?xml version="1.0" encoding="utf-8"?>
+<EnumerationResults ServiceEndpoint="https://example.com/" ContainerName="testcontainer">
+  <Blobs>
+    <Blob>
+      <Name>file.txt</Name>
+      <Properties>
+        <Last-Modified>Mon, 01 Jan 2024 00:00:00 GMT</Last-Modified>
+        <Etag>0x8D1</Etag>
+        <Content-Length>7</Content-Length>
+        <BlobType>BlockBlob</BlobType>
+      </Properties>
+    </Blob>
+    <BlobPrefix>
+      <Name>subdir/</Name>
+    </BlobPrefix>
+  </Blobs>
+  <NextMarker />
+</EnumerationResults>`
+
+func TestPagerXMLFallback(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/xml")
+		_, _ = w.Write([]byte(xmlListResponse))
+	}))
+	defer srv.Close()
+
+	client, err := NewClientWithNoCredential(srv.URL+"/testcontainer", nil)
+	require.NoError(t, err)
+
+	pager := client.NewListBlobsHierarchyPager("/", &ListBlobsHierarchyOptions{
+		UseArrowFormat: to.Ptr(true),
+	})
+	page, err := pager.NextPage(context.Background())
+	require.NoError(t, err)
+	assert.False(t, pager.More())
+	assert.Equal(t, "application/xml", *page.ContentType)
+	require.Len(t, page.Segment.BlobItems, 1)
+	assert.Equal(t, "file.txt", *page.Segment.BlobItems[0].Name)
+	assert.Equal(t, int64(7), *page.Segment.BlobItems[0].Properties.ContentLength)
+	require.Len(t, page.Segment.BlobPrefixes, 1)
+	assert.Equal(t, "subdir/", *page.Segment.BlobPrefixes[0].Name)
+}
+
+func TestPagerXMLFallbackWithEndBefore(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/xml")
+		_, _ = w.Write([]byte(xmlListResponse))
+	}))
+	defer srv.Close()
+
+	client, err := NewClientWithNoCredential(srv.URL+"/testcontainer", nil)
+	require.NoError(t, err)
+
+	pager := client.NewListBlobsHierarchyPager("/", &ListBlobsHierarchyOptions{
+		EndBefore:      to.Ptr("testcontainer/n"),
+		UseArrowFormat: to.Ptr(true),
+	})
+	_, err = pager.NextPage(context.Background())
+	require.ErrorIs(t, err, ErrEndBeforeXMLFallback)
+}
+
+func TestPagerSharedKeyAuth(t *testing.T) {
+	var gotAuth, gotDate string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotAuth = r.Header.Get("Authorization")
+		gotDate = r.Header.Get("x-ms-date")
+		w.Header().Set("Content-Type", ArrowContentType)
+		_, _ = w.Write(arrowPage(t, "blob.txt", ""))
+	}))
+	defer srv.Close()
+
+	cred, err := NewSharedKeyCredential("testaccount", base64.StdEncoding.EncodeToString([]byte("testkey")))
+	require.NoError(t, err)
+	client, err := NewClientWithSharedKeyCredential(srv.URL+"/testcontainer", cred, nil)
+	require.NoError(t, err)
+
+	pager := client.NewListBlobsHierarchyPager("/", &ListBlobsHierarchyOptions{
+		UseArrowFormat: to.Ptr(true),
+	})
+	_, err = pager.NextPage(context.Background())
+	require.NoError(t, err)
+	assert.True(t, strings.HasPrefix(gotAuth, "SharedKey testaccount:"), "Authorization = %q", gotAuth)
+	assert.NotEmpty(t, gotDate)
+}
+
+func TestPagerResponseError(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("x-ms-error-code", string(bloberror.ContainerNotFound))
+		w.WriteHeader(http.StatusNotFound)
+	}))
+	defer srv.Close()
+
+	client, err := NewClientWithNoCredential(srv.URL+"/testcontainer", nil)
+	require.NoError(t, err)
+
+	pager := client.NewListBlobsHierarchyPager("/", &ListBlobsHierarchyOptions{
+		UseArrowFormat: to.Ptr(true),
+	})
+	_, err = pager.NextPage(context.Background())
+	require.Error(t, err)
+	assert.True(t, bloberror.HasCode(err, bloberror.ContainerNotFound), "expected ContainerNotFound, got %v", err)
+}

backend/azureblob/arrowlist/challenge_policy.go

@@ -0,0 +1,121 @@
+//go:build !plan9 && !solaris && !js
+
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+//
+// Adapted from internal/shared/challenge_policy.go in the Azure SDK for Go
+// github.com/Azure/azure-sdk-for-go/sdk/storage/azblob at v1.8.0. The
+// released SDK does not export its storage bearer-challenge policy, which is
+// needed for token authentication with the same semantics as the SDK's own
+// clients.
+
+package arrowlist
+
+import (
+	"errors"
+	"net/http"
+	"strings"
+
+	"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"
+)
+
+type storageAuthorizer struct {
+	scopes   []string
+	tenantID string
+}
+
+// NewStorageChallengePolicy returns a bearer token policy for cred which
+// handles the storage service's authentication challenges.
+func NewStorageChallengePolicy(cred azcore.TokenCredential, audience string, allowHTTP bool) policy.Policy {
+	s := storageAuthorizer{scopes: []string{audience}}
+	return runtime.NewBearerTokenPolicy(cred, []string{audience}, &policy.BearerTokenOptions{
+		AuthorizationHandler: policy.AuthorizationHandler{
+			OnRequest:   s.onRequest,
+			OnChallenge: s.onChallenge,
+		},
+		InsecureAllowCredentialWithHTTP: allowHTTP,
+	})
+}
+
+func (s *storageAuthorizer) onRequest(req *policy.Request, authNZ func(policy.TokenRequestOptions) error) error {
+	return authNZ(policy.TokenRequestOptions{Scopes: s.scopes})
+}
+
+func (s *storageAuthorizer) onChallenge(req *policy.Request, resp *http.Response, authNZ func(policy.TokenRequestOptions) error) error {
+	// parse the challenge
+	err := s.parseChallenge(resp)
+	if err != nil {
+		return err
+	}
+	// TODO: Set tenantID when policy.TokenRequestOptions supports it. https://github.com/Azure/azure-sdk-for-go/issues/19841
+	return authNZ(policy.TokenRequestOptions{Scopes: s.scopes})
+}
+
+type challengePolicyError struct {
+	err error
+}
+
+func (c *challengePolicyError) Error() string {
+	return c.err.Error()
+}
+
+func (*challengePolicyError) NonRetriable() {
+	// marker method
+}
+
+func (c *challengePolicyError) Unwrap() error {
+	return c.err
+}
+
+// parses Tenant ID from auth challenge
+// https://login.microsoftonline.com/00000000-0000-0000-0000-000000000000/oauth2/authorize
+func parseTenant(url string) string {
+	if url == "" {
+		return ""
+	}
+	parts := strings.Split(url, "/")
+	if len(parts) >= 3 {
+		tenant := parts[3]
+		tenant = strings.ReplaceAll(tenant, ",", "")
+		return tenant
+	}
+	return ""
+}
+
+func (s *storageAuthorizer) parseChallenge(resp *http.Response) error {
+	authHeader := resp.Header.Get("WWW-Authenticate")
+	if authHeader == "" {
+		return &challengePolicyError{err: errors.New("response has no WWW-Authenticate header for challenge authentication")}
+	}
+
+	// Strip down to auth and resource
+	// Format is "Bearer authorization_uri=\"<site>\" resource_id=\"<site>\""
+	authHeader = strings.ReplaceAll(authHeader, "Bearer ", "")
+
+	parts := strings.Split(authHeader, " ")
+
+	vals := map[string]string{}
+	for _, part := range parts {
+		subParts := strings.Split(part, "=")
+		if len(subParts) == 2 {
+			stripped := strings.ReplaceAll(subParts[1], "\"", "")
+			stripped = strings.TrimSuffix(stripped, ",")
+			vals[subParts[0]] = stripped
+		}
+	}
+
+	s.tenantID = parseTenant(vals["authorization_uri"])
+
+	scope := vals["resource_id"]
+	if scope == "" {
+		return &challengePolicyError{err: errors.New("could not find a valid resource in the WWW-Authenticate header")}
+	}
+
+	if !strings.HasSuffix(scope, "/.default") {
+		scope += "/.default"
+	}
+	s.scopes = []string{scope}
+	return nil
+}

backend/azureblob/arrowlist/sharedkey.go

@@ -0,0 +1,318 @@
+//go:build !plan9 && !solaris && !js
+
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+//
+// Adapted from internal/exported/shared_key_credential.go in the Azure SDK
+// for Go github.com/Azure/azure-sdk-for-go/sdk/storage/azblob at v1.8.0,
+// with the internal header constants inlined. The released SDK does not
+// export its shared key signing policy, so a copy is needed to sign the
+// Arrow listing requests this package sends through its own pipeline.
+
+package arrowlist
+
+import (
+	"bytes"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"fmt"
+	"net/http"
+	"net/url"
+	"sort"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/rclone/rclone/fs"
+)
+
+// NewSharedKeyCredential creates an immutable SharedKeyCredential containing the
+// storage account's name and either its primary or secondary key.
+func NewSharedKeyCredential(accountName string, accountKey string) (*SharedKeyCredential, error) {
+	c := SharedKeyCredential{accountName: accountName}
+	if err := c.SetAccountKey(accountKey); err != nil {
+		return nil, err
+	}
+	return &c, nil
+}
+
+// SharedKeyCredential contains an account's name and its primary or secondary key.
+type SharedKeyCredential struct {
+	// Only the NewSharedKeyCredential method should set these; all other methods should treat them as read-only
+	accountName string
+	accountKey  atomic.Value // []byte
+}
+
+// AccountName returns the Storage account's name.
+func (c *SharedKeyCredential) AccountName() string {
+	return c.accountName
+}
+
+// SetAccountKey replaces the existing account key with the specified account key.
+func (c *SharedKeyCredential) SetAccountKey(accountKey string) error {
+	_bytes, err := base64.StdEncoding.DecodeString(accountKey)
+	if err != nil {
+		return fmt.Errorf("decode account key: %w", err)
+	}
+	c.accountKey.Store(_bytes)
+	return nil
+}
+
+// ComputeHMACSHA256 generates a hash signature for an HTTP request or for a SAS.
+func (c *SharedKeyCredential) computeHMACSHA256(message string) (string, error) {
+	h := hmac.New(sha256.New, c.accountKey.Load().([]byte))
+	_, err := h.Write([]byte(message))
+	return base64.StdEncoding.EncodeToString(h.Sum(nil)), err
+}
+
+func (c *SharedKeyCredential) buildStringToSign(req *http.Request) (string, error) {
+	// https://docs.microsoft.com/en-us/rest/api/storageservices/authentication-for-the-azure-storage-services
+	headers := req.Header
+	contentLength := getHeader("Content-Length", headers)
+	if contentLength == "0" {
+		contentLength = ""
+	}
+
+	canonicalizedResource, err := c.buildCanonicalizedResource(req.URL)
+	if err != nil {
+		return "", err
+	}
+
+	stringToSign := strings.Join([]string{
+		req.Method,
+		getHeader("Content-Encoding", headers),
+		getHeader("Content-Language", headers),
+		contentLength,
+		getHeader("Content-MD5", headers),
+		getHeader("Content-Type", headers),
+		"", // Empty date because x-ms-date is expected (as per web page above)
+		getHeader("If-Modified-Since", headers),
+		getHeader("If-Match", headers),
+		getHeader("If-None-Match", headers),
+		getHeader("If-Unmodified-Since", headers),
+		getHeader("Range", headers),
+		c.buildCanonicalizedHeader(headers),
+		canonicalizedResource,
+	}, "\n")
+	return stringToSign, nil
+}
+
+func getHeader(key string, headers map[string][]string) string {
+	if headers == nil {
+		return ""
+	}
+	if v, ok := headers[key]; ok {
+		if len(v) > 0 {
+			return v[0]
+		}
+	}
+
+	return ""
+}
+
+func getWeightTables() [][]int {
+	tableLv0 := [...]int{
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x71c, 0x0, 0x71f, 0x721, 0x723, 0x725,
+		0x0, 0x0, 0x0, 0x72d, 0x803, 0x0, 0x0, 0x733, 0x0, 0xd03, 0xd1a, 0xd1c, 0xd1e,
+		0xd20, 0xd22, 0xd24, 0xd26, 0xd28, 0xd2a, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0xe02, 0xe09, 0xe0a, 0xe1a, 0xe21, 0xe23, 0xe25, 0xe2c, 0xe32, 0xe35, 0xe36, 0xe48, 0xe51,
+		0xe70, 0xe7c, 0xe7e, 0xe89, 0xe8a, 0xe91, 0xe99, 0xe9f, 0xea2, 0xea4, 0xea6, 0xea7, 0xea9,
+		0x0, 0x0, 0x0, 0x743, 0x744, 0x748, 0xe02, 0xe09, 0xe0a, 0xe1a, 0xe21, 0xe23, 0xe25,
+		0xe2c, 0xe32, 0xe35, 0xe36, 0xe48, 0xe51, 0xe70, 0xe7c, 0xe7e, 0xe89, 0xe8a, 0xe91, 0xe99,
+		0xe9f, 0xea2, 0xea4, 0xea6, 0xea7, 0xea9, 0x0, 0x74c, 0x0, 0x750, 0x0,
+	}
+	tableLv2 := [...]int{
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x8012, 0x0, 0x0, 0x0, 0x0, 0x0, 0x8212, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+		0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+	}
+	tables := [][]int{tableLv0[:], tableLv2[:]}
+	return tables
+}
+
+// NewHeaderStringComparer performs a multi-level, weight-based comparison of two strings
+func compareHeaders(lhs, rhs string, tables [][]int) int {
+	currLevel, i, j := 0, 0, 0
+	n := len(tables)
+	lhsLen := len(lhs)
+	rhsLen := len(rhs)
+
+	for currLevel < n {
+		if currLevel == (n-1) && i != j {
+			if i > j {
+				return -1
+			}
+			if i < j {
+				return 1
+			}
+			return 0
+		}
+
+		var w1, w2 int
+
+		// Check bounds before accessing lhs[i]
+		if i < lhsLen {
+			w1 = tables[currLevel][lhs[i]]
+		} else {
+			w1 = 0x1
+		}
+
+		// Check bounds before accessing rhs[j]
+		if j < rhsLen {
+			w2 = tables[currLevel][rhs[j]]
+		} else {
+			w2 = 0x1
+		}
+
+		if w1 == 0x1 && w2 == 0x1 {
+			i = 0
+			j = 0
+			currLevel++
+		} else if w1 == w2 {
+			i++
+			j++
+		} else if w1 == 0 {
+			i++
+		} else if w2 == 0 {
+			j++
+		} else {
+			if w1 < w2 {
+				return -1
+			}
+			if w1 > w2 {
+				return 1
+			}
+			return 0
+		}
+	}
+	return 0
+}
+
+func (c *SharedKeyCredential) buildCanonicalizedHeader(headers http.Header) string {
+	cm := map[string][]string{}
+	for k, v := range headers {
+		headerName := strings.TrimSpace(strings.ToLower(k))
+		if strings.HasPrefix(headerName, "x-ms-") {
+			cm[headerName] = v // NOTE: the value must not have any whitespace around it.
+		}
+	}
+	if len(cm) == 0 {
+		return ""
+	}
+
+	keys := make([]string, 0, len(cm))
+	for key := range cm {
+		keys = append(keys, key)
+	}
+	tables := getWeightTables()
+	// Sort the keys using the custom comparator
+	sort.Slice(keys, func(i, j int) bool {
+		return compareHeaders(keys[i], keys[j], tables) < 0
+	})
+	ch := bytes.NewBufferString("")
+	for i, key := range keys {
+		if i > 0 {
+			ch.WriteRune('\n')
+		}
+		ch.WriteString(key)
+		ch.WriteRune(':')
+		ch.WriteString(strings.Join(cm[key], ","))
+	}
+	return ch.String()
+}
+
+func (c *SharedKeyCredential) buildCanonicalizedResource(u *url.URL) (string, error) {
+	// https://docs.microsoft.com/en-us/rest/api/storageservices/authentication-for-the-azure-storage-services
+	cr := bytes.NewBufferString("/")
+	cr.WriteString(c.accountName)
+
+	if len(u.Path) > 0 {
+		// Any portion of the CanonicalizedResource string that is derived from
+		// the resource's URI should be encoded exactly as it is in the URI.
+		// -- https://msdn.microsoft.com/en-gb/library/azure/dd179428.aspx
+		cr.WriteString(u.EscapedPath())
+	} else {
+		// a slash is required to indicate the root path
+		cr.WriteString("/")
+	}
+
+	// params is a map[string][]string; param name is key; params values is []string
+	params, err := url.ParseQuery(u.RawQuery) // Returns URL decoded values
+	if err != nil {
+		return "", fmt.Errorf("failed to parse query params: %w", err)
+	}
+
+	if len(params) > 0 { // There is at least 1 query parameter
+		var paramNames []string // We use this to sort the parameter key names
+		for paramName := range params {
+			paramNames = append(paramNames, paramName) // paramNames must be lowercase
+		}
+		sort.Strings(paramNames)
+
+		for _, paramName := range paramNames {
+			paramValues := params[paramName]
+			sort.Strings(paramValues)
+
+			// Join the sorted key values separated by ','
+			// Then prepend "keyName:"; then add this string to the buffer
+			cr.WriteString("\n" + strings.ToLower(paramName) + ":" + strings.Join(paramValues, ","))
+		}
+	}
+	return cr.String(), nil
+}
+
+// ComputeHMACSHA256 is a helper for computing the signed string outside of this package.
+func ComputeHMACSHA256(cred *SharedKeyCredential, message string) (string, error) {
+	return cred.computeHMACSHA256(message)
+}
+
+// SharedKeyCredPolicy is a pipeline policy that signs each request with the
+// account's shared key.
+type SharedKeyCredPolicy struct {
+	cred *SharedKeyCredential
+}
+
+// NewSharedKeyCredPolicy creates a SharedKeyCredPolicy signing with cred.
+func NewSharedKeyCredPolicy(cred *SharedKeyCredential) *SharedKeyCredPolicy {
+	return &SharedKeyCredPolicy{cred: cred}
+}
+
+// Do signs the request and sends it via the next policy in the pipeline.
+func (s *SharedKeyCredPolicy) Do(req *policy.Request) (*http.Response, error) {
+	// skip adding the authorization header if no SharedKeyCredential was provided.
+	// this prevents a panic that might be hard to diagnose and allows testing
+	// against http endpoints that don't require authentication.
+	if s.cred == nil {
+		return req.Next()
+	}
+
+	if d := getHeader("x-ms-date", req.Raw().Header); d == "" {
+		req.Raw().Header.Set("x-ms-date", time.Now().UTC().Format(http.TimeFormat))
+	}
+	stringToSign, err := s.cred.buildStringToSign(req.Raw())
+	if err != nil {
+		return nil, err
+	}
+	signature, err := s.cred.computeHMACSHA256(stringToSign)
+	if err != nil {
+		return nil, err
+	}
+	authHeader := strings.Join([]string{"SharedKey ", s.cred.AccountName(), ":", signature}, "")
+	req.Raw().Header.Set("Authorization", authHeader)
+
+	response, err := req.Next()
+	if err != nil && response != nil && response.StatusCode == http.StatusForbidden {
+		// Service failed to authenticate request, log it
+		fs.Debugf(nil, "azureblob arrowlist: HTTP Forbidden status, String-to-Sign: %q", stringToSign)
+	}
+	return response, err
+}

backend/azureblob/auth/auth.go

@@ -297,6 +297,13 @@ func (tr transporter) Do(req *http.Request) (*http.Response, error) {
 	return tr.RoundTripper.RoundTrip(req)
 }
 
+// Transporter returns the policy.Transporter rclone uses for Azure SDK
+// clients (an fshttp based transport with the APN user agent), for callers
+// which build their own azcore pipelines.
+func Transporter(ctx context.Context) policy.Transporter {
+	return newTransporter(ctx)
+}
+
 // NewClientOpts should be passed to configure NewClient
 type NewClientOpts[Client, ClientOptions, SharedKeyCredential any] struct {
 	DefaultBaseURL                   string // Base URL, eg blob.core.windows.net

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/adrg/xdg v0.5.3
 	github.com/anacrolix/dms v1.7.2
 	github.com/anacrolix/log v0.17.0
+	github.com/apache/arrow-go/v18 v18.7.0
 	github.com/atotto/clipboard v0.1.4
 	github.com/aws/aws-sdk-go-v2 v1.42.1
 	github.com/aws/aws-sdk-go-v2/config v1.32.30
@@ -177,9 +178,11 @@ require (
 	github.com/go-playground/validator/v10 v10.30.3 // indirect
 	github.com/go-resty/resty/v2 v2.17.2 // indirect
 	github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
+	github.com/goccy/go-json v0.10.6 // indirect
 	github.com/gofrs/flock v0.13.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/btree v1.1.3 // indirect
+	github.com/google/flatbuffers v25.12.19+incompatible // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.18 // indirect
 	github.com/googleapis/gax-go/v2 v2.22.0 // indirect

go.sum

@@ -79,6 +79,10 @@ github.com/andybalholm/brotli v1.2.2 h1:HzTuoo2ErYQqf5qvcJInB8uvqSVxRttzkFexPWtn
 github.com/andybalholm/brotli v1.2.2/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/andybalholm/cascadia v1.3.4 h1:vM2lgh0Vru9Vwyfm4cQqWP2HHMW0u0+2PAW7Q38Qufg=
 github.com/andybalholm/cascadia v1.3.4/go.mod h1:BLRmbRjpEtNKieZOCCvYj4RqN+KRA41GBe/5O+G93kM=
+github.com/apache/arrow-go/v18 v18.7.0 h1:Vw/i+cJyebUofT7JlqFpe65LrmwxULn166jjwStM4HY=
+github.com/apache/arrow-go/v18 v18.7.0/go.mod h1:PM6IigLJkdMwIpeHXnymo+xZ52f42a9EYiLtRel4p/A=
+github.com/apache/thrift v0.24.0 h1:zy31L1a49QTNB2bG1BBfMXol3yJrTH975G3pPubQVLQ=
+github.com/apache/thrift v0.24.0/go.mod h1:zPt6WxgvTOM6hF92y8C+MkEM5LMxZuk4JcQOiU4Esvs=
 github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc h1:LoL75er+LKDHDUfU5tRvFwxH0LjPpZN8OoG8Ll+liGU=
 github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc/go.mod h1:w648aMHEgFYS6xb0KVMMtZ2uMeemhiKCuD2vj6gY52A=
 github.com/atotto/clipboard v0.1.4 h1:EH0zSVneZPSuFR11BlR9YppQTVDbh5+16AmcJi4g1z4=
@@ -266,8 +270,8 @@ github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1v
 github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
 github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro=
 github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
-github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
-github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
+github.com/goccy/go-json v0.10.6 h1:p8HrPJzOakx/mn/bQtjgNjdTcN+/S6FcG2CTtQOrHVU=
+github.com/goccy/go-json v0.10.6/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
 github.com/gofrs/flock v0.13.0 h1:95JolYOvGMqeH31+FC7D2+uULf6mG61mEZ/A8dRYMzw=
 github.com/gofrs/flock v0.13.0/go.mod h1:jxeyy9R1auM5S6JYDBhDt+E2TCo7DkratH4Pgi8P+Z0=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
@@ -280,6 +284,8 @@ github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg=
 github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4=
+github.com/google/flatbuffers v25.12.19+incompatible h1:haMV2JRRJCe1998HeW/p0X9UaMTK6SDo0ffLn2+DbLs=
+github.com/google/flatbuffers v25.12.19+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=