yujun777 commented on code in PR #62606: URL: https://github.com/apache/doris/pull/62606#discussion_r3957933623
########## fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandler.java: ########## @@ -0,0 +1,490 @@ +// 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. + +package org.apache.doris.mtmv.ivm; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.MTMV; +import org.apache.doris.mtmv.ivm.agg.IvmAggApplyContext; +import org.apache.doris.mtmv.ivm.agg.IvmAggDeltaSlotRef; +import org.apache.doris.mtmv.ivm.agg.IvmAggExpressionBuilder; +import org.apache.doris.mtmv.ivm.agg.IvmAggFunctionRegistry; +import org.apache.doris.mtmv.ivm.agg.IvmAggMeta; +import org.apache.doris.mtmv.ivm.agg.IvmAggTarget; +import org.apache.doris.nereids.rules.analysis.BindRelation; +import org.apache.doris.nereids.rules.exploration.join.JoinReorderContext; +import org.apache.doris.nereids.trees.expressions.Add; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.IsNull; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Not; +import org.apache.doris.nereids.trees.expressions.NullSafeEqual; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.functions.agg.Sum; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Coalesce; +import org.apache.doris.nereids.trees.expressions.functions.scalar.If; +import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral; +import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral; +import org.apache.doris.nereids.trees.plans.JoinType; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PreAggStatus; +import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalJoin; +import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** + * Aggregate delta rewrite handler for IVM. + * + * <p>Non-aggregate nodes are handled by the linear and outer-join handlers. Aggregate + * nodes return an apply plan from {@link #rewriteAggregate(LogicalAggregate, + * IvmDeltaRewriteVisitor, IvmIncrRefreshContext)} with aggregate-level dml_factor and sequence slots. + * + * <p>Handles single-table aggregate MVs with count/sum/avg/min/max. + * Min/max use an assert_true guard: if a deleted row matches the current extreme, + * execution fails with a reason that requires full refresh recovery. + * + * <h3>Overall flow</h3> + * <ol> + * <li><b>Delta sub-plan</b>: transforms the normalized aggregate into a signed delta aggregate + * where each output is weighted by {@code dml_factor} (+1 for inserts, -1 for deletes).</li> + * <li><b>Apply plan</b>: RIGHT JOINs the MV's current state against the delta on {@code row_id}, + * computes new hidden states (COALESCE(old,0) + delta), derives visible values, and + * maps the final row state to {@code __DORIS_IVM_DML_FACTOR_COL__}.</li> + * <li><b>Insert command</b>: wraps the result in an {@code InsertIntoTableCommand} that writes + * back to the MV via MOW upsert semantics.</li> + * </ol> + * + * <h3>Visitor integration</h3> + * <p>The visitor dispatches to {@code rewriteAggregate}, which recursively rewrites its child before building + * delta + apply. Projects above the aggregate are then handled by the linear handler like other normalized projects. + */ +class IvmAggDeltaHandler { + + private final IvmDeltaRewriteHelper helper = IvmDeltaRewriteHelper.INSTANCE; + private final IvmAggFunctionRegistry aggFunctionRegistry = IvmAggFunctionRegistry.INSTANCE; + private final IvmAggExpressionBuilder aggExpressionBuilder = IvmAggExpressionBuilder.INSTANCE; + + /** + * Intermediate result from {@link #buildDeltaSubPlan}. + * Carries the delta aggregate project plus slot mappings needed by {@link #buildApplyPlan}. + */ + static final class DeltaPlanParts { + /** Top project above the delta aggregate: [row_id, group_keys, delta_agg_outputs...] */ + private final LogicalProject<?> topDeltaProject; + /** Row-id slot from the top project (hash of group keys, or 0 for scalar). */ + private final Slot rowIdSlot; + /** Delta group-count slot resolved from topDeltaProject output. */ + private final Slot deltaGroupCountSlot; + /** Per-target delta slots consumed by aggregate function processors during apply. */ + private final Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots; + /** Group key slots resolved from topDeltaProject output, keyed by column name. */ + private final Map<String, Slot> groupKeySlotsByName; + + private DeltaPlanParts(LogicalProject<?> topDeltaProject, Slot rowIdSlot, Slot deltaGroupCountSlot, + Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots, Map<String, Slot> groupKeySlotsByName) { + this.topDeltaProject = topDeltaProject; + this.rowIdSlot = rowIdSlot; + this.deltaGroupCountSlot = deltaGroupCountSlot; + this.applyDeltaSlots = applyDeltaSlots; + this.groupKeySlotsByName = groupKeySlotsByName; + } + } + + /** + * Rewrites an aggregate subtree for IVM delta refresh. + * + * <p>Recursively rewrites the child where dml_factor is injected. When the child has no pending delta, + * the aggregate also has no delta. + */ + Optional<IvmDeltaRewriteResult> rewriteAggregate(LogicalAggregate<? extends Plan> agg, + IvmDeltaRewriteVisitor visitor, IvmIncrRefreshContext context) { + Optional<IvmDeltaRewriteResult> childResult = agg.child().accept(visitor, context); + if (!childResult.isPresent()) { + return Optional.empty(); + } + IvmRewriteResult rewriteResult = context.getRewriteResult(); + if (rewriteResult == null) { + throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED, + "IVM agg delta rewrite requires normalize result"); + } + IvmAggMeta aggMeta = rewriteResult.getAggMeta(); + if (aggMeta == null) { + throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED, + "IVM agg delta rewrite requires aggregate metadata"); + } + DeltaPlanParts delta = buildDeltaSubPlan(agg, childResult.get(), aggMeta); + LogicalProject<?> applyProject = buildApplyPlan( + agg, delta, aggMeta, context, visitor.getRewriteState(), childResult.get().maxDeltaIndex); + Slot dmlFactorSlot = helper.findSlotByName(applyProject.getOutput(), Column.IVM_DML_FACTOR_COL); + Slot sequenceSlot = helper.findSlotByName(applyProject.getOutput(), Column.SEQUENCE_COL); + return Optional.of(new IvmDeltaRewriteResult(applyProject, dmlFactorSlot, sequenceSlot, + childResult.get().maxDeltaIndex)); + } + + /** + * Builds the delta sub-plan: a signed aggregate over the base table's changes. + * + * <p>Input shape (from normalize): + * <pre> + * Aggregate(normalized) → child subtree (with dml_factor injected) + * </pre> + * + * <p>Output shape: + * <pre> + * Project(row_id, group_keys, coalesced delta outputs...) + * └── Aggregate(delta: SUM(signed_expr), SUM(case_when_not_null), ...) + * └── child subtree with dml_factor + * </pre> + * + * <p>The delta aggregate replaces each original agg function with signed delta expressions: + * <ul> + * <li>COUNT(*): delta = SUM(dml_factor)</li> + * <li>COUNT(expr): delta = SUM(IF(expr IS NULL, 0, dml_factor))</li> + * <li>SUM(expr): delta_sum = SUM(IF(dml_factor > 0, expr, -expr)), + * delta_count = SUM(IF(expr IS NULL, 0, dml_factor))</li> + * <li>AVG(expr): same as SUM (visible value derived later from hidden sum/count)</li> + * </ul> + * + * <p>A top project wraps the aggregate to: + * 1. Compute row_id (hash of group keys for grouped, 0 for scalar). + * 2. Apply COALESCE to NULL-susceptible outputs (SUM may return NULL for all-NULL groups). + */ + DeltaPlanParts buildDeltaSubPlan(LogicalAggregate<?> normalizedAgg, + IvmDeltaRewriteResult childResult, IvmAggMeta aggMeta) { + Plan newAggChild = childResult.plan; + Slot dmlFactorSlot = childResult.dmlFactorSlot; + + List<NamedExpression> deltaAggOutputs = new ArrayList<>(); + int groupKeySize = aggMeta.getGroupKeySlots().size(); + for (Expression groupByExpr : normalizedAgg.getGroupByExpressions()) { + if (!(groupByExpr instanceof NamedExpression)) { + throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED, + "IVM agg delta rewrite requires slot-like group key, but got: " + + groupByExpr); + } + deltaAggOutputs.add((NamedExpression) groupByExpr); + } + + Alias deltaGroupCount = new Alias(new Sum(dmlFactorSlot), Column.IVM_DELTA_GROUP_COUNT_COL); + deltaAggOutputs.add(deltaGroupCount); + + // Dispatch each normalized aggregate target to its processor. The processor appends only the delta outputs + // needed by that aggregate function, such as signed SUM, non-NULL COUNT, or MIN/MAX insert/delete extrema. + // When multiple targets share the same hidden state column (visible or hidden), only one delta + // aggregate output is emitted per column name. + Set<String> emittedDeltaNames = new HashSet<>(); + for (IvmAggTarget target : aggMeta.getAggTargets()) { + aggFunctionRegistry.appendDeltaAggregateOutputs( + target, dmlFactorSlot, deltaAggOutputs, aggExpressionBuilder, emittedDeltaNames); + } + + LogicalAggregate<?> deltaAgg = withDeltaAggregateOutput(normalizedAgg, deltaAggOutputs, newAggChild); + List<NamedExpression> topOutputs = new ArrayList<>(); + Alias rowIdAlias = new Alias( + IvmUtil.buildRowIdHash(deltaAgg.getOutput().subList(0, groupKeySize)), Column.IVM_ROW_ID_COL); + topOutputs.add(rowIdAlias); + + Set<String> zeroDefaultDeltaOutputNames = collectZeroDefaultDeltaOutputNames(aggMeta); + for (Slot slot : deltaAgg.getOutput()) { + if (zeroDefaultDeltaOutputNames.contains(slot.getName())) { + topOutputs.add(new Alias(new Coalesce(slot, aggExpressionBuilder.zeroOf(slot.getDataType())), + slot.getName())); + } else { + topOutputs.add(slot); + } + } + + LogicalProject<?> topDeltaProject = new LogicalProject<>(ImmutableList.copyOf(topOutputs), deltaAgg); + Map<String, Slot> outputByName = indexSlotsByName(topDeltaProject.getOutput()); + Slot deltaGroupCountSlot = outputByName.get(Column.IVM_DELTA_GROUP_COUNT_COL); + Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots = new LinkedHashMap<>(); + for (IvmAggTarget target : aggMeta.getAggTargets()) { + // Convert generated delta output names into stable logical keys before the apply project starts building + // expressions. Apply expressions should depend on target ordinal + logical slot kind, not string names. + aggFunctionRegistry.mapApplyDeltaSlots( + target, outputByName, applyDeltaSlots, deltaGroupCountSlot, aggExpressionBuilder); + } + Map<String, Slot> groupKeySlotsByName = new LinkedHashMap<>(); + for (Slot groupKey : aggMeta.getGroupKeySlots()) { + Slot resolved = outputByName.get(groupKey.getName()); Review Comment: Fixed in https://github.com/apache/doris/pull/67669. Aggregate group keys are now resolved and carried by slot identity instead of by name: the delta-side key slots are matched by ExprId against the aggregate metadata and kept in an ordered list aligned with it, the apply project emits each same-named group key output (l.id / r.id) from its own delta slot, and the ivm_use_full_keys identity conjuncts resolve their delta side by slot identity first (name/sanitized fallback retained for distinct names only). A regression over a two-table join with same-named keys (INSERT, full-keys and MOW UPDATE parts, each cross-checked against COMPLETE) covers the reported (10, 20)-vs-(20, 20) shape. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java: ########## @@ -0,0 +1,1219 @@ +// 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. + +package org.apache.doris.nereids.rules.analysis; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.KeysType; +import org.apache.doris.catalog.MTMV; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.info.TableNameInfo; +import org.apache.doris.common.FeNameFormat; +import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.info.TableNameInfoUtils; +import org.apache.doris.mtmv.MTMVPartitionUtil; +import org.apache.doris.mtmv.ivm.IvmDeltaRewriteHelper; +import org.apache.doris.mtmv.ivm.IvmException; +import org.apache.doris.mtmv.ivm.IvmFailureReason; +import org.apache.doris.mtmv.ivm.IvmInfo; +import org.apache.doris.mtmv.ivm.IvmPlanSignature; +import org.apache.doris.mtmv.ivm.IvmPlanSignatureGenerator; +import org.apache.doris.mtmv.ivm.IvmRewriteContext; +import org.apache.doris.mtmv.ivm.IvmRewriteResult; +import org.apache.doris.mtmv.ivm.IvmUtil; +import org.apache.doris.mtmv.ivm.agg.IvmAggColumnKey; +import org.apache.doris.mtmv.ivm.agg.IvmAggFunctionRegistry; +import org.apache.doris.mtmv.ivm.agg.IvmAggMeta; +import org.apache.doris.mtmv.ivm.agg.IvmAggStateKey; +import org.apache.doris.mtmv.ivm.agg.IvmAggTarget; +import org.apache.doris.mtmv.ivm.agg.IvmAggTargetSpec; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.jobs.JobContext; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; +import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction; +import org.apache.doris.nereids.trees.expressions.functions.agg.Count; +import org.apache.doris.nereids.trees.expressions.functions.scalar.UuidNumeric; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; +import org.apache.doris.nereids.trees.expressions.literal.LargeIntLiteral; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral; +import org.apache.doris.nereids.trees.plans.JoinType; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier; +import org.apache.doris.nereids.trees.plans.commands.info.ColumnDefinition; +import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalJoin; +import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableSink; +import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableStreamScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat; +import org.apache.doris.nereids.trees.plans.logical.LogicalResultSink; +import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias; +import org.apache.doris.nereids.trees.plans.logical.LogicalUnion; +import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter; +import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter; +import org.apache.doris.nereids.types.DataType; +import org.apache.doris.nereids.types.LargeIntType; +import org.apache.doris.nereids.types.TinyIntType; +import org.apache.doris.nereids.types.VarcharType; +import org.apache.doris.nereids.util.TypeCoercionUtils; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.IdentityHashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Normalizes the MV define plan for IVM at both CREATE MV and REFRESH MV time. + * + * <h3>Example: aggregate MV rewrite</h3> + * <p>Given MV definition: + * <pre>{@code + * SELECT sum(v1+v2), count(v3+v4), avg(v5+v6), min(v7+v8) + * FROM t GROUP BY k1, k2 + * }</pre> + * + * <p>After IvmNormalizeMTMV the plan shape is: + * <pre>{@code + * ResultSink [row_id, visible outputs, hidden state cols] + * └── Project [ + * __DORIS_IVM_ROW_ID__ = hash(k1, k2), + * k1, k2, + * sum(v1+v2), -- ordinal 0 visible (SUM) + * count(v3+v4), -- ordinal 1 visible (COUNT(expr), no hidden col) + * avg(v5+v6), -- ordinal 2 visible (AVG) + * min(v7+v8), -- ordinal 3 visible (MIN) + * __DORIS_IVM_AGG_COUNT_COL__, -- group COUNT(*) + * __DORIS_IVM_AGG_0_COUNT__, -- SUM: hidden COUNT(v1+v2) (no hidden SUM; visible stores it) + * __DORIS_IVM_AGG_2_SUM__, -- AVG: hidden SUM(v5+v6) + * __DORIS_IVM_AGG_2_COUNT__, -- AVG: hidden COUNT(v5+v6) + * __DORIS_IVM_AGG_3_COUNT__ -- MIN: hidden COUNT(v7+v8) (no hidden MIN; visible stores it) + * ] + * └── Aggregate [GROUP BY k1, k2] + * outputs: [k1, k2, + * sum(v1+v2), count(v3+v4), avg(v5+v6), min(v7+v8), + * COUNT(*), COUNT(v1+v2), + * SUM(v5+v6), COUNT(v5+v6), + * COUNT(v7+v8)] + * └── Scan(t) with base-table row-id + * }</pre> + * + * <h3>Hidden column strategy per aggregate type</h3> + * <ul> + * <li><b>COUNT(*)</b>: no hidden columns (visible = global group count)</li> + * <li><b>COUNT(expr)</b>: no hidden columns (visible stores the count directly)</li> + * <li><b>SUM</b>: hidden COUNT only (visible stores SUM; COUNT for guard)</li> + * <li><b>AVG</b>: hidden SUM + COUNT (visible is AVG ≠ SUM or COUNT)</li> + * <li><b>MIN/MAX</b>: hidden COUNT only (visible stores extremal value)</li> + * </ul> + * + * <h3>Scan-level row-id injection</h3> + * <ul> + * <li>MOW (UNIQUE_KEYS + merge-on-write): hash(uk columns) → deterministic + * <li>Excluded AGG_KEYS table: hash(agg key columns) → deterministic + * <li>DUP_KEYS: row lsn column (__DORIS_ROW_LSN_COL__) is the identity key → deterministic. + * DUP tables with row binlog always carry this hidden column at create time, so it is + * treated like a UNIQUE table's key columns. + * <li>Other key types: not supported, throws. + * </ul> + * + * <p>Outer join null-side filling turns every column of the unmatched side to NULL. To let the + * join compose the MV row-id as {@code hash(left_row_id, right_row_id)} without relying on the + * child row-id being non-NULL (a real row may carry a NULL row-id once single-column MOW keys + * are used directly), a constant-1 match flag column is injected on each null side of the outer + * join: LOJ injects on the right, ROJ on the left, FULL on both. The join's null filling turns + * the flag NULL for unmatched rows, so {@code hash(left_row_id, [left_flag], right_row_id, + * [right_flag])} distinguishes an unmatched null-side row from a real row whose row-id is NULL. + * The preserved side needs no flag: its rows are always real, so a NULL preserved-side row-id + * can only be a real value. The flag is consumed by the compose project and never stored in the + * MV. + * + * <h3>Supported plan nodes</h3> + * OlapScan, filter, project, aggregate, inner/cross join, left/right/full outer join chain, result sink, + * logical olap table sink. Nested outer joins on a null side are supported; their pre/post + * snapshot calculation can produce a substantially larger incremental refresh plan. + */ +public class IvmNormalizeMTMV extends DefaultPlanRewriter<IvmNormalizeMTMV.NormalizeContext> + implements CustomRewriter { + private static final Logger LOG = LogManager.getLogger(IvmNormalizeMTMV.class); + + static final class NormalizeContext { + private static final NormalizeContext ROOT = new NormalizeContext(true, false, false); + + private final boolean isFirstNonSink; + private final boolean isInsideAggregate; + private final boolean isInsideJoin; + + private NormalizeContext(boolean isFirstNonSink, boolean isInsideAggregate, boolean isInsideJoin) { + this.isFirstNonSink = isFirstNonSink; + this.isInsideAggregate = isInsideAggregate; + this.isInsideJoin = isInsideJoin; + } + + private NormalizeContext afterNonSink() { + if (!isFirstNonSink) { + return this; + } + return new NormalizeContext(false, isInsideAggregate, isInsideJoin); + } + + private NormalizeContext enterAggregate() { + if (isInsideAggregate) { + return this; + } + return new NormalizeContext(isFirstNonSink, true, isInsideJoin); + } + + private NormalizeContext enterJoin() { + return new NormalizeContext(false, isInsideAggregate, true); + } + } + + // Outer-join null-side match flag columns (__DORIS_IVM_JOIN_{LEFT,RIGHT}_MATCH_COL__). Injected + // as a constant 1 on the null side of an outer join; the join's null filling turns them NULL + // for unmatched rows, so the compose hash can distinguish a real NULL child row-id from an + // unmatched null-side row. Consumed by the compose project above the join, never stored. + private static final String JOIN_LEFT_MATCH_COL = Column.IVM_HIDDEN_COLUMN_PREFIX + "JOIN_LEFT_MATCH_COL__"; + private static final String JOIN_RIGHT_MATCH_COL = Column.IVM_HIDDEN_COLUMN_PREFIX + "JOIN_RIGHT_MATCH_COL__"; + + private IvmRewriteResult rewriteResult; + private final IvmAggFunctionRegistry aggFunctionRegistry = IvmAggFunctionRegistry.INSTANCE; + private StatementContext statementContext; + private boolean useFullKeys; + private final IdentityHashMap<Plan, List<Slot>> identityKeysByNode = new IdentityHashMap<>(); + private int sinkKeyCounter; + private int unionIdxCounter; + private int baseTableRowIdRenameCounter; + + @Override + public Plan rewriteRoot(Plan plan, JobContext jobContext) { + boolean enabledByIvmRewriteContext = jobContext.getCascadesContext().getStatementContext() + .isIvmMTMVRewrite(); + if (!enabledByIvmRewriteContext) { + return plan; + } + // Idempotency: if already normalized (e.g. rewritten plan re-entering), skip. + IvmRewriteResult rewriteResult = jobContext.getCascadesContext().getOrCreateIvmRewriteResult(); + if (rewriteResult.isNormalizeRewritten()) { + return plan; + } + rewriteResult.setNormalizeRewritten(true); + this.rewriteResult = rewriteResult; + statementContext = jobContext.getCascadesContext().getStatementContext(); + this.useFullKeys = resolveUseFullKeys(); + Plan result = plan.accept(this, NormalizeContext.ROOT); + rewriteResult.setNormalizedPlan(result); + IvmPlanSignature planSignature = new IvmPlanSignatureGenerator().generate(result); + rewriteResult.setPlanSignature(planSignature); + IvmRewriteContext.Mode mode = statementContext.getIvmRewriteContext().get().getMode(); + if (mode == IvmRewriteContext.Mode.INCREMENTAL) { + // Incremental refresh relies on the stored IVM layout: a drift in the normalized + // plan (row-id generation path) would produce an unmatchable delta. Check the + // signature as soon as normalization completes so every incremental path fails + // fast instead of attempting a delta rewrite against a stale layout baseline. + validatePlanSignature(statementContext.getIvmRewriteContext().get().getMtmv(), rewriteResult); + } + if (mode == IvmRewriteContext.Mode.CREATE) { + LOG.info("IVM normalized plan, mtmvName={}, mode={}, inputRoot={}, plan={}, canonicalString={}, " + + "signature={}", + statementContext.getIvmRewriteContext().get().getMtmvName(), mode, + plan.getClass().getSimpleName(), result.treeString(), planSignature.getCanonicalString(), + planSignature.getSha256()); + } + return result; + } + + private void validatePlanSignature(MTMV mtmv, IvmRewriteResult rewriteResult) { + IvmPlanSignature currentSignature = rewriteResult.getPlanSignature(); + IvmInfo ivmInfo = mtmv.getIvmInfo(); + String storedSignature = ivmInfo.getPlanSignature(); + boolean signatureMatched = currentSignature != null + && Objects.equals(storedSignature, currentSignature.getSha256()); + if (signatureMatched) { + return; + } + String detail = "IVM layout signature mismatch for mv=" + mtmv.getName() + + ", storedSignature=" + storedSignature + + ", currentSignature=" + (currentSignature == null ? "null" : currentSignature.getSha256()) + + ", currentCanonical=" + (currentSignature == null ? "null" : currentSignature.getCanonicalString()) + + ", currentPlan=" + (rewriteResult.getNormalizedPlan() == null + ? "null" : rewriteResult.getNormalizedPlan().treeString()) + + ". Run a full refresh to rebuild IVM layout baseline."; + throw new IvmException(IvmFailureReason.PLAN_SIGNATURE_MISMATCH, detail); + } + + private boolean resolveUseFullKeys() { + if (statementContext == null || !statementContext.getIvmRewriteContext().isPresent()) { + return false; + } + IvmRewriteContext rewriteContext = statementContext.getIvmRewriteContext().get(); + if (rewriteContext.getUseFullKeys() != null) { + return rewriteContext.getUseFullKeys(); + } + if (rewriteContext.getMtmv() != null && rewriteContext.getMtmv().getIvmInfo() != null) { + return rewriteContext.getMtmv().getIvmInfo().isUseFullKeys(); + } + return false; + } + + // unsupported: any plan node not explicitly whitelisted below + @Override + public Plan visit(Plan plan, NormalizeContext context) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, "IVM does not support plan node: " + + plan.getClass().getSimpleName()); + } + + // whitelisted: only OlapScan — inject IVM row-id at index 0 + @Override + public Plan visitLogicalOlapScan(LogicalOlapScan scan, NormalizeContext context) { + OlapTable table = scan.getTable(); + ScanRowId scanRowId = computeScanRowIdAndKeys(table, scan); + validateBinlogEnabled(scan); + Alias rowIdAlias = new Alias(scanRowId.mergedRowIdExpr, Column.IVM_ROW_ID_COL); + rewriteResult.addRowId(rowIdAlias.toSlot(), scanRowId.deterministic); + // When the scanned table's only key column is its own IVM row-id (a cascading MV + // whose hidden identity-key columns are absent from the scan output) and it is under + // a join, alias that base-table row-id under a renamed column so it survives the + // project as a real output slot and can be used as an identity key without colliding + // with the injected row-id name. + Alias renamedBaseRowIdAlias = useFullKeys && !context.isInsideAggregate + && context.isInsideJoin && scanRowId.baseTableRowId.isPresent() + ? new Alias(scanRowId.baseTableRowId.get(), + Column.IVM_HIDDEN_COLUMN_PREFIX + baseTableRowIdRenameCounter++ + + Column.IVM_BASE_ROW_ID_COL_SUFFIX) + : null; + ImmutableList.Builder<NamedExpression> outputsBuilder = ImmutableList.<NamedExpression>builder() + .add(rowIdAlias); + if (renamedBaseRowIdAlias != null) { + outputsBuilder.add(renamedBaseRowIdAlias); + } + outputsBuilder.addAll(scan.getOutput().stream() + .filter(slot -> !IvmUtil.isIvmHiddenColumn(slot.getName())) + .collect(ImmutableList.toImmutableList())); + List<NamedExpression> outputs = outputsBuilder.build(); + LogicalProject<?> result = new LogicalProject<>(outputs, scan); + if (useFullKeys && !context.isInsideAggregate) { + // remainKeys excludes the base table's own row-id column, so cascading + // MVs never accumulate ancestor row-id columns in their unique keys. + // When no business key survives (only the scanned MV's own row-id is + // visible), the renamed base-table row-id is kept as an identity key, + // ahead of the remaining business keys. + List<Slot> identityKeys = renamedBaseRowIdAlias == null ? scanRowId.remainKeys + : ImmutableList.<Slot>builder() + .add(renamedBaseRowIdAlias.toSlot()) + .addAll(scanRowId.remainKeys) + .build(); + identityKeysByNode.put(result, identityKeys); + } + return result; + } + + @Override + public Plan visitLogicalOlapTableStreamScan(LogicalOlapTableStreamScan scan, NormalizeContext context) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM normalize does not support LogicalOlapTableStreamScan"); + } + + // whitelisted: one-row relation — its single row has a stable row-id within this plan node. + @Override + public Plan visitLogicalOneRowRelation(LogicalOneRowRelation oneRowRelation, NormalizeContext context) { + Alias rowIdAlias = new Alias(new LargeIntLiteral(BigInteger.ONE), Column.IVM_ROW_ID_COL); + rewriteResult.addRowId(rowIdAlias.toSlot(), true); + List<NamedExpression> outputs = ImmutableList.<NamedExpression>builder() + .add(rowIdAlias) + .addAll(oneRowRelation.getProjects()) + .build(); + return oneRowRelation.withRelationIdAndProjects(oneRowRelation.getRelationId(), outputs); + } + + // whitelisted: project — recurse into child, then propagate row-id if not already present + @Override + public Plan visitLogicalProject(LogicalProject<? extends Plan> project, NormalizeContext context) { + if (project.isDistinct()) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM does not support DISTINCT project."); + } + Plan newChild = project.child().accept(this, context); + List<NamedExpression> baseOutputs = rewriteOutputsWithIvmHiddenColumns(newChild, project.getProjects(), + context.isFirstNonSink); + + List<Slot> childKeys = useFullKeys && !context.isInsideAggregate + ? identityKeysByNode.get(newChild) : null; + List<Slot> projectKeys = ImmutableList.of(); + List<NamedExpression> finalOutputs = baseOutputs; + if (childKeys != null && !childKeys.isEmpty()) { + List<NamedExpression> extendedOutputs = new ArrayList<>(baseOutputs); + List<Slot> survivingKeys = new ArrayList<>(); + for (Slot keySlot : childKeys) { + NamedExpression projected = findProjectedKey(baseOutputs, keySlot); + if (projected != null) { + survivingKeys.add(projected.toSlot()); + } else { + extendedOutputs.add(keySlot); + survivingKeys.add(keySlot); + } + } + finalOutputs = ImmutableList.copyOf(extendedOutputs); + projectKeys = survivingKeys; + } + + Plan result; + if (newChild == project.child() && finalOutputs.equals(project.getProjects())) { + result = project; + } else { + result = project.withProjectsAndChild(finalOutputs, newChild); + } + if (useFullKeys && !context.isInsideAggregate) { + identityKeysByNode.put(result, projectKeys); + } + return result; + } + + private NamedExpression findProjectedKey(List<NamedExpression> outputs, Slot keySlot) { + for (NamedExpression output : outputs) { + if (output instanceof Slot && output.getExprId().equals(keySlot.getExprId())) { + return output; + } + if (output instanceof Alias && ((Alias) output).child().equals(keySlot)) { + return output; + } + } + return null; + } + + @Override + public Plan visitLogicalFilter(LogicalFilter<? extends Plan> filter, NormalizeContext context) { + Plan result = filter.withChildren(child -> child.accept(this, context.afterNonSink())); + if (useFullKeys && !context.isInsideAggregate) { + identityKeysByNode.put(result, identityKeysByNode.get(result.child(0))); + } + return result; + } + + @Override + public Plan visitLogicalSubQueryAlias(LogicalSubQueryAlias<? extends Plan> alias, NormalizeContext context) { + Plan result = alias.withChildren(child -> child.accept(this, context.afterNonSink())); + if (useFullKeys && !context.isInsideAggregate) { + identityKeysByNode.put(result, identityKeysByNode.get(result.child(0))); + } + return result; + } + + /** + * Handles inner join / cross join / left/right/full outer join normalization. + * + * <ol> + * <li>Validates join type is INNER_JOIN, CROSS_JOIN, LEFT/RIGHT/FULL_OUTER_JOIN</li> + * <li>Normalizes nested outer joins on either side; null-side pre/post snapshots may make the + * incremental refresh plan substantially larger</li> + * <li>Normalizes both children (first non-sink = false)</li> + * <li>Composes a single row_id = hash(left_row_id, right_row_id)</li> + * <li>Wraps with Project that replaces child row_id slots with the composed one</li> + * </ol> + * + * <p>The composed row_id is deterministic iff both children's row_ids are deterministic. + * Child row_id slots are removed from the output to prevent merge conflicts in + * {@link #collectIvmHiddenSlots} when multiple {@code __DORIS_IVM_ROW_ID_COL__} exist. + * The child entries in {@code rowIdDeterminism} are kept (not cleared) so that the + * strategy phase can look up individual child row_id determinism. + */ + @Override + public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, NormalizeContext context) { + JoinType joinType = join.getJoinType(); + if (!joinType.isInnerOrCrossJoin() && !joinType.isOuterJoin()) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM does not support join type: " + joinType + + ". Only INNER_JOIN, CROSS_JOIN, LEFT_OUTER_JOIN, RIGHT_OUTER_JOIN" + + " and FULL_OUTER_JOIN are supported."); + } + if (join.isMarkJoin()) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM does not support mark join (subquery with disjunction)."); + } + NormalizeContext childContext = context.enterJoin(); + Plan newLeft = join.left().accept(this, childContext); + Plan newRight = join.right().accept(this, childContext); + + // Inject a constant-1 match flag on each null side of an outer join (LOJ: right, + // ROJ: left, FULL: both; inner/cross: none). The join's null filling turns the flag + // NULL for unmatched rows, so the compose hash can distinguish a real NULL child + // row-id from an unmatched null-side row. The preserved side needs no flag: its rows + // are always real, so a NULL preserved-side row-id can only be a real value. + // The flag is consumed by the compose project above the join and never stored in the MV. + List<Slot> leftKeys = useFullKeys && !context.isInsideAggregate + ? identityKeysByNode.get(newLeft) : null; + List<Slot> rightKeys = useFullKeys && !context.isInsideAggregate + ? identityKeysByNode.get(newRight) : null; + boolean flagOnLeft = isNullSideOnLeft(joinType); + boolean flagOnRight = isNullSideOnRight(joinType); + if (flagOnLeft) { + newLeft = addJoinNullSideMatchedColumn(newLeft, JOIN_LEFT_MATCH_COL); + if (leftKeys != null) { + identityKeysByNode.put(newLeft, leftKeys); + } + } + if (flagOnRight) { + newRight = addJoinNullSideMatchedColumn(newRight, JOIN_RIGHT_MATCH_COL); + if (rightKeys != null) { + identityKeysByNode.put(newRight, rightKeys); + } + } + LogicalJoin<Plan, Plan> newJoin = (LogicalJoin<Plan, Plan>) join.withChildren(newLeft, newRight); + + // Find left and right row_id slots from children's output + Slot leftRowIdSlot = IvmUtil.findRowIdSlot(newLeft.getOutput(), "left child of join"); + Slot rightRowIdSlot = IvmUtil.findRowIdSlot(newRight.getOutput(), "right child of join"); + + // Look up each child's row_id determinism from the accumulated map + boolean leftDet = rewriteResult.isDeterministic(leftRowIdSlot); + boolean rightDet = rewriteResult.isDeterministic(rightRowIdSlot); + // Aggregate MVs rebuild the final MV row-id from group-by keys. Child outer join row-ids only feed signed + // aggregate input rows, so retained-side determinism is not required below the aggregate. + if (joinType.isOuterJoin() && !context.isInsideAggregate) { + // If one side may be filled as NULL by an outer join, null-side repair rows + // are keyed by the opposite side row_id plus NULL. That opposite row_id must be stable + // across refreshes. FULL OUTER JOIN applies this rule to both sides. + checkOuterJoinDeterministicRowId(joinType, leftDet, rightDet); + } + + // Compose join row_id = hash(left_row_id, [left_flag], right_row_id, [right_flag]). + // A null-side match flag is NULL exactly when that side was filled as NULL by the outer + // join, so the hash encoding distinguishes an unmatched null-side row from a real row + // whose row-id is NULL. buildRowIdHash encodes each argument as (nvl(value,''), isnull(value)), + // so flag 1 vs NULL produce different encodings. + ImmutableList.Builder<Expression> rowIdKeys = ImmutableList.builderWithExpectedSize(4); + rowIdKeys.add(leftRowIdSlot); + if (flagOnLeft) { + rowIdKeys.add(IvmDeltaRewriteHelper.INSTANCE.findSlotByName( + newJoin.getOutput(), JOIN_LEFT_MATCH_COL)); + } + rowIdKeys.add(rightRowIdSlot); + if (flagOnRight) { + rowIdKeys.add(IvmDeltaRewriteHelper.INSTANCE.findSlotByName( + newJoin.getOutput(), JOIN_RIGHT_MATCH_COL)); + } + Expression joinRowIdExpr = IvmUtil.buildRowIdHash(rowIdKeys.build()); + Alias joinRowIdAlias = new Alias(joinRowIdExpr, Column.IVM_ROW_ID_COL); + + // Build Project output: [composedRowId, joinOutput minus child row_ids and match flags] + ImmutableList.Builder<NamedExpression> projectOutputs = ImmutableList.builder(); + projectOutputs.add(joinRowIdAlias); + for (Slot slot : newJoin.getOutput()) { + if (!Column.IVM_ROW_ID_COL.equals(slot.getName()) + && !JOIN_LEFT_MATCH_COL.equals(slot.getName()) + && !JOIN_RIGHT_MATCH_COL.equals(slot.getName())) { + projectOutputs.add(slot); + } + } + + // Add composed row_id to map (don't clear — child entries are kept for strategy lookup) + rewriteResult.addRowId(joinRowIdAlias.toSlot(), leftDet && rightDet); + LogicalProject<?> result = new LogicalProject<>(projectOutputs.build(), newJoin); + if (useFullKeys && !context.isInsideAggregate) { + List<Slot> joinKeys = new ArrayList<>(); + if (leftKeys != null) { + joinKeys.addAll(leftKeys); + } + if (rightKeys != null) { + joinKeys.addAll(rightKeys); + } + identityKeysByNode.put(result, joinKeys); + } + return result; + } + + /** + * Wrap a join child with a Project that appends a constant-1 match flag column. The flag is + * injected on the null side of an outer join; the join's null filling makes it NULL for + * unmatched rows. The appended flag survives the join output and is consumed by the compose + * project above the join. + */ + private LogicalProject<?> addJoinNullSideMatchedColumn(Plan child, String flagColumnName) { + ImmutableList.Builder<NamedExpression> outputs = ImmutableList.builderWithExpectedSize( + child.getOutput().size() + 1); + outputs.addAll(child.getOutput()); + outputs.add(new Alias(new TinyIntLiteral((byte) 1), flagColumnName)); + return new LogicalProject<>(outputs.build(), child); + } + + /** + * Handles UNION ALL normalization. + * + * <p>Validates: only UNION ALL (rejects DISTINCT), no constant expression arms. + * + * <p>For each child arm: + * <ol> + * <li>Normalizes the child (injects row_id at scan/join level)</li> + * <li>Wraps with a Project that computes {@code hash(arm_index, child_row_id)} as the + * new row_id — the arm_index literal prevents cross-arm row_id collision (e.g. self-union)</li> + * <li>Strips the original child row_id from the output</li> + * </ol> + * + * <p>Then rebuilds the UNION with an additional union-level row_id output column prepended. + * The union row_id is deterministic iff all arms' row_ids are deterministic. + */ + @Override + public Plan visitLogicalUnion(LogicalUnion union, NormalizeContext context) { + if (union.getQualifier() != Qualifier.ALL) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM does not support UNION DISTINCT. Only UNION ALL is supported."); + } + if (!union.getConstantExprsList().isEmpty()) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM does not support UNION ALL with constant expressions."); + } + + NormalizeContext childContext = context.afterNonSink(); + boolean useUnionKeys = useFullKeys && !context.isInsideAggregate; + int unionIdx = useUnionKeys ? unionIdxCounter++ : -1; + + // Pass 1: normalize children and collect each arm's identity keys. + List<Plan> normalizedChildren = new ArrayList<>(); + List<List<Slot>> armKeys = new ArrayList<>(); + int maxKeys = 0; + for (int i = 0; i < union.children().size(); i++) { + Plan normalizedChild = union.child(i).accept(this, childContext); + normalizedChildren.add(normalizedChild); + List<Slot> keys = useUnionKeys ? identityKeysByNode.get(normalizedChild) : ImmutableList.of(); + if (keys == null) { + keys = ImmutableList.of(); + } + armKeys.add(keys); + maxKeys = Math.max(maxKeys, keys.size()); + } + List<DataType> posTypes = useUnionKeys ? computeUnionPositionalTypes(armKeys, maxKeys) : ImmutableList.of(); + + // Pass 2: wrap each arm with row_id + arm_index + positional keys. + List<Plan> newChildren = new ArrayList<>(); + List<List<SlotReference>> newChildrenOutputs = new ArrayList<>(); + boolean allDet = true; + for (int i = 0; i < normalizedChildren.size(); i++) { + Plan normalizedChild = normalizedChildren.get(i); + Slot childRowId = IvmUtil.findRowIdSlot(normalizedChild.getOutput(), + "child " + i + " of union"); + allDet &= rewriteResult.isDeterministic(childRowId); + + Expression hashExpr = IvmUtil.buildRowIdHash( + ImmutableList.of(new IntegerLiteral(i), childRowId)); + Alias hashAlias = new Alias(hashExpr, Column.IVM_ROW_ID_COL); + + ImmutableList.Builder<NamedExpression> projOutputs = ImmutableList.builder(); + projOutputs.add(hashAlias); + if (useUnionKeys) { + projOutputs.add(new Alias(new TinyIntLiteral((byte) i), + Column.IVM_UNION_ARM_INDEX_COL_PREFIX + unionIdx + "_COL__")); + for (int p = 0; p < maxKeys; p++) { + DataType posType = posTypes.get(p); + String posName = Column.IVM_UNION_KEY_COL_PREFIX + unionIdx + "_" + p + "_COL__"; + Expression keyExpr = p < armKeys.get(i).size() + ? armKeys.get(i).get(p) : new NullLiteral(posType); + if (!keyExpr.getDataType().equals(posType)) { + keyExpr = new Cast(keyExpr, posType); + } + projOutputs.add(new Alias(keyExpr, posName)); + } + } + for (Slot slot : normalizedChild.getOutput()) { + if (!Column.IVM_ROW_ID_COL.equals(slot.getName())) { + projOutputs.add(slot); + } + } + LogicalProject<Plan> hashedChild = new LogicalProject<>(projOutputs.build(), normalizedChild); + newChildren.add(hashedChild); + + List<SlotReference> childMapping = new ArrayList<>(); + childMapping.add((SlotReference) hashedChild.getOutput().get(0)); + if (useUnionKeys) { + childMapping.add((SlotReference) hashedChild.getOutput().get(1)); + for (int p = 0; p < maxKeys; p++) { + childMapping.add((SlotReference) hashedChild.getOutput().get(2 + p)); + } + } + childMapping.addAll(union.getRegularChildrenOutputs().get(i)); + newChildrenOutputs.add(childMapping); + } + + // Create union-level row_id output + SlotReference unionRowId = new SlotReference( + StatementScopeIdGenerator.newExprId(), + Column.IVM_ROW_ID_COL, LargeIntType.INSTANCE, false, ImmutableList.of()); + rewriteResult.addRowId(unionRowId, allDet); + + // Rebuild UNION: [union_row_id, (arm_index, positional keys), ...original_outputs] + ImmutableList.Builder<NamedExpression> newOutputs = ImmutableList.builder(); + newOutputs.add(unionRowId); + List<Slot> unionIdentityKeys = new ArrayList<>(); + if (useUnionKeys) { + SlotReference armIdxOut = new SlotReference(StatementScopeIdGenerator.newExprId(), + Column.IVM_UNION_ARM_INDEX_COL_PREFIX + unionIdx + "_COL__", + TinyIntType.INSTANCE, false, ImmutableList.of()); + newOutputs.add(armIdxOut); + unionIdentityKeys.add(armIdxOut); + for (int p = 0; p < maxKeys; p++) { + SlotReference posOut = new SlotReference(StatementScopeIdGenerator.newExprId(), + Column.IVM_UNION_KEY_COL_PREFIX + unionIdx + "_" + p + "_COL__", + posTypes.get(p), true, ImmutableList.of()); + newOutputs.add(posOut); + unionIdentityKeys.add(posOut); + } + } + newOutputs.addAll(union.getOutputs()); + + Plan result = union.withNewOutputsChildrenAndConstExprsList( + newOutputs.build(), newChildren, newChildrenOutputs, union.getConstantExprsList()); + if (useUnionKeys) { + identityKeysByNode.put(result, unionIdentityKeys); + } + return result; + } + + private List<DataType> computeUnionPositionalTypes(List<List<Slot>> armKeys, int maxKeys) { + List<DataType> posTypes = new ArrayList<>(maxKeys); + for (int p = 0; p < maxKeys; p++) { + List<DataType> typesAtPos = new ArrayList<>(); + for (List<Slot> keys : armKeys) { + if (p < keys.size()) { + typesAtPos.add(keys.get(p).getDataType()); + } + } + DataType commonType = TypeCoercionUtils.findWiderCommonType(typesAtPos, false, false) + .orElse(VarcharType.SYSTEM_DEFAULT); + posTypes.add(ColumnDefinition.isEligibleKeyType(commonType) ? commonType : VarcharType.SYSTEM_DEFAULT); + } + return posTypes; + } + + @Override + public Plan visitLogicalRepeat(LogicalRepeat<? extends Plan> repeat, NormalizeContext context) { + if (!context.isInsideAggregate) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM does not support LogicalRepeat outside aggregate."); + } + Plan newChild = repeat.child().accept(this, context.afterNonSink()); + return repeat.withChildren(ImmutableList.of(newChild)); + } + + /** + * Handles aggregate MV normalization. Post-NormalizeAggregate plan shape: + * {@code Project(top) → Aggregate(normalized) → Project(bottom) → ... → Scan} + * + * <p>This method: + * <ol> + * <li>Recurses into child (injects base scan row-id, unused at agg level)</li> + * <li>Validates and normalizes all aggregate functions via {@link IvmAggFunctionRegistry}</li> + * <li>Adds hidden state aggregate columns to the Aggregate output</li> + * <li>Wraps with a Project that computes row-id = hash(group keys) or constant</li> + * <li>Stores {@link IvmAggMeta} in {@link IvmRewriteResult}</li> + * </ol> + * + * <p>Returns: {@code Project(ivm hidden cols + original agg outputs) → Aggregate(with hidden aggs)} + */ + @Override + public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, NormalizeContext context) { + if (!context.isFirstNonSink) { + throw new IvmException(IvmFailureReason.AGG_UNSUPPORTED, + "IVM aggregate must be the top-level operator (only sinks and projects allowed above it)"); + } + Plan newChild = agg.child().accept(this, context.enterAggregate().afterNonSink()); + + // After NormalizeAggregate, outputs are: group-by key Slots + Alias(AggFunc) + List<NamedExpression> origOutputs = agg.getOutputExpressions(); + List<Expression> groupByExprs = agg.getGroupByExpressions(); + boolean scalarAgg = groupByExprs.isEmpty(); + + List<Alias> aggAliases = new ArrayList<>(); + for (NamedExpression output : origOutputs) { + if (output instanceof Slot) { + // group-by key slot — validated but not collected separately + } else if (output instanceof Alias && ((Alias) output).child() instanceof AggregateFunction) { + aggAliases.add((Alias) output); + } else { + throw new IvmException(IvmFailureReason.AGG_UNSUPPORTED, + "IVM: unexpected expression in normalized aggregate output: " + output); + } + } + + // Build hidden aggregate expressions and IvmAggTarget metadata + // __DORIS_IVM_AGG_COUNT_COL__ = COUNT(*) for group multiplicity + Alias groupCountAlias = new Alias(new Count(), Column.IVM_AGG_COUNT_COL); + + List<NamedExpression> hiddenAggOutputs = new ArrayList<>(); + hiddenAggOutputs.add(groupCountAlias); + + // Pass 1: register every visible aggregate output into the unified column pool, so a later + // target can reuse an existing visible column as its hidden state (e.g. AVG(x) reuses the + // visible SUM(x) column) instead of creating a duplicate hidden column. + Map<IvmAggColumnKey, Slot> aggColumnPool = new LinkedHashMap<>(); + for (int i = 0; i < aggAliases.size(); i++) { + AggregateFunction func = (AggregateFunction) aggAliases.get(i).child(); + Expression arg = (func instanceof Count && ((Count) func).isCountStar()) + ? null : func.child(0); + aggColumnPool.putIfAbsent( + IvmAggColumnKey.of(aggFunctionRegistry.kindOf(func), arg), + aggAliases.get(i).toSlot()); + } + + // Pass 2: build targets, reusing pooled columns where the hidden state expression matches. + List<IvmAggTarget> aggTargets = new ArrayList<>(); + for (int i = 0; i < aggAliases.size(); i++) { + Alias origAlias = aggAliases.get(i); + AggregateFunction aggFunc = (AggregateFunction) origAlias.child(); + // The registry chooses the aggregate processor. The processor owns the hidden state set for its + // function, so adding a new aggregate function should only require a new processor registration. + IvmAggTargetSpec aggTargetSpec = aggFunctionRegistry.buildTargetSpec( + i, aggFunc, origAlias, aggColumnPool, hiddenAggOutputs); + aggTargets.add(aggTargetSpec.toPlaceholderTarget()); + } + + // Build new Aggregate with hidden agg outputs AFTER original outputs + ImmutableList.Builder<NamedExpression> newAggOutputs = ImmutableList.builder(); + newAggOutputs.addAll(origOutputs); + newAggOutputs.addAll(hiddenAggOutputs); + LogicalAggregate<Plan> newAgg = agg.withAggOutputChild(newAggOutputs.build(), newChild); + if (agg.getSourceRepeat().isPresent()) { + Optional<LogicalRepeat<?>> sourceRepeat = newChild.collectFirst(LogicalRepeat.class::isInstance); + if (!sourceRepeat.isPresent()) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM aggregate source repeat is missing after normalize"); + } + newAgg = newAgg.withSourceRepeat(sourceRepeat.get()); + } + + // Build wrapping Project that computes row-id and exposes all slots + // Output order: [row_id, original visible outputs, hidden state outputs] + // groupByExprs are already Slots after NormalizeAggregate + Expression rowIdExpr = IvmUtil.buildRowIdHash(groupByExprs); + Alias rowIdAlias = new Alias(rowIdExpr, Column.IVM_ROW_ID_COL); + + // Add agg-level row-id to IvmRewriteResult (child entries are kept for strategy lookup) + rewriteResult.addRowId(rowIdAlias.toSlot(), !scalarAgg); + + // Project output: row_id first, then all Aggregate output slots (original + hidden) + ImmutableList.Builder<NamedExpression> projectOutputs = ImmutableList.builder(); + projectOutputs.add(rowIdAlias); + for (NamedExpression aggOutput : newAgg.getOutputExpressions()) { + projectOutputs.add(aggOutput.toSlot()); + } + + // Resolve IvmAggTarget slots from the new Aggregate output + List<Slot> newAggSlots = newAgg.getOutput(); + // groupCountSlot is at origOutputs.size() (first hidden output after original outputs) + Slot groupCountSlot = newAggSlots.get(origOutputs.size()); + List<IvmAggTarget> resolvedTargets = resolveAggTargetSlots(aggTargets, newAggSlots); + + // After NormalizeAggregate, group-by exprs are all Slots; cast directly + List<Slot> resolvedGroupKeys = groupByExprs.stream() + .map(expr -> (Slot) expr) + .collect(ImmutableList.toImmutableList()); + + IvmAggMeta aggMeta = new IvmAggMeta(scalarAgg, resolvedGroupKeys, + groupCountSlot, resolvedTargets); + rewriteResult.setAggMeta(aggMeta); + + LogicalProject<?> result = new LogicalProject<>(projectOutputs.build(), newAgg); + if (useFullKeys) { + identityKeysByNode.put(result, resolvedGroupKeys); + } + return result; + } + + private void checkOuterJoinDeterministicRowId(JoinType joinType, boolean leftDet, boolean rightDet) { + if (isNullSideOnLeft(joinType) && !rightDet) { + throwNonDeterministicOuterJoinRowId("right", "left"); + } + if (isNullSideOnRight(joinType) && !leftDet) { + throwNonDeterministicOuterJoinRowId("left", "right"); + } + } + + private void throwNonDeterministicOuterJoinRowId(String requiredSide, String nullSide) { + throw new IvmException(IvmFailureReason.NON_DETERMINISTIC_ROW_ID, + "IVM OUTER JOIN requires deterministic row_id on retained side (" + requiredSide + + " side) because " + nullSide + " side may be filled as NULL"); + } + + private boolean isNullSideOnLeft(JoinType joinType) { + return joinType == JoinType.RIGHT_OUTER_JOIN || joinType == JoinType.FULL_OUTER_JOIN; + } + + private boolean isNullSideOnRight(JoinType joinType) { + return joinType == JoinType.LEFT_OUTER_JOIN || joinType == JoinType.FULL_OUTER_JOIN; + } + + /** + * Resolves placeholder IvmAggTarget slots to actual slots from the rebuilt Aggregate output. + * Matching is done by column name. + */ + private List<IvmAggTarget> resolveAggTargetSlots(List<IvmAggTarget> placeholderTargets, + List<Slot> newAggSlots) { + // Build name→slot map from the new Aggregate output + Map<String, Slot> slotByName = new LinkedHashMap<>(); + for (Slot slot : newAggSlots) { + slotByName.put(slot.getName(), slot); + } + + List<IvmAggTarget> resolved = new ArrayList<>(); + for (IvmAggTarget target : placeholderTargets) { + // Resolve visible slot + Slot resolvedVisible = slotByName.get(target.getVisibleSlot().getName()); + if (resolvedVisible == null) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM: failed to resolve visible slot '" + + target.getVisibleSlot().getName() + "' from rebuilt aggregate output"); + } + + // Resolve hidden state slots + ImmutableMap.Builder<IvmAggStateKey, Slot> resolvedHidden = ImmutableMap.builder(); + for (Map.Entry<IvmAggStateKey, Slot> entry : target.getHiddenStateSlots().entrySet()) { + Slot resolvedSlot = slotByName.get(entry.getValue().getName()); + if (resolvedSlot == null) { + throw new IvmException(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED, + "IVM: failed to resolve hidden state slot '" + + entry.getValue().getName() + "' from rebuilt aggregate output"); + } + resolvedHidden.put(entry.getKey(), resolvedSlot); + } + + resolved.add(new IvmAggTarget(target.getOrdinal(), target.getFunctionKind(), + resolvedVisible, resolvedHidden.build(), target.getExprArgs())); + } + return resolved; + } + + // whitelisted: result sink — recurse into child, then prepend row-id to output exprs + @Override + public Plan visitLogicalResultSink(LogicalResultSink<? extends Plan> sink, NormalizeContext context) { + validateUserOutputColumnNames(sink.getOutputExprs()); + Plan newChild = sink.child().accept(this, context); + List<NamedExpression> baseOutputs = rewriteOutputsWithIvmHiddenColumns(newChild, sink.getOutputExprs(), + context.isFirstNonSink); + + List<Slot> childKeys = useFullKeys ? identityKeysByNode.get(newChild) : null; + List<Slot> sinkKeys = new ArrayList<>(); + List<NamedExpression> finalOutputs = new ArrayList<>(baseOutputs); + if (childKeys != null && !childKeys.isEmpty()) { + Set<String> outputNames = finalOutputs.stream() + .map(NamedExpression::getName) + .collect(Collectors.toSet()); + for (Slot keySlot : childKeys) { + String keyName = keySlot.getName(); + if (IvmUtil.isIvmHiddenColumn(keyName)) { + sinkKeys.add(keySlot); + } else if (outputNames.contains(keyName)) { Review Comment: Fixed in https://github.com/apache/doris/pull/67669. The result sink now decides whether an identity key is already projected by slot identity (the same findProjectedKey check the project layer uses) instead of by output name, so a same-named key from the other side of a join is no longer treated as projected. Materialization of an unprojected key moves from the sink down to the first project that drops it, which makes CREATE (result sink) and refresh (olap-table sink) plans share the same hidden-key layout and hidden column names; without that, the refresh rewrite could not match the stored hidden column. The new regression (GROUP BY l.id, r.id selecting only l.id with ivm_use_full_keys) asserts the hidden key column exists and that INCREMENTAL matches COMPLETE. -- 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]
