yujun777 commented on code in PR #62606: URL: https://github.com/apache/doris/pull/62606#discussion_r3955678176
########## fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java: ########## @@ -0,0 +1,143 @@ +// 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.MTMV; +import org.apache.doris.common.util.DebugPointUtil; +import org.apache.doris.common.util.DebugUtil; +import org.apache.doris.common.util.Util; +import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.mtmv.MTMVPlanUtil; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.analyzer.UnboundTableSink; +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.algebra.Sink; +import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; +import org.apache.doris.qe.OriginStatement; +import org.apache.doris.thrift.TPartialUpdateNewRowPolicy; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.List; +import java.util.Objects; +import java.util.Optional; + +/** + * Minimal orchestration entry point for incremental refresh. + */ +public class IvmIncrRefreshManager { + private static final Logger LOG = LogManager.getLogger(IvmIncrRefreshManager.class); + public static final String DEBUG_POINT_FORCE_FALLBACK_REASON = + "IvmIncrRefreshManager.doRefresh.force_fallback_reason"; + + public IvmIncrRefreshManager() { + } + + public IvmIncrRefreshResult doRefresh(IvmIncrRefreshContext context) throws Exception { + Objects.requireNonNull(context, "context can not be null"); + MTMV mtmv = context.getMtmv(); + Objects.requireNonNull(context.getAuditStmt(), "auditStmt can not be null"); + Objects.requireNonNull(context.getQueryIdConsumer(), "queryIdConsumer can not be null"); + String forceFallbackReason = DebugPointUtil.getDebugParamOrDefault( + DEBUG_POINT_FORCE_FALLBACK_REASON, "reason", ""); + if (!forceFallbackReason.isEmpty()) { + return IvmIncrRefreshResult.fallback( + IvmFailureReason.valueOf(forceFallbackReason), "forced by debug point"); + } + try { + executeInternalRefresh(context); Review Comment: Fixed in https://github.com/apache/doris/pull/67646. An aggregate delta is the only IVM refresh that reads the MV's own rows (joining the old aggregate values with the new delta). Since the refresh insert follows the generic committed-mode semantics (a COMMITTED-but-not-yet-VISIBLE txn can still end the task as SUCCESS), the guard is placed on the consumer side: when a non-empty aggregate delta is about to be computed and any MV partition's committed version is ahead of its visible version (a previous refresh txn committed but its rows are not readable yet), the delta rewriter now fails with the new MV_COMMIT_NOT_VISIBLE reason instead of joining stale old rows. The failure goes through the existing fallback chain (partition-based or COMPLETE rebuild), which recomputes from the base tables without reading old MV state, so the refresh converges once the stuck txn publishes; no delta is lost. Non-aggregate IVM refreshes never read old MV rows and are unaffected, and the check is skipped in cloud mode. The nonConcurrent regression (test_ivm_agg_previous_commit_not_visible) holds a refresh txn in COMMITTED via a new test debug point (DatabaseTransactionMgr.finishTransaction.block_visible, value = MV table name, so only the MV finish is blocked while base tables keep publishing), then verifies: the first stuck refresh still reports SUCCESS; the next strict INCREMENTAL refresh fails with MV_COMMIT_NOT_VISIBLE; a COMPLETE refresh can also report SUCCESS while its txn stays COMMITTED and its data is not readable; and after the debug point is removed the stuck txns auto-publish and the MV converges to the full aggregate. -- 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]
