yujun777 commented on code in PR #62606:
URL: https://github.com/apache/doris/pull/62606#discussion_r3915028025
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -201,108 +298,583 @@ public void run() throws JobException {
}
// Every time a task is run, the relation is regenerated because
baseTables and baseViews may change,
// such as deleting a table and creating a view with the same name
- Pair<Set<TableIf>, Set<TableIf>> tablesInPlan =
MTMVPlanUtil.getBaseTableFromQuery(mtmv.getQuerySql(), ctx);
- this.relation =
MTMVPlanUtil.generateMTMVRelation(tablesInPlan.first, tablesInPlan.second);
+ MTMVPlanUtil.QueryAnalysisResult queryAnalysis =
MTMVPlanUtil.getBaseTableFromQuery(
+ mtmv.getQuerySql(), ctx);
+ this.relation =
MTMVPlanUtil.generateMTMVRelation(queryAnalysis.getAllLevelTables(),
+ queryAnalysis.getOneLevelTables());
beforeMTMVRefresh();
- List<TableIf> tableIfs = Lists.newArrayList(tablesInPlan.first);
+ List<TableIf> tableIfs =
Lists.newArrayList(queryAnalysis.getAllLevelTables());
tableIfs.sort(Comparator.comparing(TableIf::getId));
- MTMVRefreshContext context;
- Pair<List<String>, List<PartitionKeyDesc>> syncPartitions = null;
- // lock table order by id to avoid deadlock
- MetaLockUtils.readLockTables(tableIfs);
+ // This checks whether an MV in SCHEMA_CHANGE state still matches
+ // its base-table schema and partition definition. It is not part
of
+ // refresh fallback: incompatible MV definitions must fail
directly.
+ ensureQueryUsableIfNeeded(ctx, tableIfs);
+ RefreshRequest request = resolveRefreshRequest();
+ validateIvmBaselineBeforePartitionSync(request);
+ List<RefreshAttemptType> attempts = buildAttempts(request,
queryAnalysis.containsOneRowRelation());
try {
- // if mtmv is schema_change, check if column type has changed
- // If it's not in the schema_change state, the column type
definitely won't change.
- if
(MTMVState.SCHEMA_CHANGE.equals(mtmv.getStatus().getState())) {
- MTMVPlanUtil.ensureMTMVQueryUsable(mtmv, ctx);
- }
- if (mtmv.getMvPartitionInfo().getPartitionType() !=
MTMVPartitionType.SELF_MANAGE) {
- Set<MTMVRelatedTableIf> pctTables =
mtmv.getMvPartitionInfo().getPctTables();
- for (MTMVRelatedTableIf pctTable : pctTables) {
- if (!pctTable.isValidRelatedTable()) {
- throw new JobException("MTMV " + mtmv.getName() +
"'s pct table " + pctTable.getName()
- + " is not a valid pct table anymore, stop
refreshing."
- + " e.g. Table has multiple partition
columns"
- + " or including not supported transform
functions.");
+ syncPartitionsIfNeeded(ctx, tableIfs);
+ } catch (PartitionPlanningException e) {
+ throw new JobException(e.getMessage(), e);
+ }
+ MTMVRefreshContext refreshContext = buildRefreshContext(tableIfs);
+ if (handlePendingIvmBaselineRebuild(refreshContext, request, ctx))
{
+ return;
+ }
+ boolean disablePartitionRefresh = false;
+ for (RefreshAttemptType attemptType : attempts) {
+ switch (attemptType) {
+ case IVM:
+ AttemptResultType ivmResult =
executeIvmAttempt(refreshContext, request, ctx, tableIfs);
+ if (ivmResult == AttemptResultType.SUCCESS) {
+ return;
+ }
+ if (ivmResult ==
AttemptResultType.FALLBACK_TO_COMPLETE) {
+ disablePartitionRefresh = true;
+ }
+ break;
+ case PARTITIONS:
+ if (disablePartitionRefresh) {
+ break;
+ }
+ if (executePartitionBasedRefresh(refreshContext,
request, ctx)) {
+ return;
}
+ break;
+ case COMPLETE:
+ executeCompleteAttempt(refreshContext, ctx);
+ return;
+ default:
+ throw new JobException("Unsupported refresh attempt
type: " + attemptType);
+ }
+ }
+ throw new JobException("No refresh attempt succeeded for mv=" +
mtmv.getName());
+ } catch (Throwable e) {
+ if (getStatus() == TaskStatus.RUNNING) {
+ LOG.warn("run task failed, mvName: {}, taskId: {}",
+ mtmv.getName(), getTaskId(), e);
+ throw new JobException(e.getMessage(), e);
+ } else {
+ // if status is not `RUNNING`,maybe the task was canceled,
therefore, it is a normal situation
+ LOG.info("task [{}] interruption running, because status is
[{}]", getTaskId(), getStatus());
+ }
+ }
+ }
+
+ private void ensureQueryUsableIfNeeded(ConnectContext ctx, List<TableIf>
tableIfs)
+ throws JobException, AnalysisException {
+ MetaLockUtils.readLockTables(tableIfs);
+ try {
+ if (MTMVState.SCHEMA_CHANGE.equals(mtmv.getStatus().getState())) {
+ MTMVPlanUtil.ensureMTMVQueryUsable(mtmv, ctx);
+ }
+ } finally {
+ MetaLockUtils.readUnlockTables(tableIfs);
+ }
+ }
+
+ private void syncPartitionsIfNeeded(ConnectContext ctx, List<TableIf>
tableIfs)
+ throws JobException, AnalysisException, DdlException,
PartitionPlanningException {
+ if (isSkipPartitionSyncDebugPointEnabled()) {
+ LOG.info("Skip MTMV partition synchronization for debug point,
mv={}, taskId={}",
+ mtmv.getName(), getTaskId());
+ return;
+ }
+ Pair<List<String>, List<PartitionKeyDesc>> syncPartitions = null;
+ // lock table order by id to avoid deadlock
+ MetaLockUtils.readLockTables(tableIfs);
+ try {
+ if (mtmv.getMvPartitionInfo().getPartitionType() !=
MTMVPartitionType.SELF_MANAGE) {
+ Set<MTMVRelatedTableIf> pctTables =
mtmv.getMvPartitionInfo().getPctTables();
+ for (MTMVRelatedTableIf pctTable : pctTables) {
+ if (!pctTable.isValidRelatedTable()) {
+ throw new PartitionPlanningException("MTMV " +
mtmv.getName()
+ + "'s pct table " + pctTable.getName()
+ + " is not a valid pct table anymore, stop
refreshing."
+ + " e.g. Table has multiple partition columns"
+ + " or including not supported transform
functions.");
}
+ }
+ try {
syncPartitions = MTMVPartitionUtil.alignMvPartition(mtmv);
+ } catch (Exception e) {
+ throw new PartitionPlanningException(e.getMessage(), e);
}
- } finally {
- MetaLockUtils.readUnlockTables(tableIfs);
}
- if (syncPartitions != null) {
- for (String pName : syncPartitions.first) {
- MTMVPartitionUtil.dropPartition(mtmv, pName);
+ } finally {
+ MetaLockUtils.readUnlockTables(tableIfs);
+ }
+ if (syncPartitions != null) {
+ for (String pName : syncPartitions.first) {
+ MTMVPartitionUtil.dropPartition(mtmv, pName);
+ }
+ for (PartitionKeyDesc partitionKeyDesc : syncPartitions.second) {
+ MTMVPartitionUtil.addPartition(mtmv, partitionKeyDesc);
+ }
+ }
+ }
+
+ private boolean isSkipPartitionSyncDebugPointEnabled() {
+ String targetMvName = DebugPointUtil.getDebugParamOrDefault(
+ DEBUG_POINT_SKIP_PARTITION_SYNC_FILTER, "mv_name", "");
+ return mtmv.getName().equals(targetMvName)
+ && DebugPointUtil.isEnable(DEBUG_POINT_SKIP_PARTITION_SYNC);
+ }
+
+ private RefreshRequest resolveRefreshRequest() throws JobException {
+ if (taskContext.useMvDefaultRefreshPolicy()) {
+ // Scheduled/on-commit/system tasks use the policy persisted on the
+ // MV, not the default AUTO value of a newly created task context.
+ RefreshMethod refreshMethod =
mtmv.getRefreshInfo().getRefreshMethod();
+ if (refreshMethod == null) {
+ throw new JobException("MTMV " + mtmv.getName()
+ + " has unknown refresh method, please refresh or
recreate it.");
+ }
+ return new
RefreshRequest(RefreshMode.valueOf(refreshMethod.name()),
+ mtmv.getRefreshInfo().allowFallback(),
Lists.newArrayList(), false);
+ }
+ if (!CollectionUtils.isEmpty(taskContext.getPartitions())) {
+ // A partitionSpec is an exact manual request. It never falls back
to
+ // COMPLETE because that would refresh more data than the user
asked.
+ return new RefreshRequest(RefreshMode.PARTITIONS, false,
taskContext.getPartitions(), true);
+ }
+ return new RefreshRequest(taskContext.getRefreshMode(),
taskContext.allowFallback(),
+ Lists.newArrayList(), false);
+ }
+
+ private List<RefreshAttemptType> buildAttempts(RefreshRequest request,
boolean containsOneRowRelation) {
+ if (shouldUseCompleteForInitialIvmRefresh(containsOneRowRelation)) {
+ return Lists.newArrayList(RefreshAttemptType.COMPLETE);
+ }
+ List<RefreshAttemptType> attempts = Lists.newArrayList();
+ switch (request.refreshMode) {
+ case AUTO:
+ // ALTER excluded_trigger_tables clears a successful baseline
without changing the MV state.
+ if (!mtmv.isIvm() && !mtmv.hasRefreshSnapshot()
+ && mtmv.getStatus().getState() == MTMVState.NORMAL
+ && mtmv.getStatus().getRefreshState() ==
MTMVRefreshState.SUCCESS) {
+ attempts.add(RefreshAttemptType.COMPLETE);
+ break;
+ }
+ if (mtmv.isIvm()) {
+ attempts.add(RefreshAttemptType.IVM);
}
- for (PartitionKeyDesc partitionKeyDesc :
syncPartitions.second) {
- MTMVPartitionUtil.addPartition(mtmv, partitionKeyDesc);
+ // AUTO always ends with COMPLETE. For an MV defined REFRESH
COMPLETE,
+ // skip the PARTITIONS attempt: its sync check treats base
tables that are
+ // not MTMVRelatedTableIf (external tables, views) as always
synchronous,
+ // so the refresh would be skipped forever after the first
build.
+ if (mtmv.getRefreshInfo().getRefreshMethod() !=
RefreshMethod.COMPLETE) {
+ attempts.add(RefreshAttemptType.PARTITIONS);
+ } else {
+ LOG.info("AUTO refresh of COMPLETE-method mv={} skips
partition-sync check "
+ + "and refreshes all directly, taskId={}",
mtmv.getName(), super.getTaskId());
}
+ attempts.add(RefreshAttemptType.COMPLETE);
+ break;
+ case INCREMENTAL:
+ attempts.add(RefreshAttemptType.IVM);
+ if (request.allowFallback) {
+ attempts.add(RefreshAttemptType.PARTITIONS);
+ attempts.add(RefreshAttemptType.COMPLETE);
+ }
+ break;
+ case PARTITIONS:
+ attempts.add(RefreshAttemptType.PARTITIONS);
+ if (!request.explicitPartitions && request.allowFallback) {
+ attempts.add(RefreshAttemptType.COMPLETE);
+ }
+ break;
+ case COMPLETE:
+ attempts.add(RefreshAttemptType.COMPLETE);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported refresh mode: " +
request.refreshMode);
+ }
+ return attempts;
+ }
+
+ private boolean shouldUseCompleteForInitialIvmRefresh(boolean
containsOneRowRelation) {
+ return mtmv.isIvm() && !mtmv.hasRefreshSnapshot()
+ && (taskContext.getTriggerMode() != MTMVTaskTriggerMode.MANUAL
|| containsOneRowRelation);
Review Comment:
Fixed in 10295e68806. `shouldUseCompleteForInitialIvmRefresh` now forces the
first refresh to COMPLETE whenever `excluded_trigger_tables` is set (previously
only non-manual triggers or one-row relations did): an excluded source has no
delta stream, so an INCREMENTAL first refresh could never materialize its
pre-existing rows before a baseline was recorded. Covered by the updated
`test_ivm_excluded_trigger_table` (first INCREMENTAL refresh with excluded
sources must contain the excluded table's rows).
--
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]
