/*
 * Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
 * license agreements.  See the NOTICE file distributed with this work for
 * additional information regarding copyright ownership.  Crate licenses
 * this file to you under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.  You may
 * obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
 * License for the specific language governing permissions and limitations
 * under the License.
 *
 * However, if you have executed another commercial license agreement
 * with Crate these terms will supersede the license and you may use the
 * software solely pursuant to the terms of the relevant commercial agreement.
 */

package io.crate.execution.engine.aggregation.impl;

import com.google.common.annotations.VisibleForTesting;
import io.crate.breaker.RamAccountingContext;
import io.crate.data.Input;
import io.crate.execution.engine.aggregation.AggregationFunction;
import io.crate.metadata.FunctionIdent;
import io.crate.metadata.FunctionInfo;
import io.crate.types.DataType;
import io.crate.types.DataTypes;
import org.elasticsearch.Version;
import org.elasticsearch.common.breaker.CircuitBreakingException;
import org.elasticsearch.common.util.BigArrays;

import javax.annotation.Nullable;
import java.util.Collections;
import java.util.function.BinaryOperator;

public class SumAggregation<T extends Number> extends AggregationFunction<T, T> {

    public static final String NAME = "sum";

    public static void register(AggregationImplModule mod) {

        final BinaryOperator<Long> longAddImpl = (n1, n2) -> {
            long r = n1 + n2;
            // Overflow if both arguments have the opposite sign of the result
            if (((n1 ^ r) & (n2 ^ r)) < 0) {
                throw new ArithmeticException("long overflow");
            }
            return r;
        };

        final BinaryOperator<Long> longSubImpl = (n1, n2) -> n1 - n2;

        mod.register(new SumAggregation<Float>(DataTypes.FLOAT, (n1, n2) -> n1 + n2, (n1, n2) -> n1 - n2));
        mod.register(new SumAggregation<Double>(DataTypes.DOUBLE, (n1, n2) -> n1 + n2, (n1, n2) -> n1 - n2));
        mod.register(new SumAggregation<>(DataTypes.BYTE, DataTypes.LONG, longAddImpl, longSubImpl));
        mod.register(new SumAggregation<>(DataTypes.SHORT, DataTypes.LONG, longAddImpl, longSubImpl));
        mod.register(new SumAggregation<>(DataTypes.INTEGER, DataTypes.LONG, longAddImpl, longSubImpl));
        mod.register(new SumAggregation<>(DataTypes.LONG, longAddImpl, longSubImpl));
    }

    private final FunctionInfo info;
    private final BinaryOperator<T> addition;
    private final BinaryOperator<T> subtraction;
    private final DataType<T> returnType;
    private final int bytesSize;

    @VisibleForTesting
    private SumAggregation(final DataType returnType, final BinaryOperator<T> addition, final BinaryOperator<T> subtraction) {
        this(returnType, returnType, addition, subtraction);
    }

    private SumAggregation(final DataType inputType, final DataType returnType, final BinaryOperator<T> addition, final BinaryOperator<T> subtraction) {
        this.addition = addition;
        this.subtraction = subtraction;
        this.returnType = returnType;

        if (returnType == DataTypes.FLOAT) {
            bytesSize = DataTypes.FLOAT.fixedSize();
        } else if (returnType == DataTypes.DOUBLE) {
            bytesSize = DataTypes.DOUBLE.fixedSize();
        } else {
            bytesSize = DataTypes.LONG.fixedSize();
        }

        this.info = new FunctionInfo(
            new FunctionIdent(NAME, Collections.singletonList(inputType)),
            returnType,
            FunctionInfo.Type.AGGREGATE
        );
    }

    @Nullable
    @Override
    public T newState(RamAccountingContext ramAccountingContext, Version indexVersionCreated, BigArrays bigArrays) {
        ramAccountingContext.addBytes(bytesSize);
        return null;
    }

    @Override
    public T iterate(RamAccountingContext ramAccountingContext, T state, Input[] args) throws CircuitBreakingException {
        return reduce(ramAccountingContext, state, returnType.value(args[0].value()));
    }

    @Override
    public T reduce(RamAccountingContext ramAccountingContext, T state1, T state2) {
        if (state1 == null) {
            return state2;
        }
        if (state2 == null) {
            return state1;
        }
        return addition.apply(state1, state2);
    }

    @Override
    public T terminatePartial(RamAccountingContext ramAccountingContext, T state) {
        return state;
    }

    @Override
    public DataType partialType() {
        return info.returnType();
    }

    @Override
    public FunctionInfo info() {
        return info;
    }

    @Override
    public boolean isRemovableCumulative() {
        return true;
    }

    @Override
    public T removeFromAggregatedState(RamAccountingContext ramAccountingContext, T previousAggState, Input[] stateToRemove) {
        return subtraction.apply(previousAggState, returnType.value(stateToRemove[0].value()));
    }
}

@@ -1853,17 +1853,16 @@
  mod) %7B%0A
-%0A
         
@@ -1892,370 +1892,96 @@
 ng%3E 
-longAddImpl = (n1, n2) -%3E %7B%0A            long r = n1 + n2;%0A            // Overflow if both arguments have the opposite sign of the result%0A            if (((n1 %5E r) & (n2 %5E r)) %3C 0) %7B%0A                throw new ArithmeticException(%22long overflow%22);%0A            %7D%0A            return r;%0A        %7D;%0A%0A        final BinaryOperator%3CLong%3E longSubImpl = (n1, n2) -%3E n1 - n2
+add = Math::addExact;%0A        final BinaryOperator%3CLong%3E substract = Math::subtractExact
 ;%0A%0A 
@@ -2275,32 +2275,22 @@
 NG, 
-longAddImpl, longSubImpl
+add, substract
 ));%0A
@@ -2368,32 +2368,22 @@
 NG, 
-longAddImpl, longSubImpl
+add, substract
 ));%0A
@@ -2463,32 +2463,22 @@
 NG, 
-longAddImpl, longSubImpl
+add, substract
 ));%0A
@@ -2539,32 +2539,22 @@
 NG, 
-longAddImpl, longSubImpl
+add, substract
 ));%0A

Simplify code by using java.lang.Math's available methods