@Override
   protected void mergeStagingTable () throws SQLException {
      this.getConnection().commit();

      Statement statement = this.getConnection().createStatement();

      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("MERGE INTO ")
          .append(this.getSinkTableName())
          .append(" trg USING (SELECT ")
          .append(allColls)
          .append(" FROM ")
          .append(getQualifiedStagingTableName())
          .append(" ) src ON ")
          .append(" (");

      for (int i = 0; i <= pks.length - 1; i++) {
         if (i >= 1) sql.append(" AND ");
         sql.append("src.").append(pks[i]).append("= trg.").append(pks[i]);
      }

      sql.append(" ) ");
      LOG.trace("allColls: {} \n pks: {}", allColls, pks);

      // Set all columns for UPDATE SET statement
      String allColSelect = Arrays.stream(allColls.split("\\s*,\\s*"))
          .filter(colName -> {
             boolean contains = Arrays.asList(pks).contains(colName);
             boolean containsQuoted = Arrays.asList(pks).contains("\"" + colName + "\"");
             return !contains && !containsQuoted;
          }).map(colName -> {
             return String.format("trg.%s = src.%s", colName, colName);
          }).collect(Collectors.joining(", "));

      if (allColSelect.length() > 0) {
         sql.append(" WHEN MATCHED THEN UPDATE SET ");
         sql.append(allColSelect);
      } else {
         LOG.warn("All columns in the sink table are Primary Keys. WHEN MATCHED DO NOTHING.");
      }

      sql.append(" WHEN NOT MATCHED THEN INSERT ( ").append(allColls).
          append(" ) VALUES (");

      // all columns for INSERT VALUES statement
      for (String colName : allColls.split("\\s*,\\s*")) {
         sql.append(" src.").append(colName).append(" ,");
      }
      // Delete the last comma
      sql.setLength(sql.length() - 1);

      sql.append(" ); ");

      LOG.info("Merging staging table and sink table with this command: {}", sql);
      statement.executeUpdate(sql.toString());
      statement.close();
      this.getConnection().commit();
   }
Log the error message and the column name and length that caused the error using reflection.

@param bulkCopy the SQLServerBulkCopy object
@param ex       the SQLException
src/main/java/org/replicadb/manager/SQLServerManager.java