Skip to content

Commit 1a72a8d

Browse files
dzr171712dzr171712
committed
[test](cloud) Add cumulative compaction global lock regression test
1 parent 93ebc1f commit 1a72a8d

1 file changed

Lines changed: 186 additions & 0 deletions

File tree

‎regression-test/suites/fault_injection_p0/cloud/test_cloud_cumu_compaction_global_lock_thread_count.groovy‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,186 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
import groovy.json.JsonSlurper
19+
import org.apache.doris.regression.suite.ClusterOptions
20+
import org.apache.doris.regression.util.DebugPoint
21+
import org.apache.doris.regression.util.NodeType
22+
23+
suite('test_cloud_cumu_compaction_global_lock_thread_count', 'docker') {
24+
def options = new ClusterOptions()
25+
options.cloudMode = true
26+
options.enableDebugPoints()
27+
options.beConfigs += [
28+
'enable_java_support=false',
29+
'cumulative_compaction_min_deltas=2',
30+
'cumulative_compaction_max_deltas=3',
31+
'max_cumu_compaction_threads=2',
32+
'large_cumu_compaction_task_min_thread_num=2',
33+
'large_cumu_compaction_task_row_num_threshold=1',
34+
'disable_auto_compaction=true',
35+
]
36+
options.beNum = 3
37+
38+
docker(options) {
39+
GetDebugPoint().clearDebugPointsForAllBEs()
40+
def backends = sql_return_maparray('SHOW BACKENDS')
41+
def lockHolderBe = backends[0]
42+
def lockFailedBe = backends[1]
43+
assertNotNull(lockHolderBe)
44+
assertNotNull(lockFailedBe)
45+
46+
def getCompactionStatus = { be, tabletId ->
47+
def (code, out, err) = be_get_compaction_status(be.Host, be.HttpPort, tabletId)
48+
assertEquals(0, code, "failed to get compaction status: ${out}, ${err}")
49+
return new JsonSlurper().parseText(out.trim())
50+
}
51+
52+
def waitForCompactionRunning = { be, tabletId ->
53+
def deadline = System.currentTimeMillis() + 30000
54+
while (System.currentTimeMillis() < deadline) {
55+
if (getCompactionStatus(be, tabletId).run_status) {
56+
return
57+
}
58+
sleep(100)
59+
}
60+
assertTrue(false, "compaction did not start for tablet ${tabletId}")
61+
}
62+
63+
def waitForCompactionFinished = { be, tabletId ->
64+
def deadline = System.currentTimeMillis() + 30000
65+
while (System.currentTimeMillis() < deadline) {
66+
if (!getCompactionStatus(be, tabletId).run_status) {
67+
return
68+
}
69+
sleep(100)
70+
}
71+
assertTrue(false, "compaction did not finish for tablet ${tabletId}")
72+
}
73+
74+
def getTabletStatus = { be, tabletId ->
75+
def (code, out, err) = be_show_tablet_status(be.Host, be.HttpPort, tabletId)
76+
assertEquals(0, code, "failed to get tablet status: ${out}, ${err}")
77+
return new JsonSlurper().parseText(out.trim())
78+
}
79+
80+
def getRowsetCount = { be, tabletId ->
81+
def status = getTabletStatus(be, tabletId)
82+
assertTrue(status.rowsets instanceof List)
83+
return status.rowsets.size()
84+
}
85+
86+
def createTable = { table ->
87+
sql "DROP TABLE IF EXISTS ${table} FORCE"
88+
sql """
89+
CREATE TABLE ${table} (
90+
k INT,
91+
v INT
92+
) DUPLICATE KEY(k)
93+
DISTRIBUTED BY HASH(k) BUCKETS 1
94+
PROPERTIES (
95+
"replication_num" = "1",
96+
"disable_auto_compaction" = "true"
97+
)
98+
"""
99+
for (int value = 0; value < 5; value++) {
100+
sql "INSERT INTO ${table} VALUES (${value}, ${value})"
101+
}
102+
sql "SELECT COUNT(*) FROM ${table}"
103+
def tablet = sql_return_maparray("SHOW TABLETS FROM ${table}")[0]
104+
return tablet.TabletId
105+
}
106+
107+
def runCumulativeCompaction = { be, tabletId ->
108+
def (code, out, err) = be_run_cumulative_compaction(be.Host, be.HttpPort, tabletId)
109+
assertEquals(0, code, "failed to submit cumulative compaction: ${out}, ${err}")
110+
}
111+
112+
def conflictTabletId = createTable('test_cloud_cumu_compaction_global_lock_conflict')
113+
def conflictRowsetsBefore = getRowsetCount(lockFailedBe, conflictTabletId)
114+
def holderRowsetsBefore = getRowsetCount(lockHolderBe, conflictTabletId)
115+
def blockModifyRowsets = 'CloudCumulativeCompaction::modify_rowsets.enable_spin_wait'
116+
def blockModifyRowsetsSwitch = 'CloudCumulativeCompaction::modify_rowsets.block'
117+
def holdTaskAfterExecution = 'CloudStorageEngine._submit_cumulative_compaction_task.sleep'
118+
119+
try {
120+
DebugPoint.enableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
121+
NodeType.BE, blockModifyRowsets)
122+
DebugPoint.enableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
123+
NodeType.BE, blockModifyRowsetsSwitch)
124+
125+
runCumulativeCompaction(lockHolderBe, conflictTabletId)
126+
waitForCompactionRunning(lockHolderBe, conflictTabletId)
127+
128+
runCumulativeCompaction(lockFailedBe, conflictTabletId)
129+
sleep(1000)
130+
waitForCompactionFinished(lockFailedBe, conflictTabletId)
131+
assertEquals(conflictRowsetsBefore, getRowsetCount(lockFailedBe, conflictTabletId))
132+
133+
DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
134+
NodeType.BE, blockModifyRowsetsSwitch)
135+
DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
136+
NodeType.BE, blockModifyRowsets)
137+
waitForCompactionFinished(lockHolderBe, conflictTabletId)
138+
assertTrue(getRowsetCount(lockHolderBe, conflictTabletId) < holderRowsetsBefore)
139+
140+
def runningTabletId = createTable('test_cloud_cumu_compaction_global_lock_thread_holder')
141+
def candidateTabletId = createTable('test_cloud_cumu_compaction_global_lock_thread_candidate')
142+
def candidateRowsetsBefore = getRowsetCount(lockFailedBe, candidateTabletId)
143+
def runningRowsetsBefore = getRowsetCount(lockFailedBe, runningTabletId)
144+
145+
DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
146+
NodeType.BE, holdTaskAfterExecution)
147+
DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
148+
NodeType.BE, blockModifyRowsets)
149+
DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
150+
NodeType.BE, blockModifyRowsetsSwitch)
151+
152+
runCumulativeCompaction(lockFailedBe, runningTabletId)
153+
waitForCompactionRunning(lockFailedBe, runningTabletId)
154+
155+
DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
156+
NodeType.BE, blockModifyRowsetsSwitch)
157+
DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
158+
NodeType.BE, blockModifyRowsets)
159+
def deadline = System.currentTimeMillis() + 30000
160+
while (System.currentTimeMillis() < deadline &&
161+
getRowsetCount(lockFailedBe, runningTabletId) >= runningRowsetsBefore) {
162+
sleep(100)
163+
}
164+
assertTrue(getRowsetCount(lockFailedBe, runningTabletId) < runningRowsetsBefore)
165+
assertTrue(getCompactionStatus(lockFailedBe, runningTabletId).run_status)
166+
167+
runCumulativeCompaction(lockFailedBe, candidateTabletId)
168+
sleep(1000)
169+
assertEquals(candidateRowsetsBefore, getRowsetCount(lockFailedBe, candidateTabletId),
170+
'a second large compaction must be delayed while the first task is active')
171+
waitForCompactionFinished(lockFailedBe, candidateTabletId)
172+
waitForCompactionFinished(lockFailedBe, runningTabletId)
173+
} finally {
174+
DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
175+
NodeType.BE, blockModifyRowsetsSwitch)
176+
DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
177+
NodeType.BE, blockModifyRowsets)
178+
DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
179+
NodeType.BE, holdTaskAfterExecution)
180+
DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
181+
NodeType.BE, blockModifyRowsetsSwitch)
182+
DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
183+
NodeType.BE, blockModifyRowsets)
184+
}
185+
}
186+
}

0 commit comments

Comments
 (0)