Dev: added BulkUpsertImporterBenchmark.cs

Author: KirillKurdyukov

src/Ydb.Sdk/CHANGELOG.md

@@ -3,7 +3,7 @@
 - ADO.NET: Added `MinPoolSize` setting to keep a minimum number of sessions ready in the PoolingSessionSource.
 - ADO.NET: Added `SessionIdleTimeout` to remove idle sessions from the PoolingSessionSource automatically.
 - ADO.NET: Made `PoolingSessionSource` faster and more reliable by using a lock-free LIFO stack.
-- ADO.NET: Added `BeginBulkUpsertImport` for batch upsert operations with transaction checks and integration tests.
+- ADO.NET: Added `BeginBulkUpsertImport` for batch upsert operations with transaction checks.
 - Optimization: On BadSession, do not invoke the `DeleteSession()` method.
 - Canceling AttachStream after calling the `DeleteSession` method.
 - Fixed bug: fixed issue where session was not deleted (`ClientTransportTimeout`).

src/Ydb.Sdk/src/Ado/BulkUpsert/BulkUpsertImporter.cs

@@ -11,7 +11,7 @@ public sealed class BulkUpsertImporter : IBulkUpsertImporter
     private readonly IDriver _driver;
     private readonly string _tablePath;
     private readonly IReadOnlyList<string> _columns;
-    private readonly int _maxBytes;
+    private readonly int _maxBatchByteSize;
     private readonly RepeatedField<Ydb.Value> _rows = new();
     private readonly CancellationToken _cancellationToken;
     private StructType? _structType;
@@ -21,17 +21,17 @@ internal BulkUpsertImporter(
         IDriver driver,
         string tableName,
         IReadOnlyList<string> columns,
-        int maxBytes,
+        int maxBatchByteSize,
         CancellationToken cancellationToken = default)
     {
         _driver = driver;
         _tablePath = tableName;
         _columns = columns;
-        _maxBytes = maxBytes / 2;
+        _maxBatchByteSize = maxBatchByteSize / 2;
         _cancellationToken = cancellationToken;
     }
 
