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 31936841ef5 branch-4.1: [fix](regression) Stabilize compaction
regression cases #67613 (#67691)
31936841ef5 is described below
commit 31936841ef56fce00982980a124dbf4396cb1479
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Sep 9 22:54:17 2026 +0800
branch-4.1: [fix](regression) Stabilize compaction regression cases #67613
(#67691)
Cherry-picked from #67613
Co-authored-by: Jamie <[email protected]>
---
.../test_schema_change_with_compaction1.groovy | 70 ++++++--
.../test_schema_change_with_compaction10.groovy | 94 ++++++++---
.../test_schema_change_with_compaction5.groovy | 152 +++++++----------
.../test_schema_change_with_compaction6.groovy | 179 +++++++++------------
.../test_schema_change_with_compaction7.groovy | 150 +++++++----------
.../test_schema_change_with_compaction9.groovy | 151 +++++++----------
...est_cumulative_compaction_with_format_v2.groovy | 55 +++++--
7 files changed, 413 insertions(+), 438 deletions(-)
diff --git
a/regression-test/suites/cloud_p1/schema_change/compaction1/test_schema_change_with_compaction1.groovy
b/regression-test/suites/cloud_p1/schema_change/compaction1/test_schema_change_with_compaction1.groovy
index 58b74592049..998eb2bf3ca 100644
---
a/regression-test/suites/cloud_p1/schema_change/compaction1/test_schema_change_with_compaction1.groovy
+++
b/regression-test/suites/cloud_p1/schema_change/compaction1/test_schema_change_with_compaction1.groovy
@@ -87,6 +87,39 @@ suite('test_schema_change_with_compaction1',
'p1,nonConcurrent') {
trigger_and_wait_compaction("date", "cumulative")
}
+ def triggerAndWaitCumulativeCompaction = { tabletId, latestVersionRange,
expectedVersionRange ->
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host, injectBe.HttpPort,
tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ return tabletStatus.rowsets.any { it.contains(latestVersionRange) }
+ }
+
+ logger.info("run compaction:" + tabletId)
+ def (triggerCode, triggerOut, triggerErr) =
+ be_run_cumulative_compaction(injectBe.Host, injectBe.HttpPort,
tabletId)
+ logger.info("Run compaction: code=" + triggerCode + ", out=" +
triggerOut + ", err=" + triggerErr)
+ assertEquals(0, triggerCode, "Failed to trigger cumulative compaction:
${triggerErr}")
+ def triggerResult = parseJson(triggerOut.trim())
+ assertEquals("success", triggerResult.status.toString().toLowerCase(),
+ "Unexpected cumulative compaction response: ${triggerOut}")
+
+ def tabletRowsets = []
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host, injectBe.HttpPort,
tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ tabletRowsets = tabletStatus.rowsets
+ return tabletRowsets.any { it.contains(expectedVersionRange) }
+ }
+ return tabletRowsets
+ }
+
+ def newTabletId = null
try {
load_delete_compaction()
load_delete_compaction()
@@ -101,23 +134,28 @@ suite('test_schema_change_with_compaction1',
'p1,nonConcurrent') {
sleep(5000)
array = sql_return_maparray("SHOW TABLETS FROM date")
- for (int i = 0; i < 5; i++) {
+ // NOTREADY tablets keep the latest 10 versions unmerged. Create enough
+ // double-write rowsets for older versions to remain eligible for
compaction.
+ for (int i = 0; i < 16; i++) {
load_date_once("date");
}
// base compaction
trigger_and_wait_compaction("date", "base")
- def newTabletId = array[1].TabletId
+ newTabletId = array[1].TabletId
logger.info("run compaction:" + newTabletId)
def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
logger.info("Run compaction: code=" + code + ", out=" + out + ", err="
+ err)
assertTrue(out.contains("invalid tablet state."))
- // cu compaction
- trigger_and_wait_compaction("date", "cumulative")
- } catch (Exception e) {
- logger.error("Exception: " + e)
+ triggerAndWaitCumulativeCompaction(originTabletId, "[24-24]", "[9-24]")
+ def notReadyTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[24-24]",
"[9-14]")
+ assertEquals("RUNNING", getJobState("date"))
+ for (int version = 15; version <= 24; version++) {
+ assertTrue(notReadyTabletRowsets.any {
it.contains("[${version}-${version}]") })
+ }
} finally {
if (injectBe != null) {
DebugPoint.disableDebugPoint(injectBe.Host,
injectBe.HttpPort.toInteger(), NodeType.BE, injectName)
@@ -138,7 +176,7 @@ suite('test_schema_change_with_compaction1',
'p1,nonConcurrent') {
}
assertEquals(result, "FINISHED");
def count = sql """ select count(*) from date; """
- assertEquals(count[0][0], 23004);
+ assertEquals(count[0][0], 51120);
// check rowsets
logger.info("run show:" + originTabletId)
def (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
@@ -146,7 +184,7 @@ suite('test_schema_change_with_compaction1',
'p1,nonConcurrent') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-24]"))
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -155,7 +193,7 @@ suite('test_schema_change_with_compaction1',
'p1,nonConcurrent') {
assertTrue(out.contains("[2-2]"))
assertTrue(out.contains("[7-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
// base compaction
trigger_and_wait_compaction("date", "base")
@@ -165,7 +203,7 @@ suite('test_schema_change_with_compaction1',
'p1,nonConcurrent') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
for (int i = 0; i < 3; i++) {
load_date_once("date");
@@ -173,13 +211,11 @@ suite('test_schema_change_with_compaction1',
'p1,nonConcurrent') {
sql """ select count(*) from date """
- trigger_and_wait_compaction("date", "cumulative")
- logger.info("run show:" + newTabletId)
- (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run show: code=" + code + ", out=" + out + ", err=" + err)
- assertTrue(out.contains("[0-1]"))
- assertTrue(out.contains("[2-7]"))
- assertTrue(out.contains("[8-16]"))
+ def finalTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[27-27]",
"[8-27]")
+ assertTrue(finalTabletRowsets.any { it.contains("[0-1]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[2-7]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[8-27]") })
}
}
diff --git
a/regression-test/suites/cloud_p1/schema_change/compaction10/test_schema_change_with_compaction10.groovy
b/regression-test/suites/cloud_p1/schema_change/compaction10/test_schema_change_with_compaction10.groovy
index 7e3d9d233cb..63d77e5ef71 100644
---
a/regression-test/suites/cloud_p1/schema_change/compaction10/test_schema_change_with_compaction10.groovy
+++
b/regression-test/suites/cloud_p1/schema_change/compaction10/test_schema_change_with_compaction10.groovy
@@ -89,6 +89,60 @@ suite('test_schema_change_with_compaction10', 'docker') {
trigger_and_wait_compaction("date", "cumulative")
}
+ def triggerAndWaitCumulativeCompaction = { tabletId,
latestVersionRange, expectedVersionRange ->
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ return tabletStatus.rowsets.any {
it.contains(latestVersionRange) }
+ }
+
+ logger.info("run compaction:" + tabletId)
+ def (triggerCode, triggerOut, triggerErr) =
+ be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, tabletId)
+ logger.info("Run compaction: code=" + triggerCode + ", out=" +
triggerOut + ", err=" + triggerErr)
+ assertEquals(0, triggerCode, "Failed to trigger cumulative
compaction: ${triggerErr}")
+ def triggerResult = parseJson(triggerOut.trim())
+ assertEquals("success",
triggerResult.status.toString().toLowerCase(),
+ "Unexpected cumulative compaction response: ${triggerOut}")
+
+ def tabletRowsets = []
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ tabletRowsets = tabletStatus.rowsets
+ return tabletRowsets.any { it.contains(expectedVersionRange) }
+ }
+ return tabletRowsets
+ }
+
+ def restartBackendAndRearmDebugPoint = {
+ cluster.stopBackends()
+ def rearmFuture = thread {
+ long deadline = System.currentTimeMillis() + 120000L
+ Exception lastError = null
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ DebugPoint.enableDebugPoint(injectBe.Host,
injectBe.HttpPort as int,
+ NodeType.BE, injectName)
+ return
+ } catch (Exception e) {
+ lastError = e
+ sleep(50)
+ }
+ }
+ throw new IllegalStateException("Failed to re-enable
${injectName} after BE restart", lastError)
+ }
+ cluster.startBackends()
+ rearmFuture.get()
+ }
+
+ def newTabletId = null
try {
load_delete_compaction()
load_delete_compaction()
@@ -102,26 +156,32 @@ suite('test_schema_change_with_compaction10', 'docker') {
sleep(5000)
array = sql_return_maparray("SHOW TABLETS FROM date")
- for (int i = 0; i < 5; i++) {
+ // NOTREADY tablets keep the latest 10 versions unmerged. Create
enough
+ // double-write rowsets for older versions to remain eligible for
compaction.
+ for (int i = 0; i < 16; i++) {
load_date_once("date");
}
- cluster.restartBackends()
- GetDebugPoint().enableDebugPointForAllBEs(injectName)
+ restartBackendAndRearmDebugPoint()
sleep(30000)
+ assertEquals("RUNNING", getJobState("date"),
+ "Schema change finished before the debug point was
re-enabled")
// base compaction
trigger_and_wait_compaction("date", "base")
- def newTabletId = array[1].TabletId
+ newTabletId = array[1].TabletId
logger.info("run compaction:" + newTabletId)
def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
assertTrue(out.contains("invalid tablet state."))
- // cu compaction
- trigger_and_wait_compaction("date", "cumulative")
- } catch (Exception e) {
- logger.info("Exception: " + e)
+ triggerAndWaitCumulativeCompaction(originTabletId, "[24-24]",
"[9-24]")
+ def notReadyTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[24-24]",
"[9-14]")
+ assertEquals("RUNNING", getJobState("date"))
+ for (int version = 15; version <= 24; version++) {
+ assertTrue(notReadyTabletRowsets.any {
it.contains("[${version}-${version}]") })
+ }
} finally {
if (injectBe != null) {
GetDebugPoint().disableDebugPointForAllBEs(injectName)
@@ -150,7 +210,7 @@ suite('test_schema_change_with_compaction10', 'docker') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-24]"))
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -159,7 +219,7 @@ suite('test_schema_change_with_compaction10', 'docker') {
assertTrue(out.contains("[2-2]"))
assertTrue(out.contains("[7-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
// base compaction
trigger_and_wait_compaction("date", "base")
@@ -169,7 +229,7 @@ suite('test_schema_change_with_compaction10', 'docker') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
for (int i = 0; i < 3; i++) {
load_date_once("date");
@@ -177,13 +237,11 @@ suite('test_schema_change_with_compaction10', 'docker') {
sql """ select count(*) from date """
- trigger_and_wait_compaction("date", "cumulative")
- logger.info("run show:" + newTabletId)
- (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run show: code=" + code + ", out=" + out + ", err=" +
err)
- assertTrue(out.contains("[0-1]"))
- assertTrue(out.contains("[2-7]"))
- assertTrue(out.contains("[8-16]"))
+ def finalTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[27-27]",
"[8-27]")
+ assertTrue(finalTabletRowsets.any { it.contains("[0-1]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[2-7]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[8-27]") })
}
}
}
diff --git
a/regression-test/suites/cloud_p1/schema_change/compaction5/test_schema_change_with_compaction5.groovy
b/regression-test/suites/cloud_p1/schema_change/compaction5/test_schema_change_with_compaction5.groovy
index ddd639b5fd1..eaf0621fb13 100644
---
a/regression-test/suites/cloud_p1/schema_change/compaction5/test_schema_change_with_compaction5.groovy
+++
b/regression-test/suites/cloud_p1/schema_change/compaction5/test_schema_change_with_compaction5.groovy
@@ -86,21 +86,42 @@ suite('test_schema_change_with_compaction5', 'docker') {
sql "delete from date where d_datekey < 19900000"
sql "select count(*) from date"
// cu compaction
- logger.info("run compaction:" + originTabletId)
- def (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
- boolean running = true
- do {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
+ trigger_and_wait_compaction("date", "cumulative")
}
+ def triggerAndWaitCumulativeCompaction = { tabletId,
latestVersionRange, expectedVersionRange ->
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ return tabletStatus.rowsets.any {
it.contains(latestVersionRange) }
+ }
+
+ logger.info("run compaction:" + tabletId)
+ def (triggerCode, triggerOut, triggerErr) =
+ be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, tabletId)
+ logger.info("Run compaction: code=" + triggerCode + ", out=" +
triggerOut + ", err=" + triggerErr)
+ assertEquals(0, triggerCode, "Failed to trigger cumulative
compaction: ${triggerErr}")
+ def triggerResult = parseJson(triggerOut.trim())
+ assertEquals("success",
triggerResult.status.toString().toLowerCase(),
+ "Unexpected cumulative compaction response: ${triggerOut}")
+
+ def tabletRowsets = []
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ tabletRowsets = tabletStatus.rowsets
+ return tabletRowsets.any { it.contains(expectedVersionRange) }
+ }
+ return tabletRowsets
+ }
+
+ def newTabletId = null
try {
load_delete_compaction()
load_delete_compaction()
@@ -114,59 +135,30 @@ suite('test_schema_change_with_compaction5', 'docker') {
sleep(5000)
array = sql_return_maparray("SHOW TABLETS FROM date")
- for (int i = 0; i < 5; i++) {
+ // NOTREADY tablets keep the latest 10 versions unmerged. Create
enough
+ // double-write rowsets for older versions to remain eligible for
compaction.
+ for (int i = 0; i < 16; i++) {
load_date_once("date");
}
// base compaction
- logger.info("run compaction:" + originTabletId)
- def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
-
- // wait for all compactions done
- boolean running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
- def newTabletId = array[1].TabletId
+ trigger_and_wait_compaction("date", "base")
+ newTabletId = array[1].TabletId
logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
+ def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
assertTrue(out.contains("invalid tablet state."))
- // cu compaction
- for (int i = 0; i < array.size(); i++) {
- def tabletId = array[i].TabletId
- logger.info("run compaction:" + tabletId)
- (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, tabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out +
", err=" + err)
- }
-
- for (int i = 0; i < array.size(); i++) {
- running = true
- do {
- Thread.sleep(100)
- def tabletId = array[i].TabletId
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, tabletId)
- logger.info("Get compaction status: code=" + code + ",
out=" + out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success",
compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
+ triggerAndWaitCumulativeCompaction(originTabletId, "[24-24]",
"[9-24]")
+ def notReadyTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[24-24]",
"[9-14]")
+ assertEquals("RUNNING", getJobState("date"))
+ for (int version = 15; version <= 24; version++) {
+ assertTrue(notReadyTabletRowsets.any {
it.contains("[${version}-${version}]") })
}
cluster.restartFrontends()
sleep(30000)
context.reconnectFe()
- } catch (Exception e) {
- logger.error("Exception: " + e)
} finally {
if (injectBe != null) {
GetDebugPoint().disableDebugPointForAllBEs(injectName)
@@ -187,7 +179,7 @@ suite('test_schema_change_with_compaction5', 'docker') {
}
assertEquals(result, "FINISHED");
def count = sql """ select count(*) from date; """
- assertEquals(count[0][0], 23004);
+ assertEquals(count[0][0], 51120);
// check rowsets
logger.info("run show:" + originTabletId)
def (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
@@ -195,7 +187,7 @@ suite('test_schema_change_with_compaction5', 'docker') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-24]"))
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -204,25 +196,10 @@ suite('test_schema_change_with_compaction5', 'docker') {
assertTrue(out.contains("[2-2]"))
assertTrue(out.contains("[7-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
// base compaction
- logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
-
- // wait for all compactions done
- boolean running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
+ trigger_and_wait_compaction("date", "base")
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -230,7 +207,7 @@ suite('test_schema_change_with_compaction5', 'docker') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
for (int i = 0; i < 3; i++) {
load_date_once("date");
@@ -238,28 +215,11 @@ suite('test_schema_change_with_compaction5', 'docker') {
sql """ select count(*) from date """
- logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
- // wait for all compactions done
- running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
-
- logger.info("run show:" + newTabletId)
- (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run show: code=" + code + ", out=" + out + ", err=" +
err)
- assertTrue(out.contains("[0-1]"))
- assertTrue(out.contains("[2-7]"))
- assertTrue(out.contains("[8-16]"))
+ def finalTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[27-27]",
"[8-27]")
+ assertTrue(finalTabletRowsets.any { it.contains("[0-1]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[2-7]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[8-27]") })
}
}
}
diff --git
a/regression-test/suites/cloud_p1/schema_change/compaction6/test_schema_change_with_compaction6.groovy
b/regression-test/suites/cloud_p1/schema_change/compaction6/test_schema_change_with_compaction6.groovy
index 894f06bee13..c8f5be30265 100644
---
a/regression-test/suites/cloud_p1/schema_change/compaction6/test_schema_change_with_compaction6.groovy
+++
b/regression-test/suites/cloud_p1/schema_change/compaction6/test_schema_change_with_compaction6.groovy
@@ -86,21 +86,63 @@ suite('test_schema_change_with_compaction6', 'docker') {
sql "delete from date where d_datekey < 19900000"
sql "select count(*) from date"
// cu compaction
- logger.info("run compaction:" + originTabletId)
- def (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
- boolean running = true
- do {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
+ trigger_and_wait_compaction("date", "cumulative")
+ }
+
+ def triggerAndWaitCumulativeCompaction = { tabletId,
latestVersionRange, expectedVersionRange ->
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ return tabletStatus.rowsets.any {
it.contains(latestVersionRange) }
+ }
+
+ logger.info("run compaction:" + tabletId)
+ def (triggerCode, triggerOut, triggerErr) =
+ be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, tabletId)
+ logger.info("Run compaction: code=" + triggerCode + ", out=" +
triggerOut + ", err=" + triggerErr)
+ assertEquals(0, triggerCode, "Failed to trigger cumulative
compaction: ${triggerErr}")
+ def triggerResult = parseJson(triggerOut.trim())
+ assertEquals("success",
triggerResult.status.toString().toLowerCase(),
+ "Unexpected cumulative compaction response: ${triggerOut}")
+
+ def tabletRowsets = []
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ tabletRowsets = tabletStatus.rowsets
+ return tabletRowsets.any { it.contains(expectedVersionRange) }
+ }
+ return tabletRowsets
+ }
+
+ def restartBackendAndRearmDebugPoint = {
+ cluster.stopBackends()
+ def rearmFuture = thread {
+ long deadline = System.currentTimeMillis() + 120000L
+ Exception lastError = null
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ DebugPoint.enableDebugPoint(injectBe.Host,
injectBe.HttpPort as int,
+ NodeType.BE, injectName)
+ return
+ } catch (Exception e) {
+ lastError = e
+ sleep(50)
+ }
+ }
+ throw new IllegalStateException("Failed to re-enable
${injectName} after BE restart", lastError)
+ }
+ cluster.startBackends()
+ rearmFuture.get()
}
+ def newTabletId = null
try {
load_delete_compaction()
load_delete_compaction()
@@ -114,62 +156,33 @@ suite('test_schema_change_with_compaction6', 'docker') {
sleep(5000)
array = sql_return_maparray("SHOW TABLETS FROM date")
- for (int i = 0; i < 5; i++) {
+ // NOTREADY tablets keep the latest 10 versions unmerged. Create
enough
+ // double-write rowsets for older versions to remain eligible for
compaction.
+ for (int i = 0; i < 16; i++) {
load_date_once("date");
}
- cluster.restartBackends()
- GetDebugPoint().enableDebugPointForAllBEs(injectName)
+ restartBackendAndRearmDebugPoint()
sleep(30000)
+ assertEquals("RUNNING", getJobState("date"),
+ "Schema change finished before the debug point was
re-enabled")
// base compaction
- logger.info("run compaction:" + originTabletId)
- def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
-
- // wait for all compactions done
- boolean running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
- def newTabletId = array[1].TabletId
+ trigger_and_wait_compaction("date", "base")
+ newTabletId = array[1].TabletId
logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
+ def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
assertTrue(out.contains("invalid tablet state."))
- // cu compaction
- for (int i = 0; i < array.size(); i++) {
- def tabletId = array[i].TabletId
- logger.info("run compaction:" + tabletId)
- (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, tabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out +
", err=" + err)
+ triggerAndWaitCumulativeCompaction(originTabletId, "[24-24]",
"[9-24]")
+ def notReadyTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[24-24]",
"[9-14]")
+ assertEquals("RUNNING", getJobState("date"))
+ for (int version = 15; version <= 24; version++) {
+ assertTrue(notReadyTabletRowsets.any {
it.contains("[${version}-${version}]") })
}
-
- for (int i = 0; i < array.size(); i++) {
- running = true
- do {
- Thread.sleep(100)
- def tabletId = array[i].TabletId
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, tabletId)
- logger.info("Get compaction status: code=" + code + ",
out=" + out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success",
compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
- }
-
- } catch (Exception e) {
- logger.info("Exception: " + e)
} finally {
if (injectBe != null) {
GetDebugPoint().disableDebugPointForAllBEs(injectName)
@@ -190,7 +203,7 @@ suite('test_schema_change_with_compaction6', 'docker') {
}
assertEquals(result, "FINISHED");
def count = sql """ select count(*) from date; """
- assertEquals(count[0][0], 23004);
+ assertEquals(count[0][0], 51120);
// check rowsets
logger.info("run show:" + originTabletId)
def (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
@@ -198,7 +211,7 @@ suite('test_schema_change_with_compaction6', 'docker') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-24]"))
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -207,25 +220,10 @@ suite('test_schema_change_with_compaction6', 'docker') {
assertTrue(out.contains("[2-2]"))
assertTrue(out.contains("[7-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
// base compaction
- logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
-
- // wait for all compactions done
- boolean running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
+ trigger_and_wait_compaction("date", "base")
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -233,7 +231,7 @@ suite('test_schema_change_with_compaction6', 'docker') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
for (int i = 0; i < 3; i++) {
load_date_once("date");
@@ -241,28 +239,11 @@ suite('test_schema_change_with_compaction6', 'docker') {
sql """ select count(*) from date """
- logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
- // wait for all compactions done
- running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
-
- logger.info("run show:" + newTabletId)
- (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run show: code=" + code + ", out=" + out + ", err=" +
err)
- assertTrue(out.contains("[0-1]"))
- assertTrue(out.contains("[2-7]"))
- assertTrue(out.contains("[8-16]"))
+ def finalTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[27-27]",
"[8-27]")
+ assertTrue(finalTabletRowsets.any { it.contains("[0-1]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[2-7]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[8-27]") })
}
}
}
diff --git
a/regression-test/suites/cloud_p1/schema_change/compaction7/test_schema_change_with_compaction7.groovy
b/regression-test/suites/cloud_p1/schema_change/compaction7/test_schema_change_with_compaction7.groovy
index 44a564a7bc4..80b2871ab21 100644
---
a/regression-test/suites/cloud_p1/schema_change/compaction7/test_schema_change_with_compaction7.groovy
+++
b/regression-test/suites/cloud_p1/schema_change/compaction7/test_schema_change_with_compaction7.groovy
@@ -84,21 +84,42 @@ suite('test_schema_change_with_compaction7',
'p1,nonConcurrent') {
sql "delete from date where d_datekey < 19900000"
sql "select count(*) from date"
// cu compaction
- logger.info("run compaction:" + originTabletId)
- def (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ", err="
+ err)
- boolean running = true
- do {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" + out
+ ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
+ trigger_and_wait_compaction("date", "cumulative")
}
+ def triggerAndWaitCumulativeCompaction = { tabletId, latestVersionRange,
expectedVersionRange ->
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host, injectBe.HttpPort,
tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ return tabletStatus.rowsets.any { it.contains(latestVersionRange) }
+ }
+
+ logger.info("run compaction:" + tabletId)
+ def (triggerCode, triggerOut, triggerErr) =
+ be_run_cumulative_compaction(injectBe.Host, injectBe.HttpPort,
tabletId)
+ logger.info("Run compaction: code=" + triggerCode + ", out=" +
triggerOut + ", err=" + triggerErr)
+ assertEquals(0, triggerCode, "Failed to trigger cumulative compaction:
${triggerErr}")
+ def triggerResult = parseJson(triggerOut.trim())
+ assertEquals("success", triggerResult.status.toString().toLowerCase(),
+ "Unexpected cumulative compaction response: ${triggerOut}")
+
+ def tabletRowsets = []
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host, injectBe.HttpPort,
tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ tabletRowsets = tabletStatus.rowsets
+ return tabletRowsets.any { it.contains(expectedVersionRange) }
+ }
+ return tabletRowsets
+ }
+
+ def newTabletId = null
try {
load_delete_compaction()
load_delete_compaction()
@@ -113,56 +134,27 @@ suite('test_schema_change_with_compaction7',
'p1,nonConcurrent') {
sleep(15000)
array = sql_return_maparray("SHOW TABLETS FROM date")
- for (int i = 0; i < 5; i++) {
+ // NOTREADY tablets keep the latest 10 versions unmerged. Create enough
+ // double-write rowsets for older versions to remain eligible for
compaction.
+ for (int i = 0; i < 16; i++) {
load_date_once("date");
}
// base compaction
- logger.info("run compaction:" + originTabletId)
- def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ", err="
+ err)
-
-
- // wait for all compactions done
- boolean running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" + out
+ ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
- def newTabletId = array[1].TabletId
+ trigger_and_wait_compaction("date", "base")
+ newTabletId = array[1].TabletId
logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
+ def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
logger.info("Run compaction: code=" + code + ", out=" + out + ", err="
+ err)
assertTrue(out.contains("invalid tablet state."))
- // cu compaction
- for (int i = 0; i < array.size(); i++) {
- def tabletId = array[i].TabletId
- logger.info("run compaction:" + tabletId)
- (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, tabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
+ triggerAndWaitCumulativeCompaction(originTabletId, "[24-24]", "[9-24]")
+ def notReadyTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[24-24]",
"[9-14]")
+ assertEquals("RUNNING", getJobState("date"))
+ for (int version = 15; version <= 24; version++) {
+ assertTrue(notReadyTabletRowsets.any {
it.contains("[${version}-${version}]") })
}
-
- for (int i = 0; i < array.size(); i++) {
- running = true
- do {
- Thread.sleep(100)
- def tabletId = array[i].TabletId
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, tabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
- }
- } catch (Exception e) {
- logger.error("Exception: " + e)
} finally {
if (injectBe != null) {
DebugPoint.disableDebugPoint(injectBe.Host,
injectBe.HttpPort.toInteger(), NodeType.BE, injectName)
@@ -191,7 +183,7 @@ suite('test_schema_change_with_compaction7',
'p1,nonConcurrent') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-24]"))
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -200,25 +192,10 @@ suite('test_schema_change_with_compaction7',
'p1,nonConcurrent') {
assertTrue(out.contains("[2-2]"))
assertTrue(out.contains("[7-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
// base compaction
- logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ", err="
+ err)
-
-
- // wait for all compactions done
- boolean running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" + out
+ ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
+ trigger_and_wait_compaction("date", "base")
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -226,7 +203,7 @@ suite('test_schema_change_with_compaction7',
'p1,nonConcurrent') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
for (int i = 0; i < 3; i++) {
load_date_once("date");
@@ -234,28 +211,11 @@ suite('test_schema_change_with_compaction7',
'p1,nonConcurrent') {
sql """ select count(*) from date """
- logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ", err="
+ err)
-
- // wait for all compactions done
- running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" + out
+ ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
-
- logger.info("run show:" + newTabletId)
- (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run show: code=" + code + ", out=" + out + ", err=" + err)
- assertTrue(out.contains("[0-1]"))
- assertTrue(out.contains("[2-7]"))
- assertTrue(out.contains("[8-16]"))
+ def finalTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[27-27]",
"[8-27]")
+ assertTrue(finalTabletRowsets.any { it.contains("[0-1]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[2-7]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[8-27]") })
}
}
diff --git
a/regression-test/suites/cloud_p1/schema_change/compaction9/test_schema_change_with_compaction9.groovy
b/regression-test/suites/cloud_p1/schema_change/compaction9/test_schema_change_with_compaction9.groovy
index 27f4e857007..80b59ef5a02 100644
---
a/regression-test/suites/cloud_p1/schema_change/compaction9/test_schema_change_with_compaction9.groovy
+++
b/regression-test/suites/cloud_p1/schema_change/compaction9/test_schema_change_with_compaction9.groovy
@@ -87,21 +87,42 @@ suite('test_schema_change_with_compaction9', 'docker') {
sql "delete from date where d_datekey < 19900000"
sql "select count(*) from date"
// cu compaction
- logger.info("run compaction:" + originTabletId)
- def (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
- boolean running = true
- do {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
+ trigger_and_wait_compaction("date", "cumulative")
}
+ def triggerAndWaitCumulativeCompaction = { tabletId,
latestVersionRange, expectedVersionRange ->
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ return tabletStatus.rowsets.any {
it.contains(latestVersionRange) }
+ }
+
+ logger.info("run compaction:" + tabletId)
+ def (triggerCode, triggerOut, triggerErr) =
+ be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, tabletId)
+ logger.info("Run compaction: code=" + triggerCode + ", out=" +
triggerOut + ", err=" + triggerErr)
+ assertEquals(0, triggerCode, "Failed to trigger cumulative
compaction: ${triggerErr}")
+ def triggerResult = parseJson(triggerOut.trim())
+ assertEquals("success",
triggerResult.status.toString().toLowerCase(),
+ "Unexpected cumulative compaction response: ${triggerOut}")
+
+ def tabletRowsets = []
+ awaitUntil(60, 1) {
+ def (showCode, showOut, showErr) =
+ be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, tabletId)
+ assertEquals(0, showCode, "Failed to show tablet status:
${showErr}")
+ def tabletStatus = parseJson(showOut.trim())
+ assertTrue(tabletStatus.rowsets instanceof List)
+ tabletRowsets = tabletStatus.rowsets
+ return tabletRowsets.any { it.contains(expectedVersionRange) }
+ }
+ return tabletRowsets
+ }
+
+ def newTabletId = null
try {
load_delete_compaction()
load_delete_compaction()
@@ -115,60 +136,30 @@ suite('test_schema_change_with_compaction9', 'docker') {
sleep(5000)
array = sql_return_maparray("SHOW TABLETS FROM date")
- for (int i = 0; i < 5; i++) {
+ // NOTREADY tablets keep the latest 10 versions unmerged. Create
enough
+ // double-write rowsets for older versions to remain eligible for
compaction.
+ for (int i = 0; i < 16; i++) {
load_date_once("date");
}
// base compaction
- logger.info("run compaction:" + originTabletId)
- def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
-
- // wait for all compactions done
- boolean running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, originTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
- def newTabletId = array[1].TabletId
+ trigger_and_wait_compaction("date", "base")
+ newTabletId = array[1].TabletId
logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
+ def (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
assertTrue(out.contains("invalid tablet state."))
- // cu compaction
- for (int i = 0; i < array.size(); i++) {
- def tabletId = array[i].TabletId
- logger.info("run compaction:" + tabletId)
- (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, tabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out +
", err=" + err)
- }
-
- for (int i = 0; i < array.size(); i++) {
- running = true
- do {
- Thread.sleep(100)
- def tabletId = array[i].TabletId
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, tabletId)
- logger.info("Get compaction status: code=" + code + ",
out=" + out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success",
compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
+ triggerAndWaitCumulativeCompaction(originTabletId, "[24-24]",
"[9-24]")
+ def notReadyTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[24-24]",
"[9-14]")
+ assertEquals("RUNNING", getJobState("date"))
+ for (int version = 15; version <= 24; version++) {
+ assertTrue(notReadyTabletRowsets.any {
it.contains("[${version}-${version}]") })
}
cluster.restartFrontends()
sleep(30000)
context.reconnectFe()
- } catch (Exception e) {
- logger.error("Exception: " + e.getMessage())
- assertEquals(1, 2)
} finally {
if (injectBe != null) {
GetDebugPoint().disableDebugPointForAllBEs(injectName)
@@ -197,7 +188,7 @@ suite('test_schema_change_with_compaction9', 'docker') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-24]"))
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -206,25 +197,10 @@ suite('test_schema_change_with_compaction9', 'docker') {
assertTrue(out.contains("[2-2]"))
assertTrue(out.contains("[7-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
// base compaction
- logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_base_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
-
- // wait for all compactions done
- boolean running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
+ trigger_and_wait_compaction("date", "base")
logger.info("run show:" + newTabletId)
(code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
@@ -232,7 +208,7 @@ suite('test_schema_change_with_compaction9', 'docker') {
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-7]"))
assertTrue(out.contains("[8-8]"))
- assertTrue(out.contains("[9-13]"))
+ assertTrue(out.contains("[9-14]"))
for (int i = 0; i < 3; i++) {
load_date_once("date");
@@ -240,28 +216,11 @@ suite('test_schema_change_with_compaction9', 'docker') {
sql """ select count(*) from date """
- logger.info("run compaction:" + newTabletId)
- (code, out, err) = be_run_cumulative_compaction(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
-
- // wait for all compactions done
- running = true
- while (running) {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- }
-
- logger.info("run show:" + newTabletId)
- (code, out, err) = be_show_tablet_status(injectBe.Host,
injectBe.HttpPort, newTabletId)
- logger.info("Run show: code=" + code + ", out=" + out + ", err=" +
err)
- assertTrue(out.contains("[0-1]"))
- assertTrue(out.contains("[2-7]"))
- assertTrue(out.contains("[8-16]"))
+ def finalTabletRowsets =
+ triggerAndWaitCumulativeCompaction(newTabletId, "[27-27]",
"[8-27]")
+ assertTrue(finalTabletRowsets.any { it.contains("[0-1]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[2-7]") })
+ assertTrue(finalTabletRowsets.any { it.contains("[8-27]") })
}
}
}
diff --git
a/regression-test/suites/inverted_index_p0/index_format_v2/test_cumulative_compaction_with_format_v2.groovy
b/regression-test/suites/inverted_index_p0/index_format_v2/test_cumulative_compaction_with_format_v2.groovy
index 32a1f6d6e76..a010aff5a47 100644
---
a/regression-test/suites/inverted_index_p0/index_format_v2/test_cumulative_compaction_with_format_v2.groovy
+++
b/regression-test/suites/inverted_index_p0/index_format_v2/test_cumulative_compaction_with_format_v2.groovy
@@ -16,6 +16,9 @@
// under the License.
import org.codehaus.groovy.runtime.IOGroovyMethods
+import org.awaitility.Awaitility
+
+import java.util.concurrent.TimeUnit
suite("test_cumulative_compaction_with_format_v2", "inverted_index_format_v2")
{
def tableName = "test_cumulative_compaction_with_format_v2"
@@ -180,9 +183,9 @@ suite("test_cumulative_compaction_with_format_v2",
"inverted_index_format_v2") {
backend_id = tablet.BackendId
String ip = backendId_to_backendIP.get(backend_id)
String port = backendId_to_backendHttpPort.get(backend_id)
- be_show_tablet_status(ip, port, tablet_id)
(code, out, err) = be_show_tablet_status(ip, port, tablet_id)
logger.info("Run show: code=" + code + ", out=" + out + ", err=" +
err)
+ assertEquals(0, code, "Failed to show tablet status:
stdout=${out}, stderr=${err}")
assertTrue(out.contains("[0-1]"))
assertTrue(out.contains("[2-2]"))
assertTrue(out.contains("[3-3]"))
@@ -193,24 +196,42 @@ suite("test_cumulative_compaction_with_format_v2",
"inverted_index_format_v2") {
assertTrue(out.contains("[8-8]"))
assertTrue(out.contains("[9-9]"))
logger.info("run compaction:" + tablet_id)
- (code, out, err) = be_run_cumulative_compaction(ip, port,
tablet_id)
- logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
- boolean running = true
- do {
- Thread.sleep(100)
- (code, out, err) = be_get_compaction_status(ip, port,
tablet_id)
- logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
- assertEquals(code, 0)
- def compactionStatus = parseJson(out.trim())
- assertEquals("success", compactionStatus.status.toLowerCase())
- running = compactionStatus.run_status
- } while (running)
- (code, out, err) = be_show_tablet_status(ip, port, tablet_id)
- logger.info("Run show: code=" + code + ", out=" + out + ", err=" +
err)
+ // FE can expose a newly committed version before the BE has
refreshed its
+ // partition visible version. Retry only that transient no-input
response.
+ long lastReportTime = 0
+ Awaitility.await().atMost(120, TimeUnit.SECONDS).pollInterval(1,
TimeUnit.SECONDS).until {
+ (code, out, err) = be_run_cumulative_compaction(ip, 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(ip, port.toInteger())
+ lastReportTime = now
+ }
+ return false
+ }
+
+ // run_status can be false before an asynchronous compaction
starts. Wait for
+ // the observable rowset reduction instead of treating that as
completion.
+ def rowsets = []
+ Awaitility.await().atMost(60, TimeUnit.SECONDS).pollInterval(1,
TimeUnit.SECONDS).until {
+ (code, out, err) = be_show_tablet_status(ip, port, tablet_id)
+ logger.info("Run show: code=" + code + ", out=" + out + ",
err=" + err)
+ assertEquals(0, code)
+ def tabletJson = parseJson(out.trim())
+ assertTrue(tabletJson.rowsets instanceof List)
+ rowsets = tabletJson.rowsets
+ return rowsets.size() < 9
+ }
assertTrue(out.contains("[0-1]"))
// Parse the tablet status to get rowset count after compaction
- def tabletJson = parseJson(out.trim())
- def rowsets = tabletJson.rowsets
int activeRowsetCount = rowsets.size()
// After compaction, we should have fewer rowsets than before
(originally 9 rowsets: [0-1], [2-2], ..., [9-9])
// The exact number depends on compaction strategy, but should be
less than 9
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
