 ;;----------------------------------------
 
 (def narwhals-length
-     (do-transform 
-      "length"
-      (do-filter 
+     (dtransform 
+      (dfilter 
+       whales
        "type=narwhal" 
-       whales
        '(:type)
        #(= "narwhal" %))
+      "length"
       :time
       :version
       :time
       :version
       :length)
      )
-(insert *metamodel* narwhals-length)
 
 ;;----------------------------------------
 ;; demonstrate reduction (sum)
 ;;----------------------------------------
 
-(insert *metamodel* (do-reduce-count whales String "whales"))
-(insert *metamodel* (do-reduce-sum whales :length String "whales"))
-(insert *metamodel* (do-reduce-sum whales :weight String "whales"))
-(insert *metamodel* (do-reduce-count heavier-whales-length String "heavier-whales"))
-(insert *metamodel* (do-reduce-sum heavier-whales-length :length String "heavier-whales"))
-(insert *metamodel* (do-reduce-sum longer-whales-weight :weight String "longer-whales"))
-(insert *metamodel* (do-reduce-count longer-whales-weight String "longer-whales"))
-
-;;----------------------------------------
-;; demonstrate splitting [and more reduction)
-;;----------------------------------------
-
-(let [type-count-metadata (class-metadata
-       			   "org.dada.demo.whales.TypeCount"
-       			   Object
-       			   :key
-       			   :version
-       			   [:key String :version Integer :count Integer])
-      type-count-model (model "Whales.count(types)" type-count-metadata)
+(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* type-count-model)
-  (do-split
+  (insert *metamodel* pivot-model)
+  (dsplit
    whales
    :type
    false
    list
    identity
    (fn [#^Model model value]
-       (insert *metamodel* model)
-       (let [tcm (do-reduce-count model String value type-count-metadata)]
-	 (insert *metamodel* tcm)
-	 (connect tcm type-count-model)
+       (let [tcm (dcount model String value pivot-metadata)]
+	 (connect tcm pivot-model)
 	 )
-       (insert *metamodel* (do-reduce-sum model :length String value))
-       (insert *metamodel* (do-reduce-sum model :weight String value))
+       (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 ?
        )
    ))

;; attributes...
;;----------------------------------------

(def longer-whales-weight (do-transform "weight" longer-whales :time :version :time :version :weight))
(insert *metamodel* longer-whales-weight)

(def #^Model heavier-whales-length (do-transform "length" heavier-whales :time :version :time :version :length))
(insert *metamodel* heavier-whales-length)

;;----------------------------------------
;; demonstrate transformation - a synthetic field - metric tons per metre
;;----------------------------------------

(insert *metamodel* 
	(do-transform
	 "tonsPerMetre"
	 whales
	 :time
	 :version 
	 :time
	 :version 
	 (list :tonsPerMetre Number '(:weight :length) (fn [weight length] (if (= length 0) 0 (/ weight length)))))
	)

;;----------------------------------------
;; try a transformation on top of a filtration


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

(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.

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

(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
