                 sinks (into-array
                        (for [[_ clause-type clause-id name] clauses
                              :when (not= clause-type "when")]
-                         (sink-of rule-id clause-type clause-id name var->ix)))]
-            (swap! rule->flow assoc agg-rule-id (AggregateFlow. index group-len limit-ix ascending? agg-ixes agg-funs sinks))
+                         (sink-of rule-id clause-type clause-id (str "delta-" name) var->ix)))]
+            (swap! rule->flow assoc agg-rule-id (AggregateFlow. index delta-index group-len limit-ix ascending? agg-ixes agg-funs sinks))
             (swap! kind->name->rules update-in ["know" (str "delta-" agg-index-id)] conj agg-rule-id)))))
 
     (doseq [[rule-id clauses] @rule-id->clauses]

                              (let [output (aget inputs i)]
                                (dotimes [j (alength aggs)]
                                  (apush output (aget aggs j)))
                                (apush facts&vals output)
                                (apush facts&vals 1)))
                            (aclear inputs)
                            (set!! current-key key)
                            (set!! current-limit (aget key limit-ix))

                              (for [i (range (alength out-vars))]
                                [(aget out-vars i) i]))
                index (.get-or-create-index kn "know" agg-index-id in-vars)
                delta-index (.get-or-create-index kn "know" (str "delta-" agg-index-id) in-vars)
                group-len (count (conj (set group-by-vars) limit))
                limit-ix (var->ix limit)
                ascending? (= ascending|descending "ascending")