-    public async ValueTask AddRowAsync(object?[] values)
+    public async ValueTask AddRowAsync(params object[] values)
     {
         if (values.Length != _columns.Count)
             throw new ArgumentException("Values count must match columns count", nameof(values));
@@ -46,7 +46,7 @@ public async ValueTask AddRowAsync(object?[] values)
 
         var rowSize = protoStruct.CalculateSize();
 
-        if (_currentBytes + rowSize > _maxBytes && _rows.Count > 0)
+        if (_currentBytes + rowSize > _maxBatchByteSize && _rows.Count > 0)
         {
             await FlushAsync();
         }

src/Ydb.Sdk/src/Ado/BulkUpsert/IBulkUpsertImporter.cs

@@ -2,7 +2,7 @@ namespace Ydb.Sdk.Ado.BulkUpsert;
 
 public interface IBulkUpsertImporter
 {
-    ValueTask AddRowAsync(object?[] row);
+    ValueTask AddRowAsync(params object[] row);
 
     ValueTask FlushAsync();
 }

src/Ydb.Sdk/test/Ydb.Sdk.Ado.Benchmarks/BulkUpsertImporterBenchmark.cs

@@ -0,0 +1,65 @@
+using BenchmarkDotNet.Attributes;
+using Grpc.Core;
+using Microsoft.Extensions.Logging;
+using Ydb.Sdk.Ado.BulkUpsert;
+using Ydb.Table;
+
+namespace Ydb.Sdk.Ado.Benchmarks;
+
+public class BulkUpsertImporterBenchmark
+{
+    private const int BatchSize = 15000;
+    private readonly int _maxBatchByteSize = new YdbConnectionStringBuilder().MaxSendMessageSize;
+    private IDriver _driver = null!;
+    private IList<(Guid Guid, string String, int Int, double Double, DateTime DateTime)> _rows = null!;
+
+    [GlobalSetup]
+    public void Setup()
+    {
+        _driver = new BulkUpsertMockDriver();
+        _rows = Enumerable.Range(0, BatchSize)
+            .Select(i => new ValueTuple<Guid, string, int, double, DateTime>(Guid.NewGuid(),
+                Guid.NewGuid() + "_" + Guid.NewGuid(), i, i * 1.324, DateTime.Now))
+            .ToList();
+    }
+
+    [Benchmark]
+    public async Task FlushAsync_BulkUpsertImporter()
+    {
+        var bulkUpsertImporter = new BulkUpsertImporter(_driver, "table_name", ["c1", "c2", "c3", "c4", "c5"],
+            _maxBatchByteSize, CancellationToken.None);
+
+        foreach (var row in _rows)
+        {
+            await bulkUpsertImporter.AddRowAsync(row.Guid, row.String, row.Int, row.Double, row.DateTime);
+        }
+        await bulkUpsertImporter.FlushAsync();
+    }
+}
+
+internal class BulkUpsertMockDriver : IDriver
+{
+    public ValueTask DisposeAsync() => throw new NotImplementedException();
+
+    public void Dispose() => throw new NotImplementedException();
+
+    public Task<TResponse>
+        UnaryCall<TRequest, TResponse>(Method<TRequest, TResponse> method, TRequest request,
+            GrpcRequestSettings settings) where TRequest : class where TResponse : class =>
+        Task.FromResult((TResponse)(object)new BulkUpsertResponse
+        {
+            Operation = new Operations.Operation { Status = StatusIds.Types.StatusCode.Success }
+        });
+
+    public ValueTask<IServerStream<TResponse>> ServerStreamCall<TRequest,
+        TResponse>(Method<TRequest, TResponse> method, TRequest request, GrpcRequestSettings settings) where
+        TRequest : class
+        where TResponse : class => throw new NotImplementedException();
+
+    public ValueTask<IBidirectionalStream<TRequest,
+        TResponse>> BidirectionalStreamCall<TRequest, TResponse>(Method<TRequest, TResponse> method,
+        GrpcRequestSettings settings) where TRequest : class where TResponse : class => throw new
+        NotImplementedException();
+
+    public ILoggerFactory LoggerFactory { get; }
+}

src/Ydb.Sdk/test/Ydb.Sdk.Ado.Benchmarks/README.md

@@ -54,4 +54,18 @@ DefaultJob : .NET 8.0.2 (8.0.224.6711), Arm64 RyuJIT AdvSIMD
 | HighContention_OpenClose            |     91,700.72 ns |     1,803.206 ns |     3,842.780 ns |             204.6633 |           0.0007 |   5.0049 |        - |   41951 B |
 | SessionReuse_Pattern                |    117,545.11 ns |     2,226.365 ns |     4,014.595 ns |             220.0000 |           0.0001 |   1.5869 |        - |   13656 B |
 | SessionReuse_HighContention_Pattern |  7,463,819.00 ns |   148,409.083 ns |   364,050.038 ns |           19044.6172 |           1.1719 | 765.6250 | 125.0000 | 6367528 B |
-| SessionReuse_HighIterations_Pattern | 70,844,972.06 ns | 1,400,128.942 ns | 3,589,066.009 ns |          200020.0000 |                - | 750.0000 |        - | 6407440 B |
+| SessionReuse_HighIterations_Pattern | 70,844,972.06 ns | 1,400,128.942 ns | 3,589,066.009 ns |          200020.0000 |                - | 750.0000 |        - | 6407440 B |
+
+# BulkUpsertImporter Benchmarks
+
+ButchSize = 15_000
+
+| Method                        |     Mean |    Error |   StdDev |
+|-------------------------------|---------:|---------:|---------:|
+| FlushAsync_BulkUpsertImporter | 32.41 ms | 0.546 ms | 0.456 ms | 
+
+ButchSize = 1_000
+
+| Method                        |     Mean |    Error |   StdDev |
+|-------------------------------|---------:|---------:|---------:|
+| FlushAsync_BulkUpsertImporter | 583.3 us | 11.32 us | 13.03 us |