@Override
   public void preSourceTasks () throws Exception {
      long totalRows = 0;

      if (this.options.getJobs() != 1) {
         this.getConnection();
         // get source collection
         MongoCollection<Document> collection = sourceDatabase.getCollection(this.options.getSourceTable());

         // Source Query
         if (options.getSourceQuery() != null && !options.getSourceQuery().isEmpty()) {
            String queryAggregation = options.getSourceQuery();
            List<BsonDocument> pipeline = getAggregation(queryAggregation);
            // add count
            pipeline.add(BsonDocument.parse("{ $count: \"count\" }"));
            LOG.info("Using this aggregation to count the total number of rows from the MongoDB source: {}", BsonUtils.toJsonStr(pipeline));
            // get the first document from the aggregation with 100 minutes timeout
            Document countDocument = collection.aggregate(pipeline).maxAwaitTime(100, TimeUnit.MINUTES).allowDiskUse(true).first();
            if (countDocument != null) {
               // get integer and cast to long
               totalRows = countDocument.getInteger("count",0);
            }
         } else {
            BsonDocument where = new BsonDocument();
            // Source Where
            if (options.getSourceWhere() != null && !options.getSourceWhere().isEmpty()) {
               where = BsonDocument.parse(options.getSourceWhere());
            }
            totalRows = collection.countDocuments(where);
         }

         // set chunk size for each task
         this.chunkSize = Math.abs(totalRows / this.options.getJobs());
         LOG.info("Source collection total rows: {}, chunk size per job: {}", totalRows, this.chunkSize);
      }
   }
Constructs the SqlManager.

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