                   node
                   (->> (::children node)
                        (map store!)
-                       (relink-children)
+                       (numbered-child-links)
                        (assoc node ::children)
                        (mdag/store-node! store nil)
                        (::node/data))))]


;; ## Update Functions

(defn- group-changes
  "Divide up a sorted sequence of changes into a lazy sequence of tuples of
  the first key in the subtree, the child link, and the associated changes, if
  any."
  [child-keys child-links changes]
  (lazy-seq
    (when (seq child-links)
      (if (seq changes)
        (let [next-key (first child-keys)
              [group more-changes]
                (if next-key
                  (split-with #(key/before? (first %) next-key) changes)
                  [changes nil])]
          (cons
            [(first child-links) (seq group)]
            (group-changes (next child-keys)
                           (next child-links)
                           more-changes)))
        ; No more changes
        (map vector child-keys)))))


(defn- redistribute-children
  "... accumulates child nodes and outputs valid nodes as necessary. Input
  values should be a vector of `[first-key link-or-candidate]`."
  [store limit count-children merge-nodes]
  (let [half-full (int (Math/ceil (/ limit 2)))
        valid? #(or (mdag/link? %) (<= half-full (count-children %) limit))]
    (fn [xf]
      (fn
        ([]
         [(xf) nil nil])
        ([[result prev curr]]
         (cond
           ; Current node is first in sequence.
           (nil? prev)
             (if curr
               ; Output only child node.
               (xf (xf (unreduced result) curr))
               ; No further children to process.
               (xf result))

           ; Two valid children, output directly.
           (valid? curr)
             (-> (unreduced result)
                 (xf prev)
                 (xf curr)
                 (xf))

           ; Final child is underflowing, merge with prev.
           :else
             (reduce
               xf
               (unreduced result)
               (merge-nodes
                 (if (mdag/link? prev)
                   (graph/get-link! store prev)
                   prev)
                 curr))))
        ([[result prev curr] [first-key child]]
         ; case [prev curr] input
         ; [nil nil]
         ; Input becomes current node unless it's an overflow, in which case it
         ; is split into multiple conforming nodes.
         ; TODO: might be more optimal to keep an overflowing node in curr,
         ; but would require changes to flush above.
         ;
         ; [nil link]
         ; - link => [link link]
         ; - underflow => [link underflow]
         ; - candidate => [link candidate]
         ; - overflow => split nodes, output all but last two candidates
         ;
         ; [nil underflow]
         ; - link => load link, merge with underflowing node
         ; - underflow => merge with current
         ; - candidate => redistribute with current
         ; - overflow => redistribute with current
         ;
         ; [nil candidate]
         ; - link
         ; - underflow
         ; - candidate
         ; - overflow
         ;
         ; [link link]
         ; - link
         ; - underflow
         ; - candidate
         ; - overflow
         ;
         ; [link underflow]
         ; - link
         ; - underflow
         ; - candidate
         ; - overflow
         ,,,)))))


(defn- update-index-node
  [store params carry [node-link changes]]
  (if (seq changes)
    (let [index (graph/get-link! store node-link)
          child-changes (group-changes
                          (::keys index)
                          (::children index)
                          changes)]
      (if (= 1 (::height index))
        ; Update children as partitions.
        (let [children (part/update-partitions! store params child-changes)]
          ; if carry is a virtual tablet, merge with children
          ; if carry is a partition, add it to the front (or back?) of changes list
          ; otherwise, not valid to have a carry here
          (prn :update-index-partitions children)
          (if (sequential? children)
            (if (= 1 (count children))
              ; Single partition to carry up the tree.
              (first children)
              (let [child-keys (into [] (comp (drop 1) (map ::record/first-key)) children)
                    child-links (into [] (map #(mdag/link "part" (meta %))) children)
                    record-count (reduce + 0 (map ::record/count children))]
                (assoc index
                       ::keys child-keys
                       ::children child-links
                       ::record/count record-count
                       ::record/first-key (::record/first-key (first children))
                       ::record/last-key (::record/last-key (peek children)))))
            ; Virtual tablet, return for carrying.
            children))
        ; Update child index nodes.
        (let [; if carry is an index subtree and the height is one less than
              ; ours, adopt it as a child with no changes. Otherwise, carry it
              ; into the update for the relevant child.
              children (mapv #(update-index-node store params nil %) child-changes)
              ; TODO: redistribute children
              child-keys (into [] (comp (drop 1) (map ::record/first-key)) children)
              record-count (reduce + 0 (map ::record/count children))]
          (prn :update-index-children children)
          (assoc index
                 ::keys child-keys
                 ::children children
                 ::record/count record-count
                 ::record/first-key (::record/first-key (first children))
                 ::record/last-key (::record/last-key (peek children))))))
    ; No changes, return unchanged node.
    (graph/get-link! store node-link)))


(defn- relink-children
  "Rewrite a vector of childen into numbered links."
  [children]
  (into []
        (map-indexed #(mdag/link (format "%03d" %1) %2))
        children))


(defn update-tree

    data-type
      (mapcat
        (fn read-child
          [[child-link record-keys]]
          (when (seq record-keys)
            (let [child (graph/get-link! store node child-link)]
              (read-batch store child fields (map first record-keys)))))
        (assign-records node (map vector record-keys)))

    part/data-type
      (part/read-batch store node fields record-keys)
