 
 
 (defn partition-records
-  "Divides the given records into one or more partitions. Returns a sequence of
-  node data for the persisted partitions."
+  "Divides the given (possibly lazy) sequence of records into one or more
+  partitions. Returns a sequence of node data for the persisted partitions."
   [store params records]
-  (->> records
-       (patch/remove-tombstones)
-       (split-limited (max-limit params))
-       (mapv #(from-records store params %))))
+  (let [part-size (max-limit params)
+        threshold (+ part-size (min-limit params))]
+    ; TODO: callback on partition creation?
+    (loop [partitions []
+           pending []
+           records (patch/remove-tombstones records)]
+      (cond
+        ; Enough pending records to serialize a full partition.
+        (<= threshold (count pending))
+          (let [[output remnant] (split-at part-size pending)
+                part (from-records store params output)]
+            (recur (conj partitions part) (vec remnant) records))
+        ; Pull next record into pending.
+        (seq records)
+          (recur partitions (conj pending (first records)) (next records))
+        ; No more records, but too many pending to fit in one partition.
+        (< part-size (count pending))
+          (let [[output remnant] (split-at (int (/ (count pending) 2)) pending)
+                part (from-records store params output)]
+            (recur (conj partitions part) (vec remnant) nil))
+        ; Emit one final partition.
+        :else
+          (let [part (from-records store params pending)]
+            (conj partitions part))))))
 
 
 

         (mdag/store-node! store nil tablet))])))


(defn from-records
  "Constructs a new partition from the given map of record data. The records
  will be split into tablets matching the given families, if provided. Returns
  the node data for the persisted partition."
  [store params records]
  (let [records (vec (sort-by first (patch/remove-tombstones records))) ; TODO: don't sort?
        limit (max-limit params)
        families (or (::record/families params) {})]
    (when (< limit (count records))

                       (count records) limit)
               {::record/count (count records)
                ::limit limit})))
    (validate-keys! (map first records))
    (when (seq records)
      (->>
        {:data/type data-type
