github-actions[bot] commented on code in PR #68787:
URL: https://github.com/apache/doris/pull/68787#discussion_r4217907947
##########
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:
[P1] Update the remaining FE test callers for the new constructor.
IvmAggProcessorTestBase.target() and IvmAggColumnSharingTest.targetWithHidden()
still call new IvmAggTarget(...) with five arguments, while this is now the
only constructor and requires six. Test source compilation therefore fails
before the new IVM tests can run. Pass the intended valueStateSlot (null where
appropriate) at both sites, or keep a delegating overload.
##########
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:
[P1] Preserve aggregate-state ownership across the sink's coercion projects.
For an IVM view defined as SELECT MIN(s) AS m FROM t with s STRING, the MV
stores its first visible column as VARCHAR(65533). On refresh, BindSink adds
Project(SUBSTRING(m1, 1, 65533)) -> Project(m AS m1) -> Aggregate(MIN(s) AS m).
The lower project rebinds the value state to m1; the upper project drops it,
but aggTargetOwningVisibleSlot only recognizes the original aggregate ExprId
and throws PLAN_REWRITE_FAILED. The view can be created but cannot COMPLETE or
INCREMENTAL refresh. Handle the rebinding while keeping CREATE and refresh
state layouts aligned, and cover this refresh shape.
##########
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:
[P2] Exercise a wrapped BITMAP_UNION through refresh. This bitmap case uses
BITMAP_UNION_COUNT, whose processor reads its existing hidden bitmap state; it
never tests the newly changed IvmAggBitmapUnionProcessor lookup through a
materialized value carrier. Add a view such as
BITMAP_COUNT(BITMAP_UNION(TO_BITMAP(v))), then check initial and subsequent
incremental refreshes and COMPLETE refresh against the base query. The current
unit test checks only the carrier's layout.
--
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]
