yujun777 commented on code in PR #68787:
URL: https://github.com/apache/doris/pull/68787#discussion_r4218290287
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggTarget.java:
##########
@@ -42,16 +42,23 @@ public class IvmAggTarget {
private final Slot visibleSlot;
// Persisted hidden state column slots. For example, an AVG target has
hidden SUM and COUNT states.
private final Map<IvmAggStateKey, Slot> hiddenStateSlots;
+ // Column carrying this target's own aggregate value state when the
visible column does not
+ // survive to the MV (see
IvmAggFunctionProcessor#visibleColumnHoldsValueState). Null means the
+ // visible column carries it: either because it reaches the MV as a
visible column, or because
+ // this target's old value is derived from hidden state instead (AVG,
BITMAP_UNION_COUNT,
+ // COUNT(*)). Apply reads the old value through getValueStateColumnName().
+ private final Slot valueStateSlot;
// the expression(s) from the base scan that feed this aggregate
// (empty for COUNT(*); may be Slot or compound Expression like v1+v2)
private final List<Expression> exprArgs;
public IvmAggTarget(int ordinal, IvmAggFunctionKind functionKind, Slot
visibleSlot,
- Map<IvmAggStateKey, Slot> hiddenStateSlots, List<Expression>
exprArgs) {
+ Map<IvmAggStateKey, Slot> hiddenStateSlots, Slot valueStateSlot,
List<Expression> exprArgs) {
Review Comment:
Fixed in c6426e41026. `IvmAggProcessorTestBase.target()` now passes `null`
for the value state (those targets keep their value in the visible column) and
`IvmAggColumnSharingTest.targetWithHidden()` preserves
`target.getValueStateSlot()`.
Verified by running the whole aggregate processor test set through the full
`mvn test` lifecycle: `IvmAgg*ProcessorTest`, `IvmAggColumnSharingTest`,
`IvmAggDeltaHandlerTest` and `IvmNormalizeMTMVTest`, 108 tests, all pass. My
earlier fast-path run compiled only the single test class I named, which is why
this was missed locally — thanks for catching it.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java:
##########
@@ -1069,6 +1070,143 @@ private Map<String, Slot> collectIvmHiddenSlots(Plan
normalizedChild) {
.collect(Collectors.toMap(Slot::getName, slot -> slot, (left,
right) -> left, LinkedHashMap::new));
}
+ /**
+ * Materializes the aggregate state columns this layer drops, so an
incremental refresh can still
+ * read the old aggregate state from the MV.
+ *
+ * <p>Every state slot the apply stage reads must be a persisted MV
column, because apply resolves
+ * the old state by column name from the MV physical table. Hidden state
columns are hidden-named
+ * and therefore propagate through every layer, but the aggregate
functions whose own value is
+ * their mergeable state (SUM, COUNT(expr), MIN/MAX,
COLLECT_LIST/ARRAY_AGG, BITMAP_UNION) keep
+ * that state in their visible column, which disappears as soon as an
upper layer consumes it
+ * inside an expression without projecting it, as in {@code SELECT SUM(v)
* 100}. Such a slot is
+ * materialized here as a bare pass-through hidden alias, and every target
reading it is rebound
+ * to that alias.
+ *
+ * <p>Rebinding matters as much as materializing: the column pool lets a
target reuse a visible
+ * aggregate column as its own hidden state (AVG reusing a visible SUM
column), so a reusing target
+ * must follow the column's owner onto the materialized carrier instead of
reading a column that no
+ * longer reaches the MV.
+ *
+ * <p>The materialized name is the name the delta sub-plan already
generates for that state
+ * ({@link IvmUtil#ivmAggHiddenColumnName}, keyed by the owning target's
ordinal and kind), so delta
+ * aggregate outputs and delta slot lookups are unaffected and both sides
of the merge agree on the
+ * column name. A slot an earlier layer already materialized, or that
another target already
+ * materialized for the same aggregate state, is reused instead of
materializing a duplicate.
+ */
+ private List<NamedExpression>
materializeDroppedAggState(List<NamedExpression> outputs) {
+ IvmAggMeta aggMeta = rewriteResult.getAggMeta();
+ if (aggMeta == null) {
+ // Below the aggregate no target is known yet, so no aggregate
state can be dropped here.
+ return outputs;
+ }
+ List<NamedExpression> extendedOutputs = new ArrayList<>(outputs);
+ List<IvmAggTarget> reboundTargets = new
ArrayList<>(aggMeta.getAggTargets().size());
+ boolean rebound = false;
+ for (IvmAggTarget target : aggMeta.getAggTargets()) {
+ Slot valueStateSlot = target.getValueStateSlot();
+ if (aggFunctionRegistry.visibleColumnHoldsValueState(target)) {
+ // The column currently carrying this target's value: the
carrier materialized by a lower
+ // layer if there is one, otherwise the visible column the
aggregate still produces.
+ Slot carried = materializeAggStateSlot(extendedOutputs,
+ valueStateSlot != null ? valueStateSlot :
target.getVisibleSlot(), aggMeta);
+ // The visible column reaching the MV is not a separate
carrier.
+ valueStateSlot =
carried.getExprId().equals(target.getVisibleSlot().getExprId())
+ ? null : carried;
+ }
+ ImmutableMap.Builder<IvmAggStateKey, Slot> hiddenStateSlots =
ImmutableMap.builder();
+ for (Map.Entry<IvmAggStateKey, Slot> hiddenStateSlot :
target.getHiddenStateSlots().entrySet()) {
+ hiddenStateSlots.put(hiddenStateSlot.getKey(),
+ materializeAggStateSlot(extendedOutputs,
hiddenStateSlot.getValue(), aggMeta));
+ }
+ IvmAggTarget reboundTarget = target.withStateSlots(valueStateSlot,
hiddenStateSlots.build());
+ rebound |= reboundTarget != target;
+ reboundTargets.add(reboundTarget);
+ }
+ if (rebound) {
+ // Keep the rebinding visible to the layers above: a layer that
passes a state column through
+ // under a different slot (the refresh sink rebinds the normalized
hidden columns to the MV's
+ // own slots) changes which slot carries the state without adding
any column.
+ rewriteResult.setAggMeta(aggMeta.withAggTargets(reboundTargets));
+ }
+ return extendedOutputs;
+ }
+
+ /**
+ * Returns the slot that carries {@code stateSlot} above this layer,
appending a hidden pass-through
+ * alias when this layer drops it.
+ */
+ private Slot materializeAggStateSlot(List<NamedExpression> outputs, Slot
stateSlot, IvmAggMeta aggMeta) {
+ NamedExpression projected = findProjectedKey(outputs, stateSlot);
+ if (projected != null) {
+ // Already carried above this layer: a visible column projecting
through, a hidden state
+ // column (which always propagates), or a carrier another target
needed for this same state.
+ // The projecting output's own slot is what holds the value, so an
alias that passes the
+ // state through under another name rebinds the target to that
name.
+ return projected.toSlot();
+ }
+ IvmAggTarget owner = aggTargetOwningVisibleSlot(stateSlot, aggMeta);
Review Comment:
Fixed in c6426e41026. For `SELECT MIN(s) AS m` over a STRING column the
refresh plan is `Project(SUBSTRING(m1, 1, 65533) AS m) -> Project(m AS m1) ->
... -> Aggregate(MIN(s) AS m)`. The materialization treated `m1` as a moved
state and then could not find an owner, which is exactly the path you describe.
Now a visible-named alias over a state column is treated as a binder rename, so
the state keeps its own MV column name, and a layer that emits any column under
the state column's name keeps it alive, so no carrier is materialized for a
column the MV does not have. CREATE and refresh reach the same layout.
COMPLETE refresh is verified again: `aaa`, then `aaaa` after deleting the
minimum, each matching the base query. The shape is now covered by
`test_ivm_expr_over_agg_clamped_key`.
One correction on the INCREMENTAL half: that failure is not caused by this
change. The insert-coercion project is captured by
`IvmPlanSignatureGenerator.canonicalProject` (an `Alias(SUBSTRING(m1), "m")` is
not a passthrough projection), so the CREATE-time and refresh-time signatures
differ for any MV whose key column needs clamping. An aggregate MV with no
wrapped expression at all shows the same failure on this branch: `SELECT MIN(s)
AS m, COUNT(*) AS c FROM t` over a STRING column fails INCREMENTAL with
`PLAN_SIGNATURE_MISMATCH` (the wrapped shape additionally does it; the plain
one proves it is independent of the state-carrier code). I left that
pre-existing bug out of this PR to keep it focused; happy to file a separate
issue if you agree.
##########
regression-test/suites/mtmv_p0/ivm/test_ivm_agg_expr_over_agg.groovy:
##########
@@ -0,0 +1,626 @@
+// 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.
+
+suite("test_ivm_agg_expr_over_agg") {
+
+ // =========================================================
+ // A scalar expression wrapped around an aggregate result, as
+ // in SELECT k, SUM(v) * 100 FROM t GROUP BY k, must stay
+ // incrementally maintainable.
+ //
+ // Apply merges the old MV state in the state domain and then
+ // re-applies the outer expression:
+ // new.s100 = f(apply(old_mv.sum_v, delta.sum_v))
+ // so the MV must persist a column carrying SUM(v) itself. That
+ // column is the visible aggregate output when the select list
+ // projects it (SELECT SUM(v) AS s, SUM(v) * 100) and a
+ // materialized hidden column when an upper expression consumes
+ // it without projecting it (SELECT SUM(v) * 100).
+ //
+ // These cases verify both halves of the invariant:
+ // * the hidden layout, via DESC (the dropped state column is
+ // materialized, once per aggregate state, reusing existing
+ // columns when they already carry it);
+ // * the merged values, through INSERT/UPDATE/DELETE and
+ // incremental refreshes.
+ //
+ // NOTE: set show_hidden_columns=true right before a DESC only —
+ // enabling it earlier puts the session in debug mode and blocks
+ // CREATE MATERIALIZED VIEW.
+ // =========================================================
+
+ def refreshIncremental = { mv ->
+ sql """REFRESH MATERIALIZED VIEW ${mv} INCREMENTAL"""
+ waitingMTMVTaskFinishedByMvName(mv)
+ }
+
+ sql """drop materialized view if exists test_ivm_expr_over_agg_sum;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_cnt;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_min;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_max;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_list;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_div;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_cast;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_scalar;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_sum_avg;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_plain;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_cnt_star;"""
+ sql """drop materialized view if exists
test_ivm_expr_over_agg_avg_round;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_bitmap;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_agg_arg;"""
+ sql """drop materialized view if exists test_ivm_expr_over_agg_key_expr;"""
+ sql """drop materialized view if exists
test_ivm_expr_over_agg_full_keys;"""
+ sql """drop table if exists test_ivm_expr_over_agg_base;"""
+
+ sql """
+ CREATE TABLE test_ivm_expr_over_agg_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"
+ );
+ """
+
+ // =========================================================
+ // Part 1: hidden layout of the wrapped-aggregate shapes
+ // =========================================================
+
+ // SUM(v) * 100: SUM's own value is its mergeable state and the visible
column is
+ // consumed by the outer expression, so the state is materialized as
_0_SUM_COL__
+ // next to the hidden non-NULL count.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_sum
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, SUM(v) * 100 AS s100 FROM test_ivm_expr_over_agg_base
GROUP BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_sum_desc """DESC test_ivm_expr_over_agg_sum"""
+ sql """set show_hidden_columns=false"""
+
+ // COUNT(v) + 1: same for COUNT(expr), whose visible column is the count
state.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_cnt
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, COUNT(v) + 1 AS c1 FROM test_ivm_expr_over_agg_base GROUP
BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_cnt_desc """DESC test_ivm_expr_over_agg_cnt"""
+ sql """set show_hidden_columns=false"""
+
+ // MIN(v) * 2 and MAX(v) + 1.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_min
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, MIN(v) * 2 AS m2 FROM test_ivm_expr_over_agg_base GROUP
BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_min_desc """DESC test_ivm_expr_over_agg_min"""
+ sql """set show_hidden_columns=false"""
+
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_max
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, MAX(v) + 1 AS m1 FROM test_ivm_expr_over_agg_base GROUP
BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_max_desc """DESC test_ivm_expr_over_agg_max"""
+ sql """set show_hidden_columns=false"""
+
+ // ARRAY_SIZE(COLLECT_LIST(v)): the visible array is the whole aggregate
state.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_list
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, ARRAY_SIZE(COLLECT_LIST(v)) AS n FROM
test_ivm_expr_over_agg_base GROUP BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_list_desc """DESC test_ivm_expr_over_agg_list"""
+ sql """set show_hidden_columns=false"""
+
+ // SUM(v) / COUNT(v): two states, both consumed by one expression. The
visible COUNT
+ // column of the COUNT target is also SUM's hidden non-NULL count (column
pool), so
+ // the shared count state is materialized exactly once.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_div
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, SUM(v) / COUNT(v) AS d FROM test_ivm_expr_over_agg_base
GROUP BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_div_desc """DESC test_ivm_expr_over_agg_div"""
+ sql """set show_hidden_columns=false"""
+
+ // CAST(SUM(v) AS DOUBLE) and a scalar (no GROUP BY) variant.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_cast
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, CAST(SUM(v) AS DOUBLE) AS d FROM
test_ivm_expr_over_agg_base GROUP BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_cast_desc """DESC test_ivm_expr_over_agg_cast"""
+ sql """set show_hidden_columns=false"""
+
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_scalar
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT SUM(v) * 100 AS s100 FROM test_ivm_expr_over_agg_base;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_scalar_desc """DESC test_ivm_expr_over_agg_scalar"""
+ sql """set show_hidden_columns=false"""
+
+ // SUM(v) * 100 next to AVG(v) * 200: AVG's hidden SUM state reuses the
visible SUM
+ // column, so it must follow that column onto the single materialized
carrier instead
+ // of adding a second SUM column.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_sum_avg
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, SUM(v) * 100 AS s100, AVG(v) * 200 AS a200 FROM
test_ivm_expr_over_agg_base GROUP BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_sum_avg_desc """DESC test_ivm_expr_over_agg_sum_avg"""
+ sql """set show_hidden_columns=false"""
+
+ // =========================================================
+ // Part 2: shapes that already worked must keep their exact
+ // layout — no extra column is materialized when the visible
+ // column itself carries the state.
+ // =========================================================
+
+ // The visible SUM column is projected, so it carries the state itself.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_plain
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, SUM(v) AS s, SUM(v) * 100 AS s100 FROM
test_ivm_expr_over_agg_base GROUP BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_plain_desc """DESC test_ivm_expr_over_agg_plain"""
+ sql """set show_hidden_columns=false"""
+
+ // COUNT(*) reads the group count, AVG and BITMAP_UNION_COUNT derive their
visible value
+ // from hidden state: wrapping them needs no state materialization.
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_cnt_star
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, COUNT(*) * 2 AS c2 FROM test_ivm_expr_over_agg_base GROUP
BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_cnt_star_desc """DESC test_ivm_expr_over_agg_cnt_star"""
+ sql """set show_hidden_columns=false"""
+
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_avg_round
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, ROUND(AVG(v), 2) AS a FROM test_ivm_expr_over_agg_base
GROUP BY k;
+ """
+ sql """set show_hidden_columns=true"""
+ qt_avg_round_desc """DESC test_ivm_expr_over_agg_avg_round"""
+ sql """set show_hidden_columns=false"""
+
+ sql """
+ CREATE MATERIALIZED VIEW test_ivm_expr_over_agg_bitmap
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k, BITMAP_UNION_COUNT(TO_BITMAP(v)) + 0 AS b FROM
test_ivm_expr_over_agg_base GROUP BY k;
Review Comment:
Added in 5b23aa034d5. The new case is
`BITMAP_COUNT(BITMAP_UNION(TO_BITMAP(v)))`, checked through:
- `DESC`: the bitmap state is materialized as
`__DORIS_IVM_AGG_0_BITMAP_UNION_COL__`, so `IvmAggBitmapUnionProcessor` reads
its old state through a carrier instead of the dropped visible column;
- two insert-only incremental refreshes and a COMPLETE refresh, each
compared against the same query over the base table (`2,2,0` → `3,3,0` →
`4,2,1,1`).
The `BITMAP_UNION_COUNT` case is refreshed in the insert-only window on
purpose: deleting a non-NULL bitmap element is the separate, pre-existing
limitation covered by `test_ivm_bitmap_runtime_fallback`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
