Add: BulkUpsertImporter

src/Ydb.Sdk/src/Ado/YdbConnection.cs

@@ -1,8 +1,12 @@
 using System.Data;
 using System.Data.Common;
 using System.Diagnostics.CodeAnalysis;
+using Ydb.Operations;
 using Ydb.Sdk.Ado.Session;
 using Ydb.Sdk.Services.Query;
+using Ydb.Sdk.Value;
+using Ydb.Table;
+using Ydb.Table.V1;
 using static System.Data.IsolationLevel;
 
 namespace Ydb.Sdk.Ado;
@@ -52,6 +56,52 @@ public YdbConnection(YdbConnectionStringBuilder connectionStringBuilder)
         ConnectionStringBuilder = connectionStringBuilder;
     }
 
+    public async Task BulkUpsertAsync(
+        string tablePath,
+        IReadOnlyList<string> columns,
+        IReadOnlyList<IReadOnlyList<object?>> rows,
+        CancellationToken cancellationToken = default)
+    {
+        if (columns == null || columns.Count == 0)
+            throw new ArgumentException("Columns must not be empty", nameof(columns));
+        if (rows == null || rows.Count == 0)
+            throw new ArgumentException("Rows collection is empty", nameof(rows));
+
+        var structs = rows.Select(row =>
+        {
+            if (row.Count != columns.Count)
+                throw new ArgumentException("Each row must have the same number of elements as columns");
+            var members = columns
+                .Select((col, i) => new KeyValuePair<string, YdbValue>(col, new YdbParameter { Value = row[i] }.YdbValue))
+                .ToDictionary(x => x.Key, x => x.Value);
+            return YdbValue.MakeStruct(members);
+        }).ToList();
+
+        var list = YdbValue.MakeList(structs);
+
+        var req = new BulkUpsertRequest
+        {
+            Table = tablePath,
+            Rows = list.GetProto(),
+            OperationParams = new OperationParams()
+        };
+
+        if (Session is Services.Query.Session sessionImpl)
+        {
+            var resp = await sessionImpl.Driver.UnaryCall(
+                TableService.BulkUpsertMethod,
+                req,
+                new GrpcRequestSettings { CancellationToken = cancellationToken }
+            ).ConfigureAwait(false);
+            var status = Status.FromProto(resp.Operation.Status, resp.Operation.Issues);
+            status.EnsureSuccess();
+        }
+        else
+        {
+            throw new InvalidOperationException("Underlying session does not support BulkUpsertAsync");
+        }
+    }
+
     protected override YdbTransaction BeginDbTransaction(IsolationLevel isolationLevel)
     {
         ThrowIfConnectionClosed();

src/Ydb.Sdk/src/Ado/YdbDataSource.cs

@@ -65,6 +65,16 @@ protected override async ValueTask DisposeAsyncCore() =>
         await PoolManager.ClearPool(_ydbConnectionStringBuilder.ConnectionString);
 
     protected override void Dispose(bool disposing) => DisposeAsyncCore().AsTask().GetAwaiter().GetResult();
+
+    public async Task BulkUpsertAsync(
+        string tablePath,
+        IReadOnlyList<string> columns,
+        IReadOnlyList<IReadOnlyList<object?>> rows,
+        CancellationToken cancellationToken = default)
+    {
+        await using var conn = await OpenConnectionAsync(cancellationToken);
+        await conn.BulkUpsertAsync(tablePath, columns, rows, cancellationToken);
+    }
 }
 
 #endif

src/Ydb.Sdk/test/Ydb.Sdk.Ado.Tests/YdbConnectionTests.cs

@@ -308,4 +308,156 @@
 
     protected override async Task OnDisposeAsync() =>
         await YdbConnection.ClearPool(new YdbConnection(_connectionStringTls));
