Introduce blob-based storage using ITableStorage<T> API

We modify the binary-based serializers to work on top of Stream instead of byte arrays.

And the new blob-based implementations use the table name as the container name and the partition key as the folder path for the blobs. This allows using document-storage like API without hitting the limitation of column sizes in table storage.

Author: kzu

.netconfig

@@ -431,3 +431,9 @@
 	sha = 77e83f238196d2723640abef0c7b6f43994f9747
 	etag = fcb9759a96966df40dcd24906fd328ddec05953b7e747a6bb8d0d1e4c3865274
 	weak
+[file "src/TableStorage/System/AsyncLazy.cs"]
+	url = https://github.com/devlooped/catbag/blob/main/System/Threading/Tasks/AsyncLazy.cs
+	sha = 9f3330f09713aa5f746047e3a50ee839147a5797
+
+	etag = 73320600b7a18e0eb25cadc3d687c69dc79181b0458facf526666e150c634782
+	weak

src/TableStorage.Bson/BsonDocumentSerializer.cs

@@ -32,26 +32,24 @@ partial class BsonDocumentSerializer : IBinaryDocumentSerializer
         public static IDocumentSerializer Default { get; } = new BsonDocumentSerializer();
 
         /// <inheritdoc />
-        public T? Deserialize<T>(byte[] data)
+        public T? Deserialize<T>(Stream data)
         {
             if (data.Length == 0)
                 return default;
 
-            using var mem = new MemoryStream(data);
-            using var reader = new Newtonsoft.Json.Bson.BsonDataReader(mem);
+            var reader = new Newtonsoft.Json.Bson.BsonDataReader(data);
             return (T?)serializer.Deserialize<T>(reader);
         }
 
         /// <inheritdoc />
-        public byte[] Serialize<T>(T value)
+        public void Serialize<T>(T value, Stream stream)
         {
             if (value == null)
-                return new byte[0];
+                return;
 
-            using var mem = new MemoryStream();
-            using var writer = new Newtonsoft.Json.Bson.BsonDataWriter(mem);
+            var writer = new Newtonsoft.Json.Bson.BsonDataWriter(stream);
             serializer.Serialize(writer, value);
-            return mem.ToArray();
+            writer.Flush();
         }
 
 #if NET6_0_OR_GREATER

src/TableStorage.MessagePack/MessagePackDocumentSerializer.cs

@@ -2,6 +2,7 @@
 #nullable enable
 using System;
 using System.Collections.Generic;
+using System.IO;
 using MessagePack;
 using MessagePack.Formatters;
 using MessagePack.Resolvers;
@@ -36,10 +37,18 @@ public MessagePackDocumentSerializer(MessagePackSerializerOptions? options = def
         }
 
         /// <inheritdoc />
-        public T? Deserialize<T>(byte[] data) => data.Length == 0 ? default : MessagePackSerializer.Deserialize<T>(data, options);
+        public T? Deserialize<T>(Stream data) => data.Length == 0 ? default : MessagePackSerializer.Deserialize<T>(data, options);
 
         /// <inheritdoc />
-        public byte[] Serialize<T>(T value) => value == null ? new byte[0] : MessagePackSerializer.Serialize(value.GetType(), value, options);
+        public void Serialize<T>(T value, Stream stream)
+        {
+            if (value == null)
+                return;
+            if (stream == null)
+                throw new ArgumentNullException(nameof(stream));
+
+            MessagePackSerializer.Serialize(stream, value, options);
+        }
 
 #if NET6_0_OR_GREATER
         internal class DateOnlyFormatterResolver : IFormatterResolver

src/TableStorage.Protobuf/ProtobufDocumentSerializer.cs

@@ -26,24 +26,21 @@ static ProtobufDocumentSerializer()
         }
 
         /// <inheritdoc />
-        public T? Deserialize<T>(byte[] data)
+        public T? Deserialize<T>(Stream stream)
         {
-            if (data.Length == 0)
+            if (stream.CanSeek && stream.Length == 0)
                 return default;
 
-            using var mem = new MemoryStream(data);
-            return (T?)Serializer.Deserialize(typeof(T), mem);
+            return (T?)Serializer.Deserialize(typeof(T), stream);
         }
 
         /// <inheritdoc />
-        public byte[] Serialize<T>(T value)
+        public void Serialize<T>(T value, Stream stream)
         {
             if (value == null)
-                return new byte[0];
+                return;
 
-            using var mem = new MemoryStream();
-            Serializer.Serialize(mem, value);
-            return mem.ToArray();
+            Serializer.Serialize(stream, value);
         }
     }
 }

