-- this is already defined via the install.sql script.
CREATE FUNCTION rupdate_float8(RAggregator, float8) RETURNS RAggregator AS '' LANGUAGE 'pl_R_aggregate';

CREATE AGGREGATE rmin (
 sfunc1 = rupdate_float8,
 basetype=float8,
 stype1 = RAggregator,
 finalfunc = pl_ragg_float8_result,                      
 initcond1 = '(function() {
   minVal <- Inf     
   getMinimum <- function() {
                    minVal
                 }

    update <- function(record) {
      minVal <<- min(record, minVal)
    }
                        
   return(list(rupdate = update, getMinimum = getMinimum))
 })()'
);

CREATE AGGREGATE rmedian (
 sfunc1 = rupdate,
 basetype=float8,
 stype1 = RAggregator,
 finalfunc = pl_ragg_float8_result,                      
 initcond1 = '(function(q = .5, M=10, shrink = 1) {
 
         Batch <- numeric(M)
         K <- ceiling(M/shrink)
         Dist <- numeric(K)
         Weight <- 0.
         Position <- 1
 
         medians <- seq(ceiling(shrink/2), by=shrink, len=K)
 
         merge <- function() {
            Batch <<- sort(Batch)
            Wnew <- Weight + shrink
            Dist <<- (Weight * Dist + shrink * Batch[medians])/Wnew
            Weight <<- Wnew
         }
 
        list(function(dataItem) {
                 Batch[[Position]] <<- dataItem
                 if(Position < M) {
                   Position <<- Position+1
                print(Position)
                 } else {
                print("Merging")
                    merge()
                    Position <<- 1
                 }
             },
               function() {
                  quantile(Batch, q)
               }
         )
})()');