yujun777 commented on code in PR #68787:
URL: https://github.com/apache/doris/pull/68787#discussion_r4219074727
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java:
##########
@@ -1069,6 +1070,157 @@ 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.
+ *
+ * <p>A layer keeps the state alive when it projects the state slot
itself, when it projects a
+ * hidden-named alias over it (the refresh sink rebinds the normalized
hidden columns to the MV's own
+ * slots that way), or when it emits any column under the state column's
name. That last case is the
+ * insert path: the binder renames the state column locally (for example
{@code m} to {@code m1}) and
+ * coerces it back into the MV column with the original name, so the
column apply reads from the MV is
+ * unchanged and the local rename must not move the state.
+ */
+ private Slot materializeAggStateSlot(List<NamedExpression> outputs, Slot
stateSlot, IvmAggMeta aggMeta) {
+ NamedExpression projected = findProjectedKey(outputs, stateSlot);
+ if (projected != null) {
+ if (projected instanceof Alias &&
!IvmUtil.isIvmHiddenColumn(projected.getName())) {
+ // A binder project renamed the state column locally; the MV
column keeps its own name.
+ return stateSlot;
+ }
+ // The projecting output's own slot is what carries the value: a
visible column projecting
+ // through, a hidden state column, or a carrier another target
needed for this same state.
+ return projected.toSlot();
+ }
+ if (outputs.stream().anyMatch(output ->
output.getName().equals(stateSlot.getName()))) {
Review Comment:
Fixed in db3974b1b0b. You are right that the name check on its own was
unsound. The carrier decision now accepts a same-named output only when it is
the insert path coercion of that state slot: a `CAST` or `SUBSTRING` whose only
input is the slot, so it keeps the state column's name and stores the state
value under it. `SUM(v) * 100` is a `Multiply`, so the view now materializes
`__DORIS_IVM_AGG_0_SUM_COL__` and apply merges the raw state.
Reproduced your example on the previous head (`SELECT k, SUM(v) * 100 AS
\`sum(v)\`` with SUM(v)=2 storing 200, then inserting v=3 persisted 20300) and
verified the same shape after the fix: 200 at the baseline, 500 after inserting
v=3, 1000 after replacing the row with v=7, and 1000 after COMPLETE — each
matching the source query, with the carrier present in `DESC`. The clamped-key
shape still refreshes without a carrier, which is why the coercion case is
kept, and both are now covered in `test_ivm_agg_expr_over_agg_2` (part 9 for
the colliding name, part 8 for the clamped key).
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java:
##########
@@ -1069,6 +1070,157 @@ 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.
+ *
+ * <p>A layer keeps the state alive when it projects the state slot
itself, when it projects a
+ * hidden-named alias over it (the refresh sink rebinds the normalized
hidden columns to the MV's own
+ * slots that way), or when it emits any column under the state column's
name. That last case is the
+ * insert path: the binder renames the state column locally (for example
{@code m} to {@code m1}) and
+ * coerces it back into the MV column with the original name, so the
column apply reads from the MV is
+ * unchanged and the local rename must not move the state.
+ */
+ private Slot materializeAggStateSlot(List<NamedExpression> outputs, Slot
stateSlot, IvmAggMeta aggMeta) {
+ NamedExpression projected = findProjectedKey(outputs, stateSlot);
+ if (projected != null) {
+ if (projected instanceof Alias &&
!IvmUtil.isIvmHiddenColumn(projected.getName())) {
+ // A binder project renamed the state column locally; the MV
column keeps its own name.
+ return stateSlot;
+ }
+ // The projecting output's own slot is what carries the value: a
visible column projecting
+ // through, a hidden state column, or a carrier another target
needed for this same state.
+ return projected.toSlot();
+ }
+ if (outputs.stream().anyMatch(output ->
output.getName().equals(stateSlot.getName()))) {
+ // A binder coercion project rebuilt a column under the state
column's name, so the MV column
+ // apply reads still exists and the state does not need a carrier.
+ return stateSlot;
+ }
+ IvmAggTarget owner = aggTargetOwningVisibleSlot(stateSlot, aggMeta);
+ // The carrier is named after the owning target's ordinal and kind,
which is the name the delta
+ // sub-plan already generates for the same state, so no delta-side
lookup changes.
+ Alias carrier = new Alias(stateSlot,
Review Comment:
Declining this one here, with the project's own pre-GA policy:
`fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/AGENTS.md` states that IVM
is not publicly available until October 2026 and that "breaking changes to IVM
metadata, DDL format, or internal storage layout are acceptable without
migration support" before that date. An MV created on the earlier layout is
therefore expected to be recreated rather than migrated.
Two properties do bound the failure today: an old MV cannot silently read or
write a stale layout, because the CREATE-time layout signature is revalidated
on every INCREMENTAL refresh (it fails fast with `PLAN_SIGNATURE_MISMATCH`),
and COMPLETE fails at sink initialization before writing anything. If
maintainers would prefer a clearer message or an explicit recreate hint on the
COMPLETE path, I am happy to do it as a follow-up, but I do not think migration
support belongs in this change.
--
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]
