package com.splicemachine.derby.stream.function;

import com.splicemachine.db.iapi.error.StandardException;
import com.splicemachine.db.iapi.sql.execute.ExecIndexRow;
import com.splicemachine.db.iapi.sql.execute.ExecRow;
import com.splicemachine.db.impl.sql.execute.IndexValueRow;
import com.splicemachine.derby.iapi.sql.execute.SpliceOperation;
import com.splicemachine.derby.impl.sql.execute.operations.LocatedRow;
import com.splicemachine.derby.impl.sql.execute.operations.ScalarAggregateOperation;
import com.splicemachine.derby.impl.sql.execute.operations.framework.SpliceGenericAggregator;
import com.splicemachine.derby.stream.iapi.OperationContext;
import com.splicemachine.derby.stream.spark.RDDUtils;

/**
 * Created by jleach on 5/1/15.
 */
public class ScalarAggregateFunction extends SpliceFunction2<SpliceOperation, LocatedRow, LocatedRow, LocatedRow> {
    private static final long serialVersionUID = -4150499166764796082L;
    protected boolean initialized;
    protected ScalarAggregateOperation op;
    public ScalarAggregateFunction() {
    }

    public ScalarAggregateFunction(OperationContext<SpliceOperation> operationContext) {
        super(operationContext);
    }

    @Override
    public LocatedRow call(LocatedRow t1, LocatedRow t2) throws Exception {
        if (!initialized) {
            op = (ScalarAggregateOperation) getOperation();
            initialized = true;
        }
        operationContext.recordRead();
        if (t2 == null) return t1;
        if (t1 == null) return t2;
        if (RDDUtils.LOG.isDebugEnabled())
            RDDUtils.LOG.debug(String.format("Reducing %s and %s", t1, t2));

        ExecRow r1 = t1.getRow();
        ExecRow r2 = t2.getRow();
        if (!(r1 instanceof ExecIndexRow)) {
            r1 = new IndexValueRow(r1.getClone());
        }
        if (!op.isInitialized(r1)) {
            op.initializeVectorAggregation(r1);
        }
        if (!op.isInitialized(r2)) {
            op.initializeVectorAggregation(r2);
        }
        aggregate(t2.getRow(), (ExecIndexRow) r1);
        return new LocatedRow(r1);
    }

    private void aggregate(ExecRow next, ExecIndexRow agg) throws StandardException {
        ScalarAggregateOperation op = (ScalarAggregateOperation) getOperation();
        if (RDDUtils.LOG.isDebugEnabled()) {
            RDDUtils.LOG.debug(String.format("Merging %s to %s", next, agg));
        }
        for (SpliceGenericAggregator aggregate : op.aggregates)
            aggregate.merge(next, agg);
    }

}


@@ -1958,41 +1958,49 @@
     
-op.initializeVectorAggregation(r2
+accumulate(t2.getRow(), (ExecIndexRow) r1
 );%0A 
@@ -1999,32 +1999,39 @@
 ) r1);%0A        %7D
+ else %7B
 %0A        aggrega
@@ -2023,24 +2023,24 @@
         
-aggregat
+    merg
 e(t2.get
@@ -2058,32 +2058,42 @@
 cIndexRow) r1);%0A
+        %7D%0A
         return n
@@ -2136,23 +2136,441 @@
 e void a
-ggregat
+ccumulate(ExecRow next, ExecIndexRow agg) throws StandardException %7B%0A        ScalarAggregateOperation op = (ScalarAggregateOperation) getOperation();%0A        if (RDDUtils.LOG.isDebugEnabled()) %7B%0A            RDDUtils.LOG.debug(String.format(%22Accumulating %25s to %25s%22, next, agg));%0A        %7D%0A        for (SpliceGenericAggregator aggregate : op.aggregates)%0A            aggregate.accumulate(next, agg);%0A    %7D%0A%0A    private void merg
 e(ExecRo

Optimize the accumulate and merge