src/TableStorage/BlobPartition.cs

@@ -0,0 +1,63 @@
+//<auto-generated/>
+#nullable enable
+#if NET8_0_OR_GREATER
+using System;
+using System.Collections.Concurrent;
+using System.Reflection;
+
+namespace Devlooped
+{
+    /// <summary>
+    /// Factory methods to create <see cref="ITableStorage{T}"/> instances
+    /// that store entities as a serialized document in blob storage.
+    /// </summary>
+    static partial class BlobPartition
+    {
+        /// <summary>
+        /// Creates an <see cref="ITableStorage{T}"/> for the given entity type 
+        /// <typeparamref name="T"/>, using <see cref="DocumentPartition.DefaultTableName"/> as the table name and the 
+        /// <typeparamref name="T"/> <c>Name</c> as the partition key.
+        /// </summary>
+        /// <typeparam name="T">The type of entity that the repository will manage.</typeparam>
+        /// <param name="storageAccount">The storage account to use.</param>
+        /// <param name="rowKey">Function to retrieve the row key for a given entity.</param>
+        /// <param name="serializer">Optional serializer to use instead of the default <see cref="DocumentSerializer.Default"/>.</param>
+        /// <param name="includeProperties">Whether to serialize properties as columns too, like table repositories, for easier querying.</param>
+        /// <returns>The new <see cref="ITablePartition{T}"/>.</returns>
+        public static ITableStoragePartition<T> Create<T>(
+            CloudStorageAccount storageAccount,
+            Func<T, string> rowKey,
+            IDocumentSerializer? serializer = default) where T : class
+            => Create(storageAccount, DocumentPartition.DefaultTableName, typeof(T).Name, rowKey, serializer);
+
+        /// <summary>
+        /// Creates an <see cref="ITableStorage{T}"/> for the given entity type 
+        /// <typeparamref name="T"/>.
+        /// </summary>
+        /// <typeparam name="T">The type of entity that the repository will manage.</typeparam>
+        /// <param name="storageAccount">The storage account to use.</param>
+        /// <param name="tableName">Optional table name to use. If not provided, the <typeparamref name="T"/> 
+        /// <c>Name</c> will be used, unless a <see cref="TableAttribute"/> on the type overrides it.</param>
+        /// <param name="partitionKey">Optional function to retrieve the partition key for a given entity. 
+        /// If not provided, the class will need a property annotated with <see cref="PartitionKeyAttribute"/>.</param>
+        /// <param name="rowKey">Optional function to retrieve the row key for a given entity. 
+        /// If not provided, the class will need a property annotated with <see cref="RowKeyAttribute"/>.</param>
+        /// <param name="serializer">Optional serializer to use instead of the default <see cref="DocumentSerializer.Default"/>.</param>
+        /// <returns>The new <see cref="ITableStorage{T}"/>.</returns>
+        public static ITableStoragePartition<T> Create<T>(
+            CloudStorageAccount storageAccount,
+            string? tableName = default,
+            string? partitionKey = default,
+            Func<T, string>? rowKey = default,
+            IDocumentSerializer? serializer = default) where T : class
+        {
+            tableName ??= DocumentPartition.GetDefaultTableName<T>();
+            partitionKey ??= TablePartition.GetDefaultPartitionKey<T>();
+            rowKey ??= RowKeyAttribute.CreateCompiledAccessor<T>();
+            serializer ??= DocumentSerializer.Default;
+
+            return new BlobPartition<T>(storageAccount, tableName, partitionKey, rowKey, serializer);
+        }
+    }
+}
+#endif

