Stateful ops

(defmacro statefulop
  "Lifts a function into a stateful operation by running
  `state-init` expression during operation initialization
  on the cluster, then passing its value as first
  argument of `op-fn`. The 3-arg artity allows you to specify
  the type of operation that `op-fn` reprsents. Can be one of
  mapop, mapcatop, filterop, bufferiterop."
  ([state-init op-fn] `(statefulop ~state-init identity ~op-fn))
  ([state-init op-lift op-fn]
   `(~op-lift
     (prepfn
      [x# y#]
      (let [state# ~state-init
            op# ~op-fn]
        (fn [& args#]
          (apply op# state# args#)))))))

(defmacro defstatefulfn
  "Defines `name-sym` as stateful operation.
  `state-bindings` should be a binding vector as for let, evaluated cluster
   side at initialization time. The bindigs are visible in the body of
   your operation. `op-lift` should be one of cascalog operation types,
   such as  mapop, mapcatop, filterop, or bufferiterop. For example:

  (defstatefulfn filter-words
  \"Filters words via externally loaded list\"
  [filter-words (load-filter-set)]
  (filterop [word] (not (contains? filter-words word))))"
  ([name-sym doc state-bindings [op-lift args & body]]
   `(def ^{:doc ~doc
           :arglists '(args)}
      ~name-sym (~op-lift
                 (prepfn
                  [x# y#]
                  (let ~state-bindings
                    (fn ~(vec args) ~@body)))))))