@Override
   public ResultSet readTable (String tableName, String[] columns, int nThread) throws SQLException {
      // If table name parameter is null get it from options
      String collectionName = tableName == null ? this.options.getSourceTable() : tableName;
      // if the chunk size is 0 and the current job is greater than 0, return null
      if (chunkSize == 0 && nThread > 0) {
         return null;
      }

      long skip = nThread * chunkSize;
      mongoDbResultSet = new MongoDBRowSetImpl();

      try {
         // set fetch size
         mongoDbResultSet.setFetchSize(options.getFetchSize());
         // get a handle to the collection
         MongoCollection<Document> collection = sourceDatabase.getCollection(collectionName);
         MongoCursor<Document> cursor;
         Document firstDocument;

         // if source query is specified, use it as mongodb aggregation pipeline
         if (options.getSourceQuery() != null) {
            mongoDbResultSet.setAggregation(true);
            String queryAggregation = options.getSourceQuery();
            List<BsonDocument> pipeline = getAggregation(queryAggregation);

            if (this.options.getJobs() == nThread + 1) {
               // If it's the last job, skip the first documents
               pipeline.add(BsonDocument.parse("{ $skip: " + skip + " }"));
            } else {
               // add skip and limit to the pipeline
               pipeline.add(BsonDocument.parse("{ $skip: " + skip + " }"));
               pipeline.add(BsonDocument.parse("{ $limit: " + chunkSize + " }"));
            }

            LOG.info("{}: Using this aggregation query to get data from MongoDB: {}", Thread.currentThread().getName(), BsonUtils.toJsonStr(pipeline));
            // create a MongoCursor to iterate over the results
            cursor = collection.aggregate(pipeline).batchSize(options.getFetchSize()).allowDiskUse(true).cursor();
            firstDocument = collection.aggregate(pipeline).allowDiskUse(true).maxAwaitTime(100, TimeUnit.MINUTES).first();
         } else {
            // create a MongoCursor to iterate over the results
            FindIterable<Document> findIterable = collection.find();

            // Source Where
            if (options.getSourceWhere() != null && !options.getSourceWhere().isEmpty()) {
               BsonDocument filter = BsonDocument.parse(options.getSourceWhere());
               findIterable.filter(filter);
               LOG.info("{}: Using this clause to filter data from MongoDB: {}", Thread.currentThread().getName(), filter.toJson());
            }
            // Source Fields
            if (options.getSourceColumns() != null && !options.getSourceColumns().isEmpty()) {
               BsonDocument projection = BsonDocument.parse(options.getSourceColumns());
               findIterable.projection(projection);
               mongoDbResultSet.setMongoProjection(projection);
               LOG.info("{}: Using this clause to project data from MongoDB: {}", Thread.currentThread().getName(), projection.toJson());
            }

            if (this.options.getJobs() == nThread + 1) {
               // If it's the last job, skip the first documents
               findIterable.skip((int) skip);
               LOG.info("{}: Skip {} data from source", Thread.currentThread().getName(), skip);
            } else {
               // add skip and limit to the pipeline
               findIterable.skip(Math.toIntExact(skip));
               findIterable.limit(Math.toIntExact(chunkSize));
               LOG.info("{}: Skip {}, Limit {} data from source", Thread.currentThread().getName(), skip, chunkSize);
            }

            // if it is parallel processing
            if (options.getJobs() > 1) {
               // sort by object id
               findIterable.sort(Sorts.ascending("_id"));
               LOG.info("{}: Sort by _id", Thread.currentThread().getName());
               // not compatible with 4.x version
               // findIterable.allowDiskUse(true);
            }

            findIterable.batchSize(options.getFetchSize());
            cursor = findIterable.cursor();
            firstDocument = findIterable.first();
         }

         mongoDbResultSet.setMongoFirstDocument(firstDocument);
         mongoDbResultSet.setSinkMongoDB(isSourceAndSinkMongoDB());
         mongoDbResultSet.execute();
         mongoDbResultSet.setMongoCursor(cursor);

      } catch (JsonParseException jpe) {
         LOG.error("{}: Parse JSON exception in some source parameters where, query, columns: {}", Thread.currentThread().getName(), jpe.getMessage(), jpe);
         // rethrow the exception
         throw jpe;
      } catch (Exception e) {
         LOG.error("{}: Error: {}", Thread.currentThread().getName(), e.getMessage(), e);
         // rethrow the exception
         throw e;
      }
      return mongoDbResultSet;
   }
Constructs the SqlManager.

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