         "else" "$$PRUNE"}}}
      {:$replaceRoot {:newRoot "$_root"}}]))
 
+(defn gather
+  [query-map]
+  (let [queries (into {} (map (fn [[k query]] [k (translate query)]) query-map))]
+    [{:$facet queries}]))
+
 (def steps
   {:fromEdge from-edge
    :toEdge to-edge

(defn ingest-labels-from-path!
  [graph path]
  (doseq [file (kafka/dir->files path)]
    (let [filename (.getName file)
          element (string/lower-case (kafka/path->label filename))
          lines (line-seq (io/reader file))
          parsed (map #(json/parse-string % keyword) lines)]
      (ingest-label-collections! graph element parsed filename))))

(def parse-args
  [["-c" "--config CONFIG" "path to config file"]

   :order order
   :offset offset
   :count qount
   :sort psort
   :gather gather
   :groupCount group-count})

(defn apply-step
