yujun777 commented on code in PR #68787:
URL: https://github.com/apache/doris/pull/68787#discussion_r4225835958
##########
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:
Resolved on this head (d200c6df200), and the premise no longer applies:
- IVM is not released until January 2027.
`fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/AGENTS.md` now states that
directly: no compatibility with existing IVM materialized views, nor with the
binlog/stream data they consume; such a view may be dropped and recreated
rather than migrated, and a change that only affects one is not a correctness
defect. No deployment can hold the previous layout.
- Independently, the complete-refresh recovery is restored anyway, because
the state check only concerns what an incremental refresh reads. A view whose
layout lacks the carrier completes a full refresh again: verified on a view
created before this change (`SELECT g, SUM(v) * 100 AS s100`), where COMPLETE
succeeds and matches the source query (`a 3000`, `b 7000`), while a strict
INCREMENTAL still fails loudly with the pre-existing `IVM failed to find slot:
sum(v)`. That is exactly the behaviour before the carrier change, so nothing
regresses even in-house.
- `IvmNormalizeMTMVTest#testRefreshLayoutComesFromTheMvSchema` pins both
branches of the schema-owned decision.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java:
##########
@@ -1069,6 +1071,202 @@ 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;
+ // While the layout is being created there is no MV yet (CREATE
MATERIALIZED VIEW keeps only its
+ // name), so normalize decides the layout itself; a refresh is bound
by the MV's own schema.
+ MTMV mtmv = statementContext.getIvmRewriteContext().get().getMtmv();
+ for (IvmAggTarget target : aggMeta.getAggTargets()) {
+ Slot valueStateSlot = valueStateSlotFor(target, mtmv,
extendedOutputs, aggMeta);
+ 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 this target's own value state is read from,
materializing a carrier column when the
+ * state needs a column the layout does not have yet.
+ *
+ * <p>When {@code mtmv} is null the layout is being created (CREATE
MATERIALIZED VIEW knows only its
+ * name): the state is the visible aggregate column when that column
survives into the MV, and a
+ * materialized carrier otherwise. On a refresh {@code mtmv} exists and
its schema owns the layout: the
+ * state is the carrier only when the MV really has that column, and
otherwise the MV's visible column
+ * carries it. That keeps the refresh path independent of the binder's
insert projects, which rename a
+ * state column locally and rebuild it as the MV column under its own
name, so a user column that
+ * happens to be named like the aggregate's generated alias never becomes
the state.
+ */
+ private Slot valueStateSlotFor(IvmAggTarget target, MTMV mtmv,
List<NamedExpression> outputs,
+ IvmAggMeta aggMeta) {
+ if (!aggFunctionRegistry.visibleColumnHoldsValueState(target)) {
+ // AVG, BITMAP_UNION_COUNT and COUNT(*) merge hidden state or the
group count instead.
+ return target.getValueStateSlot();
+ }
+ // 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 stateSlot = target.getValueStateSlot() != null
+ ? target.getValueStateSlot() : target.getVisibleSlot();
+ if (mtmv == null) {
+ // Otherwise the projection decides: the state keeps the column
the plan projects it as (the
+ // aggregate output itself or a rename of it), and a state the
projection drops gets a carrier.
+ return materializeAggStateSlot(outputs, stateSlot, aggMeta);
+ }
+ String carrierName =
IvmUtil.ivmAggHiddenColumnName(target.getOrdinal(),
+ target.getFunctionKind().name());
+ if (mtmv.getColumn(carrierName) != null) {
+ return materializeAggStateSlot(outputs, stateSlot, aggMeta);
+ }
+ // Without a carrier the MV keeps the state in the column the plan
projected it as. That name is
+ // only usable when the MV really has it: the binder renames a state
column to the MV's own column
+ // name for an unnamed aggregate (COUNT(*) becomes __count_0), while
for a clamped key column it
+ // renames the state to a project-local name and rebuilds the MV
column by coercion, in which case
+ // the MV's visible column carries the state.
+ NamedExpression projected = findProjectedKey(outputs, stateSlot);
+ return projected != null && mtmv.getColumn(projected.getName()) != null
+ ? projected.toSlot() : null;
+ }
+
+ /**
+ * 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
or a pure rename of it, which
+ * is how the refresh sink maps the normalized hidden columns onto the
MV's own slots. Only when the
+ * state slot is really gone from the layer's output does it need a
carrier.
+ */
+ private Slot materializeAggStateSlot(List<NamedExpression> outputs, Slot
stateSlot, IvmAggMeta aggMeta) {
+ NamedExpression projected = findProjectedKey(outputs, stateSlot);
+ if (projected != null) {
+ // The projecting output's own slot is what carries the value: the
column itself, a column it
+ // projects through, or a carrier another target needed for this
same state.
+ return projected.toSlot();
+ }
+ return materializeStateCarrier(outputs, stateSlot, aggMeta);
+ }
+
+ /**
+ * Appends a hidden carrier column for {@code stateSlot} and returns its
slot.
+ *
+ * <p>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.
+ */
+ private Slot materializeStateCarrier(List<NamedExpression> outputs, Slot
stateSlot, IvmAggMeta aggMeta) {
+ IvmAggTarget owner = aggTargetOwningVisibleSlot(stateSlot, aggMeta);
+ Alias carrier = new Alias(stateSlot,
+ IvmUtil.ivmAggHiddenColumnName(owner.getOrdinal(),
owner.getFunctionKind().name()));
Review Comment:
Not reproducible on this branch; the configuration path does not produce a
legacy column. I followed it exactly:
1. `ADMIN SET FRONTEND CONFIG ('enable_date_conversion'='false')` and
`('disable_datev1'='false')` — both accepted (both are mutable) — then `CREATE
TABLE lv_base (id INT, k INT, d DATE) UNIQUE KEY(id) ... ROW binlog` succeeds.
The stored column is `date`, not a legacy DATEV1: `information_schema.columns`
reports `d / date / date` and `SHOW CREATE TABLE` prints `` `d` date ``. So the
"restore date conversion before creating the IVM MV" step has nothing to
convert.
2. With the conversion restored, `CREATE MATERIALIZED VIEW ... AS SELECT k,
YEAR(MIN(d)) AS y FROM lv_base GROUP BY k` creates its carrier column as
`date`, and the incremental refresh succeeds with values matching the source
query.
3. Earlier checks on this head: a legacy `DATE` table is rejected under
default settings ("Disable to create table with `DATE` type columns, please use
`DATEV2`"); `MIN(CAST(d AS DATEV1))` and a `DATETIMEV1` column both refresh
with matching values; and the nested `ARRAY_SORT(ARRAY_AGG(d))` over `DATE`
records `["2023-05-05", "2024-01-01", "2024-02-01"]` and `["2024-03-01"]`,
matching the source query.
If a statement is constructible here whose plan slot type differs from the
generated MV column type, please send it and I will add the coercion for it; I
did not add one because I could not reach such a state on this branch.
--
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]
