Batch aggregation (two-level batching)

metagraph/benchmarks/benchmark_query.cpp

@@ -18,10 +18,11 @@
 
 namespace mtg::cli {
 // Forward-declared because construct_query_graph is no longer in the public
-// header; the benchmark links against it directly.
+// header; the benchmark links against it directly. Takes a getter so the
+// annotation can be loaded asynchronously while the graph is being mapped.
 std::unique_ptr<graph::AnnotatedDBG>
 construct_query_graph(const graph::DeBruijnGraph &full_dbg,
-                      const graph::AnnotatedDBG::Annotator &full_annotation,
+                      const std::function<const graph::AnnotatedDBG::Annotator *()> &get_annotation,
                       StringGenerator call_sequences,
                       size_t num_threads,
                       const Config &config);
@@ -75,7 +76,7 @@ std::unique_ptr<AnnotatedDBG> build_query_graph(const AnnotatedDBG &anno_graph,
     mtg::cli::Config config(sizeof(argv) / sizeof(argv[0]), const_cast<char**>(argv));
     return cli::construct_query_graph(
         anno_graph.get_graph(),
-        anno_graph.get_annotator(),
+        [&anno_graph]() { return &anno_graph.get_annotator(); },
         [&](std::function<void(const std::string&)> call_sequence) {
             seq_io::read_fasta_file_critical(
                 query_filename,

metagraph/src/cli/config/config.cpp

@@ -182,6 +182,8 @@ Config::Config(int argc, char *argv[]) {
             num_columns_cached = atoi(get_value(i++));
         } else if (!strcmp(argv[i], "--batch-size")) {
             query_batch_size = atoll(get_value(i++));
+        } else if (!strcmp(argv[i], "--batch-min-matches")) {
+            batch_min_matches = std::stod(get_value(i++));
         } else if (!strcmp(argv[i], "-p") || !strcmp(argv[i], "--parallel")) {
             set_num_threads(atoi(get_value(i++)));
         } else if (!strcmp(argv[i], "--parallel-nodes")) {
@@ -1371,6 +1373,7 @@ if (advanced) {
             fprintf(stderr, "\t-p --parallel [INT] \tuse multiple threads for computation [1]\n");
             // fprintf(stderr, "\t   --cache-size [INT] \tnumber of uncompressed rows to store in the cache [0]\n");
             fprintf(stderr, "\t   --batch-size [INT] \tquery batch size in bp (0 to disable batch query) [100'000'000]\n");
+            fprintf(stderr, "\t   --batch-min-matches [FLOAT] \taggregate batches unless this they have this ratio of k-mer matches [0.0]\n");
 if (advanced) {
             fprintf(stderr, "\t   --threads-each [INT]\tnumber of parallel batches [1]\n");
             fprintf(stderr, "\t   --RA-ivbuff-size [INT] \tsize (in bytes) of int_vector_buffer used in random access mode (e.g. by row disk annotator) [16384]\n");

metagraph/src/cli/config/config.hpp

@@ -146,6 +146,7 @@ class Config {
     double min_fraction = 0.0;
     double max_fraction = 1.0;
     double cleaning_threshold_percentile = 0.001;
+    double batch_min_matches = 0.0;
     std::vector<double> count_slice_quantiles;
     std::vector<double> count_quantiles;