This is an automated email from the ASF dual-hosted git repository.
starocean999 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8deb374deb1 [Chore](nereids) remove StopRoutineLoadStmt (#51830)
8deb374deb1 is described below
commit 8deb374deb11f7dd3ac4afcb9efcf595ad6108d6
Author: yaoxiao <[email protected]>
AuthorDate: Thu Jun 19 11:44:30 2025 +0800
[Chore](nereids) remove StopRoutineLoadStmt (#51830)
---
fe/fe-core/src/main/cup/sql_parser.cup | 7 ---
.../apache/doris/analysis/StopRoutineLoadStmt.java | 55 ----------------------
.../doris/load/routineload/RoutineLoadManager.java | 28 -----------
.../main/java/org/apache/doris/qe/DdlExecutor.java | 3 --
.../load/routineload/RoutineLoadManagerTest.java | 24 +++++-----
5 files changed, 12 insertions(+), 105 deletions(-)
diff --git a/fe/fe-core/src/main/cup/sql_parser.cup
b/fe/fe-core/src/main/cup/sql_parser.cup
index 5f3a598b968..a40c86603f7 100644
--- a/fe/fe-core/src/main/cup/sql_parser.cup
+++ b/fe/fe-core/src/main/cup/sql_parser.cup
@@ -2806,13 +2806,6 @@ resume_routine_load_stmt ::=
:}
;
-stop_routine_load_stmt ::=
- KW_STOP KW_ROUTINE KW_LOAD KW_FOR job_label:jobLabel
- {:
- RESULT = new StopRoutineLoadStmt(jobLabel);
- :}
- ;
-
show_routine_load_stmt ::=
KW_SHOW KW_ROUTINE KW_LOAD KW_FOR job_label:jobLabel
{:
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/analysis/StopRoutineLoadStmt.java
b/fe/fe-core/src/main/java/org/apache/doris/analysis/StopRoutineLoadStmt.java
deleted file mode 100644
index b0a340c625c..00000000000
---
a/fe/fe-core/src/main/java/org/apache/doris/analysis/StopRoutineLoadStmt.java
+++ /dev/null
@@ -1,55 +0,0 @@
-// 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.analysis;
-
-import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.UserException;
-
-/*
- Stop routine load job by name
-
- syntax:
- STOP ROUTINE LOAD [database.]name
- */
-public class StopRoutineLoadStmt extends DdlStmt implements
NotFallbackInParser {
-
- private final LabelName labelName;
-
- public StopRoutineLoadStmt(LabelName labelName) {
- this.labelName = labelName;
- }
-
- public String getName() {
- return labelName.getLabelName();
- }
-
- public String getDbFullName() {
- return labelName.getDbName();
- }
-
- @Override
- public void analyze(Analyzer analyzer) throws AnalysisException,
UserException {
- super.analyze(analyzer);
- labelName.analyze(analyzer);
- }
-
- @Override
- public StmtType stmtType() {
- return StmtType.STOP;
- }
-}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
index d0237b981ec..66beb7fd651 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
@@ -21,7 +21,6 @@ import org.apache.doris.analysis.AlterRoutineLoadStmt;
import org.apache.doris.analysis.CreateRoutineLoadStmt;
import org.apache.doris.analysis.PauseRoutineLoadStmt;
import org.apache.doris.analysis.ResumeRoutineLoadStmt;
-import org.apache.doris.analysis.StopRoutineLoadStmt;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
@@ -509,33 +508,6 @@ public class RoutineLoadManager implements Writable {
}
}
- public void stopRoutineLoadJob(StopRoutineLoadStmt stopRoutineLoadStmt)
- throws UserException {
- RoutineLoadJob routineLoadJob;
- // it needs lock when getting routine load job,
- // otherwise, it may cause the editLog out of order in the following
scenarios:
- // thread A: create job and record job meta
- // thread B: change job state and persist in editlog according to meta
- // thread A: persist in editlog
- // which will cause the null pointer exception when replaying editLog
- readLock();
- try {
- routineLoadJob =
checkPrivAndGetJob(stopRoutineLoadStmt.getDbFullName(),
- stopRoutineLoadStmt.getName());
- } finally {
- readUnlock();
- }
- routineLoadJob.updateState(RoutineLoadJob.JobState.STOPPED,
- new ErrorReason(InternalErrorCode.MANUAL_STOP_ERR,
- "User " + ConnectContext.get().getQualifiedUser() + "
stop routine load job"),
- false /* not replay */);
- LOG.info(new LogBuilder(LogKey.ROUTINE_LOAD_JOB,
routineLoadJob.getId())
- .add("current_state", routineLoadJob.getState())
- .add("user", ConnectContext.get().getQualifiedUser())
- .add("msg", "routine load job has been stopped by user")
- .build());
- }
-
public int getSizeOfIdToRoutineLoadTask() {
int sizeOfTasks = 0;
for (RoutineLoadJob routineLoadJob : idToRoutineLoadJob.values()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java
index 9828513d73f..a78f516745a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java
@@ -108,7 +108,6 @@ import org.apache.doris.analysis.RestoreStmt;
import org.apache.doris.analysis.ResumeRoutineLoadStmt;
import org.apache.doris.analysis.SetDefaultStorageVaultStmt;
import org.apache.doris.analysis.SetUserPropertyStmt;
-import org.apache.doris.analysis.StopRoutineLoadStmt;
import org.apache.doris.analysis.SyncStmt;
import org.apache.doris.analysis.TruncateTableStmt;
import org.apache.doris.analysis.UninstallPluginStmt;
@@ -189,8 +188,6 @@ public class DdlExecutor {
env.getRoutineLoadManager().pauseRoutineLoadJob((PauseRoutineLoadStmt) ddlStmt);
} else if (ddlStmt instanceof ResumeRoutineLoadStmt) {
env.getRoutineLoadManager().resumeRoutineLoadJob((ResumeRoutineLoadStmt)
ddlStmt);
- } else if (ddlStmt instanceof StopRoutineLoadStmt) {
-
env.getRoutineLoadManager().stopRoutineLoadJob((StopRoutineLoadStmt) ddlStmt);
} else if (ddlStmt instanceof AlterRoutineLoadStmt) {
env.getRoutineLoadManager().alterRoutineLoadJob((AlterRoutineLoadStmt) ddlStmt);
} else if (ddlStmt instanceof CreateJobStmt) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadManagerTest.java
index d706c52557e..f80a46b7cbc 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadManagerTest.java
@@ -23,7 +23,6 @@ import org.apache.doris.analysis.ParseNode;
import org.apache.doris.analysis.PauseRoutineLoadStmt;
import org.apache.doris.analysis.ResumeRoutineLoadStmt;
import org.apache.doris.analysis.Separator;
-import org.apache.doris.analysis.StopRoutineLoadStmt;
import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
@@ -43,6 +42,7 @@ import org.apache.doris.load.loadv2.LoadTask;
import org.apache.doris.load.routineload.kafka.KafkaConfiguration;
import org.apache.doris.mysql.privilege.AccessControllerManager;
import org.apache.doris.mysql.privilege.PrivPredicate;
+import
org.apache.doris.nereids.trees.plans.commands.load.StopRoutineLoadCommand;
import org.apache.doris.persist.EditLog;
import org.apache.doris.persist.RoutineLoadOperation;
import org.apache.doris.qe.ConnectContext;
@@ -707,10 +707,10 @@ public class RoutineLoadManagerTest {
}
@Test
- public void testStopRoutineLoadJob(@Injectable StopRoutineLoadStmt
stopRoutineLoadStmt, @Mocked Env env,
- @Mocked InternalCatalog catalog, @Mocked Database database,
@Mocked Table tbl,
- @Mocked AccessControllerManager accessManager,
- @Mocked ConnectContext connectContext) throws UserException {
+ public void testStopRoutineLoadJob(@Injectable StopRoutineLoadCommand
stopRoutineLoadCommand, @Mocked Env env,
+ @Mocked InternalCatalog catalog,
@Mocked Database database, @Mocked Table tbl,
+ @Mocked AccessControllerManager
accessManager,
+ @Mocked ConnectContext connectContext)
throws UserException {
RoutineLoadManager routineLoadManager = new RoutineLoadManager();
Map<Long, Map<String, List<RoutineLoadJob>>> dbToNameToRoutineLoadJob
= Maps.newHashMap();
Map<String, List<RoutineLoadJob>> nameToRoutineLoadJob =
Maps.newHashMap();
@@ -723,10 +723,10 @@ public class RoutineLoadManagerTest {
new Expectations() {
{
- stopRoutineLoadStmt.getDbFullName();
+ stopRoutineLoadCommand.getDbFullName();
minTimes = 0;
result = "";
- stopRoutineLoadStmt.getName();
+ stopRoutineLoadCommand.getLabel();
minTimes = 0;
result = "";
env.getInternalCatalog();
@@ -753,7 +753,7 @@ public class RoutineLoadManagerTest {
}
};
- routineLoadManager.stopRoutineLoadJob(stopRoutineLoadStmt);
+ routineLoadManager.stopRoutineLoadJob(stopRoutineLoadCommand);
Assert.assertEquals(RoutineLoadJob.JobState.STOPPED,
routineLoadJob.getState());
}
@@ -952,7 +952,7 @@ public class RoutineLoadManagerTest {
}
@Test
- public void testAlterRoutineLoadJob(@Injectable StopRoutineLoadStmt
stopRoutineLoadStmt, @Mocked Env env,
+ public void testAlterRoutineLoadJob(@Injectable StopRoutineLoadCommand
stopRoutineLoadCommand, @Mocked Env env,
@Mocked InternalCatalog catalog, @Mocked Database database,
@Mocked Table tbl,
@Mocked AccessControllerManager accessManager,
@Mocked ConnectContext connectContext) throws UserException {
@@ -968,10 +968,10 @@ public class RoutineLoadManagerTest {
new Expectations() {
{
- stopRoutineLoadStmt.getDbFullName();
+ stopRoutineLoadCommand.getDbFullName();
minTimes = 0;
result = "";
- stopRoutineLoadStmt.getName();
+ stopRoutineLoadCommand.getLabel();
minTimes = 0;
result = "";
env.getInternalCatalog();
@@ -998,7 +998,7 @@ public class RoutineLoadManagerTest {
}
};
- routineLoadManager.stopRoutineLoadJob(stopRoutineLoadStmt);
+ routineLoadManager.stopRoutineLoadJob(stopRoutineLoadCommand);
Assert.assertEquals(RoutineLoadJob.JobState.STOPPED,
routineLoadJob.getState());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
