@Override
   public int insertDataToTable (ResultSet resultSet, int taskId) throws SQLException, IOException {
      int totalRows = 0;
      try {

         ResultSetMetaData rsmd = resultSet.getMetaData();
         String tableName;

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

         String allColumns = getAllSinkColumns(rsmd);

         // Get MySQL LOAD DATA manager
         String loadDataSql = getLoadDataSql(tableName, allColumns, rsmd);
         PreparedStatement statement = this.connection.prepareStatement(loadDataSql);

         JdbcPreparedStatement mysqlStatement = null;
         MariaDbStatement mariadbStatement = null;
         if (statement.isWrapperFor(MariaDbStatement.class)) {
            mariadbStatement = statement.unwrap(MariaDbStatement.class);
         } else {
            mysqlStatement = statement.unwrap(JdbcPreparedStatement.class);
         }

         char unitSeparator = 0x1F;
         char nullAscii = 0x00;
         int columnsNumber = rsmd.getColumnCount();

         StringBuilder row = new StringBuilder();
         StringBuilder cols = new StringBuilder();

         String colValue;
         int rowCounts = 0;
         int batchSize = options.getFetchSize();

         if (resultSet.next()) {
            // Create Bandwidth Throttling
            BandwidthThrottling bt = new BandwidthThrottling(options.getBandwidthThrottling(), options.getFetchSize(), resultSet);

            do {
               bt.acquiere();

               // Get Columns values
               for (int i = 1; i <= columnsNumber; i++) {
                  if (i > 1) cols.append(unitSeparator);

                  switch (rsmd.getColumnType(i)) {
                     case Types.CLOB:
                        colValue = clobToString(resultSet.getClob(i));
                        break;
                     case Types.BINARY:
                     case Types.VARBINARY:
                        colValue = byteToHex(resultSet.getBytes(i));
                        break;
                     case Types.LONGVARBINARY:
                     case Types.BLOB:
                        colValue = blobToHex(getBlob(resultSet,i));
                        break;
                     default:
                        colValue = resultSet.getString(i);
                        if (colValue == null) colValue = String.valueOf(nullAscii);
                        break;
                  }

                  if (!resultSet.wasNull() || colValue != null) cols.append(colValue);
               }

               // Escape special chars
               if (Boolean.TRUE.equals(this.options.isSinkDisableEscape())) {
                  row.append(cols.toString().replace("\u0000", "\\N") // MySQL localInfile Null value
                  );
               } else {
                  row.append(cols.toString().replace("\\", "\\\\").replace("\n", "\\n").replace("\r", "\\r").replace("\u0000", "\\N") // MySQL localInfile Null value
                  );
               }

               // Row ends with \n
               row.append("\n");

               // Copy data to mysql
               if (++rowCounts % batchSize == 0) {
                  copyData(loadDataSql, row, mariadbStatement, mysqlStatement);

                  // Clear StringBuilders
                  row.setLength(0); // set length of buffer to 0
                  row.trimToSize();
                  rowCounts = 0;
               }

               // Clear StringBuilders
               cols.setLength(0); // set length of buffer to 0
               cols.trimToSize();
               totalRows++;
            } while (resultSet.next());
         }

         // insert remaining records
         if (rowCounts != 0) {
            copyData(loadDataSql, row, mariadbStatement, mysqlStatement);
         }

      } catch (Exception e) {
         this.connection.rollback();
         throw e;
      }

      this.getConnection().commit();
      return totalRows;
   }
Constructs the SqlManager.

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