 
 (defn create-table
   "Creates a table with options:
-    :name         - (required) table name.
-    :throughput   - (required) {:read <units> :write <units>}.
-    :hash-keydef  - (required) {:name _ :type <#{:s :n :ss :ns :b :bs}>}.
-    :range-keydef - (optional) {:name _ :type <#{:s :n :ss :ns :b :bs}>}.
-    :indexes      - (optional) [{:name _ :range-keydef _
-                                 :projection #{:all :keys-only [<attr> ...]}}].
-    :block?       - (optional) blocks for table to actually be active."
+    :name         - Table name.
+    :throughput   - {:read <units> :write <units>}.
+    :hash-keydef  - {:name _ :type <#{:s :n :ss :ns :b :bs}>}.
+    :range-keydef - {:name _ :type <#{:s :n :ss :ns :b :bs}>}.
+    :indexes      - [{:name _ :range-keydef _
+                      :projection #{:all :keys-only [<attr> ...]}}].
+    :block?       - Block for table to actually be active?"
   [creds {table-name :name
           :keys [throughput hash-keydef range-keydef indexes block?]
           :or   {throughput {:read 1 :write 1}}}]

                                                      :my-range ["0" "1"]}]
                                          :attrs []}}))

(defn- merge-unprocessed
  [more-f last-result]
  (loop [{:keys [items unprocessed consumed-capacity]} last-result]
    (if-not (seq unprocessed)
      (if items (with-meta items {:consumed-capacity consumed-capacity})
                last-result)

      (let [{items* :items unprocessed* :unprocessed
             consumed-capacity* :consumed-capacity}
            (more-f unprocessed)]

        (recur (merge {:unprocessed       unprocessed*
                       :consumed-capacity (when consumed-capacity
                                            (+ consumed-capacity
                                               consumed-capacity*))}
                      (when (or items items*)
                        {:items (into (or items []) items*)})))))))

(defn batch-get-item
  "Retrieves a batch of items in a single request.

  For automatic parallelization & segment control see `scan-parallel`.
  For indexed item retrievel see `query`."
  [creds table
   & [{:keys [attr-conds last-prim-kvs max-reqs return limit total-segments
              segment return-cc?] :as opts
       :or   {max-reqs 5}}]]
  (letfn [(run1 [last-prim-kvs]
            (as-map
             (.scan (db-client creds)
               (doto-maybe (ScanRequest.) g
                 :always (.setTableName (name table))
                 attr-conds      (.setScanFilter        (query|scan-conditions g))
                 last-prim-kvs   (.setExclusiveStartKey
                                  (clj-item->db-item last-prim-kvs))
                 limit           (.setLimit             (long g))
                 total-segments  (.setTotalSegments     (long g))
                 segment         (.setSegment           (long g))
                 (coll?* return) (.setAttributesToGet (mapv name return))
                 return-cc? (.setReturnConsumedCapacity (utils/enum :total))
                 (and return (not (coll?* return)))
                 (.setSelect (utils/enum return))))))]
    (merge-more run1 max-reqs (run1 last-prim-kvs))))

(defn scan-parallel
  "Like `scan` but starts a number of worker threads and automatically handles
