This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 0ded66a27bf [feature](ivm) Support incremental refresh for array_agg
and collect_list aggregates (#67575)
0ded66a27bf is described below
commit 0ded66a27bf9112a006999420380e6076af5c1f2
Author: yujun <[email protected]>
AuthorDate: Thu Sep 10 14:52:20 2026 +0800
[feature](ivm) Support incremental refresh for array_agg and collect_list
aggregates (#67575)
MVs containing `ARRAY_AGG` or `COLLECT_LIST` can now be refreshed
incrementally (IVM): the visible array is the full aggregate state, so
no hidden state column is needed. The delta aggregate emits insert-side
and delete-side arrays and apply merges them as a multiset
`except_all(concat(coalesce(old, []), ins), del)`, which keeps duplicate
elements and NULLs correct.
`COLLECT_LIST` skips NULL rows, so its polarity columns use the
conditional-argument idiom. `ARRAY_AGG` keeps NULL elements, so
NULL-filtering cannot build its polarity arrays; instead every change
row is packed into a single `array_agg(struct(dml_factor, elem))`
aggregate output and the delta top project derives the insert/delete
polarity columns above the aggregate with `array_filter`/`array_map`,
keeping the same two transient polarity slots for the multiset apply.
`ARRAY_AGG` over JSONB/VARIANT elements (struct/map/array constructors
all reject those types) is rejected during analysis with a precise
reason, surfaced to the user at `CREATE MATERIALIZED VIEW` time.
DISTINCT and the LIMIT variant of `collect_list` fall back to complete
refresh.
## Key changes
- add ARRAY_AGG and COLLECT_LIST aggregate kinds sharing one
`IvmAggArrayProcessor` (two transient polarity delta slots, multiset
apply, no hidden state); ARRAY_AGG derives its polarity columns above
the delta aggregate through a new per-processor top-project hook
- centralize polarity condition expressions and the typed empty array
literal in `IvmAggExpressionBuilder`
- regression suite `test_ivm_agg_array_1` covering grouped, scalar and
mixed (array_agg + count + sum) MVs over insert/update/delete windows
with NULL and non-NULL values
Tracked in https://github.com/apache/doris/issues/65418
---
.../apache/doris/mtmv/ivm/IvmAggDeltaHandler.java | 11 +
.../mtmv/ivm/agg/IvmAggArrayAggProcessor.java | 148 ++++++++++
.../doris/mtmv/ivm/agg/IvmAggArrayProcessor.java | 97 +++++++
.../mtmv/ivm/agg/IvmAggCollectListProcessor.java | 66 +++++
.../mtmv/ivm/agg/IvmAggExpressionBuilder.java | 22 +-
.../doris/mtmv/ivm/agg/IvmAggFunctionKind.java | 4 +-
.../mtmv/ivm/agg/IvmAggFunctionProcessor.java | 13 +
.../doris/mtmv/ivm/agg/IvmAggFunctionRegistry.java | 13 +-
.../mtmv/ivm/agg/IvmAggArrayAggProcessorTest.java | 120 +++++++++
.../ivm/agg/IvmAggCollectListProcessorTest.java | 67 +++++
.../nereids/trees/plans/CreateMTMVCommandTest.java | 23 ++
.../data/mtmv_p0/ivm/test_ivm_agg_array_1.out | 199 ++++++++++++++
.../suites/mtmv_p0/ivm/test_ivm_agg_array_1.groovy | 299 +++++++++++++++++++++
13 files changed, 1078 insertions(+), 4 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandler.java
index a72e073413f..d12c94a15f6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandler.java
@@ -228,6 +228,17 @@ class IvmAggDeltaHandler {
IvmUtil.buildRowIdHash(deltaAgg.getOutput().subList(0,
groupKeySize)), Column.IVM_ROW_ID_COL);
topOutputs.add(rowIdAlias);
+ // Processors that pack every change row into one aggregate column
(ARRAY_AGG packs
+ // (dml_factor, elem) into a struct array) derive their polarity
columns here, above the
+ // aggregate: an aggregate output cannot reference a sibling aggregate
output, but a column
+ // of the top delta project can. Derived outputs keep the same
transient names the apply
+ // stage resolves below.
+ Map<String, Slot> deltaAggOutputByName =
indexSlotsByName(deltaAgg.getOutput());
+ for (IvmAggTarget target : aggMeta.getAggTargets()) {
+ aggFunctionRegistry.appendDeltaTopProjectOutputs(
+ target, deltaAggOutputByName, topOutputs,
aggExpressionBuilder);
+ }
+
Set<String> zeroDefaultDeltaOutputNames =
collectZeroDefaultDeltaOutputNames(aggMeta);
for (Slot slot : deltaAgg.getOutput()) {
if (zeroDefaultDeltaOutputNames.contains(slot.getName())) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayAggProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayAggProcessor.java
new file mode 100644
index 00000000000..9e362b8a5ce
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayAggProcessor.java
@@ -0,0 +1,148 @@
+// 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.doris.mtmv.ivm.agg;
+
+import org.apache.doris.mtmv.ivm.IvmException;
+import org.apache.doris.mtmv.ivm.IvmFailureReason;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ArrayItemReference;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
+import org.apache.doris.nereids.trees.expressions.LessThan;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.ArrayAgg;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayFilter;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayMap;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.CreateStruct;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ElementAt;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Lambda;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StructLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
+import org.apache.doris.nereids.types.DataType;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Processor for ARRAY_AGG(expr).
+ *
+ * <p>The visible array keeps NULL elements, so polarity columns cannot use
the NULL-filtering
+ * conditional-argument idiom (rows of the other polarity would leak into the
array as NULL
+ * elements). Instead every change row is packed into one struct field pair
and aggregated once:
+ *
+ * <pre>array_agg(struct(dml_factor, elem))</pre>
+ *
+ * <p>The insert/delete polarity columns are then derived above the delta
aggregate (see
+ * {@link #appendDeltaTopProjectOutputs}) by splitting the packed pair array
on the sign of the
+ * factor field and projecting the elem field, one scalar pass per side. Both
derived columns keep
+ * the transient names the apply stage resolves, so the multiset merge is
unchanged.
+ */
+class IvmAggArrayAggProcessor extends IvmAggArrayProcessor {
+ private static final String PAIRS_SLOT = "ARRAY_PAIRS";
+ /**
+ * Field name of {@code struct(dml_factor, elem)} carrying the dml factor.
{@link CreateStruct}
+ * names unnamed fields {@code StructLiteral.COL_PREFIX + (i + 1)}, so the
first field is "col1";
+ * keep the string composed from the shared prefix so a rename cannot
silently desync the
+ * {@code element_at(x, ...)} field lookups below.
+ */
+ private static final String FACTOR_FIELD = StructLiteral.COL_PREFIX + "1";
+ /**
+ * Field name of {@code struct(dml_factor, elem)} carrying the packed
element (second field).
+ */
+ private static final String ELEM_FIELD = StructLiteral.COL_PREFIX + "2";
+
+ @Override
+ public boolean supportsOriginalFunction(AggregateFunction function) {
+ if (!(function instanceof ArrayAgg)) {
+ return false;
+ }
+ // The packed delta aggregates struct(dml_factor, elem) per change
row, and struct fields
+ // cannot carry JSONB/VARIANT (CreateStruct rejects them). ARRAY_AGG
over such element types
+ // is not incrementally maintainable. Throw with the precise reason
instead of returning
+ // false: false would surface only the registry's generic "unsupported
aggregate for IVM:
+ // array_agg" error, which misleadingly blames the whole function
while ARRAY_AGG itself is
+ // supported. This check also runs while CREATE MATERIALIZED VIEW
analyzes the query, so the
+ // user sees the exact cause at create time.
+ DataType elemType = function.child(0).getDataType();
+ if (elemType.isJsonType() || elemType.isVariantType()) {
+ throw new IvmException(IvmFailureReason.AGG_UNSUPPORTED,
+ "IVM: ARRAY_AGG over JSONB/VARIANT element type " +
elemType.toSql()
+ + " is not incrementally maintainable (the delta
packs struct(dml_factor,"
+ + " elem) and struct fields cannot carry
JSONB/VARIANT), create the MV"
+ + " with a COMPLETE refresh instead: " +
function.toSql());
+ }
+ return true;
+ }
+
+ @Override
+ public IvmAggFunctionKind handledFunctionKind() {
+ return IvmAggFunctionKind.ARRAY_AGG;
+ }
+
+ @Override
+ void appendDeltaAggregateOutputs(IvmAggTarget target, Slot dmlFactorSlot,
+ List<NamedExpression> outputs, IvmAggExpressionBuilder ctx) {
+ Expression elem = target.getExprArgs().get(0);
+ outputs.add(new Alias(new ArrayAgg(new CreateStruct(dmlFactorSlot,
elem)),
+ ctx.transientDeltaColumnName(target, PAIRS_SLOT)));
+ }
+
+ @Override
+ void appendDeltaTopProjectOutputs(IvmAggTarget target, Map<String, Slot>
deltaAggOutputByName,
+ List<NamedExpression> topOutputs, IvmAggExpressionBuilder ctx) {
+ String pairsColumnName = ctx.transientDeltaColumnName(target,
PAIRS_SLOT);
+ Slot pairs = deltaAggOutputByName.get(pairsColumnName);
+ if (pairs == null) {
+ throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED,
+ "IVM agg delta rewrite failed to resolve packed array
pairs output: "
+ + pairsColumnName + " for target " + target);
+ }
+ topOutputs.add(new Alias(splitPolarityPairs(true, pairs),
+ ctx.transientDeltaColumnName(target, polaritySlotName(true))));
+ topOutputs.add(new Alias(splitPolarityPairs(false, pairs),
+ ctx.transientDeltaColumnName(target,
polaritySlotName(false))));
+ }
+
+ /**
+ * Splits the packed {@code array<struct<dml_factor, elem>>} into the
elements of one polarity
+ * side, mirroring a conditional aggregate without filtering NULL elements
first:
+ *
+ * <pre>array_map(x -> element_at(x, 'col2'),
+ * array_filter(x -> element_at(x, 'col1') > 0, pairs)) //
insert side
+ * array_filter(x -> element_at(x, 'col1') < 0, pairs)) //
delete side</pre>
+ *
+ * <p>The factor field of a change row is never NULL, so every pair is
classified exactly once
+ * and rows with factor zero are dropped on both sides, as before.
+ */
+ private Expression splitPolarityPairs(boolean insertSide, Slot pairs) {
+ ArrayItemReference pair = new ArrayItemReference("x", pairs);
+ Expression factor = new ElementAt(pair.toSlot(), new
StringLiteral(FACTOR_FIELD));
+ Expression condition = insertSide ? new GreaterThan(factor, new
TinyIntLiteral((byte) 0))
+ : new LessThan(factor, new TinyIntLiteral((byte) 0));
+ Expression filtered = new ArrayFilter(new Lambda(
+ ImmutableList.of("x"), condition, ImmutableList.of(pair)));
+ ArrayItemReference kept = new ArrayItemReference("y", filtered);
+ Expression value = new ElementAt(kept.toSlot(), new
StringLiteral(ELEM_FIELD));
+ return new ArrayMap(new Lambda(ImmutableList.of("y"), value,
ImmutableList.of(kept)));
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayProcessor.java
new file mode 100644
index 00000000000..269f298069d
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayProcessor.java
@@ -0,0 +1,97 @@
+// 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.doris.mtmv.ivm.agg;
+
+import org.apache.doris.mtmv.ivm.IvmException;
+import org.apache.doris.mtmv.ivm.IvmFailureReason;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayConcat;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayExceptAll;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Coalesce;
+
+import java.util.Map;
+
+/**
+ * Shared delta/apply logic for array-valued aggregate functions (ARRAY_AGG /
COLLECT_LIST).
+ *
+ * <p>The stored visible array is the full aggregate state, so no hidden state
column is needed: empty
+ * groups and fully-deleted groups are handled by the central group-count
machinery. Each subclass
+ * decides how the delta aggregate emits its polarity columns, one over insert
rows and one over delete
+ * rows: COLLECT_LIST emits two conditional aggregates (NULL-filtering is safe
because it skips NULL
+ * rows), while ARRAY_AGG keeps NULL elements and therefore packs every change
row into a single
+ * {@code struct(dml_factor, elem)} array aggregate, deriving the two polarity
columns above the delta
+ * aggregate with scalar array functions. Apply merges them into the new
visible array as a multiset:
+ *
+ * <pre>new = except_all(concat(coalesce(old, []), coalesce(ins, [])),
coalesce(del, []))</pre>
+ *
+ * <p>The merge is a multiset identity: element order is irrelevant and
duplicate elements cancel by
+ * occurrence, which keeps updates and repeated values correct. NULL handling
is delegated to each
+ * underlying aggregate (ARRAY_AGG keeps NULL elements; COLLECT_LIST skips
NULL rows).
+ */
+abstract class IvmAggArrayProcessor extends IvmAggFunctionProcessor {
+ private static final String INS_SLOT = "ARRAY_INS";
+ private static final String DEL_SLOT = "ARRAY_DEL";
+
+ /** Transient output-name suffix of the insert/delete polarity column for
one side. */
+ String polaritySlotName(boolean insertSide) {
+ return insertSide ? INS_SLOT : DEL_SLOT;
+ }
+
+ @Override
+ void mapApplyDeltaSlots(IvmAggTarget target, Map<String, Slot>
outputByName,
+ Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots, Slot
deltaGroupCountSlot,
+ IvmAggExpressionBuilder ctx) {
+ super.mapApplyDeltaSlots(target, outputByName, applyDeltaSlots,
deltaGroupCountSlot, ctx);
+ resolveDeltaSlot(target, INS_SLOT, outputByName, applyDeltaSlots, ctx);
+ resolveDeltaSlot(target, DEL_SLOT, outputByName, applyDeltaSlots, ctx);
+ }
+
+ @Override
+ public void appendApplyExpressions(IvmAggTarget target, IvmAggApplyContext
applyContext) {
+ IvmAggExpressionBuilder ctx = applyContext.expressions();
+ Slot oldArray =
applyContext.rawMvSlot(target.getVisibleSlot().getName());
+ Expression empty =
ctx.emptyArrayLiteral(target.getVisibleSlot().getDataType());
+ // The old MV side is genuinely NULL for groups that appear only in
the delta (new groups).
+ // The ins/del transient columns can never be NULL -- every delta
group holds at least one
+ // change row, and array_agg/collect_list yield an empty array (not
NULL) for zero kept
+ // elements -- so their COALESCE is defensive only.
+ Expression insArray = applyContext.deltaSlotValue(target,
deltaSlotRef(target, INS_SLOT));
+ Expression delArray = applyContext.deltaSlotValue(target,
deltaSlotRef(target, DEL_SLOT));
+ applyContext.putFinalExpression(target,
target.getVisibleSlot().getName(),
+ new ArrayExceptAll(
+ new ArrayConcat(new Coalesce(oldArray, empty), new
Coalesce(insArray, empty)),
+ new Coalesce(delArray, empty)));
+ }
+
+ private void resolveDeltaSlot(IvmAggTarget target, String slotName,
Map<String, Slot> outputByName,
+ Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots,
IvmAggExpressionBuilder ctx) {
+ String columnName = ctx.transientDeltaColumnName(target, slotName);
+ Slot slot = outputByName.get(columnName);
+ if (slot == null) {
+ throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED,
+ "IVM agg delta rewrite failed to resolve delta output
slot: "
+ + columnName + " for target " + target);
+ }
+ applyDeltaSlots.put(deltaSlotRef(target, slotName), slot);
+ }
+
+ private IvmAggDeltaSlotRef deltaSlotRef(IvmAggTarget target, String
slotName) {
+ return new IvmAggDeltaSlotRef(target.getOrdinal(), slotName);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggCollectListProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggCollectListProcessor.java
new file mode 100644
index 00000000000..fe618f89e1d
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggCollectListProcessor.java
@@ -0,0 +1,66 @@
+// 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.doris.mtmv.ivm.agg;
+
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CollectList;
+
+import java.util.List;
+
+/**
+ * Processor for COLLECT_LIST(expr) with exactly one argument.
+ *
+ * <p>COLLECT_LIST skips NULL rows, so the polarity columns use the plain
conditional-argument idiom:
+ * rows of the other polarity become NULL and are skipped by the aggregate
itself.
+ *
+ * <p>The two-argument LIMIT variant is not incrementally maintainable and is
intentionally not
+ * supported (falls back to complete refresh).
+ */
+class IvmAggCollectListProcessor extends IvmAggArrayProcessor {
+ @Override
+ public boolean supportsOriginalFunction(AggregateFunction function) {
+ return function instanceof CollectList && function.children().size()
== 1;
+ }
+
+ @Override
+ public IvmAggFunctionKind handledFunctionKind() {
+ return IvmAggFunctionKind.COLLECT_LIST;
+ }
+
+ @Override
+ void appendDeltaAggregateOutputs(IvmAggTarget target, Slot dmlFactorSlot,
+ List<NamedExpression> outputs, IvmAggExpressionBuilder ctx) {
+ Expression arg = target.getExprArgs().get(0);
+ outputs.add(new Alias(buildPolarityAggregate(true, arg, dmlFactorSlot,
ctx),
+ ctx.transientDeltaColumnName(target, polaritySlotName(true))));
+ outputs.add(new Alias(buildPolarityAggregate(false, arg,
dmlFactorSlot, ctx),
+ ctx.transientDeltaColumnName(target,
polaritySlotName(false))));
+ }
+
+ private AggregateFunction buildPolarityAggregate(boolean insertSide,
Expression elem, Slot dmlFactorSlot,
+ IvmAggExpressionBuilder ctx) {
+ Expression filtered = insertSide
+ ? ctx.insertOnlyValue(elem, dmlFactorSlot)
+ : ctx.deleteOnlyValue(elem, dmlFactorSlot);
+ return new CollectList(filtered);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExpressionBuilder.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExpressionBuilder.java
index 82cc99e06df..24990b7a72b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExpressionBuilder.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExpressionBuilder.java
@@ -31,6 +31,7 @@ import org.apache.doris.nereids.trees.expressions.Subtract;
import org.apache.doris.nereids.trees.expressions.functions.scalar.AssertTrue;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Coalesce;
import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral;
import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
@@ -38,6 +39,8 @@ import
org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.types.DataType;
+import com.google.common.collect.ImmutableList;
+
/**
* Stateless expression builder shared by aggregate processors.
*
@@ -52,8 +55,18 @@ public class IvmAggExpressionBuilder {
/** Returns {@code expr} for inserts and {@code -expr} for deletes. */
Expression signedDeltaValue(Expression expr, Slot dmlFactorSlot) {
- return new If(new GreaterThan(dmlFactorSlot, new TinyIntLiteral((byte)
0)),
- expr, new Subtract(zeroOf(expr.getDataType()), expr));
+ return new If(factorPositive(dmlFactorSlot), expr,
+ new Subtract(zeroOf(expr.getDataType()), expr));
+ }
+
+ /** Returns {@code dmlFactorSlot > 0}; selects insert rows of the delta
stream. */
+ Expression factorPositive(Slot dmlFactorSlot) {
+ return new GreaterThan(dmlFactorSlot, new TinyIntLiteral((byte) 0));
+ }
+
+ /** Returns {@code dmlFactorSlot < 0}; selects delete rows of the delta
stream. */
+ Expression factorNegative(Slot dmlFactorSlot) {
+ return new LessThan(dmlFactorSlot, new TinyIntLiteral((byte) 0));
}
/** Returns {@code dml_factor} for non-NULL input values and zero for NULL
input values. */
@@ -99,6 +112,11 @@ public class IvmAggExpressionBuilder {
return new TinyIntLiteral((byte) 0).checkedCastTo(dataType);
}
+ /** Builds a typed empty array literal used to merge possibly-NULL array
sides. */
+ public Expression emptyArrayLiteral(DataType arrayDataType) {
+ return new ArrayLiteral(ImmutableList.of(), arrayDataType);
+ }
+
/** Keeps an expression only for inserted rows; deleted rows become NULL
and are ignored by MIN/MAX. */
Expression insertOnlyValue(Expression expr, Slot dmlFactorSlot) {
return new If(new GreaterThan(dmlFactorSlot, new TinyIntLiteral((byte)
0)),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionKind.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionKind.java
index 42324242bb1..b7f216b2a32 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionKind.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionKind.java
@@ -30,5 +30,7 @@ public enum IvmAggFunctionKind {
MIN,
MAX,
BITMAP_UNION,
- BITMAP_UNION_COUNT
+ BITMAP_UNION_COUNT,
+ ARRAY_AGG,
+ COLLECT_LIST
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java
index e5bd1336ff6..8042b852286 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java
@@ -86,6 +86,19 @@ public abstract class IvmAggFunctionProcessor {
abstract void appendDeltaAggregateOutputs(IvmAggTarget target, Slot
dmlFactorSlot, List<NamedExpression> outputs,
IvmAggExpressionBuilder ctx);
+ /**
+ * Appends processor-owned derived (non-aggregate) delta columns on top of
the delta aggregate.
+ *
+ * <p>Some processors pack every change row into one aggregate column and
then derive the polarity
+ * columns (for example ARRAY_AGG's insert/delete views) with scalar
functions over that packed
+ * aggregate output. Such derived columns cannot be aggregate outputs of
the same node, so this
+ * hook appends them to the delta top project's output list, where {@code
deltaAggOutputByName}
+ * exposes the delta aggregate's own output slots.
+ */
+ void appendDeltaTopProjectOutputs(IvmAggTarget target, Map<String, Slot>
deltaAggOutputByName,
+ List<NamedExpression> topOutputs, IvmAggExpressionBuilder ctx) {
+ }
+
/** Resolves delta output slots and registers the slots that apply
expressions will read later. */
void mapApplyDeltaSlots(IvmAggTarget target, Map<String, Slot>
outputByName,
Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots, Slot
deltaGroupCountSlot, IvmAggExpressionBuilder ctx) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionRegistry.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionRegistry.java
index d7a4b08d09f..f10a23c934f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionRegistry.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionRegistry.java
@@ -64,7 +64,9 @@ public class IvmAggFunctionRegistry {
new IvmAggMinProcessor(),
new IvmAggMaxProcessor(),
new IvmAggBitmapUnionProcessor(),
- new IvmAggBitmapUnionCountProcessor());
+ new IvmAggBitmapUnionCountProcessor(),
+ new IvmAggArrayAggProcessor(),
+ new IvmAggCollectListProcessor());
processorByKind = new EnumMap<>(IvmAggFunctionKind.class);
for (IvmAggFunctionProcessor processor : processors) {
processorByKind.put(processor.handledFunctionKind(), processor);
@@ -97,6 +99,15 @@ public class IvmAggFunctionRegistry {
}
}
+ /**
+ * Appends processor-owned derived (non-aggregate) columns to the delta
top project, computed
+ * over the delta aggregate's own output slots.
+ */
+ public void appendDeltaTopProjectOutputs(IvmAggTarget target, Map<String,
Slot> deltaAggOutputByName,
+ List<NamedExpression> topOutputs, IvmAggExpressionBuilder ctx) {
+ processorFor(target).appendDeltaTopProjectOutputs(target,
deltaAggOutputByName, topOutputs, ctx);
+ }
+
/**
* Maps the delta plan output slots to stable logical keys consumed by the
apply expressions.
*/
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayAggProcessorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayAggProcessorTest.java
new file mode 100644
index 00000000000..28ba8142abf
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggArrayAggProcessorTest.java
@@ -0,0 +1,120 @@
+// 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.doris.mtmv.ivm.agg;
+
+import org.apache.doris.mtmv.ivm.IvmException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.ArrayAgg;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayConcat;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayExceptAll;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayFilter;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayMap;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Coalesce;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.CreateStruct;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ElementAt;
+import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.types.JsonType;
+import org.apache.doris.nereids.types.VariantType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class IvmAggArrayAggProcessorTest extends IvmAggProcessorTestBase {
+ @Test
+ void testArrayAggPacksStructPairsAndSplitsPolarityViews() {
+ IvmAggArrayAggProcessor processor = new IvmAggArrayAggProcessor();
+ Assertions.assertTrue(processor.supportsOriginalFunction(new
ArrayAgg(value)));
+ Assertions.assertEquals(IvmAggFunctionKind.ARRAY_AGG,
processor.handledFunctionKind());
+ Assertions.assertTrue(processor.hiddenStateKeys(new
ArrayAgg(value)).isEmpty());
+
+ IvmAggTarget target = target(0, IvmAggFunctionKind.ARRAY_AGG, "arr",
ArrayType.of(IntegerType.INSTANCE),
+ ImmutableMap.of(), valueArg());
+
+ // The delta aggregate packs every change row into one struct array;
NULL elements stay packed
+ // inside the structs, so no conditional (NULL-filtering) aggregate is
involved.
+ List<NamedExpression> aggOutputs = deltaOutputs(processor, target);
+ Assertions.assertEquals(1, aggOutputs.size());
+ Assertions.assertTrue(aggOutputs.get(0).child(0) instanceof ArrayAgg);
+ Assertions.assertTrue(aggOutputs.get(0).child(0).anyMatch(node -> node
instanceof CreateStruct));
+ Assertions.assertFalse(aggOutputs.get(0).anyMatch(node -> node
instanceof ElementAt));
+
+ // The insert/delete polarity columns are derived above the aggregate,
mirroring
+ // IvmAggDeltaHandler.buildDeltaSubPlan: the handler builds the top
project from the delta
+ // aggregate's output slots, then asks every processor to append its
derived outputs -- so
+ // here the agg outputs plus the derived ones must be resolvable by
name exactly like the
+ // top project's columns, each polarity view a map over a filter on
the factor field.
+ Map<String, Slot> aggOutputByName = new HashMap<>();
+ for (NamedExpression output : aggOutputs) {
+ aggOutputByName.put(output.getName(), output.toSlot());
+ }
+ List<NamedExpression> topOutputs = new ArrayList<>(aggOutputs);
+ processor.appendDeltaTopProjectOutputs(target, aggOutputByName,
topOutputs,
+ IvmAggExpressionBuilder.INSTANCE);
+ Assertions.assertEquals(3, topOutputs.size());
+ NamedExpression ins = topOutputs.get(1);
+ NamedExpression del = topOutputs.get(2);
+ Assertions.assertTrue(ins.getName().contains("ARRAY_INS"));
+ Assertions.assertTrue(del.getName().contains("ARRAY_DEL"));
+ Assertions.assertNotEquals(ins.getName(), del.getName());
+ for (NamedExpression polarity : ImmutableList.of(ins, del)) {
+ Assertions.assertTrue(polarity.child(0) instanceof ArrayMap);
+ Assertions.assertTrue(polarity.child(0).anyMatch(node -> node
instanceof ArrayFilter));
+ Assertions.assertTrue(polarity.child(0).anyMatch(node -> node
instanceof ElementAt));
+ }
+
+ Map<String, Expression> finalByName = apply(processor, target,
+ ImmutableList.of(slot("arr",
ArrayType.of(IntegerType.INSTANCE))),
+ mappedDeltaSlots(processor, target, topOutputs),
+ slot("delta_group_count", IntegerType.INSTANCE));
+ Expression visible = finalByName.get("arr");
+ Assertions.assertNotNull(visible);
+ Assertions.assertTrue(visible instanceof ArrayExceptAll);
+ Assertions.assertTrue(visible.anyMatch(node -> node instanceof
ArrayConcat));
+ // old, ins and del sides each merge NULL into an empty array
+ Assertions.assertEquals(3, visible.collect(node -> node instanceof
Coalesce).size());
+ }
+
+ @Test
+ void testArrayAggRejectsVariantAndJsonbElementsWithPreciseReason() {
+ IvmAggArrayAggProcessor processor = new IvmAggArrayAggProcessor();
+ Assertions.assertTrue(processor.supportsOriginalFunction(new
ArrayAgg(value)));
+ // struct(dml_factor, elem) cannot carry JSONB/VARIANT fields, so
ARRAY_AGG over those
+ // element types must be rejected with the precise reason (and that
reason must reach the
+ // user when CREATE MATERIALIZED VIEW analyzes the query) instead of
returning false, which
+ // would only produce the registry's generic "unsupported aggregate"
error.
+ for (Slot variantSlot : ImmutableList.of(
+ new SlotReference("jv", VariantType.INSTANCE, true),
+ new SlotReference("jb", JsonType.INSTANCE, true))) {
+ IvmException ex = Assertions.assertThrows(IvmException.class,
+ () -> processor.supportsOriginalFunction(new
ArrayAgg(variantSlot)));
+ Assertions.assertTrue(ex.getMessage().contains("not incrementally
maintainable"),
+ "unexpected message: " + ex.getMessage());
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggCollectListProcessorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggCollectListProcessorTest.java
new file mode 100644
index 00000000000..5f31ea814f6
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggCollectListProcessorTest.java
@@ -0,0 +1,67 @@
+// 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.doris.mtmv.ivm.agg;
+
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CollectList;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayConcat;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayExceptAll;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.IntegerType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+class IvmAggCollectListProcessorTest extends IvmAggProcessorTestBase {
+ @Test
+ void testCollectListUsesNullFilteringIdiomAndMultisetApply() {
+ IvmAggCollectListProcessor processor = new
IvmAggCollectListProcessor();
+ Assertions.assertTrue(processor.supportsOriginalFunction(new
CollectList(value)));
+ // two-argument LIMIT variant is not incrementally maintainable
+ Assertions.assertFalse(processor.supportsOriginalFunction(
+ new CollectList(value, new IntegerLiteral(10))));
+ Assertions.assertEquals(IvmAggFunctionKind.COLLECT_LIST,
processor.handledFunctionKind());
+ Assertions.assertTrue(processor.hiddenStateKeys(new
CollectList(value)).isEmpty());
+
+ IvmAggTarget target = target(0, IvmAggFunctionKind.COLLECT_LIST, "arr",
+ ArrayType.of(IntegerType.INSTANCE), ImmutableMap.of(),
valueArg());
+ List<NamedExpression> outputs = deltaOutputs(processor, target);
+ Assertions.assertEquals(2, outputs.size());
+ Assertions.assertTrue(outputs.get(0).child(0) instanceof CollectList);
+ Assertions.assertTrue(outputs.get(1).child(0) instanceof CollectList);
+ Assertions.assertTrue(outputs.get(0).child(0).anyMatch(node -> node
instanceof If));
+ Assertions.assertTrue(outputs.get(1).child(0).anyMatch(node -> node
instanceof If));
+
+ Map<String, Expression> finalByName = apply(processor, target,
+ ImmutableList.of(slot("arr",
ArrayType.of(IntegerType.INSTANCE))),
+ mappedDeltaSlots(processor, target, outputs),
+ slot("delta_group_count", IntegerType.INSTANCE));
+ Expression visible = finalByName.get("arr");
+ Assertions.assertNotNull(visible);
+ Assertions.assertTrue(visible instanceof ArrayExceptAll);
+ Assertions.assertTrue(visible.anyMatch(node -> node instanceof
ArrayConcat));
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/CreateMTMVCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/CreateMTMVCommandTest.java
index 112b4aafe71..33c29674863 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/CreateMTMVCommandTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/CreateMTMVCommandTest.java
@@ -369,6 +369,29 @@ public class CreateMTMVCommandTest extends
TestWithFeService {
assertCreateMtmvFails(explicitColumnMv, "column name can't start with
'__DORIS_'");
}
+ @Test
+ public void testCreateIvmMvRejectsArrayAggOverUnsupportedElementTypes()
throws Exception {
+ // ARRAY_AGG incremental maintenance packs struct(dml_factor, elem)
per change row, and
+ // struct fields cannot carry JSONB/VARIANT (CreateStruct rejects
both, as does CreateMap,
+ // and the array constructor rejects them too). CREATE MATERIALIZED
VIEW analyzes the
+ // query with the IVM normalize rewrite, so the unsupported element
type must surface here
+ // with the precise reason (not a generic "unsupported aggregate" that
blames ARRAY_AGG
+ // itself).
+ // JSONB (not VARIANT) is used as the element type because variant
columns cannot be
+ // created on tables with ROW binlog, which IVM base tables require.
+ createTable("create table test.mtmv_arr_agg_jsonb_base (k1 int, v1
jsonb)\n"
+ + "duplicate key(k1)\n"
+ + "distributed by hash(k1) buckets 1\n"
+ + "properties('replication_num' = '1', 'binlog.enable' =
'true',"
+ + " 'binlog.format' = 'ROW');");
+ String mv = "CREATE MATERIALIZED VIEW mtmv_arr_agg_jsonb\n"
+ + " BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + " PROPERTIES ('replication_num' = '1')\n"
+ + " AS SELECT k1, array_agg(v1) AS arr FROM
mtmv_arr_agg_jsonb_base GROUP BY k1;";
+ assertCreateMtmvFails(mv, "ARRAY_AGG over JSONB/VARIANT element type
JSON"
+ + " is not incrementally maintainable");
+ }
+
@Test
public void testCreateMTMVWithIncrementalFallback() throws Exception {
String mv = "CREATE MATERIALIZED VIEW mtmv_increment_fallback\n"
diff --git a/regression-test/data/mtmv_p0/ivm/test_ivm_agg_array_1.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_agg_array_1.out
new file mode 100644
index 00000000000..855297ac3a1
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_agg_array_1.out
@@ -0,0 +1,199 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !agg_initial --
+1 [10, 20]
+2 [30, 40]
+3 [null]
+
+-- !agg_initial_source --
+1 [10, 20]
+2 [30, 40]
+3 [null]
+
+-- !agg_after_insert --
+1 [10, 15, 20]
+2 [30, 40]
+3 [null]
+4 [70]
+
+-- !agg_after_insert_source --
+1 [10, 15, 20]
+2 [30, 40]
+3 [null]
+4 [70]
+
+-- !agg_after_update --
+1 [15, 15, 20]
+2 [30, 40]
+3 [null]
+4 [70]
+
+-- !agg_after_update_source --
+1 [15, 15, 20]
+2 [30, 40]
+3 [null]
+4 [70]
+
+-- !agg_after_delete --
+1 [15, 15]
+2 [null, 30]
+4 [70]
+5 [50]
+
+-- !agg_after_delete_source --
+1 [15, 15]
+2 [null, 30]
+4 [70]
+5 [50]
+
+-- !agg_after_complete --
+1 [15, 15]
+2 [null, 30]
+4 [70]
+5 [50]
+
+-- !agg_after_complete_source --
+1 [15, 15]
+2 [null, 30]
+4 [70]
+5 [50]
+
+-- !list_initial --
+1 [15, 15]
+2 [30]
+4 [70]
+5 [50]
+
+-- !list_initial_source --
+1 [15, 15]
+2 [30]
+4 [70]
+5 [50]
+
+-- !list_after_insert --
+1 [15, 15]
+2 [30, 55]
+4 [70]
+5 [50]
+6 []
+
+-- !list_after_insert_source --
+1 [15, 15]
+2 [30, 55]
+4 [70]
+5 [50]
+6 []
+
+-- !list_after_delete --
+1 [15, 15]
+2 [60]
+4 [70]
+5 [50]
+6 []
+
+-- !list_after_delete_source --
+1 [15, 15]
+2 [60]
+4 [70]
+5 [50]
+6 []
+
+-- !list_size_after_delete --
+1 2
+2 1
+4 1
+5 1
+6 0
+
+-- !list_size_after_delete_source --
+1 2
+2 1
+4 1
+5 1
+6 0
+
+-- !list_after_group_delete --
+1 [15, 15]
+2 [60, 70]
+4 [70]
+5 [50]
+
+-- !list_after_group_delete_source --
+1 [15, 15]
+2 [60, 70]
+4 [70]
+5 [50]
+
+-- !list_after_complete --
+1 [15, 15]
+2 [60, 70]
+4 [70]
+5 [50]
+
+-- !list_after_complete_source --
+1 [15, 15]
+2 [60, 70]
+4 [70]
+5 [50]
+
+-- !scalar_after_refresh --
+[null, null, 15, 15, 50, 60, 70, 70]
+
+-- !scalar_after_refresh_source --
+[null, null, 15, 15, 50, 60, 70, 70]
+
+-- !scalar_empty_table --
+[]
+
+-- !scalar_empty_table_source --
+[]
+
+-- !scalar_back_to_null_row --
+[null]
+
+-- !scalar_back_to_null_row_source --
+[null]
+
+-- !scalar_after_final_complete --
+[null]
+
+-- !scalar_after_final_complete_source --
+[null]
+
+-- !mixed_initial --
+1 [10, 20] 2 30
+2 [null, 30] 2 30
+
+-- !mixed_initial_source --
+1 [10, 20] 2 30
+2 [null, 30] 2 30
+
+-- !mixed_after_update --
+1 [10, 15] 2 25
+2 [null, 30] 2 30
+3 [40] 1 40
+
+-- !mixed_after_update_source --
+1 [10, 15] 2 25
+2 [null, 30] 2 30
+3 [40] 1 40
+
+-- !mixed_after_delete --
+1 [10, 15] 2 25
+2 [30] 1 30
+3 [40, 50] 2 90
+
+-- !mixed_after_delete_source --
+1 [10, 15] 2 25
+2 [30] 1 30
+3 [40, 50] 2 90
+
+-- !mixed_after_complete --
+1 [10, 15] 2 25
+2 [30] 1 30
+3 [40, 50] 2 90
+
+-- !mixed_after_complete_source --
+1 [10, 15] 2 25
+2 [30] 1 30
+3 [40, 50] 2 90
+
diff --git a/regression-test/suites/mtmv_p0/ivm/test_ivm_agg_array_1.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_agg_array_1.groovy
new file mode 100644
index 00000000000..6e109fb5289
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_agg_array_1.groovy
@@ -0,0 +1,299 @@
+// 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.
+
+// IVM incremental refresh for ARRAY_AGG / COLLECT_LIST aggregate MVs.
+// Reading the MV array columns is wrapped in array_sort because element order
is not guaranteed;
+// array_sort keeps the .out output stable.
+
+suite("test_ivm_agg_array_1") {
+
+ def base = "test_ivm_array_agg_collect_base"
+ def aggMv = "test_ivm_array_agg_collect_agg_mv"
+ def listMv = "test_ivm_array_agg_collect_list_mv"
+ def scalarMv = "test_ivm_array_agg_collect_scalar_mv"
+
+ def refresh = { mv ->
+ sql """REFRESH MATERIALIZED VIEW ${mv} INCREMENTAL"""
+ waitingMTMVTaskFinishedByMvName(mv)
+ }
+
+ // =========================================================
+ // Setup: MOW base table with one row per unique id
+ // =========================================================
+
+ sql """drop materialized view if exists ${scalarMv};"""
+ sql """drop materialized view if exists ${listMv};"""
+ sql """drop materialized view if exists ${aggMv};"""
+ sql """drop table if exists ${base};"""
+
+ sql """
+ CREATE TABLE ${base} (
+ id INT,
+ k INT,
+ v INT
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 2
+ PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW", "binlog.need_historical_value" = "true",
+ "enable_unique_key_merge_on_write" = "true"
+ );
+ """
+
+ // Initial rows: k=1 has two values, k=2 has two values, k=3 has one NULL
value.
+ sql """
+ INSERT INTO ${base} VALUES
+ (1, 1, 10),
+ (2, 1, 20),
+ (3, 2, 30),
+ (4, 2, 40),
+ (5, 3, NULL);
+ """
+
+ // =========================================================
+ // Part 1: grouped ARRAY_AGG (keeps NULL elements)
+ // =========================================================
+
+ sql """
+ CREATE MATERIALIZED VIEW ${aggMv}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES (
+ 'replication_num' = '1'
+ )
+ AS SELECT k, array_agg(v) AS arr FROM ${base} GROUP BY k;
+ """
+
+ refresh(aggMv)
+ order_qt_agg_initial """SELECT k, array_sort(arr) FROM ${aggMv} ORDER BY
k"""
+ order_qt_agg_initial_source """SELECT k, array_sort(array_agg(v)) FROM
${base} GROUP BY k ORDER BY k"""
+
+ // Insert a new row into k=1 and a new group k=4.
+ sql """INSERT INTO ${base} VALUES (6, 1, 15), (7, 4, 70);"""
+
+ refresh(aggMv)
+ order_qt_agg_after_insert """SELECT k, array_sort(arr) FROM ${aggMv} ORDER
BY k"""
+ order_qt_agg_after_insert_source """SELECT k, array_sort(array_agg(v))
FROM ${base} GROUP BY k ORDER BY k"""
+
+ // Update row id=1 (k=1: 10 -> 15): MOW emits delete + insert for the same
id.
+ sql """INSERT INTO ${base} VALUES (1, 1, 15);"""
+
+ refresh(aggMv)
+ order_qt_agg_after_update """SELECT k, array_sort(arr) FROM ${aggMv} ORDER
BY k"""
+ order_qt_agg_after_update_source """SELECT k, array_sort(array_agg(v))
FROM ${base} GROUP BY k ORDER BY k"""
+
+ // Delete one row of k=1 and the whole group k=3 (its only row had a NULL
value).
+ sql """DELETE FROM ${base} WHERE id = 2;"""
+ sql """DELETE FROM ${base} WHERE k = 3;"""
+ // Update (MOW upsert) id=4's value to NULL in the same window: ARRAY_AGG
must drop the old
+ // element 40 and keep a NULL element for the new row value.
+ sql """INSERT INTO ${base} VALUES (4, 2, NULL);"""
+ // Dirty another partition so the incremental refresh picks up the deletes.
+ sql """INSERT INTO ${base} VALUES (8, 5, 50);"""
+
+ refresh(aggMv)
+ order_qt_agg_after_delete """SELECT k, array_sort(arr) FROM ${aggMv} ORDER
BY k"""
+ order_qt_agg_after_delete_source """SELECT k, array_sort(array_agg(v))
FROM ${base} GROUP BY k ORDER BY k"""
+
+ // Complete refresh must agree with the incremental result.
+ sql """REFRESH MATERIALIZED VIEW ${aggMv} COMPLETE"""
+ waitingMTMVTaskFinishedByMvName(aggMv)
+ order_qt_agg_after_complete """SELECT k, array_sort(arr) FROM ${aggMv}
ORDER BY k"""
+ order_qt_agg_after_complete_source """SELECT k, array_sort(array_agg(v))
FROM ${base} GROUP BY k ORDER BY k"""
+
+ // =========================================================
+ // Part 2: grouped COLLECT_LIST (skips NULL rows)
+ // =========================================================
+
+ sql """
+ CREATE MATERIALIZED VIEW ${listMv}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES (
+ 'replication_num' = '1'
+ )
+ AS SELECT k, collect_list(v) AS lst FROM ${base} GROUP BY k;
+ """
+
+ refresh(listMv)
+ order_qt_list_initial """SELECT k, array_sort(lst) FROM ${listMv} ORDER BY
k"""
+ order_qt_list_initial_source """SELECT k, array_sort(collect_list(v)) FROM
${base} GROUP BY k ORDER BY k"""
+
+ // Same change windows as Part 1 (base is currently at the "after delete"
state).
+ // Inserts a non-NULL value into k=2 and a NULL row into the new group
k=6: COLLECT_LIST must
+ // skip the NULL row (group k=6 stays an empty list), and the group itself
stays alive.
+ sql """INSERT INTO ${base} VALUES (9, 2, 55), (10, 6, NULL);"""
+
+ refresh(listMv)
+ order_qt_list_after_insert """SELECT k, array_sort(lst) FROM ${listMv}
ORDER BY k"""
+ order_qt_list_after_insert_source """SELECT k, array_sort(collect_list(v))
FROM ${base} GROUP BY k ORDER BY k"""
+
+ // Delete a non-NULL row of k=2, insert a new non-NULL value, and update
(MOW upsert) the
+ // remaining row id=9 from 55 to NULL in the same window: COLLECT_LIST
must drop 55 and ignore
+ // the NULL replacement, keeping the other non-NULL elements.
+ sql """DELETE FROM ${base} WHERE id = 3;"""
+ sql """INSERT INTO ${base} VALUES (11, 2, 60);"""
+ sql """INSERT INTO ${base} VALUES (9, 2, NULL);"""
+
+ refresh(listMv)
+ order_qt_list_after_delete """SELECT k, array_sort(lst) FROM ${listMv}
ORDER BY k"""
+ order_qt_list_after_delete_source """SELECT k, array_sort(collect_list(v))
FROM ${base} GROUP BY k ORDER BY k"""
+
+ // Element-count sanity independent of ordering assumptions.
+ order_qt_list_size_after_delete """SELECT k, array_size(lst) FROM
${listMv} ORDER BY k"""
+ order_qt_list_size_after_delete_source """
+ SELECT k, array_size(collect_list(v)) FROM ${base} GROUP BY k ORDER BY
k"""
+
+ // Delete the only row of k=6 (a NULL row, so COLLECT_LIST never saw it)
together with an
+ // insert on k=2 in the same window to dirty the partition: the empty
group disappears while
+ // the other lists keep their non-NULL elements.
+ sql """DELETE FROM ${base} WHERE id = 10;"""
+ sql """INSERT INTO ${base} VALUES (12, 2, 70);"""
+
+ refresh(listMv)
+ order_qt_list_after_group_delete """SELECT k, array_sort(lst) FROM
${listMv} ORDER BY k"""
+ order_qt_list_after_group_delete_source """
+ SELECT k, array_sort(collect_list(v)) FROM ${base} GROUP BY k ORDER BY
k"""
+
+ sql """REFRESH MATERIALIZED VIEW ${listMv} COMPLETE"""
+ waitingMTMVTaskFinishedByMvName(listMv)
+ order_qt_list_after_complete """SELECT k, array_sort(lst) FROM ${listMv}
ORDER BY k"""
+ order_qt_list_after_complete_source """SELECT k,
array_sort(collect_list(v)) FROM ${base} GROUP BY k ORDER BY k"""
+
+ // =========================================================
+ // Part 3: scalar ARRAY_AGG over the whole table
+ // =========================================================
+
+ sql """
+ CREATE MATERIALIZED VIEW ${scalarMv}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 1
+ PROPERTIES (
+ 'replication_num' = '1'
+ )
+ AS SELECT array_agg(v) AS arr FROM ${base};
+ """
+
+ refresh(scalarMv)
+ order_qt_scalar_after_refresh """SELECT array_sort(arr) FROM ${scalarMv}"""
+ order_qt_scalar_after_refresh_source """SELECT array_sort(array_agg(v))
FROM ${base}"""
+
+ // A change window that contains both the last inserts and their deletes
leaves the base
+ // table empty; the scalar MV must collapse to an empty array (not a NULL).
+ sql """INSERT INTO ${base} VALUES (12, 7, 1), (13, 8, 2);"""
+ sql """DELETE FROM ${base} WHERE id > 0;"""
+
+ refresh(scalarMv)
+ order_qt_scalar_empty_table """SELECT array_sort(arr) FROM ${scalarMv}"""
+ order_qt_scalar_empty_table_source """SELECT array_sort(array_agg(v)) FROM
${base}"""
+
+ // Sanity: a later non-empty window still upserts a real array.
+ sql """INSERT INTO ${base} VALUES (14, 9, NULL);"""
+ refresh(scalarMv)
+ order_qt_scalar_back_to_null_row """SELECT array_sort(arr) FROM
${scalarMv}"""
+ order_qt_scalar_back_to_null_row_source """SELECT array_sort(array_agg(v))
FROM ${base}"""
+
+ // Final COMPLETE refresh of every part (waitingMTMVTaskFinishedByMvName
asserts the task
+ // status is SUCCESS) must reproduce the incremental result.
+ sql """REFRESH MATERIALIZED VIEW ${scalarMv} COMPLETE"""
+ waitingMTMVTaskFinishedByMvName(scalarMv)
+ order_qt_scalar_after_final_complete """SELECT array_sort(arr) FROM
${scalarMv}"""
+ order_qt_scalar_after_final_complete_source """SELECT
array_sort(array_agg(v)) FROM ${base}"""
+
+ // =========================================================
+ // Part 4: mixed aggregate MV -- ARRAY_AGG next to COUNT(*) / SUM in one MV
+ // =========================================================
+
+ sql """drop materialized view if exists test_ivm_agg_array_mixed_mv;"""
+ sql """drop table if exists test_ivm_agg_array_mixed_base;"""
+
+ sql """
+ CREATE TABLE test_ivm_agg_array_mixed_base (
+ id INT,
+ k INT,
+ v INT
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 2
+ PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW", "binlog.need_historical_value" = "true",
+ "enable_unique_key_merge_on_write" = "true"
+ );
+ """
+
+ sql """INSERT INTO test_ivm_agg_array_mixed_base VALUES (1, 1, 10), (2, 1,
20), (3, 2, 30), (4, 2, NULL);"""
+
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_agg_array_mixed_mv
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES (
+ 'replication_num' = '1'
+ )
+ AS SELECT k, array_agg(v) AS arr, COUNT(*) AS cnt, SUM(v) AS sv
+ FROM test_ivm_agg_array_mixed_base GROUP BY k;
+ """
+
+ refresh("test_ivm_agg_array_mixed_mv")
+ order_qt_mixed_initial """
+ SELECT k, array_sort(arr), cnt, sv FROM test_ivm_agg_array_mixed_mv
ORDER BY k"""
+ order_qt_mixed_initial_source """
+ SELECT k, array_sort(array_agg(v)), COUNT(*), SUM(v)
+ FROM test_ivm_agg_array_mixed_base GROUP BY k ORDER BY k"""
+
+ // Update id=2 (k=1: 20 -> 15) and insert a new group k=3 in one window:
array multiset,
+ // COUNT and SUM signed deltas must all merge consistently in the same
refresh.
+ sql """INSERT INTO test_ivm_agg_array_mixed_base VALUES (2, 1, 15), (5, 3,
40);"""
+
+ refresh("test_ivm_agg_array_mixed_mv")
+ order_qt_mixed_after_update """
+ SELECT k, array_sort(arr), cnt, sv FROM test_ivm_agg_array_mixed_mv
ORDER BY k"""
+ order_qt_mixed_after_update_source """
+ SELECT k, array_sort(array_agg(v)), COUNT(*), SUM(v)
+ FROM test_ivm_agg_array_mixed_base GROUP BY k ORDER BY k"""
+
+ // Delete the NULL row of k=2 in a delete-plus-insert window.
+ sql """DELETE FROM test_ivm_agg_array_mixed_base WHERE id = 4;"""
+ sql """INSERT INTO test_ivm_agg_array_mixed_base VALUES (6, 3, 50);"""
+
+ refresh("test_ivm_agg_array_mixed_mv")
+ order_qt_mixed_after_delete """
+ SELECT k, array_sort(arr), cnt, sv FROM test_ivm_agg_array_mixed_mv
ORDER BY k"""
+ order_qt_mixed_after_delete_source """
+ SELECT k, array_sort(array_agg(v)), COUNT(*), SUM(v)
+ FROM test_ivm_agg_array_mixed_base GROUP BY k ORDER BY k"""
+
+ sql """REFRESH MATERIALIZED VIEW test_ivm_agg_array_mixed_mv COMPLETE"""
+ waitingMTMVTaskFinishedByMvName("test_ivm_agg_array_mixed_mv")
+ order_qt_mixed_after_complete """
+ SELECT k, array_sort(arr), cnt, sv FROM test_ivm_agg_array_mixed_mv
ORDER BY k"""
+ order_qt_mixed_after_complete_source """
+ SELECT k, array_sort(array_agg(v)), COUNT(*), SUM(v)
+ FROM test_ivm_agg_array_mixed_base GROUP BY k ORDER BY k"""
+
+ sql """drop materialized view if exists test_ivm_agg_array_mixed_mv;"""
+ sql """drop table if exists test_ivm_agg_array_mixed_base;"""
+
+ sql """drop materialized view if exists ${scalarMv};"""
+ sql """drop materialized view if exists ${listMv};"""
+ sql """drop materialized view if exists ${aggMv};"""
+ sql """drop table if exists ${base};"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
