-
Notifications
You must be signed in to change notification settings - Fork 3.9k
[feat](mtmv) support ALTER MATERIALIZED VIEW ADD COLUMN #65539
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -51,6 +51,7 @@ | |
| import org.apache.doris.datasource.iceberg.IcebergExternalTable; | ||
| import org.apache.doris.info.TableNameInfoUtils; | ||
| import org.apache.doris.mtmv.BaseTableInfo; | ||
| import org.apache.doris.mtmv.MTMVAlterOpType; | ||
| import org.apache.doris.nereids.trees.plans.commands.AlterSystemCommand; | ||
| import org.apache.doris.nereids.trees.plans.commands.AlterTableCommand; | ||
| import org.apache.doris.nereids.trees.plans.commands.AlterViewCommand; | ||
|
|
@@ -1313,6 +1314,11 @@ public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) { | |
| case ALTER_PROPERTY: | ||
| mtmv.alterMvProperties(alterMTMV.getMvProperties()); | ||
| break; | ||
| case ALTER_ADD_COLUMN: | ||
| mtmv.alterAddColumn( | ||
| alterMTMV.getNewColumn(), | ||
| alterMTMV.getExprSql(), alterMTMV.getNewQuerySql()); | ||
| break; | ||
| case ADD_TASK: | ||
| alterSuccess = mtmv.addTaskResult(alterMTMV.getTask(), alterMTMV.getRelation(), | ||
| alterMTMV.getPartitionSnapshots(), | ||
|
|
@@ -1329,6 +1335,10 @@ public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) { | |
| if (alterMTMV.isNeedRebuildJob()) { | ||
| Env.getCurrentEnv().getMtmvService().alterJob(mtmv, isReplay); | ||
| } | ||
| // keep FE-side plan/cache consistent after MTMV schema change | ||
| if (alterSuccess && alterMTMV.getOpType() == MTMVAlterOpType.ALTER_ADD_COLUMN) { | ||
| Env.getCurrentEnv().notifyTableMetaChange(mtmv); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. CRITICAL — Compilation error: This code will not compile. The intended behavior (per the comment "keep FE-side plan/cache consistent after MTMV schema change") should be achieved by:
Alternatively, if a new |
||
| } | ||
| // 4. log it and replay it in the follower | ||
| if (!isReplay && alterSuccess) { | ||
| Env.getCurrentEnv().getEditLog().logAlterMTMV(alterMTMV); | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,6 +22,7 @@ | |
| import org.apache.doris.catalog.info.TableNameInfo; | ||
| import org.apache.doris.common.AnalysisException; | ||
| import org.apache.doris.common.Config; | ||
| import org.apache.doris.common.Pair; | ||
| import org.apache.doris.common.util.PropertyAnalyzer; | ||
| import org.apache.doris.datasource.CatalogMgr; | ||
| import org.apache.doris.job.common.TaskStatus; | ||
|
|
@@ -44,18 +45,26 @@ | |
| import org.apache.doris.mtmv.MTMVRelation; | ||
| import org.apache.doris.mtmv.MTMVSnapshotIf; | ||
| import org.apache.doris.mtmv.MTMVStatus; | ||
| import org.apache.doris.nereids.DorisParser; | ||
| import org.apache.doris.nereids.DorisParserBaseVisitor; | ||
| import org.apache.doris.nereids.parser.NereidsParser; | ||
| import org.apache.doris.nereids.rules.analysis.SessionVarGuardRewriter; | ||
| import org.apache.doris.qe.ConnectContext; | ||
|
|
||
| import com.google.common.collect.Maps; | ||
| import com.google.common.collect.Sets; | ||
| import com.google.gson.annotations.SerializedName; | ||
| import org.apache.commons.collections4.MapUtils; | ||
| import org.antlr.v4.runtime.ParserRuleContext; | ||
| import org.antlr.v4.runtime.tree.ParseTree; | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. test |
||
| import org.antlr.v4.runtime.tree.RuleNode; | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. test |
||
| import org.apache.commons.collections.MapUtils; | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. test
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Accidental library downgrade: The import changed from
Fix: Revert this import to |
||
| import org.apache.commons.lang3.StringUtils; | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. test |
||
| import org.apache.logging.log4j.LogManager; | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. test |
||
| import org.apache.logging.log4j.Logger; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Map.Entry; | ||
| import java.util.Optional; | ||
|
|
@@ -197,6 +206,80 @@ public MTMVStatus alterStatus(MTMVStatus newStatus) { | |
| } | ||
| } | ||
|
|
||
| public void alterAddColumn(Column newColumn, String exprSql, String newQuerySql) { | ||
| writeMvLock(); | ||
| try { | ||
| querySql = StringUtils.isBlank(newQuerySql) | ||
| ? rewriteQuerySqlForAddColumn(querySql, exprSql, newColumn.getName()) | ||
| : newQuerySql; | ||
|
|
||
| MaterializedIndexMeta baseIndexMeta = getIndexMetaByIndexId(getBaseIndexId()); | ||
| List<Column> newSchema = new ArrayList<>(baseIndexMeta.getSchema()); | ||
| Column copied = new Column(newColumn); | ||
| copied.setUniqueId(baseIndexMeta.incAndGetMaxColUniqueId()); | ||
| newSchema.add(copied); | ||
| baseIndexMeta.setSchema(newSchema); | ||
| baseIndexMeta.setSchemaVersion(baseIndexMeta.getSchemaVersion() + 1); | ||
| rebuildFullSchema(); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. BUG — Missing rewrite cache invalidation: Between the ALTER ADD COLUMN and the next successful refresh, the MTMV's rewrite caches will serve stale results to query rewriting — matching incoming queries against the old schema and potentially returning incorrect results. The existing pattern in rewriteCacheGeneration++;
cacheWithGuard = null;
cacheWithoutGuard = null;Add these three lines here alongside |
||
| this.schemaChangeVersion++; | ||
| this.refreshSnapshot = new MTMVRefreshSnapshot(); | ||
| } catch (Exception e) { | ||
| throw new RuntimeException("alter materialized view add column failed", e); | ||
| } finally { | ||
| writeMvUnlock(); | ||
| } | ||
| } | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Duplicated SQL-rewriting pattern: There are now three copies of this pattern in the codebase (MTMV, BaseViewInfo, and LogicalPlanBuilderForSyncMv). Consider extracting a shared |
||
|
|
||
| public static String rewriteQuerySqlForAddColumn(String originQuerySql, String exprSql, String columnName) { | ||
| SelectColumnClauseIndexFinder finder = new SelectColumnClauseIndexFinder(); | ||
| ParserRuleContext tree = NereidsParser.toAst(originQuerySql, DorisParser::singleStatement); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. NPE risk: This is caught by Add a null check before line 135: if (selectColumnClauseIndex == null) {
throw new IllegalStateException("Could not locate SELECT column clause in query SQL");
} |
||
| finder.visit(tree); | ||
| Pair<Integer, Integer> selectColumnClauseIndex = finder.getIndex(); | ||
| String escapedColumnName = columnName.replace("`", "``"); | ||
| String newSelectColumnClause = originQuerySql.substring(selectColumnClauseIndex.first, | ||
| selectColumnClauseIndex.second + 1) | ||
| + ", " + exprSql + " AS `" + escapedColumnName + "`"; | ||
| return StringUtils.overlay(originQuerySql, newSelectColumnClause, | ||
| selectColumnClauseIndex.first, selectColumnClauseIndex.second + 1); | ||
| } | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Code duplication: The only difference is that Extract a shared |
||
|
|
||
| private static class SelectColumnClauseIndexFinder extends DorisParserBaseVisitor<Void> { | ||
| private boolean found = false; | ||
| private Pair<Integer, Integer> index; | ||
|
|
||
| @Override | ||
| public Void visitChildren(RuleNode node) { | ||
| if (found) { | ||
| return null; | ||
| } | ||
| int n = node.getChildCount(); | ||
| for (int i = 0; i < n; ++i) { | ||
| ParseTree child = node.getChild(i); | ||
| child.accept(this); | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public Void visitCte(DorisParser.CteContext ctx) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public Void visitSelectColumnClause(DorisParser.SelectColumnClauseContext ctx) { | ||
| if (found) { | ||
| return null; | ||
| } | ||
| found = true; | ||
| index = Pair.of(ctx.getStart().getStartIndex(), ctx.getStop().getStopIndex()); | ||
| return null; | ||
| } | ||
|
|
||
| public Pair<Integer, Integer> getIndex() { | ||
| return index; | ||
| } | ||
| } | ||
|
|
||
| public void processBaseViewChange(String schemaChangeDetail) { | ||
| writeMvLock(); | ||
| try { | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,5 +21,6 @@ public enum MTMVAlterOpType { | |
| ALTER_REFRESH_INFO, | ||
| ALTER_STATUS, | ||
| ALTER_PROPERTY, | ||
| ALTER_ADD_COLUMN, | ||
| ADD_TASK; | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| // 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.trees.plans.commands.info; | ||
|
|
||
| import org.apache.doris.catalog.Env; | ||
| import org.apache.doris.catalog.MTMV; | ||
| import org.apache.doris.catalog.TableIf.TableType; | ||
| import org.apache.doris.common.DdlException; | ||
| import org.apache.doris.common.UserException; | ||
| import org.apache.doris.mtmv.MTMVPlanUtil; | ||
| import org.apache.doris.nereids.exceptions.AnalysisException; | ||
| import org.apache.doris.qe.ConnectContext; | ||
|
|
||
| import org.apache.commons.collections.CollectionUtils; | ||
| import org.apache.commons.lang3.StringUtils; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Objects; | ||
|
|
||
| /** | ||
| * alter mtmv add column info. | ||
| * Grammar: ALTER MATERIALIZED VIEW mv ADD COLUMN colName AS expression. | ||
| * The new column type is fully inferred from the expression, users do NOT specify a type. | ||
| */ | ||
| public class AlterMTMVAddColumnInfo extends AlterMTMVInfo { | ||
| private final String columnName; | ||
| private final String exprSql; | ||
| private ColumnDefinition analyzedColumn; | ||
| private String rewrittenQuerySql; | ||
|
|
||
| public AlterMTMVAddColumnInfo(TableNameInfo mvName, String columnName, String exprSql) { | ||
| super(mvName); | ||
| this.columnName = Objects.requireNonNull(columnName, "require columnName"); | ||
| this.exprSql = Objects.requireNonNull(exprSql, "require exprSql object"); | ||
| } | ||
|
|
||
| @Override | ||
| public void analyze(ConnectContext ctx) throws AnalysisException { | ||
| super.analyze(ctx); | ||
| if (StringUtils.isBlank(columnName)) { | ||
| throw new AnalysisException("ADD COLUMN must specify a column name"); | ||
| } | ||
| if (StringUtils.isBlank(exprSql)) { | ||
| throw new AnalysisException("ADD COLUMN must specify expression by `AS expression`"); | ||
| } | ||
| try { | ||
| MTMV mtmv = (MTMV) Env.getCurrentInternalCatalog().getDbOrAnalysisException(mvName.getDb()) | ||
| .getTableOrDdlException(mvName.getTbl(), TableType.MATERIALIZED_VIEW); | ||
| if (mtmv.getColumn(columnName) != null) { | ||
| throw new AnalysisException("Column already exists: " + columnName); | ||
| } | ||
| inferColumnFromExpression(mtmv, ctx); | ||
| } catch (DdlException | org.apache.doris.common.AnalysisException e) { | ||
| throw new AnalysisException(e.getMessage(), e); | ||
| } | ||
| } | ||
|
|
||
| private void inferColumnFromExpression(MTMV mtmv, ConnectContext ctx) { | ||
| rewrittenQuerySql = MTMV.rewriteQuerySqlForAddColumn(mtmv.getQuerySql(), exprSql, columnName); | ||
| List<ColumnDefinition> allColumns = MTMVPlanUtil.generateColumnsBySql( | ||
| rewrittenQuerySql, | ||
| ctx, | ||
| mtmv.getMvPartitionInfo() == null ? null : mtmv.getMvPartitionInfo().getPartitionCol(), | ||
| mtmv.getDistributionColumnNames(), | ||
| null, | ||
| mtmv.getTableProperty() == null ? null : mtmv.getTableProperty().getProperties()); | ||
| if (CollectionUtils.isEmpty(allColumns) || allColumns.size() != mtmv.getBaseSchema().size() + 1) { | ||
| throw new AnalysisException("failed to infer added column by expression"); | ||
| } | ||
| analyzedColumn = allColumns.get(allColumns.size() - 1); | ||
| } | ||
|
|
||
| @Override | ||
| public void run() throws UserException { | ||
| Env.getCurrentEnv().alterMTMVAddColumnInfo(this); | ||
| } | ||
|
|
||
| public ColumnDefinition getColumn() { | ||
| return analyzedColumn; | ||
| } | ||
|
|
||
| public String getExprSql() { | ||
| return exprSql; | ||
| } | ||
|
|
||
| public String getRewrittenQuerySql() { | ||
| return rewrittenQuerySql; | ||
| } | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Exception escape:
MTMV.alterAddColumn()wraps all exceptions inRuntimeException(seecatch (Exception e)at the end of the method). However,processAlterMTMVonly catchesUserExceptionat line 1336, notRuntimeException.If
alterAddColumn()fails (e.g., concurrent modification, corrupted index metadata, or the NPE from Finding 6 above), theRuntimeExceptionwill propagate uncaught past theprocessAlterMTMVhandler, potentially crashing the statement execution.Either:
alterAddColumn()throw a checked exception thatprocessAlterMTMVcan handle, orcatch (RuntimeException e)inprocessAlterMTMVfor this case, orRuntimeExceptionas well asUserExceptionat line 1336.