| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF 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. |
| */ |
| package org.apache.calcite.interpreter; |
| |
| import org.apache.calcite.DataContext; |
| import org.apache.calcite.adapter.enumerable.AggAddContext; |
| import org.apache.calcite.adapter.enumerable.AggImpState; |
| import org.apache.calcite.adapter.enumerable.JavaRowFormat; |
| import org.apache.calcite.adapter.enumerable.PhysType; |
| import org.apache.calcite.adapter.enumerable.PhysTypeImpl; |
| import org.apache.calcite.adapter.enumerable.RexToLixTranslator; |
| import org.apache.calcite.adapter.enumerable.impl.AggAddContextImpl; |
| import org.apache.calcite.adapter.java.JavaTypeFactory; |
| import org.apache.calcite.interpreter.Row.RowBuilder; |
| import org.apache.calcite.linq4j.tree.BlockBuilder; |
| import org.apache.calcite.linq4j.tree.Expression; |
| import org.apache.calcite.linq4j.tree.Expressions; |
| import org.apache.calcite.linq4j.tree.ParameterExpression; |
| import org.apache.calcite.rel.core.Aggregate; |
| import org.apache.calcite.rel.core.AggregateCall; |
| import org.apache.calcite.rel.type.RelDataTypeFactory; |
| import org.apache.calcite.rex.RexInputRef; |
| import org.apache.calcite.rex.RexNode; |
| import org.apache.calcite.runtime.FunctionContexts; |
| import org.apache.calcite.schema.FunctionContext; |
| import org.apache.calcite.schema.impl.AggregateFunctionImpl; |
| import org.apache.calcite.sql.SqlAggFunction; |
| import org.apache.calcite.sql.fun.SqlInternalOperators; |
| import org.apache.calcite.sql.fun.SqlStdOperatorTable; |
| import org.apache.calcite.sql.validate.SqlConformance; |
| import org.apache.calcite.sql.validate.SqlConformanceEnum; |
| import org.apache.calcite.util.ImmutableBitSet; |
| import org.apache.calcite.util.Util; |
| |
| import com.google.common.collect.ImmutableList; |
| |
| import org.checkerframework.checker.nullness.qual.Nullable; |
| |
| import java.lang.reflect.Constructor; |
| import java.lang.reflect.InvocationTargetException; |
| import java.math.BigDecimal; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.function.BiFunction; |
| import java.util.function.Supplier; |
| |
| import static org.apache.calcite.linq4j.Nullness.castNonNull; |
| |
| import static java.util.Objects.requireNonNull; |
| |
| /** |
| * Interpreter node that implements an |
| * {@link org.apache.calcite.rel.core.Aggregate}. |
| */ |
| public class AggregateNode extends AbstractSingleNode<Aggregate> { |
| private final List<Grouping> groups = new ArrayList<>(); |
| private final ImmutableBitSet unionGroups; |
| private final int outputRowLength; |
| private final ImmutableList<AccumulatorFactory> accumulatorFactories; |
| private final DataContext dataContext; |
| |
| public AggregateNode(Compiler compiler, Aggregate rel) { |
| super(compiler, rel); |
| this.dataContext = compiler.getDataContext(); |
| |
| ImmutableBitSet union = ImmutableBitSet.of(); |
| |
| for (ImmutableBitSet group : rel.getGroupSets()) { |
| union = union.union(group); |
| groups.add(new Grouping(group)); |
| } |
| |
| this.unionGroups = union; |
| this.outputRowLength = unionGroups.cardinality() |
| + rel.getAggCallList().size(); |
| |
| ImmutableList.Builder<AccumulatorFactory> builder = ImmutableList.builder(); |
| for (AggregateCall aggregateCall : rel.getAggCallList()) { |
| @SuppressWarnings("method.invocation.invalid") |
| AccumulatorFactory accumulator = |
| getAccumulator(compiler, aggregateCall, false); |
| builder.add(accumulator); |
| } |
| accumulatorFactories = builder.build(); |
| } |
| |
| @Override public void run() throws InterruptedException { |
| Row r; |
| while ((r = source.receive()) != null) { |
| for (Grouping group : groups) { |
| group.send(r); |
| } |
| } |
| |
| for (Grouping group : groups) { |
| group.end(sink); |
| } |
| } |
| |
| private AccumulatorFactory getAccumulator(Compiler compiler, |
| final AggregateCall call, boolean ignoreFilter) { |
| if (call.filterArg >= 0 && !ignoreFilter) { |
| final AccumulatorFactory factory = getAccumulator(compiler, call, true); |
| return () -> { |
| final Accumulator accumulator = factory.get(); |
| return new FilterAccumulator(accumulator, call.filterArg); |
| }; |
| } |
| final SqlAggFunction op = call.getAggregation(); |
| if (op == SqlStdOperatorTable.COUNT) { |
| return () -> new CountAccumulator(call); |
| } else if (op == SqlStdOperatorTable.SUM |
| || op == SqlStdOperatorTable.SUM0) { |
| final Class<?> clazz = sumClass(call); |
| return new UdaAccumulatorFactory(getAggFunction(clazz), call, |
| op == SqlStdOperatorTable.SUM, dataContext); |
| } else if (op == SqlStdOperatorTable.MAX |
| || op == SqlStdOperatorTable.MIN) { |
| final Class<?> clazz = maxMinClass(call); |
| return new UdaAccumulatorFactory(getAggFunction(clazz), call, true, |
| dataContext); |
| } else if (op == SqlInternalOperators.LITERAL_AGG) { |
| final Scalar scalar = compiler.compile(call.rexList, null); |
| final Object value = scalar.execute(compiler.createContext()); |
| return () -> new LiteralAccumulator(value); |
| } else { |
| final JavaTypeFactory typeFactory = |
| (JavaTypeFactory) rel.getCluster().getTypeFactory(); |
| int stateOffset = 0; |
| final AggImpState agg = new AggImpState(0, call, false); |
| int stateSize = requireNonNull(agg.state, "agg.state").size(); |
| |
| final BlockBuilder builder2 = new BlockBuilder(); |
| final PhysType inputPhysType = |
| PhysTypeImpl.of(typeFactory, rel.getInput().getRowType(), |
| JavaRowFormat.ARRAY); |
| final RelDataTypeFactory.Builder builder = typeFactory.builder(); |
| for (Expression expression : agg.state) { |
| builder.add("a", |
| typeFactory.createJavaType((Class) expression.getType())); |
| } |
| final PhysType accPhysType = |
| PhysTypeImpl.of(typeFactory, builder.build(), JavaRowFormat.ARRAY); |
| final ParameterExpression inParameter = |
| Expressions.parameter(inputPhysType.getJavaRowType(), "in"); |
| final ParameterExpression acc_ = |
| Expressions.parameter(accPhysType.getJavaRowType(), "acc"); |
| |
| List<Expression> accumulator = new ArrayList<>(stateSize); |
| for (int j = 0; j < stateSize; j++) { |
| accumulator.add(accPhysType.fieldReference(acc_, j + stateOffset)); |
| } |
| agg.state = accumulator; |
| |
| AggAddContext addContext = |
| new AggAddContextImpl(builder2, accumulator) { |
| @Override public List<RexNode> rexArguments() { |
| List<RexNode> args = new ArrayList<>(); |
| for (int index : agg.call.getArgList()) { |
| args.add(RexInputRef.of(index, inputPhysType.getRowType())); |
| } |
| return args; |
| } |
| |
| @Override public @Nullable RexNode rexFilterArgument() { |
| return agg.call.filterArg < 0 |
| ? null |
| : RexInputRef.of(agg.call.filterArg, |
| inputPhysType.getRowType()); |
| } |
| |
| @Override public RexToLixTranslator rowTranslator() { |
| final SqlConformance conformance = |
| SqlConformanceEnum.DEFAULT; // TODO: get this from implementor |
| return RexToLixTranslator.forAggregation(typeFactory, |
| currentBlock(), |
| new RexToLixTranslator.InputGetterImpl(inParameter, |
| inputPhysType), |
| conformance); |
| } |
| }; |
| |
| agg.implementor.implementAdd(requireNonNull(agg.context, "agg.context"), addContext); |
| |
| final ParameterExpression context_ = |
| Expressions.parameter(Context.class, "context"); |
| final ParameterExpression outputValues_ = |
| Expressions.parameter(Object[].class, "outputValues"); |
| final Scalar.Producer addScalarProducer = |
| JaninoRexCompiler.baz(context_, outputValues_, builder2.toBlock(), |
| ImmutableList.of()); |
| final Scalar initScalar = castNonNull(null); |
| final Scalar addScalar = addScalarProducer.apply(dataContext); |
| final Scalar endScalar = castNonNull(null); |
| return new ScalarAccumulatorDef(initScalar, addScalar, endScalar, |
| rel.getInput().getRowType().getFieldCount(), stateSize, dataContext); |
| } |
| } |
| |
| private static Class<?> maxMinClass(AggregateCall call) { |
| boolean max = call.getAggregation() == SqlStdOperatorTable.MAX; |
| switch (call.getType().getSqlTypeName()) { |
| case INTEGER: |
| return max ? MaxInt.class : MinInt.class; |
| case REAL: |
| return max ? MaxFloat.class : MinFloat.class; |
| case FLOAT: |
| case DOUBLE: |
| return max ? MaxDouble.class : MinDouble.class; |
| case DECIMAL: |
| return max ? MaxBigDecimal.class : MinBigDecimal.class; |
| case BOOLEAN: |
| return max ? MaxBoolean.class : MinBoolean.class; |
| default: |
| return max ? MaxLong.class : MinLong.class; |
| } |
| } |
| |
| private static Class<?> sumClass(AggregateCall call) { |
| switch (call.type.getSqlTypeName()) { |
| case DOUBLE: |
| case REAL: |
| case FLOAT: |
| return DoubleSum.class; |
| case DECIMAL: |
| return BigDecimalSum.class; |
| case INTEGER: |
| return IntSum.class; |
| case BIGINT: |
| default: |
| return LongSum.class; |
| } |
| } |
| |
| private static AggregateFunctionImpl getAggFunction(Class<?> clazz) { |
| return requireNonNull( |
| AggregateFunctionImpl.create(clazz), |
| () -> "Unable to create AggregateFunctionImpl for " + clazz); |
| } |
| |
| /** Accumulator for calls to the COUNT function. */ |
| private static class CountAccumulator implements Accumulator { |
| private final AggregateCall call; |
| long cnt; |
| |
| CountAccumulator(AggregateCall call) { |
| this.call = call; |
| cnt = 0; |
| } |
| |
| @Override public void send(Row row) { |
| boolean notNull = true; |
| for (Integer i : call.getArgList()) { |
| if (row.getObject(i) == null) { |
| notNull = false; |
| break; |
| } |
| } |
| if (notNull) { |
| cnt++; |
| } |
| } |
| |
| @Override public Object end() { |
| return cnt; |
| } |
| } |
| |
| /** Accumulator for calls to the LITERAL_AGG function. */ |
| private static class LiteralAccumulator implements Accumulator { |
| private final @Nullable Object value; |
| |
| LiteralAccumulator(@Nullable Object value) { |
| this.value = value; |
| } |
| |
| @Override public void send(Row row) { |
| } |
| |
| @Override public @Nullable Object end() { |
| return value; |
| } |
| } |
| |
| /** Creates an {@link Accumulator}. */ |
| private interface AccumulatorFactory extends Supplier<Accumulator> { |
| } |
| |
| /** Accumulator powered by {@link Scalar} code fragments. */ |
| private static class ScalarAccumulatorDef implements AccumulatorFactory { |
| final Scalar initScalar; |
| final Scalar addScalar; |
| final Scalar endScalar; |
| final Context sendContext; |
| final Context endContext; |
| final int rowLength; |
| final int accumulatorLength; |
| |
| private ScalarAccumulatorDef(Scalar initScalar, Scalar addScalar, |
| Scalar endScalar, int rowLength, int accumulatorLength, |
| DataContext root) { |
| this.initScalar = initScalar; |
| this.addScalar = addScalar; |
| this.endScalar = endScalar; |
| this.accumulatorLength = accumulatorLength; |
| this.rowLength = rowLength; |
| this.sendContext = new Context(root); |
| this.sendContext.values = new Object[rowLength + accumulatorLength]; |
| this.endContext = new Context(root); |
| this.endContext.values = new Object[accumulatorLength]; |
| } |
| |
| @Override public Accumulator get() { |
| return new ScalarAccumulator(this, new Object[accumulatorLength]); |
| } |
| } |
| |
| /** Accumulator powered by {@link Scalar} code fragments. */ |
| private static class ScalarAccumulator implements Accumulator { |
| final ScalarAccumulatorDef def; |
| final Object[] values; |
| |
| private ScalarAccumulator(ScalarAccumulatorDef def, Object[] values) { |
| this.def = def; |
| this.values = values; |
| } |
| |
| @Override public void send(Row row) { |
| @Nullable Object[] sendValues = |
| requireNonNull(def.sendContext.values, "def.sendContext.values"); |
| System.arraycopy(row.getValues(), 0, sendValues, 0, |
| def.rowLength); |
| System.arraycopy(this.values, 0, sendValues, def.rowLength, |
| this.values.length); |
| def.addScalar.execute(def.sendContext, this.values); |
| } |
| |
| @Override public @Nullable Object end() { |
| Context endContext = requireNonNull(def.endContext, "def.endContext"); |
| @Nullable Object[] values = requireNonNull(endContext.values, "endContext.values"); |
| System.arraycopy(this.values, 0, values, 0, this.values.length); |
| return def.endScalar.execute(endContext); |
| } |
| } |
| |
| /** |
| * Internal class to track groupings. |
| */ |
| private class Grouping { |
| private final ImmutableBitSet grouping; |
| private final Map<Row, AccumulatorList> accumulators = new HashMap<>(); |
| |
| private Grouping(ImmutableBitSet grouping) { |
| this.grouping = grouping; |
| } |
| |
| public void send(Row row) { |
| // TODO: fix the size of this row. |
| RowBuilder builder = Row.newBuilder(grouping.cardinality()); |
| int j = 0; |
| for (Integer i : grouping) { |
| builder.set(j++, row.getObject(i)); |
| } |
| Row key = builder.build(); |
| |
| if (!accumulators.containsKey(key)) { |
| AccumulatorList list = new AccumulatorList(); |
| for (AccumulatorFactory factory : accumulatorFactories) { |
| list.add(factory.get()); |
| } |
| accumulators.put(key, list); |
| } |
| |
| accumulators.get(key).send(row); |
| } |
| |
| public void end(Sink sink) throws InterruptedException { |
| for (Map.Entry<Row, AccumulatorList> e : accumulators.entrySet()) { |
| final Row key = e.getKey(); |
| final AccumulatorList list = e.getValue(); |
| |
| RowBuilder rb = Row.newBuilder(outputRowLength); |
| int index = 0; |
| for (Integer groupPos : unionGroups) { |
| if (grouping.get(groupPos)) { |
| rb.set(index, key.getObject(index)); |
| } |
| // need to set false when not part of grouping set. |
| |
| index++; |
| } |
| |
| list.end(rb); |
| |
| sink.send(rb.build()); |
| } |
| } |
| } |
| |
| /** |
| * A list of accumulators used during grouping. |
| */ |
| private static class AccumulatorList extends ArrayList<Accumulator> { |
| public void send(Row row) { |
| for (Accumulator a : this) { |
| a.send(row); |
| } |
| } |
| |
| public void end(RowBuilder r) { |
| for (int accIndex = 0, rowIndex = r.size() - size(); |
| rowIndex < r.size(); rowIndex++, accIndex++) { |
| r.set(rowIndex, get(accIndex).end()); |
| } |
| } |
| } |
| |
| /** |
| * Defines function implementation for |
| * things like {@code count()} and {@code sum()}. |
| */ |
| private interface Accumulator { |
| void send(Row row); |
| @Nullable Object end(); |
| } |
| |
| /** Implementation of {@code SUM} over INTEGER values as a user-defined |
| * aggregate. */ |
| public static class IntSum { |
| public IntSum() { |
| } |
| public int init() { |
| return 0; |
| } |
| public int add(int accumulator, int v) { |
| return accumulator + v; |
| } |
| public int merge(int accumulator0, int accumulator1) { |
| return accumulator0 + accumulator1; |
| } |
| public int result(int accumulator) { |
| return accumulator; |
| } |
| } |
| |
| /** Implementation of {@code SUM} over BIGINT values as a user-defined |
| * aggregate. */ |
| public static class LongSum { |
| public LongSum() { |
| } |
| public long init() { |
| return 0L; |
| } |
| public long add(long accumulator, long v) { |
| return accumulator + v; |
| } |
| public long merge(long accumulator0, long accumulator1) { |
| return accumulator0 + accumulator1; |
| } |
| public long result(long accumulator) { |
| return accumulator; |
| } |
| } |
| |
| /** Implementation of {@code SUM} over DOUBLE values as a user-defined |
| * aggregate. */ |
| public static class DoubleSum { |
| public DoubleSum() { |
| } |
| public double init() { |
| return 0D; |
| } |
| public double add(double accumulator, double v) { |
| return accumulator + v; |
| } |
| public double merge(double accumulator0, double accumulator1) { |
| return accumulator0 + accumulator1; |
| } |
| public double result(double accumulator) { |
| return accumulator; |
| } |
| } |
| |
| /** Implementation of {@code SUM} over BigDecimal values as a user-defined |
| * aggregate. */ |
| public static class BigDecimalSum { |
| public BigDecimalSum(){ |
| } |
| |
| public BigDecimal init() { |
| return new BigDecimal("0"); |
| } |
| |
| public BigDecimal add(BigDecimal accumulator, BigDecimal v) { |
| return accumulator.add(v); |
| } |
| |
| public BigDecimal merge(BigDecimal accumulator0, BigDecimal accumulator01) { |
| return add(accumulator0, accumulator01); |
| } |
| |
| public BigDecimal result(BigDecimal accumulator) { |
| return accumulator; |
| } |
| } |
| |
| /** Common implementation of comparison aggregate methods over numeric |
| * values as a user-defined aggregate. |
| * |
| * @param <T> The numeric type |
| */ |
| public static class NumericComparison<T> { |
| private final T initialValue; |
| private final BiFunction<T, T, T> comparisonFunction; |
| |
| public NumericComparison(T initialValue, BiFunction<T, T, T> comparisonFunction) { |
| this.initialValue = initialValue; |
| this.comparisonFunction = comparisonFunction; |
| } |
| |
| public T init() { |
| return this.initialValue; |
| } |
| |
| public T add(T accumulator, T value) { |
| return this.comparisonFunction.apply(accumulator, value); |
| } |
| |
| public T merge(T accumulator0, T accumulator1) { |
| return add(accumulator0, accumulator1); |
| } |
| |
| public T result(T accumulator) { |
| return accumulator; |
| } |
| } |
| |
| /** Implementation of {@code MIN} function to calculate the minimum of |
| * {@code integer} values as a user-defined aggregate. |
| */ |
| public static class MinInt extends NumericComparison<Integer> { |
| public MinInt() { |
| super(Integer.MAX_VALUE, Math::min); |
| } |
| } |
| |
| /** Implementation of {@code MIN} function to calculate the minimum of |
| * {@code long} values as a user-defined aggregate. |
| */ |
| public static class MinLong extends NumericComparison<Long> { |
| public MinLong() { |
| super(Long.MAX_VALUE, Math::min); |
| } |
| } |
| |
| /** Implementation of {@code MIN} function to calculate the minimum of |
| * {@code float} values as a user-defined aggregate. |
| */ |
| public static class MinFloat extends NumericComparison<Float> { |
| public MinFloat() { |
| super(Float.MAX_VALUE, Math::min); |
| } |
| } |
| |
| /** Implementation of {@code MIN} function to calculate the minimum of |
| * {@code double} and {@code real} values as a user-defined aggregate. |
| */ |
| public static class MinDouble extends NumericComparison<Double> { |
| public MinDouble() { |
| super(Double.MAX_VALUE, Math::min); |
| } |
| } |
| |
| /** Implementation of {@code MIN} function to calculate the minimum of |
| * {@code BigDecimal} values as a user-defined aggregate. |
| */ |
| public static class MinBigDecimal extends NumericComparison<BigDecimal> { |
| public MinBigDecimal() { |
| super(new BigDecimal(Double.MAX_VALUE), MinBigDecimal::min); |
| } |
| |
| public static BigDecimal min(BigDecimal a, BigDecimal b) { |
| return a.min(b); |
| } |
| } |
| |
| /** Implementation of {@code MIN} function to calculate the minimum of |
| * {@code boolean} values as a user-defined aggregate. |
| */ |
| public static class MinBoolean { |
| public MinBoolean() { } |
| |
| public Boolean init() { |
| return Boolean.TRUE; |
| } |
| |
| public Boolean add(Boolean accumulator, Boolean value) { |
| return accumulator && value; |
| } |
| |
| public Boolean merge(Boolean accumulator0, Boolean accumulator1) { |
| return add(accumulator0, accumulator1); |
| } |
| |
| public Boolean result(Boolean accumulator) { |
| return accumulator; |
| } |
| } |
| |
| /** Implementation of {@code MAX} function to calculate the maximum of |
| * {@code integer} values as a user-defined aggregate. |
| */ |
| public static class MaxInt extends NumericComparison<Integer> { |
| public MaxInt() { |
| super(Integer.MIN_VALUE, Math::max); |
| } |
| } |
| |
| /** Implementation of {@code MAX} function to calculate the maximum of |
| * {@code long} values as a user-defined aggregate. |
| */ |
| public static class MaxLong extends NumericComparison<Long> { |
| public MaxLong() { |
| super(Long.MIN_VALUE, Math::max); |
| } |
| } |
| |
| /** Implementation of {@code MAX} function to calculate the maximum of |
| * {@code float} values as a user-defined aggregate. |
| */ |
| public static class MaxFloat extends NumericComparison<Float> { |
| public MaxFloat() { |
| super(Float.MIN_VALUE, Math::max); |
| } |
| } |
| |
| /** Implementation of {@code MAX} function to calculate the maximum of |
| * {@code double} and {@code real} values as a user-defined aggregate. |
| */ |
| public static class MaxDouble extends NumericComparison<Double> { |
| public MaxDouble() { |
| super(Double.MIN_VALUE, Math::max); |
| } |
| } |
| |
| /** Implementation of {@code MAX} function to calculate the maximum of |
| * {@code BigDecimal} values as a user-defined aggregate. |
| */ |
| public static class MaxBigDecimal extends NumericComparison<BigDecimal> { |
| public MaxBigDecimal() { |
| super(new BigDecimal(Double.MIN_VALUE), MaxBigDecimal::max); |
| } |
| |
| public static BigDecimal max(BigDecimal a, BigDecimal b) { |
| return a.max(b); |
| } |
| } |
| |
| /** Implementation of {@code MAX} function to calculate the maximum of |
| * {@code boolean} values as a user-defined aggregate. |
| */ |
| public static class MaxBoolean { |
| public MaxBoolean() { } |
| |
| public Boolean init() { |
| return Boolean.FALSE; |
| } |
| |
| public Boolean add(Boolean accumulator, Boolean value) { |
| return accumulator || value; |
| } |
| |
| public Boolean merge(Boolean accumulator0, Boolean accumulator1) { |
| return add(accumulator0, accumulator1); |
| } |
| |
| public Boolean result(Boolean accumulator) { |
| return accumulator; |
| } |
| } |
| |
| /** Accumulator factory based on a user-defined aggregate function. */ |
| private static class UdaAccumulatorFactory implements AccumulatorFactory { |
| final AggregateFunctionImpl aggFunction; |
| final int argOrdinal; |
| public final @Nullable Object instance; |
| public final boolean nullIfEmpty; |
| |
| UdaAccumulatorFactory(AggregateFunctionImpl aggFunction, |
| AggregateCall call, boolean nullIfEmpty, DataContext dataContext) { |
| this.aggFunction = aggFunction; |
| if (call.getArgList().size() != 1) { |
| throw new UnsupportedOperationException("in current implementation, " |
| + "aggregate must have precisely one argument"); |
| } |
| argOrdinal = call.getArgList().get(0); |
| instance = createInstance(aggFunction, dataContext); |
| this.nullIfEmpty = nullIfEmpty; |
| } |
| |
| static @Nullable Object createInstance(AggregateFunctionImpl aggFunction, |
| DataContext dataContext) { |
| if (aggFunction.isStatic) { |
| return null; |
| } |
| try { |
| final Constructor<?> constructor = |
| aggFunction.declaringClass.getConstructor(); |
| return constructor.newInstance(); |
| } catch (InstantiationException | IllegalAccessException |
| | NoSuchMethodException | InvocationTargetException e) { |
| // ignore, and try next constructor |
| } |
| try { |
| final Constructor<?> constructor = |
| aggFunction.declaringClass.getConstructor(FunctionContext.class); |
| final Object[] args = new Object[aggFunction.getParameters().size()]; |
| final FunctionContext functionContext = |
| FunctionContexts.of(dataContext, args); |
| return constructor.newInstance(functionContext); |
| } catch (InstantiationException | IllegalAccessException |
| | NoSuchMethodException | InvocationTargetException e) { |
| throw Util.toUnchecked(e); |
| } |
| } |
| |
| @Override public Accumulator get() { |
| return new UdaAccumulator(this); |
| } |
| } |
| |
| /** Accumulator based upon a user-defined aggregate. */ |
| private static class UdaAccumulator implements Accumulator { |
| private final UdaAccumulatorFactory factory; |
| private @Nullable Object value; |
| private boolean empty; |
| |
| UdaAccumulator(UdaAccumulatorFactory factory) { |
| this.factory = factory; |
| try { |
| this.value = factory.aggFunction.initMethod.invoke(factory.instance); |
| } catch (IllegalAccessException | InvocationTargetException e) { |
| throw new RuntimeException(e); |
| } |
| this.empty = true; |
| } |
| |
| @Override public void send(Row row) { |
| final @Nullable Object[] args = {value, row.getValues()[factory.argOrdinal]}; |
| for (int i = 1; i < args.length; i++) { |
| if (args[i] == null) { |
| return; // one of the arguments is null; don't add to the total |
| } |
| } |
| try { |
| value = factory.aggFunction.addMethod.invoke(factory.instance, args); |
| } catch (IllegalAccessException | InvocationTargetException e) { |
| throw new RuntimeException(e); |
| } |
| empty = false; |
| } |
| |
| @Override public @Nullable Object end() { |
| if (factory.nullIfEmpty && empty) { |
| return null; |
| } |
| final @Nullable Object[] args = {value}; |
| try { |
| AggregateFunctionImpl aggFunction = |
| requireNonNull(factory.aggFunction, "factory.aggFunction"); |
| return requireNonNull(aggFunction.resultMethod, "aggFunction.resultMethod") |
| .invoke(factory.instance, args); |
| } catch (IllegalAccessException | InvocationTargetException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| } |
| |
| /** Accumulator that applies a filter to another accumulator. |
| * The filter is a BOOLEAN field in the input row. */ |
| private static class FilterAccumulator implements Accumulator { |
| private final Accumulator accumulator; |
| private final int filterArg; |
| |
| FilterAccumulator(Accumulator accumulator, int filterArg) { |
| this.accumulator = accumulator; |
| this.filterArg = filterArg; |
| } |
| |
| @Override public void send(Row row) { |
| if (row.getValues()[filterArg] == Boolean.TRUE) { |
| accumulator.send(row); |
| } |
| } |
| |
| @Override public @Nullable Object end() { |
| return accumulator.end(); |
| } |
| } |
| } |
