@Override
   protected void mergeStagingTable () throws SQLException {
      Statement statement = this.getConnection().createStatement();

      try {
         String[] pks = this.getSinkPrimaryKeys(this.getSinkTableName());
         // Primary key is required
         if (pks == null || pks.length == 0) {
            throw new IllegalArgumentException("Sink table must have at least one primary key column for incremental mode.");
         }

         // options.sinkColumns was set during the insertDataToTable
         String allColls = getAllSinkColumns(null);

         StringBuilder sql = new StringBuilder();
         sql.append("INSERT INTO ").append(this.getSinkTableName()).append(" (").append(allColls).append(" ) ").append(" SELECT ").append(allColls).append(" FROM ").append(this.getSinkStagingTableName()).append(" as excluded ON DUPLICATE KEY UPDATE ");

         // Set all columns for DO UPDATE SET statement
         for (String colName : allColls.split(",")) {
            sql.append(" ").append(colName).append(" = excluded.").append(colName).append(" ,");
         }
         // Delete the last comma
         sql.setLength(sql.length() - 1);

         LOG.info("Merging staging table and sink table with this command: {}", sql);
         statement.executeUpdate(sql.toString());
         statement.close();
         this.getConnection().commit();

      } catch (Exception e) {
         statement.close();
         this.connection.rollback();
         throw e;
      }
   }
Constructs the SqlManager.

@param opts the ReplicaDB ToolOptions describing the user's requested action.
src/main/java/org/replicadb/manager/MySQLManager.java