diff --git a/backend/azureblob/arrowlist/arrow.go b/backend/azureblob/arrowlist/arrow.go new file mode 100644 index 000000000..f5b0e7b5d --- /dev/null +++ b/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 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 +} diff --git a/backend/azureblob/arrowlist/arrow_test.go b/backend/azureblob/arrowlist/arrow_test.go new file mode 100644 index 000000000..a0000f2a4 --- /dev/null +++ b/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) +} diff --git a/backend/azureblob/arrowlist/arrowlist.go b/backend/azureblob/arrowlist/arrowlist.go new file mode 100644 index 000000000..d80a9d47f --- /dev/null +++ b/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, ",") +} diff --git a/backend/azureblob/arrowlist/arrowlist_test.go b/backend/azureblob/arrowlist/arrowlist_test.go new file mode 100644 index 000000000..d74389284 --- /dev/null +++ b/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 = ` + + + + file.txt + + Mon, 01 Jan 2024 00:00:00 GMT + 0x8D1 + 7 + BlockBlob + + + + subdir/ + + + +` + +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) +} diff --git a/backend/azureblob/arrowlist/challenge_policy.go b/backend/azureblob/arrowlist/challenge_policy.go new file mode 100644 index 000000000..df6698ddc --- /dev/null +++ b/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=\"\" resource_id=\"\"" + 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 +} diff --git a/backend/azureblob/arrowlist/sharedkey.go b/backend/azureblob/arrowlist/sharedkey.go new file mode 100644 index 000000000..bbee8b8a4 --- /dev/null +++ b/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 +} diff --git a/backend/azureblob/auth/auth.go b/backend/azureblob/auth/auth.go index bbea6a4fb..5fa74ceb8 100644 --- a/backend/azureblob/auth/auth.go +++ b/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 diff --git a/go.mod b/go.mod index 7034c3986..50a64b8e5 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 20b902b2b..883cb8c18 100644 --- a/go.sum +++ b/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=