-}
+
+    [Fact]
+    public async Task BulkUpsert_HappyPath_C()
+    {
+        var tableName = $"BulkTest_{Guid.NewGuid():N}";
+        var database = new YdbConnectionStringBuilder(_connectionStringTls).Database.TrimEnd('/');
+        var absTablePath = string.IsNullOrEmpty(database) ? tableName : $"{database}/{tableName}";
+
+        await using var conn = new YdbConnection(_connectionStringTls);
+        await conn.OpenAsync();
+
+        await using (var createCmd = conn.CreateCommand())
+        {
+            createCmd.CommandText = $@"
+            CREATE TABLE {tableName} (
+                Id Int32,
+                Name Utf8,
+                PRIMARY KEY (Id)
+            )";
+            await createCmd.ExecuteNonQueryAsync();
+        }
+
+        await Task.Delay(500);
+
+        var columns = new[] { "Id", "Name" };
+        var rows = Enumerable.Range(1, 10)
+            .Select(i => new object?[] { i, $"Name {i}" })
+            .ToList();
+
+        await conn.BulkUpsertAsync(absTablePath, columns, rows);
+
+        await using (var checkCmd = conn.CreateCommand())
+        {
+            checkCmd.CommandText = $"SELECT COUNT(*) FROM {tableName}";
+            var count = Convert.ToInt32(await checkCmd.ExecuteScalarAsync());
+            Assert.Equal(rows.Count, count);
+        }
+
+        await using (var dropCmd = conn.CreateCommand())
+        {
+            dropCmd.CommandText = $"DROP TABLE {tableName}";
+            await dropCmd.ExecuteNonQueryAsync();
+        }
+    }
+
+    [Fact]
+    public async Task BulkUpsert_InsertsNewRows_C()
+    {
+        var tableName = $"BulkTest_{Guid.NewGuid():N}";
+        var database = new YdbConnectionStringBuilder(_connectionStringTls).Database.TrimEnd('/');
+        var absTablePath = string.IsNullOrEmpty(database) ? tableName : $"{database}/{tableName}";
+
+        await using var conn = new YdbConnection(_connectionStringTls);
+        await conn.OpenAsync();
+
+        await using (var createCmd = conn.CreateCommand())
+        {
+            createCmd.CommandText = $@"
+            CREATE TABLE {tableName} (
+                Id Int32,
+                Name Utf8,
+                PRIMARY KEY (Id)
+            )";
+            await createCmd.ExecuteNonQueryAsync();
+        }
+
+        await Task.Delay(500);
+
+        var columns = new[] { "Id", "Name" };
+
+        var firstRows = new List<object[]>
+        {
+            new object?[] { 1, "Alice" },
+            new object?[] { 2, "Bob" }
+        };
+        await conn.BulkUpsertAsync(absTablePath, columns, firstRows);
+
+        var newRows = new List<object[]>
+        {
+            new object?[] { 3, "Charlie" },
+            new object?[] { 4, "Diana" }
+        };
+        await conn.BulkUpsertAsync(absTablePath, columns, newRows);
+
+        await using (var selectCmd = conn.CreateCommand())
+        {
+            selectCmd.CommandText = $"SELECT Id, Name FROM {tableName}";
+            var results = new Dictionary<int, string>();
+            await using var reader = await selectCmd.ExecuteReaderAsync();
+            while (await reader.ReadAsync())
+            {
+                results[reader.GetInt32(0)] = reader.GetString(1);
+            }
+
+            Assert.Equal(4, results.Count);
+            Assert.Equal("Alice", results[1]);
+            Assert.Equal("Bob", results[2]);
+            Assert.Equal("Charlie", results[3]);
+            Assert.Equal("Diana", results[4]);
+        }
+
+        await using (var dropCmd = conn.CreateCommand())
+        {
+            dropCmd.CommandText = $"DROP TABLE {tableName};";
+            await dropCmd.ExecuteNonQueryAsync();
+        }
+    }
+
+    [Fact]
+    public async Task BulkUpsert_UpdatesExistingRows_C()
+    {
+        var tableName = $"BulkTest_{Guid.NewGuid():N}";
+        var database = new YdbConnectionStringBuilder(_connectionStringTls).Database.TrimEnd('/');
+        var absTablePath = string.IsNullOrEmpty(database) ? tableName : $"{database}/{tableName}";
+
+        await using var conn = new YdbConnection(_connectionStringTls);
+        await conn.OpenAsync();
+
+        await using (var createCmd = conn.CreateCommand())
+        {
+            createCmd.CommandText = $@"
+            CREATE TABLE {tableName} (
+                Id Int32,
+                Name Utf8,
+                PRIMARY KEY (Id)
+            )";
+            await createCmd.ExecuteNonQueryAsync();
+        }
+
+        await Task.Delay(500);
+
+        var columns = new[] { "Id", "Name" };
+
+        var row = new object?[] { 1, "Alice" };
+        await conn.BulkUpsertAsync(absTablePath, columns, new[] { row });
+
+        var updated = new object?[] { 1, "Alice Updated" };
+        await conn.BulkUpsertAsync(absTablePath, columns, new[] { updated });
+
+        await using (var selectCmd = conn.CreateCommand())
+        {
+            selectCmd.CommandText = $"SELECT Name FROM {tableName} WHERE Id = 1;";
+            var name = (string)(await selectCmd.ExecuteScalarAsync())!;
+            Assert.Equal("Alice Updated", name);
+        }
+
+        await using (var dropCmd = conn.CreateCommand())
+        {
+            dropCmd.CommandText = $"DROP TABLE {tableName};";
+            await dropCmd.ExecuteNonQueryAsync();
+        }
+    }
+}