This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 2cb678f080d IGNITE-28924 SQL Calcite: Add the ability to implement a
custom DistinctAccumulator (#13405)
2cb678f080d is described below
commit 2cb678f080dcd0c1fa97353de48ebef95073ad19
Author: Kirill Tkalenko <[email protected]>
AuthorDate: Sat Aug 1 10:29:17 2026 +0300
IGNITE-28924 SQL Calcite: Add the ability to implement a custom
DistinctAccumulator (#13405)
---
.../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..4d4e402f1e5 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 cnt;
+
+ /** */
+ 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
+ cnt++;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator<Row> other) {
+ TestCountPairs<Row> other0 = (TestCountPairs<Row>)other;
+
+ if (aggregateCall().isDistinct())
+ distinctPairs.addAll(other0.distinctPairs);
+ else
+ cnt += other0.cnt;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return aggregateCall().isDistinct() ? (long)distinctPairs.size() :
cnt;
+ }
+
+ /** {@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;
+ }
+ }
}
