Fix SqlServerBulkInsert

Unit Test Successful

Author: Philipp Wagner

JSqlServerBulkInsert/src/main/java/de/bytefish/jsqlserverbulkinsert/SqlServerBulkInsert.java

@@ -5,10 +5,12 @@
 
 import com.microsoft.sqlserver.jdbc.ISQLServerBulkRecord;
 import com.microsoft.sqlserver.jdbc.SQLServerBulkCopy;
+import com.microsoft.sqlserver.jdbc.SQLServerBulkCopyOptions;
 import de.bytefish.jsqlserverbulkinsert.functional.Func2;
 import de.bytefish.jsqlserverbulkinsert.model.ColumnDefinition;
 import de.bytefish.jsqlserverbulkinsert.model.ColumnMetaData;
 import de.bytefish.jsqlserverbulkinsert.model.TableDefinition;
+import de.bytefish.jsqlserverbulkinsert.records.SqlServerRecord;
 import de.bytefish.jsqlserverbulkinsert.records.SqlServerRecordBuilder;
 
 import java.math.BigDecimal;
@@ -36,17 +38,26 @@ public SqlServerBulkInsert(String schemaName, String tableName)
     }
 
     public void saveAll(Connection connection, Stream<TEntity> entities) throws SQLException {
+        saveAll(connection, new SQLServerBulkCopyOptions(), entities);
+    }
+
+    public void saveAll(Connection connection, SQLServerBulkCopyOptions options, Stream<TEntity> entities) throws SQLException {
 
         final SqlServerRecordBuilder<TEntity> sqlServerRecordBuilder = new SqlServerRecordBuilder<>(columns);
 
         try (SQLServerBulkCopy sqlServerBulkCopy = new SQLServerBulkCopy(connection)) {
 
+            // Set the Options:
+            sqlServerBulkCopy.setBulkCopyOptions(options);
+
             // The Destination Table to write to:
             sqlServerBulkCopy.setDestinationTableName(table.GetFullQualifiedTableName());
 
-            entities
-                    .map(x -> sqlServerRecordBuilder.build(x))
-                    .forEach(x -> internalWriteToServer(sqlServerBulkCopy, x));
+            // The SQL Records to insert:
+            ISQLServerBulkRecord record = new SqlServerRecord<TEntity>(columns, entities.iterator());
+
+            // Finally start the Bulk Copy Process:
+            internalWriteToServer(sqlServerBulkCopy, record);
         }
     }
 

JSqlServerBulkInsert/src/main/java/de/bytefish/jsqlserverbulkinsert/records/SqlServerRecord.java

@@ -5,27 +5,40 @@
 
 import com.microsoft.sqlserver.jdbc.ISQLServerBulkRecord;
 import com.microsoft.sqlserver.jdbc.SQLServerException;
+import de.bytefish.jsqlserverbulkinsert.model.ColumnDefinition;
 import de.bytefish.jsqlserverbulkinsert.model.ColumnMetaData;
 
+import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 
-public class SqlServerRecord implements ISQLServerBulkRecord {
+public class SqlServerRecord<TEntity> implements ISQLServerBulkRecord {
+
+    private final Iterator<TEntity> entities;
 
     private final List<ColumnMetaData> columnMetaData;
-    private final Object[] values;
 
-    public SqlServerRecord(List<ColumnMetaData> columnMetaData, Object[] values) {
-        if(columnMetaData == null) {
-            throw new IllegalArgumentException("columnMetaData");
+    private final SqlServerRecordBuilder<TEntity> builder;
+
+    public SqlServerRecord(List<ColumnDefinition<TEntity>> columnDefinition, Iterator<TEntity> entities) {
+        if(columnDefinition == null) {
+            throw new IllegalArgumentException("columnDefinition");
         }
-        if(values == null) {
-            throw new IllegalArgumentException("values");
+        if(entities == null) {
+            throw new IllegalArgumentException("entities");
         }
-        this.columnMetaData = columnMetaData;
-        this.values = values;
+        this.entities = entities;
+
+        // Cache the Column Meta Data, so we don't calculate it for each Record:
+        this.columnMetaData = columnDefinition.stream()
+                .map(x -> x.getColumnMetaData())
+                .collect(Collectors.toList());
+
+        // Cache a Values Builder to populate Records faster:
+        this.builder = new SqlServerRecordBuilder<TEntity>(columnDefinition);
     }
 
     @Override
@@ -64,11 +77,13 @@ public boolean isAutoIncrement(int i) {
 
     @Override
     public Object[] getRowData() throws SQLServerException {
-        return values;
+        TEntity entity = entities.next();
+
+        return builder.build(entity);
     }
 
     @Override
     public boolean next() throws SQLServerException {
-        return false;
+        return entities.hasNext();
     }
 }

JSqlServerBulkInsert/src/main/java/de/bytefish/jsqlserverbulkinsert/records/SqlServerRecordBuilder.java

@@ -12,32 +12,19 @@
 public class SqlServerRecordBuilder<TEntity> {
 
     private final List<ColumnDefinition<TEntity>> columns;
-    private final List<ColumnMetaData> columnMetaData;
 
     public SqlServerRecordBuilder(List<ColumnDefinition<TEntity>> columns) {
         if(columns == null) {
             throw new IllegalArgumentException("columns");
         }
-
-        // Holds the Columns:
         this.columns = columns;
-
-        // Cache the Column Meta Data, so we don't calculate it for each Record:
-        this.columnMetaData = columns.stream()
-                .map(x -> x.getColumnMetaData())
-                .collect(Collectors.toList());
     }
 
-    public SqlServerRecord build(final TEntity entity) {
-
+    public Object[] build(final TEntity entity) {
         // Get the Values for each column:
-        Object[] values = columns.stream()
+        return columns.stream()
                 .map(x -> x.getPropertyValue(entity))
                 .toArray();
-
-        // And finally build the SqlServerRecord:
-        return new SqlServerRecord(columnMetaData, values);
-
     }
 
 }

JSqlServerBulkInsert/src/test/java/de/bytefish/jsqlserverbulkinsert/IntegrationTest.java

@@ -33,14 +33,16 @@ protected void onSetUpInTransaction() throws Exception {
 
     @Test
     public void bulkInsertPersonDataTest() throws SQLException {
+        // The Number of Entities to insert:
+        int numEntities = 1000000;
         // Create a large list of Persons:
-        List<Person> persons = getPersonList(100000);
+        List<Person> persons = getPersonList(numEntities);
         // Create the BulkInserter:
         PersonBulkInserter personBulkInserter = new PersonBulkInserter();
         // Now save all entities of a given stream:
         personBulkInserter.saveAll(connection, persons.stream());
         // And assert all have been written to the database:
-        Assert.assertEquals(100000, getRowCount());
+        Assert.assertEquals(numEntities, getRowCount());
     }
 
     private List<Person> getPersonList(int numPersons) {
@@ -77,9 +79,9 @@ private int getRowCount() throws SQLException {
 
         Statement s = connection.createStatement();
 
-        ResultSet r = s.executeQuery("SELECT COUNT(*) AS rowcount FROM [dbo].[UnitTest]");
+        ResultSet r = s.executeQuery("SELECT COUNT(*) AS total FROM [dbo].[UnitTest];");
         r.next();
-        int count = r.getInt("rowcount");
+        int count = r.getInt("total");
         r.close();
 
         return count;