Ticket #12 : Support bulk insert

Author: habarthierry-hue

samples/EFCore.Cassandra.Samples/Program.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See LICENSE in the project root for license information.
 using Cassandra;
 using EFCore.Cassandra.Samples.Models;
+using Microsoft.EntityFrameworkCore;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -18,6 +19,10 @@ static void Main(string[] args)
         {
             using (var dbContext = new FakeDbContext())
             {
+                Console.WriteLine("Bulk insert");
+                var applicants = Enumerable.Repeat(1, 1).Select(_ => BuildApplicant()).ToList();
+                dbContext.BulkInsert(applicants);
+
                 Console.WriteLine("Add applicant");
                 var timeUuid = TimeUuid.NewId();
                 dbContext.Applicants.Add(BuildApplicant());

src/EFCore.Cassandra.Benchmarks/InsertData.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See LICENSE in the project root for license information.
 using BenchmarkDotNet.Attributes;
 using EFCore.Cassandra.Benchmarks.Models;
+using Microsoft.EntityFrameworkCore;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -16,7 +17,7 @@ public class InsertData : IDisposable
         public InsertData()
         {
             _dbContext = new FakeDbContext();
-            _applicants = Enumerable.Range(1, 5).Select(_ => BuildApplicant()).ToArray();
+            _applicants = Enumerable.Range(1, 20).Select(_ => BuildApplicant()).ToArray();
         }
 
         public void Dispose()
@@ -25,7 +26,7 @@ public void Dispose()
         }
 
         [Benchmark]
-        public void Add100Applicants()
+        public void AddApplicants()
         {
             foreach (var applicant in _applicants)
             {
@@ -36,12 +37,18 @@ public void Add100Applicants()
         }
 
         [Benchmark]
-        public void AddRange100Applicants()
+        public void AddRangeApplicants()
         {
             _dbContext.Applicants.AddRange(_applicants);
             _dbContext.SaveChanges();
         }
 
+        [Benchmark]
+        public void BulkInsertApplicants()
+        {
+            _dbContext.BulkInsert(_applicants.ToList());
+        }
+
         private static Applicant BuildApplicant()
         {
             return new Applicant

src/EFCore.Cassandra/Bulk/DbContextBulkExtensions.cs

@@ -0,0 +1,21 @@
+// Copyright (c) SimpleIdServer. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See LICENSE in the project root for license information.
+using EFCore.Cassandra.Bulk;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Microsoft.EntityFrameworkCore
+{
+    public static class DbContextBulkExtensions
+    {
+        public static void BulkInsert<T>(this DbContext dbContext, List<T> entities)
+        {
+            SqlBulkOperation.Insert(dbContext, entities);
+        }
+
+        public static Task BulkInsertAsync<T>(this DbContext dbContext, List<T> entities)
+        {
+            return SqlBulkOperation.InsertAsync(dbContext, entities);
+        }
+    }
+}

src/EFCore.Cassandra/Bulk/SqlBulkOperation.cs

@@ -0,0 +1,85 @@
+// Copyright (c) SimpleIdServer. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See LICENSE in the project root for license information.
+using Cassandra;
+using Cassandra.Data;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Metadata;
+using Microsoft.EntityFrameworkCore.Storage;
+using Microsoft.EntityFrameworkCore.Update;
+using System.Collections.Generic;
+using System.Data;
+using System.Linq;
+using System.Net;
+using System.Reflection;
+using System.Threading.Tasks;
+
+namespace EFCore.Cassandra.Bulk
+{
+    internal class SqlBulkOperation
+    {
+        public static void Insert<T>(DbContext dbContext, IList<T> entities)
+        {
+            var result = BuildBatch(dbContext, entities);
+            result.session.Execute(result.batchStatement);
+        }
+
+        public static async Task InsertAsync<T>(DbContext dbContext, IList<T> entities)
+        {
+            var result = BuildBatch(dbContext, entities);
+            await result.session.ExecuteAsync(result.batchStatement);
+        }
+
+        private static (BatchStatement batchStatement, ISession session) BuildBatch<T>(DbContext dbContext, IList<T> entities)
+        {
+            var service = dbContext.GetService<ICommandBatchPreparer>();
+            var sqlGenerationHelper = dbContext.GetService<ISqlGenerationHelper>();
+            var database = dbContext.Database.GetDbConnection() as CqlConnection;
+            if (database.State != ConnectionState.Open)
+            {
+                database.Open();
+            }
+
+            var prop = typeof(CqlConnection).GetField("ManagedConnection", BindingFlags.NonPublic | BindingFlags.Instance);
+            var session = (ISession)prop.GetValue(database);
+            var batch = new BatchStatement();
+            foreach (var entity in entities)
+            {
+                var name = entity.GetType().FullName;
+                var entityType = dbContext.Model.FindEntityType(name);
+                var propertyNames = new List<string>();
+                var propertyValues = new List<object>();
+                foreach (var property in entityType.GetProperties())
+                {
+                    propertyNames.Add(sqlGenerationHelper.DelimitIdentifier(property.GetColumnName()));
+                    var propValue = property.PropertyInfo.GetValue(entity);
+                    propValue = GetValue(property, propValue);
+                    propertyValues.Add(propValue);
+                }
+
+                var tableName = entityType.GetTableName();
+                var schema = entityType.GetSchema();
+                var cqlQuery = $"INSERT INTO \"{schema}\".\"{tableName}\" ({string.Join(',', propertyNames)}) VALUES ({string.Join(',', Enumerable.Repeat(1, propertyNames.Count()).Select(_ => "?"))})";
+                var smt = session.Prepare(cqlQuery);
+                batch.Add(smt.Bind(propertyValues.ToArray()));
+            }
+
+            return (batch, session);
+        }
+
+        private static object GetValue(IProperty property, object value)
+        {
+            if(property.ClrType == typeof(IPAddress))
+            {
+                if (value == null)
+                {
+                    return new byte[0];
+                }
+
+                return ((IPAddress)value).GetAddressBytes();
+            }
+
+            return value;
+        }
+    }
+}