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

      String collectionName = getInsertDataCollection();

      MongoCollection<Document> sinkCollection = sinkDatabase.getCollection(collectionName);
      List<WriteModel<Document>> writeOperations = new ArrayList<>();
      // unordered bulk write
      BulkWriteOptions bulkWriteOptions = new BulkWriteOptions().ordered(false);

      if (resultSet != null && resultSet.next()) {
         // Create Bandwidth Throttling
         BandwidthThrottling bt = new BandwidthThrottling(options.getBandwidthThrottling(), options.getFetchSize(), resultSet);
         do {
            bt.acquiere();
            if (Boolean.TRUE.equals(isSourceAndSinkMongoDB())) {
               // Add document to bulk
               writeOperations.add(new InsertOneModel<>((Document) resultSet.getObject(1)));
            } else {
               // iterate columns
               Document document = new Document();
               for (int i = 1; i <= resultSet.getMetaData().getColumnCount(); i++) {
                  String columnName = resultSet.getMetaData().getColumnName(i);
                  switch (resultSet.getMetaData().getColumnType(i)) {
                     case -104: //Oracle INTERVALDS
                     case -103: //Oracle INTERVALYM
                     case Types.SQLXML:
                        document.put(columnName, resultSet.getString(i));
                        break;
                     case Types.TIMESTAMP:
                     case Types.TIMESTAMP_WITH_TIMEZONE:
                     case -101:
                     case -102:
                        document.put(columnName, resultSet.getTimestamp(i));
                        break;
                     case Types.BINARY:
                     case Types.VARBINARY:
                     case Types.LONGVARBINARY:
                        document.put(columnName, resultSet.getBytes(i));
                        break;
                     case Types.BLOB:
                        Blob blobData = resultSet.getBlob(i);
                        if (blobData != null) {
                           document.put(columnName, blobData.getBytes(1, (int) blobData.length()));
                           blobData.free();
                        }
                        break;
                     case Types.CLOB:
                        Clob clobData = resultSet.getClob(i);
                        document.put(columnName, clobToString(clobData));
                        if (clobData != null) clobData.free();
                        break;
                     case 1111: // Postgres JSON, intervals and others
                        Object object = resultSet.getObject(i);
                        if (object instanceof PGobject) {
                           PGobject pgObject = (PGobject) object;
                           // if Document.parse fails, will be saved as a string
                           try {
                              document.put(columnName, Document.parse(pgObject.getValue()));
                           } catch (Exception e) {
                              document.put(columnName, pgObject.getValue());
                           }
                        } else {
                           document.put(columnName, object);
                        }
                        break;
                     default:
                        document.put(columnName, resultSet.getObject(i));
                        break;
                  }
               }

               // Add document to bulk
               writeOperations.add(new InsertOneModel<>(document));
            }

            totalRows++;
            if (writeOperations.size() == options.getFetchSize()) {
               sinkCollection.bulkWrite(writeOperations, bulkWriteOptions);
               writeOperations.clear();
            }
         } while (resultSet.next());

         // insert remaining documents
         if (!writeOperations.isEmpty()) {
            sinkCollection.bulkWrite(writeOperations, bulkWriteOptions);
            writeOperations.clear();
         }
      }

      return totalRows;
   }
Constructs the SqlManager.

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