yujun777 commented on code in PR #68787:
URL: https://github.com/apache/doris/pull/68787#discussion_r4219498641
##########
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:
Correction to my previous reply here, after 362ea855566: the failure mode
changed with the layout rework.
The sink-arity path no longer exists. A refresh now takes the state column
from the MV schema, so an MV created before that carrier column existed does
not get the carrier added to its plan at all; instead normalization reports the
missing state column by name ("aggregate state column 'X' is missing from the
normalized MV output"). Same outcome for such an MV (it has to be recreated,
per the pre-GA layout policy above), but the diagnostic fires earlier and names
the column instead of surfacing as an output/destination tuple mismatch at sink
initialization.
`IvmNormalizeMTMVTest#testRefreshLayoutComesFromTheMvSchema` pins both
branches: an MV that has the carrier keeps it in the normalized plan, and an MV
without it raises the missing-state-column error rather than adding a column
the MV does not have.
--
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]
