 
 ;;----------------------------------------
 
+(defn dfilter [#^Model src-model #^String tgt-view #^Collection attr-keys #^IFn filter-fn]
+  ;; TODO: accept ready-made View
+  (insert *metamodel* (do-filter tgt-view src-model attr-keys filter-fn)))
+
+(defn dtransform [#^Model src-model #^String suffix & #^Collection attribute-descrips]
+  ;; TOODO: accept ready-made View ?
+  (insert *metamodel* (apply do-transform suffix src-model attribute-descrips)))
+
+(defn dcount 
+  ([#^Model src-model key-value]
+   (let [key-type (type key-value)]
+     (dcount src-model key-type key-value)))
+  ([#^Model src-model #^Class key-type key-value]
+   (dcount src-model key-type key-value (count-reducer-metadata key-type)))
+  ([#^Model src-model #^Class key-type key-value #^Metadata metadata]
+   (insert *metamodel* (do-reduce-count src-model key-type key-value metadata)))
+  )
+
+(defn dsum
+  ([#^Model src-model #^Keyword attribute-key key-value]
+   (dsum src-model attribute-key (type key-value) key-value))
+  ([#^Model src-model #^Keyword attribute-key #^Class key-type key-value]
+   (insert *metamodel* (do-reduce-sum src-model attribute-key key-type key-value)))
+  )
+
+;; try to lose key-to-valua and vice-versa then define dpivot...
+(defn dsplit [#^Model src-model #^Keyword key #^boolean mutable #^IFn value-to-keys #^IFn key-to-value #^IFn tgt-hook]
+  (do-split
+   src-model
+   key
+   mutable
+   value-to-keys
+   key-to-value
+   (fn [#^Model model value]
+       (insert *metamodel* model)
+       (tgt-hook model value)
+       )
+   )
+  )
+
+;;----------------------------------------
+
 ;; demonstrate filtration - only selecting a subset of the values from
 ;; one model into another - based on their contents.
 
 ;;----------------------------------------
 
-(let [half-max-length (/ max-length 2)]
-  (def longer-whales
-       (do-filter (str "length>" half-max-length) whales '(:length) #(> % half-max-length))))
-
-(insert *metamodel* longer-whales)
-
-(let [half-max-weight (/ max-weight 2)]
-  (def heavier-whales
-       (do-filter (str "weight>" half-max-weight)  whales '(:weight) #(> % half-max-weight))))
-
-(insert *metamodel* heavier-whales)
+(def median-length (/ max-length 2))
+(def median-weight (/ max-weight 2))
+
+(def longer-whales 
+     (dfilter whales (str "length>" median-length) '(:length) #(> % median-length)))
+
+(def heavier-whales
+     (dfilter whales (str "weight>" median-weight) '(:weight) #(> % median-weight)))
 
 ;;----------------------------------------
 ;; demonstrate transformation - only selecting a subset of the

      :length (Float/TYPE)
      :weight (Float/TYPE)))

(def #^Class Whale (apply make-class "org.dada.demo.whales.Whale" Object attributes))
(def #^Metadata whale-metadata (apply metadata Whale :time :version attributes))
(def #^Model whales (model "Whales" whale-metadata))

(def num-whales 10000)
(def start-time (System/currentTimeMillis))

(insert *metamodel* whales)

;;----------------------------------------

(def narwhals-length
     (dtransform 
      (dfilter 
       whales
       "type=narwhal" 
       '(:type)
       #(= "narwhal" %))
      "length"
      :time
      :version
      :time
      :version
      :length)
     )

;;----------------------------------------
;; demonstrate reduction (sum)
;;----------------------------------------

(dcount whales "whales")
(dsum whales :length "whales")
(dsum whales :weight "whales")
(dcount heavier-whales-length "heavier-whales")
(dsum heavier-whales-length :length "heavier-whales")
(dsum longer-whales-weight :weight "longer-whales")
(dcount longer-whales-weight "longer-whales")

;;----------------------------------------
;; demonstrate splitting [and more reduction] - a pivot...
;;----------------------------------------

(let [pivot-metadata (class-metadata
		      "org.dada.demo.whales.TypeCount"
		      Object
		      :key
		      :version
		      [:key String :version Integer :count Integer])
      pivot-model (model "Whales.count(split(type))" pivot-metadata)
      ]
  (insert *metamodel* pivot-model)
  (dsplit
   whales
   :type
   false
   list
   identity
   (fn [#^Model model value]
       (let [tcm (dcount model String value pivot-metadata)]
	 (connect tcm pivot-model)
	 )
       (dsum model :length value)
       (dsum model :weight value)
       model ; TODO: do we really need to be able to override the model ? should threading go here ?
       )
   ))