src/TableStorage/BlobPartition`1.cs

@@ -0,0 +1,60 @@
+//<auto-generated/>
+#nullable enable
+#if NET8_0_OR_GREATER
+using System;
+using System.Collections.Generic;
+using System.Linq.Expressions;
+using System.Threading;
+using System.Threading.Tasks;
+using Mono.Linq.Expressions;
+
+namespace Devlooped
+{
+    /// <inheritdoc />
+    partial class BlobPartition<T> : ITableStoragePartition<T> where T : class
+    {
+        readonly BlobStorage<T> storage;
+
+        /// <summary>
+        /// Initializes the repository with the given storage account and optional table name.
+        /// </summary>
+        /// <param name="storageAccount">The <see cref="CloudStorageAccount"/> to use to connect to the table.</param>
+        /// <param name="tableName">The table that backs this table partition.</param>
+        /// <param name="partitionKey">The fixed partition key that backs this table partition.</param>
+        /// <param name="rowKey">A function to determine the row key for an entity of type <typeparamref name="T"/> within the partition.</param>
+        /// <param name="serializer">The serializer to use.</param>
+        /// <param name="includeProperties">Whether to serialize properties as columns too, like table repositories, for easier querying.</param>
+        protected internal BlobPartition(CloudStorageAccount storageAccount, string tableName, string partitionKey, Func<T, string> rowKey, IDocumentSerializer serializer, bool includeProperties = false)
+        {
+            PartitionKey = partitionKey ?? TablePartition.GetDefaultPartitionKey<T>();
+            storage = new BlobStorage<T>(storageAccount, tableName, _ => PartitionKey, rowKey, serializer);
+        }
+
+        /// <inheritdoc />
+        public string TableName => storage.TableName;
+
+        /// <inheritdoc />
+        public string PartitionKey { get; }
+
+        /// <inheritdoc />
+        public Task<bool> DeleteAsync(T entity, CancellationToken cancellation = default)
+            => storage.DeleteAsync(entity, cancellation);
+
+        /// <inheritdoc />
+        public Task<bool> DeleteAsync(string rowKey, CancellationToken cancellation = default)
+            => storage.DeleteAsync(PartitionKey, rowKey, cancellation);
+        
+        /// <inheritdoc />
+        public IAsyncEnumerable<T> EnumerateAsync(CancellationToken cancellation = default) 
+            => storage.EnumerateAsync(PartitionKey, cancellation);
+
+        /// <inheritdoc />
+        public Task<T?> GetAsync(string rowKey, CancellationToken cancellation = default)
+            => storage.GetAsync(PartitionKey, rowKey, cancellation);
+
+        /// <inheritdoc />
+        public Task<T> PutAsync(T entity, CancellationToken cancellation = default)
+            => storage.PutAsync(entity, cancellation);
+    }
+}
+#endif

src/TableStorage/BlobStorage.cs

@@ -0,0 +1,44 @@
+//<auto-generated/>
+#nullable enable
+#if NET8_0_OR_GREATER
+using System;
+
+namespace Devlooped
+{
+    /// <summary>
+    /// Factory methods to create <see cref="ITableStorage{T}"/> instances
+    /// that store entities as a serialized document in blob storage.
+    /// </summary>
+    static partial class BlobStorage
+    {
+        /// <summary>
+        /// Creates an <see cref="ITableStorage{T}"/> for the given entity type 
+        /// <typeparamref name="T"/>.
+        /// </summary>
+        /// <typeparam name="T">The type of entity that the repository will manage.</typeparam>
+        /// <param name="storageAccount">The storage account to use.</param>
+        /// <param name="tableName">Optional table name to use. If not provided, the <typeparamref name="T"/> 
+        /// <c>Name</c> will be used, unless a <see cref="TableAttribute"/> on the type overrides it.</param>
+        /// <param name="partitionKey">Optional function to retrieve the partition key for a given entity. 
+        /// If not provided, the class will need a property annotated with <see cref="PartitionKeyAttribute"/>.</param>
+        /// <param name="rowKey">Optional function to retrieve the row key for a given entity. 
+        /// If not provided, the class will need a property annotated with <see cref="RowKeyAttribute"/>.</param>
+        /// <param name="serializer">Optional serializer to use instead of the default <see cref="DocumentSerializer.Default"/>.</param>
+        /// <returns>The new <see cref="ITableStorage{T}"/>.</returns>
+        public static ITableStorage<T> Create<T>(
+            CloudStorageAccount storageAccount,
+            string? tableName = default,
+            Func<T, string>? partitionKey = default,
+            Func<T, string>? rowKey = default,
+            IDocumentSerializer? serializer = default) where T : class
+        {
+            tableName ??= DocumentPartition.GetDefaultTableName<T>();
+            partitionKey ??= PartitionKeyAttribute.CreateCompiledAccessor<T>();
+            rowKey ??= RowKeyAttribute.CreateCompiledAccessor<T>();
+            serializer ??= DocumentSerializer.Default;
+
+            return new BlobStorage<T>(storageAccount, tableName, partitionKey, rowKey, serializer);
+        }
+    }
+}
+#endif