From 17fcc91154153ea38a87c2678559846b3b90365a Mon Sep 17 00:00:00 2001 From: LiBinfeng Date: Thu, 21 Nov 2024 14:11:46 +0800 Subject: [PATCH 1/3] support cancel commands --- .../org/apache/doris/nereids/DorisParser.g4 | 10 +- .../doris/analysis/CancelExportStmt.java | 9 ++ .../apache/doris/analysis/CancelLoadStmt.java | 9 ++ .../nereids/parser/LogicalPlanBuilder.java | 38 +++++ .../doris/nereids/trees/plans/PlanType.java | 3 + .../trees/plans/commands/CancelCommand.java | 152 ++++++++++++++++++ .../plans/commands/CancelExportCommand.java | 102 ++++++++++++ .../plans/commands/CancelLoadCommand.java | 107 ++++++++++++ .../commands/CancelWarmUpJobCommand.java | 116 +++++++++++++ .../trees/plans/visitor/CommandVisitor.java | 15 ++ 10 files changed, 558 insertions(+), 3 deletions(-) create mode 100644 fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelCommand.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java diff --git a/fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 b/fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 index d7ed0598d19ddb..c806beed9db04a 100644 --- a/fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 +++ b/fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 @@ -59,6 +59,7 @@ statementBase | supportedRefreshStatement #supportedRefreshStatementAlias | supportedShowStatement #supportedShowStatementAlias | supportedLoadStatement #supportedLoadStatementAlias + | supportedCancelStatement #supportedCancelStatementAlias | supportedRecoverStatement #supportedRecoverStatementAlias | unsupportedStatement #unsupported ; @@ -452,10 +453,14 @@ unsupportedCleanStatement | CLEAN ALL QUERY STATS #cleanAllQueryStats ; -unsupportedCancelStatement +supportedCancelStatement : CANCEL LOAD ((FROM | IN) database=identifier)? wildWhere? #cancelLoad | CANCEL EXPORT ((FROM | IN) database=identifier)? wildWhere? #cancelExport - | CANCEL ALTER TABLE (ROLLUP | (MATERIALIZED VIEW) | COLUMN) + | CANCEL WARM UP JOB wildWhere? #cancelWarmUpJob + ; + +unsupportedCancelStatement + : CANCEL ALTER TABLE (ROLLUP | (MATERIALIZED VIEW) | COLUMN) FROM tableName=multipartIdentifier (LEFT_PAREN jobIds+=INTEGER_VALUE (COMMA jobIds+=INTEGER_VALUE)* RIGHT_PAREN)? #cancelAlterTable | CANCEL BUILD INDEX ON tableName=multipartIdentifier @@ -465,7 +470,6 @@ unsupportedCancelStatement (COMMA hostPorts+=STRING_LITERAL)* #cancelDecommisionBackend | CANCEL BACKUP ((FROM | IN) database=identifier)? #cancelBackup | CANCEL RESTORE ((FROM | IN) database=identifier)? #cancelRestore - | CANCEL WARM UP JOB wildWhere? #cancelWarmUp ; supportedRecoverStatement diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CancelExportStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CancelExportStmt.java index 63448cf35a027a..fa6d7640781e17 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CancelExportStmt.java +++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CancelExportStmt.java @@ -58,6 +58,15 @@ public CancelExportStmt(String dbName, Expr whereClause) { this.whereClause = whereClause; } + public CancelExportStmt(String dbName, Expr whereClause, String label, CompoundPredicate.Operator operator, + String state) { + this.dbName = dbName; + this.whereClause = whereClause; + this.label = label; + this.operator = operator; + this.state = state; + } + private void checkColumn(Expr expr, boolean like) throws AnalysisException { String inputCol = ((SlotRef) expr.getChild(0)).getColumnName(); if (!SUPPORT_COLUMNS.contains(inputCol.toLowerCase())) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CancelLoadStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CancelLoadStmt.java index 1371e308675708..bbffb2649738fb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CancelLoadStmt.java +++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CancelLoadStmt.java @@ -59,6 +59,15 @@ public CancelLoadStmt(String dbName, Expr whereClause) { this.SUPPORT_COLUMNS.add("state"); } + public CancelLoadStmt(String dbName, Expr whereClause, String label, CompoundPredicate.Operator operator, + String state) { + this.dbName = dbName; + this.whereClause = whereClause; + this.label = label; + this.operator = operator; + this.state = state; + } + private void checkColumn(Expr expr, boolean like) throws AnalysisException { String inputCol = ((SlotRef) expr.getChild(0)).getColumnName(); if (!SUPPORT_COLUMNS.contains(inputCol)) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java index 6e86b4073f5792..268a4882f53620 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java @@ -448,8 +448,11 @@ import org.apache.doris.nereids.trees.plans.commands.AlterWorkloadGroupCommand; import org.apache.doris.nereids.trees.plans.commands.AlterWorkloadPolicyCommand; import org.apache.doris.nereids.trees.plans.commands.CallCommand; +import org.apache.doris.nereids.trees.plans.commands.CancelExportCommand; import org.apache.doris.nereids.trees.plans.commands.CancelJobTaskCommand; +import org.apache.doris.nereids.trees.plans.commands.CancelLoadCommand; import org.apache.doris.nereids.trees.plans.commands.CancelMTMVTaskCommand; +import org.apache.doris.nereids.trees.plans.commands.CancelWarmUpJobCommand; import org.apache.doris.nereids.trees.plans.commands.CleanAllProfileCommand; import org.apache.doris.nereids.trees.plans.commands.Command; import org.apache.doris.nereids.trees.plans.commands.Constraint; @@ -1073,6 +1076,41 @@ public ShowCreateMTMVCommand visitShowCreateMTMV(ShowCreateMTMVContext ctx) { return new ShowCreateMTMVCommand(new ShowCreateMTMVInfo(new TableNameInfo(nameParts))); } + @Override + public CancelExportCommand visitCancelExport(DorisParser.CancelExportContext ctx) { + String databaseName = null; + if (ctx.database != null) { + databaseName = stripQuotes(ctx.database.getText()); + } + Expression wildWhere = null; + if (ctx.wildWhere() != null) { + wildWhere = getWildWhere(ctx.wildWhere()); + } + return new CancelExportCommand(databaseName, wildWhere); + } + + @Override + public CancelLoadCommand visitCancelLoad(DorisParser.CancelLoadContext ctx) { + String databaseName = null; + if (ctx.database != null) { + databaseName = stripQuotes(ctx.database.getText()); + } + Expression wildWhere = null; + if (ctx.wildWhere() != null) { + wildWhere = getWildWhere(ctx.wildWhere()); + } + return new CancelLoadCommand(databaseName, wildWhere); + } + + @Override + public CancelWarmUpJobCommand visitCancelWarmUpJob(DorisParser.CancelWarmUpJobContext ctx) { + Expression wildWhere = null; + if (ctx.wildWhere() != null) { + wildWhere = getWildWhere(ctx.wildWhere()); + } + return new CancelWarmUpJobCommand(wildWhere); + } + @Override public CancelMTMVTaskCommand visitCancelMTMVTask(CancelMTMVTaskContext ctx) { List nameParts = visitMultipartIdentifier(ctx.mvName); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java index 57eb2bbb6b42c2..ac67900adc9fb3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java @@ -158,6 +158,9 @@ public enum PlanType { PAUSE_MTMV_COMMAND, RESUME_MTMV_COMMAND, SHOW_CREATE_MTMV_COMMAND, + CANCEL_EXPORT_COMMAND, + CANCEL_LOAD_COMMAND, + CANCEL_WARM_UP_JOB_COMMAND, CANCEL_MTMV_TASK_COMMAND, CALL_COMMAND, CREATE_PROCEDURE_COMMAND, diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelCommand.java new file mode 100644 index 00000000000000..33eda33a45b3a7 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelCommand.java @@ -0,0 +1,152 @@ +// 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; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.nereids.CascadesContext; +import org.apache.doris.nereids.analyzer.UnboundSlot; +import org.apache.doris.nereids.glue.translator.ExpressionTranslator; +import org.apache.doris.nereids.glue.translator.PlanTranslatorContext; +import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.BinaryOperator; +import org.apache.doris.nereids.trees.expressions.CompoundPredicate; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.Like; +import org.apache.doris.nereids.trees.expressions.Not; +import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +import com.google.common.base.Strings; + +import java.util.ArrayList; +import java.util.Map; + +/** + * cancel load command + */ +public abstract class CancelCommand extends Command implements ForwardWithSync { + public CancelCommand(PlanType type) { + super(type); + } + + @Override + public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + + } + + @Override + public R accept(PlanVisitor visitor, C context) { + return null; + } + + /** + * translate to legacy expr, which do not need complex expression and table columns + */ + public Expr translateToLegacyExpr(ConnectContext ctx, Expression expression) { + LogicalEmptyRelation plan = new LogicalEmptyRelation( + ConnectContext.get().getStatementContext().getNextRelationId(), + new ArrayList<>()); + CascadesContext cascadesContext = CascadesContext.initContext(ctx.getStatementContext(), plan, + PhysicalProperties.ANY); + PlanTranslatorContext planTranslatorContext = new PlanTranslatorContext(cascadesContext); + return ExpressionTranslator.translate(expression, planTranslatorContext); + } + + /** + * check where filter for cancel load/export commands + * @param expression where clause + * @param supportedColumns only these kind of columns is supported + * @throws AnalysisException analyze exceptions + */ + public void checkWhereFilter(Expression expression, Map supportedColumns) throws AnalysisException { + if (null == expression) { + throw new AnalysisException("Where clause can't be null"); + } else if (expression instanceof Like) { + likeCheck(expression, supportedColumns); + } else if (expression instanceof BinaryOperator) { + binaryCheck(expression, supportedColumns); + } else if (expression instanceof CompoundPredicate) { + compoundCheck(expression, supportedColumns); + } else { + throw new AnalysisException("Only support like/binary/compound predicate"); + } + } + + private void checkColumn(Expression expr, boolean like, Map supportedColumns) + throws AnalysisException { + if (!(expr.child(0) instanceof UnboundSlot)) { + throw new AnalysisException("Current only support label and state, invalid column: " + + expr.child(0).toSql()); + } + String inputCol = ((UnboundSlot) expr.child(0)).getName(); + if (!supportedColumns.keySet().contains(inputCol.toLowerCase())) { + throw new AnalysisException("Current only support label and state, invalid column: " + inputCol); + } + if (!(expr.child(1) instanceof StringLikeLiteral)) { + throw new AnalysisException("Value must be a string"); + } + + String inputValue = ((StringLikeLiteral) expr.child(1)).getStringValue(); + if (Strings.isNullOrEmpty(inputValue)) { + throw new AnalysisException("Value can't be null"); + } + + if (inputCol.equalsIgnoreCase("label")) { + supportedColumns.put("label", inputValue); + } + + if (inputCol.equalsIgnoreCase("state")) { + if (like) { + throw new AnalysisException("Only label can use like"); + } + supportedColumns.put("state", inputValue); + } + } + + private void likeCheck(Expression expr, Map supportedColumns) throws AnalysisException { + checkColumn(expr, true, supportedColumns); + } + + private void binaryCheck(Expression expr, Map supportedColumns) throws AnalysisException { + checkColumn(expr, false, supportedColumns); + } + + private void compoundCheck(Expression expr, Map supportedColumns) throws AnalysisException { + // current only support label and state + if (expr instanceof Not) { + throw new AnalysisException("Current not support NOT operator"); + } + for (int i = 0; i < 2; i++) { + Expression child = expr.child(i); + if (child instanceof CompoundPredicate) { + throw new AnalysisException("Current not support nested clause"); + } else if (child instanceof Like) { + likeCheck(child, supportedColumns); + } else if (child instanceof BinaryOperator) { + binaryCheck(child, supportedColumns); + } else { + throw new AnalysisException("Only support like/binary predicate"); + } + } + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java new file mode 100644 index 00000000000000..6f603043cc596e --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java @@ -0,0 +1,102 @@ +// 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; + +import org.apache.doris.analysis.CancelExportStmt; +import org.apache.doris.analysis.Expr; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.UserException; +import org.apache.doris.load.ExportJobState; +import org.apache.doris.nereids.trees.expressions.CompoundPredicate; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +import com.google.common.base.Strings; + +import java.util.HashMap; +import java.util.Map; + +/** + * cancel export command + */ +public class CancelExportCommand extends CancelCommand implements ForwardWithSync { + private Map supportedColumns = new HashMap<>(); + + private String dbName; + + private String label; + + private String state; + + private Expression whereClause; + + private Expr legacyWhereClause; + + public CancelExportCommand(String dbName, Expression whereClause) { + super(PlanType.CANCEL_EXPORT_COMMAND); + this.dbName = dbName; + this.whereClause = whereClause; + } + + @Override + public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + validate(ctx); + CancelExportStmt cancelStmt = null; + if (whereClause instanceof CompoundPredicate) { + cancelStmt = new CancelExportStmt(dbName, legacyWhereClause, label, null, state); + } else { + cancelStmt = new CancelExportStmt(dbName, legacyWhereClause, label, + ((org.apache.doris.analysis.CompoundPredicate) legacyWhereClause).getOp(), state); + } + ctx.getEnv().getExportMgr().cancelExportJob(cancelStmt); + } + + private void validate(ConnectContext ctx) throws UserException { + if (Strings.isNullOrEmpty(dbName)) { + dbName = ctx.getDatabase(); + if (Strings.isNullOrEmpty(dbName)) { + throw new AnalysisException("No database selected"); + } + } + + supportedColumns.put("label", ""); + supportedColumns.put("state", ""); + checkWhereFilter(whereClause, supportedColumns); + if (!Strings.isNullOrEmpty(supportedColumns.get("label"))) { + label = supportedColumns.get("label"); + } + if (!Strings.isNullOrEmpty(supportedColumns.get("state"))) { + state = supportedColumns.get("state"); + ExportJobState jobState = ExportJobState.valueOf(state); + if (jobState != ExportJobState.PENDING + && jobState != ExportJobState.EXPORTING) { + throw new AnalysisException("Only support PENDING/EXPORTING, invalid state: " + state); + } + } + + legacyWhereClause = translateToLegacyExpr(ctx, whereClause); + } + + @Override + public R accept(PlanVisitor visitor, C context) { + return visitor.visitCancelExportCommand(this, context); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java new file mode 100644 index 00000000000000..245eea131a1fed --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java @@ -0,0 +1,107 @@ +// 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; + +import org.apache.doris.analysis.CancelLoadStmt; +import org.apache.doris.analysis.Expr; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.UserException; +import org.apache.doris.job.exception.JobException; +import org.apache.doris.load.ExportJobState; +import org.apache.doris.nereids.trees.expressions.CompoundPredicate; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +import com.google.common.base.Strings; + +import java.util.HashMap; +import java.util.Map; + +/** + * cancel load command + */ +public class CancelLoadCommand extends CancelCommand implements ForwardWithSync { + private Map supportedColumns = new HashMap<>(); + + private String dbName; + + private String label; + + private String state; + + private Expression whereClause; + + private Expr legacyWhereClause; + + public CancelLoadCommand(String dbName, Expression whereClause) { + super(PlanType.CANCEL_EXPORT_COMMAND); + this.dbName = dbName; + this.whereClause = whereClause; + } + + @Override + public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + validate(ctx); + CancelLoadStmt cancelStmt = null; + if (whereClause instanceof CompoundPredicate) { + cancelStmt = new CancelLoadStmt(dbName, legacyWhereClause, label, null, state); + } else { + cancelStmt = new CancelLoadStmt(dbName, legacyWhereClause, label, + ((org.apache.doris.analysis.CompoundPredicate) legacyWhereClause).getOp(), state); + } + try { + ctx.getEnv().getJobManager().cancelLoadJob(cancelStmt); + } catch (JobException e) { + ctx.getEnv().getLoadManager().cancelLoadJob(cancelStmt); + } + } + + private void validate(ConnectContext ctx) throws UserException { + if (Strings.isNullOrEmpty(dbName)) { + dbName = ctx.getDatabase(); + if (Strings.isNullOrEmpty(dbName)) { + throw new AnalysisException("No database selected"); + } + } + + supportedColumns.put("label", ""); + supportedColumns.put("state", ""); + checkWhereFilter(whereClause, supportedColumns); + if (!Strings.isNullOrEmpty(supportedColumns.get("label"))) { + label = supportedColumns.get("label"); + } + if (!Strings.isNullOrEmpty(supportedColumns.get("state"))) { + state = supportedColumns.get("state"); + ExportJobState jobState = ExportJobState.valueOf(state); + if (jobState != ExportJobState.PENDING + && jobState != ExportJobState.EXPORTING) { + throw new AnalysisException("Only support PENDING/EXPORTING, invalid state: " + state); + } + } + + legacyWhereClause = translateToLegacyExpr(ctx, whereClause); + } + + @Override + public R accept(PlanVisitor visitor, C context) { + return visitor.visitCancelLoadCommand(this, context); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java new file mode 100644 index 00000000000000..b5d26b4c4213ff --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java @@ -0,0 +1,116 @@ +// 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; + +import org.apache.doris.analysis.CancelCloudWarmUpStmt; +import org.apache.doris.analysis.Expr; +import org.apache.doris.cloud.catalog.CloudEnv; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.Config; +import org.apache.doris.nereids.CascadesContext; +import org.apache.doris.nereids.analyzer.UnboundSlot; +import org.apache.doris.nereids.glue.translator.ExpressionTranslator; +import org.apache.doris.nereids.glue.translator.PlanTranslatorContext; +import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +import java.util.ArrayList; + +/** + * cancel warm up job command + */ +public class CancelWarmUpJobCommand extends Command implements ForwardWithSync { + private Expression whereClause; + private long jobId; + private Expr legacyWhereClause; + + public CancelWarmUpJobCommand(Expression whereClause) { + super(PlanType.CANCEL_WARM_UP_JOB_COMMAND); + this.whereClause = whereClause; + } + + @Override + public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + validate(ctx); + if (Config.isCloudMode()) { + CancelCloudWarmUpStmt stmt = new CancelCloudWarmUpStmt(legacyWhereClause); + ((CloudEnv) ctx.getEnv()).cancelCloudWarmUp(stmt); + } + } + + public long getJobId() { + return jobId; + } + + /** + * validate cloud warm up job + * @param ctx connect context + * @throws AnalysisException check whether this sql is legal + */ + public void validate(ConnectContext ctx) throws AnalysisException { + if (!Config.isCloudMode()) { + throw new AnalysisException("The sql is illegal in disk mode "); + } + if (whereClause == null) { + throw new AnalysisException("Missing job id"); + } + boolean valid = true; + CHECK: { + if (!(whereClause instanceof EqualTo)) { + valid = false; + break CHECK; + } + + // left child + if (!(whereClause.child(0) instanceof UnboundSlot)) { + valid = false; + break CHECK; + } + String leftKey = ((UnboundSlot) whereClause.child(0)).getName(); + if (leftKey.equalsIgnoreCase("id") && (whereClause.child(1) instanceof IntegerLiteral)) { + jobId = ((IntegerLiteral) whereClause.child(1)).getLongValue(); + } else { + valid = false; + } + } + + if (!valid) { + throw new AnalysisException("Where clause should looks like one of them: id = 123"); + } + + LogicalEmptyRelation plan = new LogicalEmptyRelation( + ConnectContext.get().getStatementContext().getNextRelationId(), + new ArrayList<>()); + CascadesContext cascadesContext = CascadesContext.initContext(ctx.getStatementContext(), plan, + PhysicalProperties.ANY); + PlanTranslatorContext planTranslatorContext = new PlanTranslatorContext(cascadesContext); + legacyWhereClause = ExpressionTranslator.translate(whereClause, planTranslatorContext); + } + + @Override + public R accept(PlanVisitor visitor, C context) { + return visitor.visitCancelWarmUpJobCommand(this, context); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/CommandVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/CommandVisitor.java index 85633cb798b06c..0bc6d9002ca9d3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/CommandVisitor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/CommandVisitor.java @@ -25,8 +25,11 @@ import org.apache.doris.nereids.trees.plans.commands.AlterWorkloadGroupCommand; import org.apache.doris.nereids.trees.plans.commands.AlterWorkloadPolicyCommand; import org.apache.doris.nereids.trees.plans.commands.CallCommand; +import org.apache.doris.nereids.trees.plans.commands.CancelExportCommand; import org.apache.doris.nereids.trees.plans.commands.CancelJobTaskCommand; +import org.apache.doris.nereids.trees.plans.commands.CancelLoadCommand; import org.apache.doris.nereids.trees.plans.commands.CancelMTMVTaskCommand; +import org.apache.doris.nereids.trees.plans.commands.CancelWarmUpJobCommand; import org.apache.doris.nereids.trees.plans.commands.CleanAllProfileCommand; import org.apache.doris.nereids.trees.plans.commands.Command; import org.apache.doris.nereids.trees.plans.commands.CreateJobCommand; @@ -221,6 +224,18 @@ default R visitShowCreateMTMVCommand(ShowCreateMTMVCommand showCreateMTMVCommand return visitCommand(showCreateMTMVCommand, context); } + default R visitCancelLoadCommand(CancelLoadCommand cancelLoadCommand, C context) { + return visitCommand(cancelLoadCommand, context); + } + + default R visitCancelExportCommand(CancelExportCommand cancelExportCommand, C context) { + return visitCommand(cancelExportCommand, context); + } + + default R visitCancelWarmUpJobCommand(CancelWarmUpJobCommand cancelWarmUpJobCommand, C context) { + return visitCommand(cancelWarmUpJobCommand, context); + } + default R visitCancelMTMVTaskCommand(CancelMTMVTaskCommand cancelMTMVTaskCommand, C context) { return visitCommand(cancelMTMVTaskCommand, context); } From e6e01aca8144eccacfbe6d0d17a980ab316f61b1 Mon Sep 17 00:00:00 2001 From: LiBinfeng Date: Tue, 26 Nov 2024 15:00:17 +0800 Subject: [PATCH 2/3] fix traslate bug --- .../nereids/trees/plans/commands/CancelCommand.java | 12 +++++++++++- .../trees/plans/commands/CancelExportCommand.java | 6 +++--- .../trees/plans/commands/CancelLoadCommand.java | 4 ++-- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelCommand.java index 33eda33a45b3a7..88ecf3cba1e6d8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelCommand.java @@ -18,6 +18,7 @@ package org.apache.doris.nereids.trees.plans.commands; import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.SlotRef; import org.apache.doris.common.AnalysisException; import org.apache.doris.nereids.CascadesContext; import org.apache.doris.nereids.analyzer.UnboundSlot; @@ -69,7 +70,16 @@ public Expr translateToLegacyExpr(ConnectContext ctx, Expression expression) { CascadesContext cascadesContext = CascadesContext.initContext(ctx.getStatementContext(), plan, PhysicalProperties.ANY); PlanTranslatorContext planTranslatorContext = new PlanTranslatorContext(cascadesContext); - return ExpressionTranslator.translate(expression, planTranslatorContext); + ExpressionToExpr translator = new ExpressionToExpr(); + return expression.accept(translator, planTranslatorContext); + } + + private static class ExpressionToExpr extends ExpressionTranslator { + @Override + public Expr visitUnboundSlot(UnboundSlot unboundSlot, PlanTranslatorContext context) { + String inputCol = unboundSlot.getName(); + return new SlotRef(null, inputCol); + } } /** diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java index 6f603043cc596e..785d3ffd7f7640 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java @@ -61,10 +61,10 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { validate(ctx); CancelExportStmt cancelStmt = null; if (whereClause instanceof CompoundPredicate) { - cancelStmt = new CancelExportStmt(dbName, legacyWhereClause, label, null, state); - } else { cancelStmt = new CancelExportStmt(dbName, legacyWhereClause, label, - ((org.apache.doris.analysis.CompoundPredicate) legacyWhereClause).getOp(), state); + ((org.apache.doris.analysis.CompoundPredicate) legacyWhereClause).getOp(), state); + } else { + cancelStmt = new CancelExportStmt(dbName, legacyWhereClause, label, null, state); } ctx.getEnv().getExportMgr().cancelExportJob(cancelStmt); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java index 245eea131a1fed..21d10e3ce56fef 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java @@ -62,10 +62,10 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { validate(ctx); CancelLoadStmt cancelStmt = null; if (whereClause instanceof CompoundPredicate) { - cancelStmt = new CancelLoadStmt(dbName, legacyWhereClause, label, null, state); - } else { cancelStmt = new CancelLoadStmt(dbName, legacyWhereClause, label, ((org.apache.doris.analysis.CompoundPredicate) legacyWhereClause).getOp(), state); + } else { + cancelStmt = new CancelLoadStmt(dbName, legacyWhereClause, label, null, state); } try { ctx.getEnv().getJobManager().cancelLoadJob(cancelStmt); From 0a7d7fa3af4164207a958626e25b93911f5c9ece Mon Sep 17 00:00:00 2001 From: LiBinfeng <1204975323@qq.com> Date: Wed, 27 Nov 2024 20:51:54 +0800 Subject: [PATCH 3/3] add interface of cancel load and export commands --- .../apache/doris/job/manager/JobManager.java | 89 ++++++++++++++++++ .../java/org/apache/doris/load/ExportMgr.java | 74 +++++++++++++++ .../apache/doris/load/loadv2/LoadManager.java | 94 +++++++++++++++++-- .../plans/commands/CancelExportCommand.java | 17 +--- .../plans/commands/CancelJobTaskCommand.java | 2 +- .../plans/commands/CancelLoadCommand.java | 19 +--- 6 files changed, 255 insertions(+), 40 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java b/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java index 47a3a0c5c19d5c..db2ca6eb81d156 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java @@ -41,6 +41,8 @@ import org.apache.doris.job.scheduler.JobScheduler; import org.apache.doris.load.loadv2.JobState; import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.BinaryOperator; import org.apache.doris.qe.ConnectContext; import com.google.common.collect.Lists; @@ -522,4 +524,91 @@ private static void addNeedCancelLoadJob(String label, String state, // job.updateLoadingStatus(beId, loadId, fragmentId, scannedRows, scannedBytes, isDone); // } // } + + /** + * used for nereids planner + */ + public void cancelLoadJob(String dbName, String label, String state, BinaryOperator operator) + throws JobException, AnalysisException, DdlException { + Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(dbName); + // List of load jobs waiting to be cancelled + List unfinishedLoadJob; + readLock(); + try { + List loadJobs = Env.getCurrentEnv().getLabelProcessor().getJobs(db); + List matchLoadJobs = Lists.newArrayList(); + addNeedCancelLoadJob(label, state, operator, loadJobs, matchLoadJobs); + if (matchLoadJobs.isEmpty()) { + throw new JobException("Load job does not exist"); + } + // check state here + unfinishedLoadJob = + matchLoadJobs.stream().filter(InsertJob::isRunning) + .collect(Collectors.toList()); + if (unfinishedLoadJob.isEmpty()) { + throw new JobException("There is no uncompleted job"); + } + } finally { + readUnlock(); + } + // check auth + if (unfinishedLoadJob.size() > 1 || unfinishedLoadJob.get(0).getTableNames().isEmpty()) { + if (Env.getCurrentEnv().getAccessManager() + .checkDbPriv(ConnectContext.get(), InternalCatalog.INTERNAL_CATALOG_NAME, dbName, + PrivPredicate.LOAD)) { + ErrorReport.reportAnalysisException(ErrorCode.ERR_DBACCESS_DENIED_ERROR, "LOAD", + ConnectContext.get().getQualifiedUser(), + ConnectContext.get().getRemoteIP(), dbName); + } + } else { + for (String tableName : unfinishedLoadJob.get(0).getTableNames()) { + if (Env.getCurrentEnv().getAccessManager() + .checkTblPriv(ConnectContext.get(), InternalCatalog.INTERNAL_CATALOG_NAME, dbName, + tableName, + PrivPredicate.LOAD)) { + ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "LOAD", + ConnectContext.get().getQualifiedUser(), + ConnectContext.get().getRemoteIP(), dbName + ":" + tableName); + } + } + } + for (InsertJob loadJob : unfinishedLoadJob) { + try { + alterJobStatus(loadJob.getJobId(), JobStatus.STOPPED); + } catch (JobException e) { + log.warn("Fail to cancel job, its label: {}", loadJob.getLabelName()); + } + } + } + + private static void addNeedCancelLoadJob(String label, String state, + BinaryOperator operator, List loadJobs, + List matchLoadJobs) + throws AnalysisException { + PatternMatcher matcher = PatternMatcherWrapper.createMysqlPattern(label, + CaseSensibility.LABEL.getCaseSensibility()); + matchLoadJobs.addAll( + loadJobs.stream() + .filter(job -> !job.isCancelled()) + .filter(job -> { + if (operator != null) { + // compound + boolean labelFilter = + label.contains("%") ? matcher.match(job.getLabelName()) + : job.getLabelName().equalsIgnoreCase(label); + boolean stateFilter = job.getJobStatus().name().equalsIgnoreCase(state); + return operator instanceof And ? labelFilter && stateFilter : + labelFilter || stateFilter; + } + if (StringUtils.isNotEmpty(label)) { + return label.contains("%") ? matcher.match(job.getLabelName()) + : job.getLabelName().equalsIgnoreCase(label); + } + if (StringUtils.isNotEmpty(state)) { + return job.getJobStatus().name().equalsIgnoreCase(state); + } + return false; + }).collect(Collectors.toList()) + ); + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java index 49ebbfe7dcddb0..398cc8f4ac2518 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java @@ -38,6 +38,8 @@ import org.apache.doris.common.util.TimeUtils; import org.apache.doris.datasource.InternalCatalog; import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.trees.expressions.BinaryOperator; +import org.apache.doris.nereids.trees.expressions.Or; import org.apache.doris.qe.ConnectContext; import org.apache.doris.scheduler.exception.JobException; @@ -160,6 +162,78 @@ public void cancelExportJob(CancelExportStmt stmt) throws DdlException, Analysis } } + private List getWaitingCancelJobs(String label, String state, BinaryOperator operator) + throws AnalysisException { + Predicate jobFilter = buildCancelJobFilter(label, state, operator); + readLock(); + try { + return getJobs().stream().filter(jobFilter).collect(Collectors.toList()); + } finally { + readUnlock(); + } + } + + @VisibleForTesting + public static Predicate buildCancelJobFilter(String label, String state, BinaryOperator operator) + throws AnalysisException { + PatternMatcher matcher = PatternMatcherWrapper.createMysqlPattern(label, + CaseSensibility.LABEL.getCaseSensibility()); + + return job -> { + boolean labelFilter = true; + boolean stateFilter = true; + if (StringUtils.isNotEmpty(label)) { + labelFilter = label.contains("%") ? matcher.match(job.getLabel()) : + job.getLabel().equalsIgnoreCase(label); + } + if (StringUtils.isNotEmpty(state)) { + stateFilter = job.getState().name().equalsIgnoreCase(state); + } + + if (operator != null && operator instanceof Or) { + return labelFilter || stateFilter; + } + + return labelFilter && stateFilter; + }; + } + + /** + * used for Nereids planner + */ + public void cancelExportJob(String label, String state, BinaryOperator operator, String dbName) + throws DdlException, AnalysisException { + // List of export jobs waiting to be cancelled + List matchExportJobs = getWaitingCancelJobs(label, state, operator); + if (matchExportJobs.isEmpty()) { + throw new DdlException("Export job(s) do not exist"); + } + matchExportJobs = matchExportJobs.stream() + .filter(job -> !job.isFinalState()).collect(Collectors.toList()); + if (matchExportJobs.isEmpty()) { + throw new DdlException("All export job(s) are at final state (CANCELLED/FINISHED)"); + } + + // check auth + checkCancelExportJobAuth(InternalCatalog.INTERNAL_CATALOG_NAME, dbName, matchExportJobs); + // Must add lock to protect export job. + // Because job may be cancelled when generating task executors, + // the cancel process may clear the task executor list at same time, + // which will cause ConcurrentModificationException + writeLock(); + try { + for (ExportJob exportJob : matchExportJobs) { + // exportJob.cancel(ExportFailMsg.CancelType.USER_CANCEL, "user cancel"); + exportJob.updateExportJobState(ExportJobState.CANCELLED, 0L, null, + ExportFailMsg.CancelType.USER_CANCEL, "user cancel"); + } + } catch (JobException e) { + throw new AnalysisException(e.getMessage()); + } finally { + writeUnlock(); + } + } + public void checkCancelExportJobAuth(String ctlName, String dbName, List jobs) throws AnalysisException { if (jobs.size() > 1) { if (!Env.getCurrentEnv().getAccessManager() diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java index 07c459d61cf056..5c39a27e24f828 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java @@ -46,6 +46,8 @@ import org.apache.doris.load.FailMsg.CancelType; import org.apache.doris.load.Load; import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.BinaryOperator; import org.apache.doris.persist.CleanLabelOperationLog; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.OriginStatement; @@ -246,23 +248,22 @@ public void recordFinishedLoadJob(String label, long transactionId, String dbNam * Match need cancel loadJob by stmt. **/ @VisibleForTesting - public static void addNeedCancelLoadJob(CancelLoadStmt stmt, List loadJobs, List matchLoadJobs) + public static void addNeedCancelLoadJob(String label, String state, BinaryOperator operator, + List loadJobs, List matchLoadJobs) throws AnalysisException { - String label = stmt.getLabel(); - String state = stmt.getState(); PatternMatcher matcher = PatternMatcherWrapper.createMysqlPattern(label, CaseSensibility.LABEL.getCaseSensibility()); matchLoadJobs.addAll( loadJobs.stream() .filter(job -> job.getState() != JobState.CANCELLED) .filter(job -> { - if (stmt.getOperator() != null) { + if (operator != null) { // compound boolean labelFilter = label.contains("%") ? matcher.match(job.getLabel()) : job.getLabel().equalsIgnoreCase(label); boolean stateFilter = job.getState().name().equalsIgnoreCase(state); - return Operator.AND.equals(stmt.getOperator()) ? labelFilter && stateFilter : + return operator instanceof And ? labelFilter && stateFilter : labelFilter || stateFilter; } if (StringUtils.isNotEmpty(label)) { @@ -280,8 +281,9 @@ public static void addNeedCancelLoadJob(CancelLoadStmt stmt, List loadJ /** * Cancel load job by stmt. **/ - public void cancelLoadJob(CancelLoadStmt stmt) throws DdlException, AnalysisException { - Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(stmt.getDbName()); + public void cancelLoadJob(String dbName, String label, String state, BinaryOperator operator) + throws DdlException, AnalysisException { + Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(dbName); // List of load jobs waiting to be cancelled List unfinishedLoadJob; readLock(); @@ -291,7 +293,7 @@ public void cancelLoadJob(CancelLoadStmt stmt) throws DdlException, AnalysisExce throw new DdlException("Load job does not exist"); } List matchLoadJobs = Lists.newArrayList(); - addNeedCancelLoadJob(stmt, + addNeedCancelLoadJob(label, state, operator, labelToLoadJobs.values().stream().flatMap(Collection::stream).collect(Collectors.toList()), matchLoadJobs); if (matchLoadJobs.isEmpty()) { @@ -318,6 +320,82 @@ public void cancelLoadJob(CancelLoadStmt stmt) throws DdlException, AnalysisExce } } + /** + * Match need cancel loadJob by stmt. + **/ + @VisibleForTesting + public static void addNeedCancelLoadJob(CancelLoadStmt stmt, List loadJobs, List matchLoadJobs) + throws AnalysisException { + String label = stmt.getLabel(); + String state = stmt.getState(); + PatternMatcher matcher = PatternMatcherWrapper.createMysqlPattern(label, + CaseSensibility.LABEL.getCaseSensibility()); + matchLoadJobs.addAll( + loadJobs.stream() + .filter(job -> job.getState() != JobState.CANCELLED) + .filter(job -> { + if (stmt.getOperator() != null) { + // compound + boolean labelFilter = + label.contains("%") ? matcher.match(job.getLabel()) + : job.getLabel().equalsIgnoreCase(label); + boolean stateFilter = job.getState().name().equalsIgnoreCase(state); + return Operator.AND.equals(stmt.getOperator()) ? labelFilter && stateFilter : + labelFilter || stateFilter; + } + if (StringUtils.isNotEmpty(label)) { + return label.contains("%") ? matcher.match(job.getLabel()) + : job.getLabel().equalsIgnoreCase(label); + } + if (StringUtils.isNotEmpty(state)) { + return job.getState().name().equalsIgnoreCase(state); + } + return false; + }).collect(Collectors.toList()) + ); + } + + /** + * Cancel load job by stmt. + **/ + public void cancelLoadJob(CancelLoadStmt stmt) throws DdlException, AnalysisException { + Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(stmt.getDbName()); + // List of load jobs waiting to be cancelled + List unfinishedLoadJob; + readLock(); + try { + Map> labelToLoadJobs = dbIdToLabelToLoadJobs.get(db.getId()); + if (labelToLoadJobs == null) { + throw new DdlException("Load job does not exist"); + } + List matchLoadJobs = Lists.newArrayList(); + addNeedCancelLoadJob(stmt, + labelToLoadJobs.values().stream().flatMap(Collection::stream).collect(Collectors.toList()), + matchLoadJobs); + if (matchLoadJobs.isEmpty()) { + throw new DdlException("Load job does not exist"); + } + // check state here + unfinishedLoadJob = + matchLoadJobs.stream().filter(entity -> !entity.isTxnDone()).collect(Collectors.toList()); + if (unfinishedLoadJob.isEmpty()) { + throw new DdlException("There is no uncompleted job"); + } + } finally { + readUnlock(); + } + for (LoadJob loadJob : unfinishedLoadJob) { + try { + loadJob.cancelJob(new FailMsg(FailMsg.CancelType.USER_CANCEL, "user cancel")); + } catch (DdlException e) { + throw new DdlException( + "Cancel load job [" + loadJob.getId() + "] fail, " + "label=[" + loadJob.getLabel() + + + "] failed msg=" + e.getMessage()); + } + } + } + /** * Replay end load job. **/ diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java index 785d3ffd7f7640..f7defefb422476 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelExportCommand.java @@ -17,12 +17,10 @@ package org.apache.doris.nereids.trees.plans.commands; -import org.apache.doris.analysis.CancelExportStmt; -import org.apache.doris.analysis.Expr; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.UserException; import org.apache.doris.load.ExportJobState; -import org.apache.doris.nereids.trees.expressions.CompoundPredicate; +import org.apache.doris.nereids.trees.expressions.BinaryOperator; import org.apache.doris.nereids.trees.expressions.Expression; import org.apache.doris.nereids.trees.plans.PlanType; import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; @@ -48,8 +46,6 @@ public class CancelExportCommand extends CancelCommand implements ForwardWithSyn private Expression whereClause; - private Expr legacyWhereClause; - public CancelExportCommand(String dbName, Expression whereClause) { super(PlanType.CANCEL_EXPORT_COMMAND); this.dbName = dbName; @@ -59,14 +55,7 @@ public CancelExportCommand(String dbName, Expression whereClause) { @Override public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { validate(ctx); - CancelExportStmt cancelStmt = null; - if (whereClause instanceof CompoundPredicate) { - cancelStmt = new CancelExportStmt(dbName, legacyWhereClause, label, - ((org.apache.doris.analysis.CompoundPredicate) legacyWhereClause).getOp(), state); - } else { - cancelStmt = new CancelExportStmt(dbName, legacyWhereClause, label, null, state); - } - ctx.getEnv().getExportMgr().cancelExportJob(cancelStmt); + ctx.getEnv().getExportMgr().cancelExportJob(label, state, (BinaryOperator) whereClause, dbName); } private void validate(ConnectContext ctx) throws UserException { @@ -91,8 +80,6 @@ private void validate(ConnectContext ctx) throws UserException { throw new AnalysisException("Only support PENDING/EXPORTING, invalid state: " + state); } } - - legacyWhereClause = translateToLegacyExpr(ctx, whereClause); } @Override diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelJobTaskCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelJobTaskCommand.java index a9ea241e3b63ec..e02c731911a18e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelJobTaskCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelJobTaskCommand.java @@ -36,7 +36,7 @@ /** * base class for all drop commands */ -public class CancelJobTaskCommand extends Command implements ForwardWithSync { +public class CancelJobTaskCommand extends CancelCommand implements ForwardWithSync { private static final String jobNameKey = "jobName"; private static final String taskIdKey = "taskId"; diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java index 21d10e3ce56fef..4d3068ed08f559 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelLoadCommand.java @@ -17,13 +17,11 @@ package org.apache.doris.nereids.trees.plans.commands; -import org.apache.doris.analysis.CancelLoadStmt; -import org.apache.doris.analysis.Expr; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.UserException; import org.apache.doris.job.exception.JobException; import org.apache.doris.load.ExportJobState; -import org.apache.doris.nereids.trees.expressions.CompoundPredicate; +import org.apache.doris.nereids.trees.expressions.BinaryOperator; import org.apache.doris.nereids.trees.expressions.Expression; import org.apache.doris.nereids.trees.plans.PlanType; import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; @@ -49,8 +47,6 @@ public class CancelLoadCommand extends CancelCommand implements ForwardWithSync private Expression whereClause; - private Expr legacyWhereClause; - public CancelLoadCommand(String dbName, Expression whereClause) { super(PlanType.CANCEL_EXPORT_COMMAND); this.dbName = dbName; @@ -60,17 +56,10 @@ public CancelLoadCommand(String dbName, Expression whereClause) { @Override public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { validate(ctx); - CancelLoadStmt cancelStmt = null; - if (whereClause instanceof CompoundPredicate) { - cancelStmt = new CancelLoadStmt(dbName, legacyWhereClause, label, - ((org.apache.doris.analysis.CompoundPredicate) legacyWhereClause).getOp(), state); - } else { - cancelStmt = new CancelLoadStmt(dbName, legacyWhereClause, label, null, state); - } try { - ctx.getEnv().getJobManager().cancelLoadJob(cancelStmt); + ctx.getEnv().getJobManager().cancelLoadJob(dbName, label, state, (BinaryOperator) whereClause); } catch (JobException e) { - ctx.getEnv().getLoadManager().cancelLoadJob(cancelStmt); + ctx.getEnv().getLoadManager().cancelLoadJob(dbName, label, state, (BinaryOperator) whereClause); } } @@ -96,8 +85,6 @@ private void validate(ConnectContext ctx) throws UserException { throw new AnalysisException("Only support PENDING/EXPORTING, invalid state: " + state); } } - - legacyWhereClause = translateToLegacyExpr(ctx, whereClause); } @Override