Implement simple BulkUpsert in YdbConnection and YdbDataSource (delete module)

Author: LiamHamsters

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

@@ -2,9 +2,9 @@
 using System.Data.Common;
 using System.Diagnostics.CodeAnalysis;
 using Ydb.Operations;
-using Ydb.Sdk.Ado.BulkUpsert;
 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;
@@ -56,20 +56,26 @@ public YdbConnection(YdbConnectionStringBuilder connectionStringBuilder)
         ConnectionStringBuilder = connectionStringBuilder;
     }
 
-    internal async Task BulkUpsertInternalAsync<T>(
+    public async Task BulkUpsertAsync(
         string tablePath,
-        IReadOnlyCollection<T> rows,
         IReadOnlyList<string> columns,
-        IReadOnlyList<Func<T, object?>> selectors,
-        CancellationToken cancellationToken)
+        IReadOnlyList<IReadOnlyDictionary<string, object?>> rows,
+        CancellationToken cancellationToken = default)
     {
-        if (CurrentTransaction is { Completed: false })
-            throw new InvalidOperationException("BulkUpsert does not support working within a transaction");
+        var structs = rows.Select(dict =>
+            YdbValue.MakeStruct(
+                columns.ToDictionary(
+                    col => col,
+                    col => new YdbParameter { Value = dict[col] }.YdbValue
+                ))).ToList();
+
+        var list = YdbValue.MakeList(structs);
+
         var req = new BulkUpsertRequest
         {
             Table = tablePath,
-            OperationParams = new OperationParams(),
-            Rows = TypedValueFactory.FromObjects(rows, columns, selectors)
+            Rows = list.GetProto(),
+            OperationParams = new OperationParams()
         };
 
         if (Session is Services.Query.Session sessionImpl)

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

@@ -1,6 +1,5 @@
 #if NET7_0_OR_GREATER
 using System.Data.Common;
-using Ydb.Sdk.Ado.BulkUpsert;
 using Ydb.Sdk.Value;
 
 namespace Ydb.Sdk.Ado;
@@ -68,21 +67,14 @@ protected override async ValueTask DisposeAsyncCore() =>
 
     protected override void Dispose(bool disposing) => DisposeAsyncCore().AsTask().GetAwaiter().GetResult();
 
-    public async Task<YdbBulkUpsertImporter<T>> BeginBulkUpsertAsync<T>(
+    public async Task BulkUpsertAsync(
         string tablePath,
         IReadOnlyList<string> columns,
-        IReadOnlyList<Func<T, YdbValue>> selectors,
-        int maxRowsInBatch = 1000,
+        IReadOnlyList<IReadOnlyDictionary<string, object?>> rows,
         CancellationToken cancellationToken = default)
     {
-        var conn = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
-
-        return new YdbBulkUpsertImporter<T>(
-            conn,
-            tablePath,
-            columns,
-            maxRowsInBatch
-        );
+        await using var conn = await OpenConnectionAsync(cancellationToken);
+        await conn.BulkUpsertAsync(tablePath, columns, rows, cancellationToken);
     }
 }
 

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

@@ -316,6 +316,13 @@ private class TestEntity
         public string Name { get; set; }
     }
 
+    private static IReadOnlyList<IReadOnlyDictionary<string, object?>> ToDicts(IEnumerable<TestEntity> rows)
+        => rows.Select(x => (IReadOnlyDictionary<string, object?>)new Dictionary<string, object?>
+        {
+            ["Id"] = x.Id,
+            ["Name"] = x.Name
+        }).ToList();
+
     [Fact]
     public async Task BulkUpsert_HappyPath_C()
     {
@@ -329,11 +336,11 @@ public async Task BulkUpsert_HappyPath_C()
         await using (var createCmd = conn.CreateCommand())
         {
             createCmd.CommandText = $@"
-CREATE TABLE {tableName} (
-    Id Int32,
-    Name Utf8,
-    PRIMARY KEY (Id)
-)";
+            CREATE TABLE {tableName} (
+                Id Int32,
+                Name Utf8,
+                PRIMARY KEY (Id)
+            )";
             await createCmd.ExecuteNonQueryAsync();
         }
 
@@ -344,8 +351,7 @@ PRIMARY KEY (Id)
             .ToList();
 
         var columns = new[] { "Id", "Name" };
-
-        await conn.BulkUpsertWithRetry(absTablePath, rows, columns, default);
+        await conn.BulkUpsertAsync(absTablePath, columns, ToDicts(rows));
 
         await using (var checkCmd = conn.CreateCommand())
         {
@@ -374,11 +380,11 @@ public async Task BulkUpsert_InsertsNewRows_С()
         await using (var createCmd = conn.CreateCommand())
         {
             createCmd.CommandText = $@"
-CREATE TABLE {tableName} (
-    Id Int32,
-    Name Utf8,
-    PRIMARY KEY (Id)
-)";
+            CREATE TABLE {tableName} (
+                Id Int32,
+                Name Utf8,
+                PRIMARY KEY (Id)
+            )";
             await createCmd.ExecuteNonQueryAsync();
         }
 
@@ -391,14 +397,14 @@ PRIMARY KEY (Id)
             new() { Id = 1, Name = "Alice" },
             new() { Id = 2, Name = "Bob" }
         };
-        await conn.BulkUpsertWithRetry(absTablePath, firstRows, columns, default);
+        await conn.BulkUpsertAsync(absTablePath, columns, ToDicts(firstRows));
 
         var newRows = new List<TestEntity>
         {
             new() { Id = 3, Name = "Charlie" },
             new() { Id = 4, Name = "Diana" }
         };
-        await conn.BulkUpsertWithRetry(absTablePath, newRows, columns, default);
+        await conn.BulkUpsertAsync(absTablePath, columns, ToDicts(newRows));
 
         await using (var selectCmd = conn.CreateCommand())
         {
@@ -437,11 +443,11 @@ public async Task BulkUpsert_UpdatesExistingRows_С()
         await using (var createCmd = conn.CreateCommand())
         {
             createCmd.CommandText = $@"
-CREATE TABLE {tableName} (
-    Id Int32,
-    Name Utf8,
-    PRIMARY KEY (Id)
-)";
+            CREATE TABLE {tableName} (
+                Id Int32,
+                Name Utf8,
+                PRIMARY KEY (Id)
+            )";
             await createCmd.ExecuteNonQueryAsync();
         }
 
@@ -450,15 +456,15 @@ PRIMARY KEY (Id)
         var columns = new[] { "Id", "Name" };
 
         var row = new TestEntity { Id = 1, Name = "Alice" };
-        await conn.BulkUpsertWithRetry(absTablePath, new[] { row }, columns, default);
+        await conn.BulkUpsertAsync(absTablePath, columns, ToDicts([row]));
 
         var updated = new TestEntity { Id = 1, Name = "Alice Updated" };
-        await conn.BulkUpsertWithRetry(absTablePath, new[] { updated }, columns, default);
+        await conn.BulkUpsertAsync(absTablePath, columns, ToDicts([updated]));
 
         await using (var selectCmd = conn.CreateCommand())
         {
             selectCmd.CommandText = $"SELECT Name FROM {tableName} WHERE Id = 1;";
-            var name = (string)(await selectCmd.ExecuteScalarAsync())!;
+            var name = await selectCmd.ExecuteScalarAsync() as string;
             Assert.Equal("Alice Updated", name);
         }
 
@@ -468,5 +474,4 @@ PRIMARY KEY (Id)
             await dropCmd.ExecuteNonQueryAsync();
         }
     }
-
 }