             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)

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

