[WIP] Use inferred return type for all accumulator functions.

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/exp/agg/Accumulator.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.sql.engine.exec.exp.agg;
 
 import java.util.List;
+import org.apache.calcite.rel.core.AggregateCall;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
 
@@ -55,6 +56,8 @@ public interface Accumulator {
      *
      * @param typeFactory Type factory.
      * @return A result type.
+     * @deprecated Use {@link AggregateCall#getType()} instead.
      */
+    @Deprecated
     RelDataType returnType(IgniteTypeFactory typeFactory);
 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/exp/agg/Accumulators.java

@@ -35,6 +35,8 @@
 import org.apache.calcite.rel.core.AggregateCall;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.ignite.internal.catalog.commands.CatalogUtils;
+import org.apache.ignite.internal.sql.engine.exec.exp.IgniteSqlFunctions;
 import org.apache.ignite.internal.sql.engine.type.IgniteCustomType;
 import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
 import org.apache.ignite.internal.sql.engine.util.IgniteMath;
@@ -98,10 +100,10 @@ private Supplier<Accumulator> avgFactory(AggregateCall call) {
             case SMALLINT:
             case INTEGER:
             case BIGINT:
-                return () -> DecimalAvg.FACTORY.apply(0);
+                return () -> new DecimalAvg(call.type.getPrecision(), call.type.getScale());
             case DECIMAL:
                 // TODO: https://issues.apache.org/jira/browse/IGNITE-17373 Add support for interval types.
-                return () -> DecimalAvg.FACTORY.apply(call.type.getScale());
+                return () -> new DecimalAvg(call.type.getPrecision(), call.type.getScale());
             case DOUBLE:
             case REAL:
             case FLOAT:
@@ -117,8 +119,9 @@ private Supplier<Accumulator> avgFactory(AggregateCall call) {
     private Supplier<Accumulator> sumFactory(AggregateCall call) {
         switch (call.type.getSqlTypeName()) {
             case BIGINT:
+                return () -> new Sum(new LongSumEmptyIsZero());
             case DECIMAL:
-                return () -> new Sum(new DecimalSumEmptyIsZero());
+                return () -> new Sum(new DecimalSumEmptyIsZero(call.type.getPrecision(), call.type.getScale()));
 
             case DOUBLE:
             case REAL:
@@ -143,7 +146,7 @@ private Supplier<Accumulator> sumEmptyIsZeroFactory(AggregateCall call) {
                 // Used by REDUCE phase of COUNT aggregate.
                 return LongSumEmptyIsZero.FACTORY;
             case DECIMAL:
-                return DecimalSumEmptyIsZero.FACTORY;
+                return () -> new DecimalSumEmptyIsZero(call.type.getPrecision(), call.type.getScale());
 
             case DOUBLE:
             case REAL:
@@ -333,11 +336,17 @@ public static class DecimalAvgState {
 
         private final int scale;
 
+        @Deprecated
         DecimalAvg(int scale) {
             this.precision = RelDataType.PRECISION_NOT_SPECIFIED;
             this.scale = scale;
         }
 
+        public DecimalAvg(int precision, int scale) {
+            this.precision = precision;
+            this.scale = scale;
+        }
+
         /** {@inheritDoc} */
         @Override
         public void add(AccumulatorsState state, Object... args) {
@@ -431,7 +440,6 @@ public void end(AccumulatorsState state, AccumulatorsState result) {
                     result.set(null);
                 }
             }
-
         }
 
         /** {@inheritDoc} */
@@ -625,7 +633,22 @@ public RelDataType returnType(IgniteTypeFactory typeFactory) {
 
     /** SUM(DECIMAL) accumulator. */
     public static class DecimalSumEmptyIsZero implements Accumulator {
-        public static final Supplier<Accumulator> FACTORY = DecimalSumEmptyIsZero::new;
+        public static final IntFunction<Accumulator> FACTORY = DecimalSumEmptyIsZero::new;
+
+        private final int precision;
+
+        private final int scale;
+
+        @Deprecated
+        private DecimalSumEmptyIsZero(int scale) {
+            this.precision = CatalogUtils.MAX_DECIMAL_PRECISION;
+            this.scale = scale;
+        }
+
+        public DecimalSumEmptyIsZero(int precision, int scale) {
+            this.precision = precision;
+            this.scale = scale;
+        }
 
         /** {@inheritDoc} */
         @Override
@@ -650,7 +673,8 @@ public void end(AccumulatorsState state, AccumulatorsState result) {
             if (!state.hasValue()) {
                 result.set(BigDecimal.ZERO);
             } else {
-                result.set(state.get());
+                BigDecimal value = (BigDecimal) state.get();
+                result.set(IgniteSqlFunctions.toBigDecimal(value, precision, scale));
             }
         }
 
@@ -663,7 +687,7 @@ public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
         /** {@inheritDoc} */
         @Override
         public RelDataType returnType(IgniteTypeFactory typeFactory) {
-            return typeFactory.createTypeWithNullability(typeFactory.createSqlType(DECIMAL), false);
+            return typeFactory.createTypeWithNullability(typeFactory.createSqlType(DECIMAL, precision, scale), false);
         }
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/exp/agg/AccumulatorsFactory.java

@@ -195,7 +195,7 @@ private Accumulator accumulator() {
             Accumulator accumulator = accFactory.get();
 
             inAdapter = createInAdapter(accumulator);
-            outAdapter = createOutAdapter(accumulator);
+            outAdapter = Function.identity();
 
             return accumulator;
         }
@@ -235,11 +235,9 @@ private Function<Object, Object> createOutAdapter(Accumulator accumulator) {
             if (type == AggregateType.MAP) {
                 return Function.identity();
             }
-
-            RelDataType inType = accumulator.returnType(ctx.getTypeFactory());
             RelDataType outType = call.getType();
 
-            return cast(inType, outType);
+            return cast(outType, outType);
         }
 
         private RelDataType nonNull(RelDataType type) {

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/exp/agg/AccumulatorsState.java

@@ -17,7 +17,11 @@
 
 package org.apache.ignite.internal.sql.engine.exec.exp.agg;
 
+import java.util.Arrays;
 import java.util.BitSet;
+import java.util.List;
+import java.util.Objects;
+import org.apache.ignite.internal.tostring.S;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -36,6 +40,12 @@ public AccumulatorsState(int rowSize) {
         this.row = new Object[rowSize];
     }
 
+    /** Creates a copy from the given state. */
+    public AccumulatorsState(AccumulatorsState src) {
+        this.row = new Object[src.row.length];
+        System.arraycopy(src.row, 0, this.row, 0, src.row.length);
+    }
+
     /** Sets current field index. */
     public void setIndex(int i) {
         this.index = i;
@@ -61,4 +71,39 @@ public void set(@Nullable Object value) {
     public boolean hasValue() {
         return set.get(index);
     }
+
+    /** The number of elements. */
+    public int size() {
+        return row.length;
+    }
+
+    /** Elements of this state as list. */
+    public List<Object> toList() {
+        return Arrays.asList(row);
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        AccumulatorsState state = (AccumulatorsState) o;
+        return Objects.deepEquals(row, state.row);
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public int hashCode() {
+        return Arrays.hashCode(row);
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public String toString() {
+        return S.toString(this);
+    }
 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/PlanUtils.java

@@ -30,8 +30,6 @@
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.calcite.util.mapping.Mapping;
-import org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulator;
-import org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators;
 import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
 
 /**
@@ -80,7 +78,7 @@ public static RelDataType createSortAggRowType(ImmutableBitSet grpKeys,
             builder.add(fld);
         }
 
-        addAccumulatorFields(typeFactory, aggregateCalls, builder);
+        addAccumulatorFields(aggregateCalls, builder);
 
         return builder.build();
     }
@@ -115,32 +113,17 @@ public static RelDataType createHashAggRowType(List<ImmutableBitSet> groupSets,
             builder.add(fld);
         }
 
-        addAccumulatorFields(typeFactory, aggregateCalls, builder);
+        addAccumulatorFields(aggregateCalls, builder);
 
         builder.add("_GROUP_ID", SqlTypeName.TINYINT);
 
         return builder.build();
     }
 
-    private static void addAccumulatorFields(IgniteTypeFactory typeFactory, List<AggregateCall> aggregateCalls, Builder builder) {
-        Accumulators accumulators = new Accumulators(typeFactory);
-
+    private static void addAccumulatorFields(List<AggregateCall> aggregateCalls, Builder builder) {
         for (int i = 0; i < aggregateCalls.size(); i++) {
             AggregateCall call = aggregateCalls.get(i);
-            Accumulator acc = accumulators.accumulatorFactory(call).get();
-            RelDataType fieldType;
-            // For a decimal type Accumulator::returnType returns a type with default precision and scale,
-            // that can cause precision loss when a tuple is sent over the wire by an exchanger/outbox.
-            // Outbox uses its input type as wire format, so if a scale is 0, then the scale is lost
-            // (see Outbox::sendBatch -> RowHandler::toBinaryTuple -> BinaryTupleBuilder::appendDecimalNotNull).
-            if (call.getType().getSqlTypeName().allowsScale()) {
-                fieldType = call.type;
-            } else {
-                fieldType = acc.returnType(typeFactory);
-            }
-            String fieldName = "_ACC" + i;
-
-            builder.add(fieldName, fieldType);
+            builder.add("_ACC" + i, call.type);
         }
     }
 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/TypeUtils.java

@@ -341,6 +341,7 @@ public static ColumnType columnType(RelDataType type) {
                 return ColumnType.FLOAT;
             case BINARY:
             case VARBINARY:
+                return ColumnType.BYTE_ARRAY;
             case ANY:
                 if (type instanceof IgniteCustomType) {
                     IgniteCustomType customType = (IgniteCustomType) type;
@@ -652,14 +653,15 @@ public static RowSchema rowSchemaFromRelTypes(List<RelDataType> types) {
         RowSchema.Builder fieldTypes = RowSchema.builder();
 
         for (RelDataType relType : types) {
-            TypeSpec typeSpec = convertToTypeSpec(relType);
+            TypeSpec typeSpec = relational2rowSchemaType(relType);
             fieldTypes.addField(typeSpec);
         }
 
         return fieldTypes.build();
     }
 
-    private static TypeSpec convertToTypeSpec(RelDataType type) {
+    /** Converts the given relational data type to its {@link TypeSpec runtime type}. */
+    public static TypeSpec relational2rowSchemaType(RelDataType type) {
         boolean simpleType = type instanceof BasicSqlType;
         boolean nullable = type.isNullable();
 
@@ -689,7 +691,7 @@ private static TypeSpec convertToTypeSpec(RelDataType type) {
             List<TypeSpec> fields = new ArrayList<>();
 
             for (RelDataTypeField field : type.getFieldList()) {
-                TypeSpec fieldTypeSpec = convertToTypeSpec(field.getType());
+                TypeSpec fieldTypeSpec = relational2rowSchemaType(field.getType());
                 fields.add(fieldTypeSpec);
             }
 
@@ -811,6 +813,20 @@ public static boolean typesRepresentTheSameColumnTypes(RelDataType lhs, RelDataT
         }
     }
 
+    /**
+     * Converts relational type to a natively supported type if it is possible.
+     * This method returns {@code null} when the given relational type is {@link SqlTypeName#NULL}.
+     */
+    public static @Nullable NativeType relational2nativeType(RelDataType relDataType) {
+        TypeSpec typeSpec = relational2rowSchemaType(relDataType);
+        if (typeSpec == RowSchemaTypes.NULL) {
+            return null;
+        } else {
+            BaseTypeSpec baseTypeSpec = (BaseTypeSpec) typeSpec;
+            return baseTypeSpec.nativeType();
+        }
+    }
+
     private static boolean isCustomType(RelDataType type) {
         return type instanceof IgniteCustomType;
     }

modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/exp/agg/SumAccumulatorTest.java

@@ -54,8 +54,19 @@ private static Stream<Arguments> testArgs() {
         return Stream.of(
                 Arguments.of(namedAccumulator(DoubleSumEmptyIsZero.FACTORY), 4.0d, new Object[]{3.0d, 1.0d}),
                 Arguments.of(namedAccumulator(LongSumEmptyIsZero.FACTORY), 4L, new Object[]{3L, 1L}),
-                Arguments.of(namedAccumulator(DecimalSumEmptyIsZero.FACTORY), new BigDecimal("3.4"),
-                        new Object[]{new BigDecimal("1.3"), new BigDecimal("2.1")})
+
+                Arguments.of(namedAccumulator(() -> DecimalSumEmptyIsZero.FACTORY.apply(1)), new BigDecimal("3.4"),
+                        new Object[]{new BigDecimal("1.3"), new BigDecimal("2.1")}),
+
+                Arguments.of(namedAccumulator(() -> DecimalSumEmptyIsZero.FACTORY.apply(1)), new BigDecimal("3.4"),
+                        new Object[]{new BigDecimal("1.31"), new BigDecimal("2.13")}),
+                Arguments.of(namedAccumulator(() -> DecimalSumEmptyIsZero.FACTORY.apply(1)), new BigDecimal("3.5"),
+                        new Object[]{new BigDecimal("1.32"), new BigDecimal("2.13")}),
+
+                Arguments.of(namedAccumulator(() -> DecimalSumEmptyIsZero.FACTORY.apply(2)), new BigDecimal("3.44"),
+                        new Object[]{new BigDecimal("1.31"), new BigDecimal("2.13")}),
+                Arguments.of(namedAccumulator(() -> DecimalSumEmptyIsZero.FACTORY.apply(2)), new BigDecimal("3.45"),
+                        new Object[]{new BigDecimal("1.32"), new BigDecimal("2.13")})
         );
     }
 
@@ -71,7 +82,7 @@ private static Stream<Arguments> emptyArgs() {
         return Stream.of(
                 Arguments.of(namedAccumulator(DoubleSumEmptyIsZero.FACTORY)),
                 Arguments.of(namedAccumulator(LongSumEmptyIsZero.FACTORY)),
-                Arguments.of(namedAccumulator(DecimalSumEmptyIsZero.FACTORY))
+                Arguments.of(namedAccumulator(() -> DecimalSumEmptyIsZero.FACTORY.apply(1)))
         );
     }
 
@@ -81,6 +92,6 @@ private StatefulAccumulator newCall(Accumulator accumulator) {
 
     private static Named<Accumulator> namedAccumulator(Supplier<Accumulator> supplier) {
         Accumulator accumulator = supplier.get();
-        return Named.of(accumulator.getClass().getName(), accumulator);
+        return Named.of(accumulator.getClass().getSimpleName(), accumulator);
     }
 }