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 c6cde0c7148 branch-4.1: [fix](regression) Stabilize BE compaction
tasks test #65637 (#66543)
c6cde0c7148 is described below
commit c6cde0c7148bd8fee9c5f0f4c82edc682228b9ad
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Aug 9 07:26:07 2026 +0800
branch-4.1: [fix](regression) Stabilize BE compaction tasks test #65637
(#66543)
Cherry-picked from #65637
Co-authored-by: Jamie <[email protected]>
---
.../compaction/test_be_compaction_tasks.groovy | 93 ++++++++++++++--------
1 file changed, 61 insertions(+), 32 deletions(-)
diff --git a/regression-test/suites/compaction/test_be_compaction_tasks.groovy
b/regression-test/suites/compaction/test_be_compaction_tasks.groovy
index 69ec6a5cc6a..04334b31640 100644
--- a/regression-test/suites/compaction/test_be_compaction_tasks.groovy
+++ b/regression-test/suites/compaction/test_be_compaction_tasks.groovy
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-import org.codehaus.groovy.runtime.IOGroovyMethods
+import org.awaitility.Awaitility
+
+import java.util.concurrent.TimeUnit
suite("test_be_compaction_tasks", "p0") {
def tableName = "test_be_compaction_tasks_tbl"
@@ -25,8 +27,6 @@ suite("test_be_compaction_tasks", "p0") {
def backendId_to_backendHttpPort = [:]
getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort)
- String backend_id = backendId_to_backendIP.keySet()[0]
-
try {
// Step 2: Create table with disable_auto_compaction
sql """ DROP TABLE IF EXISTS ${tableName} """
@@ -72,35 +72,64 @@ suite("test_be_compaction_tasks", "p0") {
assertTrue(tablets.size() > 0)
def tablet = tablets[0]
String tablet_id = tablet.TabletId
+ String backend_id = tablet.BackendId
+ String backend_ip = backendId_to_backendIP[backend_id]
+ String backend_http_port = backendId_to_backendHttpPort[backend_id]
+ assertTrue(backend_ip != null && backend_http_port != null,
+ "Backend address should exist for backend " + backend_id)
+
+ // Step 6: A new partition's visible version reaches BE
asynchronously. Trigger tablet reports and
+ // retry only E-2000 until cumulative compaction has visible input
rowsets.
+ long lastReportTime = 0
+ Awaitility.await().atMost(120, TimeUnit.SECONDS).pollInterval(1,
TimeUnit.SECONDS).until {
+ def (code, out, err) = be_run_cumulative_compaction(backend_ip,
backend_http_port, tablet_id)
+ assertEquals(0, code, "Failed to trigger compaction:
stdout=${out}, stderr=${err}")
+ def triggerResult = parseJson(out.trim())
+ String triggerStatus = triggerResult.status.toString()
+ if (triggerStatus.equalsIgnoreCase("success")) {
+ return true
+ }
+ assertEquals("E-2000", triggerStatus,
+ "Unexpected compaction response: stdout=${out},
stderr=${err}")
+
+ long now = System.currentTimeMillis()
+ if (now - lastReportTime >= 5000) {
+ be_report_tablet(backend_ip, backend_http_port.toInteger())
+ lastReportTime = now
+ }
+ return false
+ }
- // Step 6: Trigger cumulative compaction via HTTP API and wait for
completion
- trigger_and_wait_compaction(tableName, "cumulative")
-
- // Step 7: After compaction - query system table with WHERE STATUS =
'FINISHED'
- def finishedResult = sql """ SELECT * FROM
information_schema.be_compaction_tasks WHERE STATUS = 'FINISHED' """
- logger.info("Finished compaction tasks: " + finishedResult.size())
- assertTrue(finishedResult.size() > 0, "Expected at least one FINISHED
compaction task")
-
- // Step 8: Field validation - verify key fields are non-null and
reasonable
- // Query specific fields for the completed compaction on our tablet
- def fieldResult = sql_return_maparray """
- SELECT BACKEND_ID, COMPACTION_ID, TABLE_ID, PARTITION_ID,
TABLET_ID,
- COMPACTION_TYPE, STATUS, TRIGGER_METHOD, COMPACTION_SCORE,
- SCHEDULED_TIME, START_TIME, END_TIME, ELAPSED_TIME_MS,
- INPUT_ROWSETS_COUNT, INPUT_ROW_NUM, INPUT_DATA_SIZE,
- INPUT_INDEX_SIZE, INPUT_TOTAL_SIZE, INPUT_SEGMENTS_NUM,
- INPUT_VERSION_RANGE,
- OUTPUT_ROW_NUM, OUTPUT_DATA_SIZE, OUTPUT_SEGMENTS_NUM,
- IS_VERTICAL
- FROM information_schema.be_compaction_tasks
- WHERE TABLET_ID = ${tablet_id} AND STATUS = 'FINISHED'
- ORDER BY COMPACTION_ID DESC
- LIMIT 1
- """
+ // Step 7: Wait for the tracker record itself to reach a terminal
state. The HTTP endpoint can return
+ // success after two seconds while compaction is still running.
+ def fieldResult = []
+ Awaitility.await().atMost(60, TimeUnit.SECONDS).pollInterval(1,
TimeUnit.SECONDS).until {
+ fieldResult = sql_return_maparray """
+ SELECT BACKEND_ID, COMPACTION_ID, TABLE_ID, PARTITION_ID,
TABLET_ID,
+ COMPACTION_TYPE, STATUS, TRIGGER_METHOD,
COMPACTION_SCORE,
+ SCHEDULED_TIME, START_TIME, END_TIME, ELAPSED_TIME_MS,
+ INPUT_ROWSETS_COUNT, INPUT_ROW_NUM, INPUT_DATA_SIZE,
+ INPUT_INDEX_SIZE, INPUT_TOTAL_SIZE, INPUT_SEGMENTS_NUM,
+ INPUT_VERSION_RANGE,
+ OUTPUT_ROW_NUM, OUTPUT_DATA_SIZE, OUTPUT_SEGMENTS_NUM,
+ IS_VERTICAL, STATUS_MSG
+ FROM information_schema.be_compaction_tasks
+ WHERE TABLET_ID = ${tablet_id}
+ AND COMPACTION_TYPE = 'cumulative'
+ AND TRIGGER_METHOD = 'MANUAL'
+ ORDER BY COMPACTION_ID DESC
+ LIMIT 1
+ """
+ return fieldResult.size() == 1 &&
+ (fieldResult[0].STATUS.toString() in ["FINISHED",
"FAILED"])
+ }
logger.info("Field validation result: " + fieldResult)
- assertTrue(fieldResult.size() > 0, "Expected FINISHED record for
tablet " + tablet_id)
def record = fieldResult[0]
+ assertEquals("FINISHED", record.STATUS.toString(),
+ "Compaction failed for tablet ${tablet_id}:
${record.STATUS_MSG}")
+
+ // Verify key fields are non-null and reasonable
// Verify key fields are non-null
assertTrue(record.BACKEND_ID != null &&
Long.parseLong(record.BACKEND_ID.toString()) > 0,
"BACKEND_ID should be positive")
@@ -140,11 +169,11 @@ suite("test_be_compaction_tasks", "p0") {
assertTrue(record.COMPACTION_SCORE != null &&
Long.parseLong(record.COMPACTION_SCORE.toString()) >= 0,
"COMPACTION_SCORE should be non-negative")
- // Step 9: TRIGGER_METHOD check - manual triggered should show 'MANUAL'
+ // Step 8: TRIGGER_METHOD check - manual triggered should show 'MANUAL'
assertTrue(record.TRIGGER_METHOD != null &&
record.TRIGGER_METHOD.toString() == "MANUAL",
"TRIGGER_METHOD should be MANUAL for manually triggered
compaction, got: " + record.TRIGGER_METHOD)
- // Step 10: Filter test - WHERE tablet_id = X AND status = 'FINISHED'
+ // Step 9: Filter test - WHERE tablet_id = X AND status = 'FINISHED'
def filterResult = sql """
SELECT COUNT(*) FROM information_schema.be_compaction_tasks
WHERE TABLET_ID = ${tablet_id} AND STATUS = 'FINISHED'
@@ -152,7 +181,7 @@ suite("test_be_compaction_tasks", "p0") {
logger.info("Filter result (tablet_id + status): " + filterResult)
assertTrue(filterResult[0][0] > 0, "Expected at least one record
matching filter")
- // Step 11: Non-existent tablet - WHERE tablet_id = 999999999 returns
empty
+ // Step 10: Non-existent tablet - WHERE tablet_id = 999999999 returns
empty
def emptyResult = sql """
SELECT * FROM information_schema.be_compaction_tasks
WHERE TABLET_ID = 999999999
@@ -161,7 +190,7 @@ suite("test_be_compaction_tasks", "p0") {
assertEquals(0, emptyResult.size())
} finally {
- // Step 12: Cleanup
+ // Step 11: Cleanup
try_sql("DROP TABLE IF EXISTS ${tableName} FORCE")
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
