This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 5cf524587c6 branch-4.1: [fix](regression) Stop MTMV task waits from
latching onto the previous task #67710 (#67821)
5cf524587c6 is described below
commit 5cf524587c6dfd132ab995533546e8c189e601e8
Author: yujun <[email protected]>
AuthorDate: Sat Sep 12 00:55:26 2026 +0800
branch-4.1: [fix](regression) Stop MTMV task waits from latching onto the
previous task #67710 (#67821)
cherry-pick: #67710
---
.../org/apache/doris/regression/suite/Suite.groovy | 224 +++++++++++++--------
.../regression/suite/SuiteMTMVTaskWaitTest.groovy | 76 +++++++
2 files changed, 214 insertions(+), 86 deletions(-)
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
index cbae49d7b10..d894e961e85 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
@@ -113,6 +113,10 @@ class Suite implements GroovyInterceptable {
private AmazonS3 s3Client = null
private FileSystem fs = null
+ // MTMV task ids that a wait in this suite already returned as terminal. A
task that just
+ // finished is briefly absent from tasks(), so a later wait can fall back
to one of these.
+ private final Set<String> finishedMTMVTaskIds =
Collections.synchronizedSet(new HashSet<>())
+
Suite(String name, String group, SuiteContext context, SuiteCluster
cluster) {
this.name = name
this.group = group
@@ -2003,73 +2007,135 @@ class Suite implements GroovyInterceptable {
return debugPoint
}
- def waitingMTMVTaskFinishedByMvName = { mvName, dbName = context.dbName ->
- Thread.sleep(2000);
- String showTasks = "select
TaskId,JobId,JobName,MvId,Status,MvName,MvDatabaseName,ErrorMsg from
tasks('type'='mv') where MvDatabaseName = '${dbName}' and MvName = '${mvName}'
order by CreateTime ASC"
+ /**
+ * Poll a tasks('type'='mv') query until the newest row is terminal and
return that row.
+ * A task that just finished is briefly missing from tasks(): the job
removes it from its
+ * running list before the MV history gets it, so while that window is
open the newest row
+ * is the previous task of the same MV. A terminal row is therefore
trusted only when the
+ * same task was already seen running by this wait, or is seen terminal
twice in a row and
+ * no earlier wait returned it; the window lasts tens of milliseconds, far
less than the
+ * poll interval, so one extra poll is enough to tell the new task from
the previous one
+ * even when the previous task was not waited for before.
+ * poll() runs the query once, log() receives the progress messages, and
the two intervals
+ * exist so tests can shorten them. The last row seen is returned when the
timeout expires,
+ * so callers can report a status.
+ */
+ static List<Object> pollMTMVTaskTerminal(String showTasks, String caller,
Set<String> finishedTaskIds,
+ Closure<List<List<Object>>> poll, Closure<String> log, long
pollIntervalMs, long skipFinishedMs) {
String status = "NULL"
- List<List<Object>> result
- long startTime = System.currentTimeMillis()
- long timeoutTimestamp = startTime + 5 * 60 * 1000 // 5 min
- List<String> toCheckTaskRow = new ArrayList<>();
- while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL')) {
- result = sql(showTasks)
- logger.info("current db is " + dbName + ", showTasks is " +
result.toString())
+ String runningTaskId = null
+ String confirmedTaskId = null
+ String lastLoggedStatus = null
+ List<Object> taskRow = null
+ long skipFinishedDeadline = 0
+ long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 //
5 min
+ while (timeoutTimestamp > System.currentTimeMillis()) {
+ List<List<Object>> result = poll()
if (result.isEmpty()) {
- logger.info("waitingMTMVTaskFinishedByMvName toCheckTaskRow is
empty")
- Thread.sleep(1000);
+ if (lastLoggedStatus != "NULL") {
+ log("${caller} task row is empty")
+ lastLoggedStatus = "NULL"
+ }
+ confirmedTaskId = null
+ Thread.sleep(pollIntervalMs);
continue;
}
- toCheckTaskRow = result.last();
- status = toCheckTaskRow.get(4)
- logger.info("The state of ${showTasks} is ${status}")
- Thread.sleep(1000);
+ taskRow = result[0]
+ String taskId = taskRow.get(0).toString()
+ status = taskRow.get(1).toString()
+ if (lastLoggedStatus != status) {
+ log("The state of ${showTasks} is ${status}, taskId is
${taskId}")
+ lastLoggedStatus = status
+ }
+ if (status == 'PENDING' || status == 'RUNNING') {
+ runningTaskId = taskId
+ confirmedTaskId = null
+ Thread.sleep(pollIntervalMs);
+ continue;
+ }
+ if (finishedTaskIds.contains(taskId)) {
+ // A wait already returned this task, so it is the previous
task and the one
+ // being waited for is still in the gap. Skip it for as long
as any real gap
+ // could last; a caller that waits twice for the same task
(nothing new was
+ // submitted) then falls back to it instead of hitting the 5
minute timeout.
+ if (skipFinishedDeadline == 0) {
+ skipFinishedDeadline = System.currentTimeMillis() +
skipFinishedMs
+ }
+ if (System.currentTimeMillis() < skipFinishedDeadline) {
+ confirmedTaskId = null
+ Thread.sleep(pollIntervalMs);
+ continue;
+ }
+ }
+ if (taskId == runningTaskId || taskId == confirmedTaskId) {
+ finishedTaskIds.add(taskId)
+ return taskRow
+ }
+ // The first sighting of a terminal task may be the previous task
seen through the
+ // gap described above; confirm the same task once more before
trusting it.
+ confirmedTaskId = taskId
+ Thread.sleep(pollIntervalMs);
}
+ return taskRow
+ }
+
+ /**
+ * Wait for the newest MTMV task matching showTasks to reach a terminal
state and return
+ * its row. See pollMTMVTaskTerminal for why a terminal row needs to be
confirmed.
+ */
+ List<Object> waitMTMVTaskTerminal(String showTasks, String caller,
+ long pollIntervalMs = 500, long skipFinishedMs = 30 * 1000) {
+ return pollMTMVTaskTerminal(showTasks, caller, finishedMTMVTaskIds,
+ { -> sql(showTasks) }, { String message ->
logger.info(message) },
+ pollIntervalMs, skipFinishedMs)
+ }
+
+ def waitingMTMVTaskFinishedByMvName = { mvName, dbName = context.dbName ->
+ // Wait for the newly submitted MTMV task to become visible in tasks().
+ Thread.sleep(2000);
+ String showTasks = """
+ select TaskId, Status, MvName, MvDatabaseName from
tasks('type'='mv')
+ where MvDatabaseName = '${dbName}' and MvName = '${mvName}'
+ order by CreateTime DESC limit 1
+ """
+ List<Object> toCheckTaskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinishedByMvName")
+ String status = toCheckTaskRow == null ? "NULL" :
toCheckTaskRow.get(1).toString()
if (status != "SUCCESS") {
- logger.info("status is not success")
+ logger.info("status is ${status}")
}
Assert.assertEquals("SUCCESS", status)
def show_tables = sql """
- show tables from ${toCheckTaskRow.get(6)};
+ show tables from ${toCheckTaskRow.get(3)};
"""
- def db_id = getDbId(toCheckTaskRow.get(6))
- def table_id = getTableId(toCheckTaskRow.get(6), mvName)
+ def db_id = getDbId(toCheckTaskRow.get(3))
+ def table_id = getTableId(toCheckTaskRow.get(3), mvName)
logger.info("waitingMTMVTaskFinished analyze mv name is " + mvName
- + ", db name is " + toCheckTaskRow.get(6)
+ + ", db name is " + toCheckTaskRow.get(3)
+ ", show_tables are " + show_tables
+ ", db_id is " + db_id
+ ", table_id " + table_id)
- sql "analyze table ${toCheckTaskRow.get(6)}.${mvName} with sync;"
+ sql "analyze table ${toCheckTaskRow.get(3)}.${mvName} with sync;"
}
def waitingMTMVTaskFinishedByMvNameAllowCancel = {mvName, dbName =
context.dbName ->
+ // Wait for the newly submitted MTMV task to become visible in tasks().
Thread.sleep(2000);
- String showTasks = "select
TaskId,JobId,JobName,MvId,Status,MvName,MvDatabaseName,ErrorMsg from
tasks('type'='mv') where MvDatabaseName = '${dbName}' and MvName = '${mvName}'
order by CreateTime ASC"
-
- String status = "NULL"
- List<List<Object>> result
- long startTime = System.currentTimeMillis()
- long timeoutTimestamp = startTime + 5 * 60 * 1000 // 5 min
- List<String> toCheckTaskRow = new ArrayList<>();
- while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL' || status == 'CANCELED'))
{
- result = sql(showTasks)
- logger.info("current db is " + dbName + ", showTasks result: " +
result.toString())
- if (result.isEmpty()) {
- logger.info("waitingMTMVTaskFinishedByMvName toCheckTaskRow is
empty")
- Thread.sleep(1000);
- continue;
- }
- toCheckTaskRow = result.last()
- status = toCheckTaskRow.get(4)
- logger.info("The state of ${showTasks} is ${status}")
- Thread.sleep(1000);
- }
+ String showTasks = """
+ select TaskId, Status, MvName, MvDatabaseName, ErrorMsg from
tasks('type'='mv')
+ where MvDatabaseName = '${dbName}' and MvName = '${mvName}'
+ order by CreateTime DESC limit 1
+ """
+
+ List<Object> toCheckTaskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinishedByMvNameAllowCancel")
+ String status = toCheckTaskRow == null ? "NULL" :
toCheckTaskRow.get(1).toString()
if (status != "SUCCESS") {
logger.info("status is not success")
- assertTrue(result.toString().contains("same table"))
+ Assert.assertNotNull(toCheckTaskRow)
+ assertTrue(String.valueOf(toCheckTaskRow.get(4)).contains("same
table"))
}
// Need to analyze materialized view for cbo to choose the
materialized view accurately
- logger.info("waitingMTMVTaskFinished analyze mv name is " +
toCheckTaskRow.get(5))
- sql "analyze table ${toCheckTaskRow.get(6)}.${mvName} with sync;"
+ logger.info("waitingMTMVTaskFinished analyze mv name is " +
toCheckTaskRow.get(2))
+ sql "analyze table ${toCheckTaskRow.get(3)}.${mvName} with sync;"
}
void waitingMVTaskFinishedByMvName(String dbName, String tableName, String
indexName) {
@@ -2134,61 +2200,47 @@ class Suite implements GroovyInterceptable {
}
void waitingMTMVTaskFinished(String jobName) {
+ // Wait for the newly submitted MTMV task to become visible in tasks().
Thread.sleep(2000);
- String showTasks = "select
TaskId,JobId,JobName,MvId,Status,MvName,MvDatabaseName,ErrorMsg from
tasks('type'='mv') where JobName = '${jobName}' order by CreateTime ASC"
- String status = "NULL"
- List<List<Object>> result
- long startTime = System.currentTimeMillis()
- long timeoutTimestamp = startTime + 5 * 60 * 1000 // 5 min
- do {
- result = sql(showTasks)
- logger.info("result: " + result.toString())
- if (!result.isEmpty()) {
- status = result.last().get(4)
- }
- logger.info("The state of ${showTasks} is ${status}")
- Thread.sleep(1000);
- } while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL'))
+ String showTasks = """
+ select TaskId, Status, MvName, MvDatabaseName from
tasks('type'='mv')
+ where JobName = '${jobName}' order by CreateTime DESC limit 1
+ """
+ List<Object> taskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinished")
+ String status = taskRow == null ? "NULL" : taskRow.get(1).toString()
if (status != "SUCCESS") {
- logger.info("status is not success")
+ logger.info("status is ${status}")
}
Assert.assertEquals("SUCCESS", status)
// Need to analyze materialized view for cbo to choose the
materialized view accurately
def show_tables = sql """
- show tables from ${result.last().get(6)};
+ show tables from ${taskRow.get(3)};
"""
- def db_id = getDbId(result.last().get(6))
- def table_id = getTableId(result.last().get(6), result.last().get(5))
- logger.info("waitingMTMVTaskFinished analyze mv name is " +
result.last().get(5)
- + ", db name is " + result.last().get(6)
+ def db_id = getDbId(taskRow.get(3))
+ def table_id = getTableId(taskRow.get(3), taskRow.get(2))
+ logger.info("waitingMTMVTaskFinished analyze mv name is " +
taskRow.get(2)
+ + ", db name is " + taskRow.get(3)
+ ", show_tables are " + show_tables
+ ", db_id is " + db_id
+ ", table_id " + table_id)
- sql "analyze table ${result.last().get(6)}.${result.last().get(5)}
with sync;"
- String db = result.last().get(6)
- String table = result.last().get(5)
- result = sql("show table stats ${db}.${table}")
- logger.info("table stats: " + result.toString())
- result = sql("show index stats ${db}.${table} ${table}")
- logger.info("index stats: " + result.toString())
+ sql "analyze table ${taskRow.get(3)}.${taskRow.get(2)} with sync;"
+ String db = taskRow.get(3)
+ String table = taskRow.get(2)
+ def stats = sql("show table stats ${db}.${table}")
+ logger.info("table stats: " + stats.toString())
+ stats = sql("show index stats ${db}.${table} ${table}")
+ logger.info("index stats: " + stats.toString())
}
void waitingMTMVTaskFinishedNotNeedSuccess(String jobName) {
+ // Wait for the newly submitted MTMV task to become visible in tasks().
Thread.sleep(2000);
- String showTasks = "select
TaskId,JobId,JobName,MvId,Status,MvName,MvDatabaseName,ErrorMsg from
tasks('type'='mv') where JobName = '${jobName}' order by CreateTime ASC"
- String status = "NULL"
- List<List<Object>> result
- long startTime = System.currentTimeMillis()
- long timeoutTimestamp = startTime + 5 * 60 * 1000 // 5 min
- do {
- result = sql(showTasks)
- logger.info("result: " + result.toString())
- if (!result.isEmpty()) {
- status = result.last().get(4)
- }
- logger.info("The state of ${showTasks} is ${status}")
- Thread.sleep(1000);
- } while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL'))
+ String showTasks = """
+ select TaskId, Status from tasks('type'='mv')
+ where JobName = '${jobName}' order by CreateTime DESC limit 1
+ """
+ List<Object> taskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinishedNotNeedSuccess")
+ String status = taskRow == null ? "NULL" : taskRow.get(1).toString()
if (status != "SUCCESS") {
logger.info("status is not success")
}
diff --git
a/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SuiteMTMVTaskWaitTest.groovy
b/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SuiteMTMVTaskWaitTest.groovy
new file mode 100644
index 00000000000..39e5a7d8ea6
--- /dev/null
+++
b/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SuiteMTMVTaskWaitTest.groovy
@@ -0,0 +1,76 @@
+// 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.regression.suite
+
+import org.junit.jupiter.api.Test
+
+import static org.junit.jupiter.api.Assertions.assertEquals
+
+class SuiteMTMVTaskWaitTest {
+ private static final String SHOW_TASKS = "select TaskId, Status from
tasks('type'='mv')"
+ private static final Closure<String> IGNORE_LOG = { String message -> }
+
+ private static List<Object> task(String taskId, String status) {
+ return [taskId, status] as List<Object>
+ }
+
+ private static Closure<List<List<Object>>> polls(List<List<List<Object>>>
script) {
+ Deque<List<List<Object>>> queue = new ArrayDeque<>(script)
+ return { ->
+ if (queue.isEmpty()) {
+ throw new IllegalStateException("poll script exhausted, the
wait did not converge")
+ }
+ return queue.poll()
+ }
+ }
+
+ private static String waitFor(Set<String> finished,
List<List<List<Object>>> script) {
+ return Suite.pollMTMVTaskTerminal(SHOW_TASKS, "test", finished,
polls(script), IGNORE_LOG, 1, 5)
+ .get(0).toString()
+ }
+
+ @Test
+ void followsATaskThatWasSeenRunning() {
+ assertEquals("X", waitFor(new HashSet<>(), [[task("X", "RUNNING")],
[task("X", "SUCCESS")]]))
+ }
+
+ @Test
+ void confirmsATerminalRowThatWasNeverSeenRunning() {
+ assertEquals("X", waitFor(new HashSet<>(), [[task("X", "SUCCESS")],
[task("X", "SUCCESS")]]))
+ }
+
+ @Test
+ void ignoresThePreviousTaskWhileTheNewTaskIsMissing() {
+ Set<String> finished = new HashSet<>()
+ assertEquals("P", waitFor(finished, [[task("P", "RUNNING")],
[task("P", "SUCCESS")]]))
+
+ // X runs and finishes; while it moves from the running list to the MV
history the
+ // newest row of the MV is P again, and P must not be returned a
second time.
+ assertEquals("X", waitFor(finished, [[task("X", "RUNNING")],
[task("P", "SUCCESS")],
+ [task("X", "SUCCESS")]]))
+ }
+
+ @Test
+ void fallsBackToTheSameTaskWhenNothingNewIsSubmitted() {
+ Set<String> finished = new HashSet<>()
+ assertEquals("P", waitFor(finished, [[task("P", "RUNNING")],
[task("P", "SUCCESS")]]))
+
+ // Nothing new was submitted: the wait must return P instead of
polling to the timeout.
+ assertEquals("P", waitFor(finished, (1..100).collect { [task("P",
"SUCCESS")] }))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
