@Override
   public int insertDataToTable (ResultSet resultSet, int taskId) throws SQLException {

      String tableName;

      // Get table name and columns
      if (options.getMode().equals(ReplicationMode.COMPLETE.getModeText())) {
         tableName = getSinkTableName();
      } else {
         tableName = getQualifiedStagingTableName();
      }

      ResultSetMetaData rsmd = resultSet.getMetaData();
      int columnCount = rsmd.getColumnCount();

      SQLServerBulkCopy bulkCopy = new SQLServerBulkCopy(this.getConnection());
      // BulkCopy Options
      SQLServerBulkCopyOptions copyOptions = new SQLServerBulkCopyOptions();
      copyOptions.setBulkCopyTimeout(0);
      copyOptions.setBatchSize(options.getFetchSize());
      bulkCopy.setBulkCopyOptions(copyOptions);

      bulkCopy.setDestinationTableName(tableName);

      // Columns Mapping
      if (this.options.getSinkColumns() != null && !this.options.getSinkColumns().isEmpty()) {
         String sinkColumns = getAllSinkColumns(rsmd);
         // Remove quotes from column names, which are not supported by SQLServerBulkCopy
         String[] sinkColumnsArray = sinkColumns.replace("\"", "").split(",");
         LOG.trace("Mapping columns: source --> sink");
         for (int i = 1; i <= sinkColumnsArray.length; i++) {
            bulkCopy.addColumnMapping(rsmd.getColumnName(i), sinkColumnsArray[i - 1]);
            LOG.trace("{} --> {}", rsmd.getColumnName(i), sinkColumnsArray[i - 1]);
         }
      } else {
         for (int i = 1; i <= columnCount; i++) {
            LOG.trace("source {} - {} sink", rsmd.getColumnName(i), i);
            bulkCopy.addColumnMapping(rsmd.getColumnName(i), i);
         }
      }

      LOG.debug("Performing BulkCopy into {} ", tableName);
      try {
         // Write from the source to the destination.
         // If the source ResulSet is an implementation of RowSet (e.g. csv file) cast it.
         if (resultSet instanceof RowSet) {
            bulkCopy.writeToServer((RowSet) resultSet);
         } else {
            bulkCopy.writeToServer(resultSet);
         }


      } catch (SQLException ex) {
         LOG.error("Error while performing BulkCopy into {} ", tableName, ex);
         // log the error
         logColumnLengthError(bulkCopy, ex);

         throw ex;
      }

      bulkCopy.close();

      // TODO: getAllSinkColumns should not update the sinkColumns property. Change it in Oracle and check it in Postgres
      // Set Sink columns
      getAllSinkColumns(rsmd);
      
      this.getConnection().commit();
      return getNumFetchedRows(resultSet);
   }
Constructs the SqlManager.

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