feat: switch BulkUpsertImporter to plain proto, remove extra wrappers and disposables

Author: LiamHamsters

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

@@ -0,0 +1,90 @@
+using Google.Protobuf.Collections;
+using Ydb.Sdk.Ado.Internal;
+using Ydb.Sdk.Value;
+using Ydb.Table;
+using Ydb.Table.V1;
+
+namespace Ydb.Sdk.Ado.BulkUpsert;
+
+public sealed class BulkUpsertImporter : IBulkUpsertImporter
+{
+    private readonly IDriver _driver;
+    private readonly string _tablePath;
+    private readonly IReadOnlyList<string> _columns;
+    private readonly RepeatedField<Ydb.Value> _rows = new();
+    private readonly CancellationToken _cancellationToken;
+    private StructType? _structType;
+
+    public BulkUpsertImporter(
+        IDriver driver,
+        string tableName,
+        IReadOnlyList<string> columns,
+        CancellationToken cancellationToken = default)
+    {
+        _driver = driver;
+        _tablePath = tableName;
+        _columns = columns;
+        _cancellationToken = cancellationToken;
+    }
+
+    public async ValueTask AddRowsAsync(IEnumerable<object?[]> rows)
+    {
+        foreach (var values in rows)
+        {
+            if (values.Length != _columns.Count)
+                throw new ArgumentException("Values count must match columns count", nameof(values));
+
+            var ydbValues = values.Select(v =>
+                v as YdbValue ?? (v is YdbParameter param ? param.YdbValue :
+                    new YdbParameter { Value = v }.YdbValue)
+            ).ToArray();
+
+            var protoStruct = new Ydb.Value();
+            foreach (var value in ydbValues)
+                protoStruct.Items.Add(value.GetProto().Value);
+
+            _rows.Add(protoStruct);
+
+            _structType ??= new StructType
+            {
+                Members =
+                {
+                    _columns.Select((col, i) => new StructMember
+                    {
+                        Name = col,
+                        Type = ydbValues[i].GetProto().Type
+                    })
+                }
+            };
+
+            await FlushAsync(_cancellationToken);
+        }
+    }
+
+    public async ValueTask FlushAsync(CancellationToken cancellationToken = default)
+    {
+        if (_rows.Count == 0) return;
+        if (_structType == null)
+            throw new InvalidOperationException("structType is undefined");
+
+        var typedValue = new TypedValue
+        {
+            Type = new Type { ListType = new ListType { Item = new Type { StructType = _structType } } },
+            Value = new Ydb.Value { Items = { _rows } }
+        };
+
+        var req = new BulkUpsertRequest { Table = _tablePath, Rows = typedValue };
+
+        var resp = await _driver.UnaryCall(
+            TableService.BulkUpsertMethod,
+            req,
+            new GrpcRequestSettings { CancellationToken = cancellationToken }
+        ).ConfigureAwait(false);
+
+        var operation = resp.Operation;
+        if (operation.Status.IsNotSuccess())
+            throw YdbException.FromServer(operation.Status, operation.Issues);
+
+        _rows.Clear();
+    }
+}

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

@@ -1,11 +1,7 @@
-using Ydb.Sdk.Value;
-
 namespace Ydb.Sdk.Ado.BulkUpsert;
 
-public interface IBulkUpsertImporter : IAsyncDisposable
+public interface IBulkUpsertImporter
 {
-    ValueTask AddRowsAsync(IEnumerable<YdbValue[]> rows);
     ValueTask AddRowsAsync(IEnumerable<object?[]> rows);
-
     ValueTask FlushAsync(CancellationToken cancellationToken = default);
-}
+}

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

@@ -66,7 +66,7 @@ public IBulkUpsertImporter BeginBulkUpsertImport(
         var database = ConnectionStringBuilder.Database.TrimEnd('/');
         var tablePath = string.IsNullOrEmpty(database) ? name : $"{database}/{name}";
 
-        return new BulkUpsertImporter(this, tablePath, columns, cancellationToken);
+        return new BulkUpsertImporter(Session.Driver, tablePath, columns, cancellationToken);
     }
 
     protected override YdbTransaction BeginDbTransaction(IsolationLevel isolationLevel)