yujun777 commented on code in PR #68787:
URL: https://github.com/apache/doris/pull/68787#discussion_r4219449163
##########
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 properly in 362ea855566, which removes the heuristic instead of
narrowing it.
The carrier decision no longer looks at the plan's expression shape at all.
On a refresh the MV schema owns the layout: the state is the carrier when the
MV has that column, and the MV's visible column otherwise. CREATE has no MV
yet, so it keeps the original rule of materializing a carrier when the
aggregate output does not survive into the sink. The insert-coercion check and
its helper are gone, so a `CAST`/`SUBSTRING` over the state slot can never be
taken for it, no matter whether the coercion came from the binder or from the
user.
Verified with your example and the earlier ones:
- `SELECT k, CAST(SUM(v) AS FLOAT) AS \`sum(v)\`` with SUM(v)=16777217: the
MV materializes `__DORIS_IVM_AGG_0_SUM_COL__`, and after inserting v=1 it
returns 16777218, matching the source query (the old code returned 16777216).
- `SELECT k, CAST(SUM(d) AS DECIMAL(20, 0)) AS \`sum(d)\`` over two rows of
1.55 returns 3 (the old code returned 157).
- `SELECT k, SUM(v) * 100 AS \`sum(v)\`` returns 200, then 500, then 1000.
Both colliding shapes now have regression coverage in
`test_ivm_agg_expr_over_agg_2` (part 9 multiplication alias, part 10 lossy
cast), and both branches of the new decision are unit-tested in
`IvmNormalizeMTMVTest#testRefreshLayoutComesFromTheMvSchema`.
On the sink-arity path you mentioned in the same area: an MV created before
the carrier column existed now fails during normalization with the missing
state column named, rather than at sink initialization, so an incomplete layout
is still reported and can never be merged against silently.
--
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]
