@Override
   protected void mergeStagingTable () throws Exception {

      this.getConnection();

      // merge staging table with sink table
      try {
         MongoCollection<Document> sinkStagingCollection = sinkDatabase.getCollection(getQualifiedStagingTableName());

         String aggregationQuery = "[ {$project: {_id:0}},{$merge:{ into:\"${SINK_COLLECTION}\", on:[${PRIMARY_KEYS}], whenMatched: \"replace\", whenNotMatched: \"insert\" }} ]";
         aggregationQuery = aggregationQuery.replace("${SINK_COLLECTION}", getSinkTableName());
         // replace primary keys separated by commas and surrounded by quotes
         aggregationQuery = aggregationQuery.replace("${PRIMARY_KEYS}", primaryKeys.stream().map(s -> "\"" + s + "\"").collect(Collectors.joining(",")));

         // merge collections
         LOG.info("Merging staging collection with sink collection using aggregation query: {}", aggregationQuery);
         List<BsonDocument> pipeline = getAggregation(aggregationQuery);
         sinkStagingCollection.aggregate(pipeline).allowDiskUse(true).maxAwaitTime(100, TimeUnit.MINUTES).first();


      } catch (Exception e) {
         LOG.error("Error merging staging table with sink table: {}", e.getMessage(), e);
         throw e;
      }
      this.close();
   }
Constructs the SqlManager.

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