This is an automated email from the ASF dual-hosted git repository. tkalkirill pushed a commit to branch ignite-28924 in repository https://gitbox.apache.org/repos/asf/ignite.git
commit 5e19fe579747d6cca3203bce5e4cd8c8c8525c16 Author: Kirill Tkalenko <[email protected]> AuthorDate: Tue Jul 28 13:04:23 2026 +0300 IGNITE-28924 Wip --- .../query/calcite/exec/exp/agg/Accumulator.java | 5 ++ .../query/calcite/exec/exp/agg/Accumulators.java | 18 +++- .../OperatorsExtensionIntegrationTest.java | 100 +++++++++++++++++++++ 3 files changed, 119 insertions(+), 4 deletions(-) diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulator.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulator.java index 749db6a2226..fd8e25e825e 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulator.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulator.java @@ -41,4 +41,9 @@ public interface Accumulator<Row> extends Serializable { /** */ RelDataType returnType(IgniteTypeFactory typeFactory); + + /** Returns {@code true} if the accumulator handles DISTINCT. */ + default boolean handlesDistinct() { + return false; + } } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulators.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulators.java index 085d212be25..0fa998bedc4 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulators.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulators.java @@ -58,8 +58,13 @@ public class Accumulators { public static <Row> Supplier<Accumulator<Row>> accumulatorFactory(AggregateCall call, ExecutionContext<Row> ctx) { Supplier<Accumulator<Row>> supplier = accumulatorFunctionFactory(call, ctx); - if (call.isDistinct()) - return () -> new DistinctAccumulator<>(call, ctx.rowHandler(), supplier); + if (call.isDistinct()) { + return () -> { + Accumulator<Row> acc = supplier.get(); + + return acc.handlesDistinct() ? acc : new DistinctAccumulator<>(call, ctx.rowHandler(), acc); + }; + } return supplier; } @@ -1472,10 +1477,10 @@ public class Accumulators { private final List<Integer> args; /** */ - private DistinctAccumulator(AggregateCall aggCall, RowHandler<Row> hnd, Supplier<Accumulator<Row>> accSup) { + private DistinctAccumulator(AggregateCall aggCall, RowHandler<Row> hnd, Accumulator<Row> acc) { super(aggCall, hnd); - acc = accSup.get(); + this.acc = acc; args = super.arguments().isEmpty() ? List.of(0) : super.arguments(); } @@ -1519,5 +1524,10 @@ public class Accumulators { @Override public RelDataType returnType(IgniteTypeFactory typeFactory) { return acc.returnType(typeFactory); } + + /** {@inheritDoc} */ + @Override public boolean handlesDistinct() { + return true; + } } } diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/OperatorsExtensionIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/OperatorsExtensionIntegrationTest.java index d08259d0877..7a527fca074 100644 --- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/OperatorsExtensionIntegrationTest.java +++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/OperatorsExtensionIntegrationTest.java @@ -18,7 +18,9 @@ package org.apache.ignite.internal.processors.query.calcite.integration; import java.math.BigDecimal; import java.sql.Timestamp; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.function.Supplier; import com.google.common.collect.ImmutableList; import org.apache.calcite.adapter.enumerable.NullPolicy; @@ -40,6 +42,7 @@ import org.apache.calcite.sql.fun.SqlTrimFunction; import org.apache.calcite.sql.parser.SqlParserPos; import org.apache.calcite.sql.type.OperandTypes; import org.apache.calcite.sql.type.ReturnTypes; +import org.apache.calcite.sql.type.SqlTypeFamily; import org.apache.calcite.sql.type.SqlTypeName; import org.apache.calcite.sql.util.ReflectiveSqlOperatorTable; import org.apache.calcite.sql.util.SqlOperatorTables; @@ -158,6 +161,23 @@ public class OperatorsExtensionIntegrationTest extends AbstractBasicIntegrationT .check(); } + /** */ + @Test + public void testCustomAggregateHandlesDistinct() { + assertQuery("SELECT TEST_COUNT_PAIRS(DISTINCT x, y) " + + "FROM (VALUES (1, 10), (1, 20), (1, 20), (2, 10)) t(x, y)") + .returns(3L) + .check(); + } + + /** */ + @Test + public void testCustomAggregateUsesDefaultDistinctHandling() { + assertQuery("SELECT TEST_SUM(DISTINCT x) FROM (VALUES (1), (1), (2)) t(x)") + .returns(3L) + .check(); + } + /** Rewrites LTRIM with 2 parameters. */ public static SqlCall rewriteLtrim(SqlValidator validator, SqlCall call) { if (call.operandCount() != 2) @@ -220,6 +240,9 @@ public class OperatorsExtensionIntegrationTest extends AbstractBasicIntegrationT /** */ public static final SqlAggFunction TEST_SUM = new SqlTestSumAggFunction(); + + /** */ + public static final SqlAggFunction TEST_COUNT_PAIRS = new SqlTestCountPairsAggFunction(); } /** Extended convertlet table. */ @@ -264,6 +287,9 @@ public class OperatorsExtensionIntegrationTest extends AbstractBasicIntegrationT if (call.getAggregation().getName().equals(OperatorTable.TEST_SUM.getName())) return () -> new TestSum<>(call, ctx.rowHandler()); + if (call.getAggregation().getName().equals(OperatorTable.TEST_COUNT_PAIRS.getName())) + return () -> new TestCountPairs<>(call, ctx.rowHandler()); + return null; } } @@ -287,6 +313,25 @@ public class OperatorsExtensionIntegrationTest extends AbstractBasicIntegrationT } } + /** */ + public static class SqlTestCountPairsAggFunction extends SqlAggFunction { + /** */ + public SqlTestCountPairsAggFunction() { + super( + "TEST_COUNT_PAIRS", + null, + SqlKind.SUM, + opBinding -> opBinding.getTypeFactory().createSqlType(SqlTypeName.BIGINT), + null, + OperandTypes.family(SqlTypeFamily.NUMERIC, SqlTypeFamily.NUMERIC), + SqlFunctionCategory.NUMERIC, + false, + false, + Optionality.FORBIDDEN + ); + } + } + /** */ private static class TestSum<Row> extends Accumulators.AbstractAccumulator<Row> { /** */ @@ -325,4 +370,59 @@ public class OperatorsExtensionIntegrationTest extends AbstractBasicIntegrationT return typeFactory.createSqlType(org.apache.calcite.sql.type.SqlTypeName.BIGINT); } } + + /** */ + private static class TestCountPairs<Row> extends Accumulators.AbstractAccumulator<Row> { + /** */ + private long count; + + /** */ + private final Set<List<Object>> distinctPairs = new HashSet<>(); + + /** */ + protected TestCountPairs(AggregateCall aggCall, RowHandler<Row> hnd) { + super(aggCall, hnd); + } + + /** {@inheritDoc} */ + @Override public void add(Row row) { + if (aggregateCall().isDistinct()) + distinctPairs.add(List.of(get(0, row), get(1, row))); + else + count++; + } + + /** {@inheritDoc} */ + @Override public void apply(Accumulator<Row> other) { + TestCountPairs<Row> other0 = (TestCountPairs<Row>)other; + + if (aggregateCall().isDistinct()) + distinctPairs.addAll(other0.distinctPairs); + else + count += other0.count; + } + + /** {@inheritDoc} */ + @Override public Object end() { + return aggregateCall().isDistinct() ? (long)distinctPairs.size() : count; + } + + /** {@inheritDoc} */ + @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) { + RelDataType type = + typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.BIGINT), true); + + return List.of(type, type); + } + + /** {@inheritDoc} */ + @Override public RelDataType returnType(IgniteTypeFactory typeFactory) { + return typeFactory.createSqlType(SqlTypeName.BIGINT); + } + + /** {@inheritDoc} */ + @Override public boolean handlesDistinct() { + return true; + } + } }
