Merge pull request #105 from avalori/mergeDB

Add a new Database consumer which merges the records to the target DB rather than truncating and inserting as the current DatabaseDataSetConsumer does

Author: capgen628

git

@@ -15,6 +15,10 @@
 
 package org.alfasoftware.morf.dataset;
 
+import static java.util.stream.Collectors.toList;
+import static org.alfasoftware.morf.sql.SqlUtils.parameter;
+import static org.alfasoftware.morf.sql.SqlUtils.tableRef;
+
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 
@@ -25,8 +29,9 @@
 import org.alfasoftware.morf.metadata.SchemaUtils;
 import org.alfasoftware.morf.metadata.Table;
 import org.alfasoftware.morf.sql.InsertStatement;
+import org.alfasoftware.morf.sql.MergeStatement;
+import org.alfasoftware.morf.sql.SelectStatement;
 import org.alfasoftware.morf.sql.element.SqlParameter;
-import org.alfasoftware.morf.sql.element.TableReference;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -43,12 +48,12 @@ public class TableLoader {
   private final Connection connection;
   private final SqlDialect sqlDialect;
   private final boolean explicitCommit;
+  private final boolean merge;
   private final SqlScriptExecutor sqlExecutor;
   private final Table table;
   private final boolean insertingWithPresetAutonums;
   private final boolean insertingUnderAutonumLimit;
   private final boolean truncateBeforeLoad;
-  private final String insertStatement;
   private final int batchSize;
 
 
@@ -64,28 +69,26 @@ public static TableLoaderBuilder builder() {
    * Constructor.
    */
   TableLoader(Connection connection,
-                     SqlScriptExecutor sqlScriptExecutor,
-                     SqlDialect dialect,
-                     boolean explicitCommit,
-                     Table table,
-                     boolean insertingWithPresetAutonums,
-                     boolean insertingUnderAutonumLimit,
-                     boolean truncateBeforeLoad,
-                     int batchSize) {
+              SqlScriptExecutor sqlScriptExecutor,
+              SqlDialect dialect,
+              boolean explicitCommit,
+              boolean merge,
+              Table table,
+              boolean insertingWithPresetAutonums,
+              boolean insertingUnderAutonumLimit,
+              boolean truncateBeforeLoad,
+              int batchSize) {
     super();
     this.connection = connection;
     this.sqlExecutor = sqlScriptExecutor;
     this.sqlDialect = dialect;
     this.explicitCommit = explicitCommit;
+    this.merge = merge;
     this.table = table;
     this.insertingWithPresetAutonums = insertingWithPresetAutonums;
     this.insertingUnderAutonumLimit = insertingUnderAutonumLimit;
     this.truncateBeforeLoad = truncateBeforeLoad;
     this.batchSize = batchSize;
-    this.insertStatement = sqlDialect.convertStatementToSQL(
-      new InsertStatement().into(new TableReference(table.getName())),
-      SchemaUtils.schema(table)
-    );
   }
 
 
@@ -100,7 +103,7 @@ public void load(final Iterable<Record> records) {
       truncate(table, connection);
     }
 
-    insertRecords(records);
+    insertOrMergeRecords(records);
   }
 
 
@@ -131,13 +134,13 @@ private void truncate(Table table, Connection connection) {
    * @param records The records to insert
    * @param bulk Indicates if we should call {@link SqlDialect#preInsertStatements(Table)} and {@link SqlDialect#postInsertStatements(Table)}
    */
-  private void insertRecords(Iterable<Record> records) {
+  private void insertOrMergeRecords(Iterable<Record> records) {
 
     if (insertingWithPresetAutonums) {
       sqlExecutor.execute(sqlDialect.preInsertWithPresetAutonumStatements(table, insertingUnderAutonumLimit), connection);
     }
 
-    sqlInsertLoad(table, records, connection);
+    sqlInsertOrMergeLoad(table, records, connection);
 
     if (insertingWithPresetAutonums) {
       sqlDialect.postInsertWithPresetAutonumStatements(table, sqlExecutor, connection,insertingUnderAutonumLimit);
@@ -152,11 +155,26 @@ private void insertRecords(Iterable<Record> records) {
    * @param records The data to insert.
    * @param connection The connection.
    */
-  private void sqlInsertLoad(Table table, Iterable<Record> records, Connection connection) {
+  private void sqlInsertOrMergeLoad(Table table, Iterable<Record> records, Connection connection) {
     try {
-      sqlExecutor.executeStatementBatch(insertStatement, SqlParameter.parametersFromColumns(table.columns()), records, connection, explicitCommit, batchSize);
-    } catch (Exception e) {
-      throw new RuntimeException(String.format("Failure in batch insert for table [%s]", table.getName()), e);
+      if (merge && !table.primaryKey().isEmpty()) { // if the table has no primary we don't have a merge ON criterion
+        SelectStatement selectStatement = SelectStatement.select()
+            .fields(table.columns().stream().filter(c -> !c.isAutoNumbered()).map(c -> parameter(c)).collect(toList()))
+            .build();
+        MergeStatement mergeStatement = MergeStatement.merge()
+            .from(selectStatement)
+            .into(tableRef(table.getName()))
+            .tableUniqueKey(table.primaryKey().stream().map(c -> parameter(c)).collect(toList()))
+            .build();
+        String mergeSQL = sqlDialect.convertStatementToSQL(mergeStatement);
+        sqlExecutor.executeStatementBatch(mergeSQL, SqlParameter.parametersFromColumns(table.columns()), records, connection, explicitCommit, batchSize);
+      } else {
+        InsertStatement insertStatement = InsertStatement.insert().into(tableRef(table.getName())).build();
+        String insertSQL = sqlDialect.convertStatementToSQL(insertStatement, SchemaUtils.schema(table));
+        sqlExecutor.executeStatementBatch(insertSQL, SqlParameter.parametersFromColumns(table.columns()), records, connection, explicitCommit, batchSize);
+      }
+    } catch (Exception exceptionOnBatch) {
+      throw new RuntimeException(String.format("Failure in batch insert for table [%s]", table.getName()), exceptionOnBatch);
     }
   }
 }

morf-core/src/main/java/org/alfasoftware/morf/dataset/TableLoader.java

@@ -82,6 +82,16 @@ public interface TableLoaderBuilder {
    */
   TableLoaderBuilder explicitCommit(boolean explicitCommit);
 
+
+  /**
+   * Should we merge records rather than inserting?
+   *
+   * <p>Merging will overwrite records matched by primary key with the new data.</p>
+   *
+   * <p>Defaults to false if not specified.</p>
+   */
+  TableLoaderBuilder merge(boolean ignoreDuplicateKeys);
+
   /**
    * Should the table be truncated before the load?
    *
@@ -142,6 +152,7 @@ class TableLoaderBuilderImpl implements TableLoaderBuilder {
     private boolean insertingUnderAutonumLimit;
     private Provider<SqlDialect> sqlDialect;
     private int batchSize = 1000;
+    private boolean merge;
 
     TableLoaderBuilderImpl() {
       super();
@@ -188,6 +199,12 @@ public TableLoaderBuilder explicitCommit(boolean explicitCommit) {
       return this;
     }
 
+    @Override
+    public TableLoaderBuilder merge(boolean merge) {
+      this.merge = merge;
+      return this;
+    }
+
     @Override
     public TableLoaderBuilder truncateBeforeLoad() {
       this.truncateBeforeLoad = true;
@@ -222,15 +239,16 @@ public TableLoader forTable(Table table) {
         executor = sqlScriptExecutorProvider.get();
       }
       return new TableLoader(
-        connection,
-        executor,
-        sqlDialect.get(),
-        explicitCommit,
-        table,
-        insertingWithPresetAutonums,
-        insertingUnderAutonumLimit,
-        truncateBeforeLoad,
-        batchSize);
+          connection,
+          executor,
+          sqlDialect.get(),
+          explicitCommit,
+          merge,
+          table,
+          insertingWithPresetAutonums,
+          insertingUnderAutonumLimit,
+          truncateBeforeLoad,
+          batchSize);
     }
   }
 }

morf-core/src/main/java/org/alfasoftware/morf/dataset/TableLoaderBuilder.java

@@ -48,17 +48,17 @@ public class DatabaseDataSetConsumer implements DataSetConsumer {
   /**
    * Database connection to which the data is written.
    */
-  private Connection connection;
+  protected Connection connection;
 
   /**
    * The SQL dialect
    */
-  private SqlDialect sqlDialect;
+  protected SqlDialect sqlDialect;
 
   /**
    * Used to run SQL statements.
    */
-  private final SqlScriptExecutor sqlExecutor;
+  protected final SqlScriptExecutor sqlExecutor;
 
   /**
    * DataSource providing database connections.

morf-core/src/main/java/org/alfasoftware/morf/jdbc/DatabaseDataSetConsumer.java

@@ -0,0 +1,79 @@
+/* Copyright 2020 Alfa Financial Software
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.alfasoftware.morf.jdbc;
+
+import javax.sql.DataSource;
+
+import org.alfasoftware.morf.dataset.DataSetConsumer;
+import org.alfasoftware.morf.dataset.Record;
+import org.alfasoftware.morf.dataset.TableLoader;
+import org.alfasoftware.morf.metadata.Table;
+
+import com.google.inject.Inject;
+
+
+/**
+ * Implementation of {@link DataSetConsumer} that pipes records into a JDBC
+ * database connection without truncating tables and merging duplicate records.
+ *
+ *
+ * @author Copyright (c) Alfa Financial Software 2020
+ */
+public class MergingDatabaseDataSetConsumer extends DatabaseDataSetConsumer implements DataSetConsumer {
+
+
+  /**
+   * Creates an instance of the database data set consumer.
+   *
+   * @param connectionResources Provides database connection interfaces.
+   * @param sqlScriptExecutorProvider a provider of {@link SqlScriptExecutor}s.
+   */
+  public MergingDatabaseDataSetConsumer(ConnectionResources connectionResources, SqlScriptExecutorProvider sqlScriptExecutorProvider) {
+    this(connectionResources, sqlScriptExecutorProvider, connectionResources.getDataSource());
+  }
+
+
+  /**
+   * Creates an instance of the database data set consumer.
+   *
+   * <p>Can also be injected from Guice.</p>
+   *
+   * @param connectionResources Provides database connection interfaces.
+   * @param sqlScriptExecutorProvider a provider of {@link SqlScriptExecutor}s.
+   * @param dataSource DataSource providing database connections.
+   */
+  @Inject
+  public MergingDatabaseDataSetConsumer(ConnectionResources connectionResources, SqlScriptExecutorProvider sqlScriptExecutorProvider, DataSource dataSource) {
+    super(connectionResources, sqlScriptExecutorProvider, dataSource);
+  }
+
+
+  /**
+   * @see DataSetConsumer#table(Table, Iterable)
+   */
+  @Override
+  public void table(Table table, Iterable<Record> records) {
+    TableLoader.builder()
+      .withConnection(connection)
+      .withSqlScriptExecutor(sqlExecutor)
+      .withDialect(sqlDialect)
+      .explicitCommit(true)
+      .merge(true)
+      .insertingWithPresetAutonums()
+      .forTable(table)
+      .load(records);
+  }
+}