             a-denorms (->> isars
                            (s/buffer 5)
                            (s/map (fn [isar]
-                                    (rel/denormalize cassandra sa isar :upsert (ts/default-timestamp-opt))))
+                                    (rel/denormalize cassandra
+                                                     sa
+                                                     isar
+                                                     :upsert
+                                                     (ts/default-timestamp-opt
+                                                      {:fetch-size fetch-size}))))
                            (s/realize-each)
                            (s/reduce conj [])
                            deref)

            b-denorms (->> isbrs
                           (s/buffer 5)
                           (s/map (fn [isbr]
                                    (rel/denormalize cassandra sb isbr :upsert (ts/default-timestamp-opt))))
                           (s/realize-each)
                           (s/reduce conj [])
                           deref)

        sbrs (map (fn [sbid v] {:idb sbid :value v}) sbids (iterate inc 1))
        pstep (+ (quot bcnt acnt) (if (> (mod bcnt acnt) 0) 1 0))
        ;; bcnt records divided into acnt partitions
        sbrs-parts (partition pstep pstep nil sbrs)

        ;; make sure we page
        fetch-size (max 1 (quot bcnt 10))]

    (testing "write initial records"
      (let [trs (flatten
