     "Remove some records from the table, identified by a collection of id keys.
     Returns an updated table.")
 
+  (flush!
+    [table]
+    [table apply-patches?]
+    "Ensure that all local state has been persisted to the storage backend and
+    return an updated persisted table.
+
+    If `apply-patches?` is true, the current patch data will be merged into the
+    main data tree and the returned table will have no patch link.")
+
   ;; Partitions
 
   (list-partitions

  (.dirty table))


; TODO: put this in protocol?
(defn flush!
  "Ensure that all local state has been persisted to the storage backend and
  return an updated non-dirty table."
  ([table]
   (flush! table false))
  ([^Table table apply-patches?]
   (if (dirty? table)
     (let [[patch-link data-link]
             (if (or (seq (.pending table)) (::patch table))
               (if apply-patches?
                 ; Combine pending changes and patch tablet and update data tree.
                 [nil (throw (UnsupportedOperationException. "updated data-tree link"))]
                 ; flush any pending changes to the patch tablet
                 ; if patch tablet overflows, update main data tree
                 (throw (UnsupportedOperationException. "NYI")))
               ; No patch data or tablet.
               [nil (::data table)])
           root-data (-> (.root-data table)



(defn- -scan
  ([table]
   (-scan table nil))
  ([^Table table opts]
   (let [lexicoder (table-lexicoder table)
         start-key (some->> (:start-key opts) (key/encode lexicoder))
         end-key (some->> (:end-key opts) (key/encode lexicoder))]
     (->>
       (patch/patch-seq
         ; Merged patch data to apply to the records.
         (patch/filter-patch
           (.pending table) ; TODO: merge ::patch
           {:fields (:fields opts)
            :start-key start-key
            :end-key end-key})
         ; Lazy sequence of matching records from the index tree.
         (when-let [data-node (mdag/get-data
                                (.store table)
                                (::data (.root-data table)))]
           (if (or start-key end-key)
             (index/read-range
               (.store table)
               data-node
               (:fields opts)
               start-key
               end-key)
             (index/read-all
               (.store table)
               data-node
               (:fields opts)))))
       (patch/remove-tombstones)
       (map (key-decoder lexicoder))))))


(defn- -get-records
  ([table id-keys]
   (-get-records table id-keys nil))
  ([^Table table id-keys opts]
   (let [lexicoder (table-lexicoder table)
         id-keys (into #{} (map (partial key/encode lexicoder)) id-keys)
         patch-map (.pending table) ; TODO: merge ::patch
         patch-entries (select-keys patch-map id-keys)
         extra-keys (apply disj id-keys (keys patch-entries))
         patch-entries (patch/filter-patch patch-entries {:fields (:fields opts)})
         data-entries (when-let [data-node (and (seq extra-keys)
                                                (mdag/get-data
                                                  (.store table)
                                                  (::data (.root-data table))))]
                        (index/read-batch
                          (.store table)
                          data-node
                          (:fields opts)
                          extra-keys))]
     (->> (concat patch-entries data-entries)
          (patch/remove-tombstones)
          (map (key-decoder lexicoder))))))


(defn- -insert
  ([table records]
   (-insert table records nil))
  ([table records opts]
   (let [{:keys [merge-record merge-field]} opts
         lexicoder (table-lexicoder table)
         update-record (cond
                         merge-record merge-record
                         merge-field
                         (fn merge-fields
                           [_ old-data new-data]
                           (reduce
                             (fn [data field-key]
                               (assoc data
                                      field-key
                                      (merge-field field-key
                                                   (get old-data field-key)
                                                   (get new-data field-key))))
                             {} (distinct (concat (keys old-data) (keys new-data)))))
                         :else
                         (fn merge-simple
                           [_ old-data new-data]
                           (into {}
                                 (filter (comp some? val))
                                 (merge old-data new-data))))
         extant (into {} (get-records table (map first records)))
         new-records (map
                       (fn [[k data]]
                         [(key/encode lexicoder k)
                          (update-record k (get extant k) data)])
                       records)]
     ; Add new data maps to pending changes.
     (-> table
         (update-pending into new-records)
         (update ::data/count + (- (count records) (count extant)))))))


(defn- -delete
