github-actions[bot] commented on code in PR #68449:
URL: https://github.com/apache/doris/pull/68449#discussion_r4091146430
##########
regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/table/test_warm_up_table.groovy:
##########
@@ -113,17 +125,95 @@ suite("test_warm_up_table") {
}
}
- def getMetricsMethod = { ip, port, check_func ->
- httpTest {
- endpoint ip + ":" + port
- uri "/brpc_metrics"
- op "get"
- check check_func
+ def clusterBackends = ["regression_cluster_name0",
"regression_cluster_name1"].collectEntries { name ->
+ def members = sql_return_maparray("SHOW BACKENDS").findAll { be ->
+ "${be.Alive}".equalsIgnoreCase("true") &&
+ parseJson(be.Tag.toString()).compute_group_name == name
+ }.collect { be ->
+ [id: be.BackendId as Long, ip: be.Host.toString(),
+ httpPort: be.HttpPort.toString(), brpcPort:
be.BrpcPort.toString()]
+ }
+ assertTrue(!members.isEmpty(), "No alive backends in ${name}")
+ [(name): members]
+ }
+ def tabletIds = []
+ def waitUntil = { String stage, long timeoutMs, Closure ready ->
+ long deadline = System.currentTimeMillis() + timeoutMs
+ while (System.currentTimeMillis() < deadline) {
+ if (ready()) { return }
+ sleep(1000)
+ }
+ assertTrue(false, "Timeout waiting for ${stage}; see last cache state
in log")
+ }
+ def getGlobalTtl = {
+ clusterBackends.collectEntries { name, members ->
+ [(name): WarmupMetricsUtils.getBackendMetricSum(members,
"ttl_cache_size")]
+ }
+ }
+ def getTableCache = { String clusterName ->
+ sql "use @${clusterName}"
+ def beIds = clusterBackends[clusterName].collect { it.id }
+ def rows = sql """select tablet_id, lower(type), sum(size)
+ from information_schema.file_cache_info
+ where tablet_id in (${tabletIds.join(',')}) and be_id in
(${beIds.join(',')})
+ group by tablet_id, lower(type)"""
+ def ttlByTablet = rows.findAll { it[1]?.toString() == "ttl" }
+ .collectEntries { [(it[0] as Long): it[2] as Long] }
+ [ttlByTablet: ttlByTablet, complete: tabletIds.every {
(ttlByTablet[it] ?: 0L) > 0L },
+ nonTtlBytes: rows.findAll { it[1]?.toString() != "ttl" }.sum(0L) {
it[2] as Long }]
+ }
+ def waitForTargetTablets = {
+ sql "use @regression_cluster_name1"
+ def targetBes =
clusterBackends.regression_cluster_name1.collectEntries { [(it.id): it] }
+ waitUntil("target primary tablet metadata", 120000L) {
+ def ready = [] as Set
+ sql_return_maparray("show tablets from ${table}").each { tablet ->
+ long id = tablet.TabletId as Long
+ long primary = tablet.PrimaryBackendId as Long
+ if (id in tabletIds && targetBes.containsKey(primary) &&
(tablet.BackendId as Long) == primary) {
+ def be = targetBes[primary]
+ try {
+ // Metadata only: a business read here would populate
the target cache.
+ def meta = parseJson(new
URL("http://${be.ip}:${be.httpPort}/api/meta/header/${id}";)
Review Comment:
[P1] Route this required metadata poll through `Http.openConnection`. When
`enableTLS=true`, the BE webserver is reached via HTTPS and may require the
configured client certificate, but direct `URL.getText` sends cleartext and
bypasses the installed SSL context. The exception is swallowed, so `ready`
never fills and the mandatory 120-second barrier fails.
##########
regression-test/suites/cloud_p0/cache/test_file_cache_info.groovy:
##########
@@ -15,121 +15,214 @@
// specific language governing permissions and limitations
// under the License.
-suite("test_file_cache_info") {
- def custoBeConfig = [
- enable_evict_file_cache_in_advance : false,
- file_cache_enter_disk_resource_limit_mode_percent : 99
+suite("test_file_cache_info", "nonConcurrent") {
+ def customBeConfig = [
+ enable_evict_file_cache_in_advance: false,
+ file_cache_enter_disk_resource_limit_mode_percent: 99
]
-
- setBeConfigTemporary(custoBeConfig) {
-
- String[][] backends = sql """ show backends """
- def backendSockets = []
- def backendIdToBackendIP = [:]
- def backendIdToBackendHttpPort = [:]
- for (String[] backend in backends) {
- if (backend[9].equals("true")) {
- backendIdToBackendIP.put(backend[0], backend[1])
- backendIdToBackendHttpPort.put(backend[0], backend[4])
+ setBeConfigTemporary(customBeConfig) {
+ String tableName = "test_file_cache_info_lifecycle"
+ def clusters = sql "SHOW CLUSTERS"
+ assertTrue(!clusters.isEmpty(), "No compute group found")
+ String clusterName = clusters[0][0].toString()
+ String dbName = sql("SELECT DATABASE()")[0][0].toString()
+ def backends = sql_return_maparray("SHOW BACKENDS").findAll { be ->
+ "${be.Alive}".equalsIgnoreCase("true") &&
+ parseJson(be.Tag.toString()).compute_group_name ==
clusterName
+ }.collectEntries { be -> [(be.BackendId as Long): be] }
+ assertTrue(!backends.isEmpty(), "No alive backends in ${clusterName}")
+ def backendIds = backends.keySet()
+ Long tabletId = null
+ def lastState = [:]
+ def clearResponses = [:]
+ def originalQueryCache = sql("select @@enable_sql_cache,
@@enable_query_cache")[0].collect { value ->
+ String setting = value.toString().toLowerCase(Locale.ROOT)
+ assertTrue(setting in ["true", "false", "0", "1"], "Unexpected
cache setting: ${value}")
+ setting
}
- }
- assertTrue(backendIdToBackendIP.size() > 0, "No alive backends found")
-
- backendIdToBackendIP.each { backendId, ip ->
- def socket = ip + ":" + backendIdToBackendHttpPort.get(backendId)
- backendSockets.add(socket)
- }
-
- sql "drop table IF EXISTS customer"
-
- sql """
- CREATE TABLE IF NOT EXISTS customer (
- `c_custkey` int NULL,
- `c_name` string NULL,
- `c_address` string NULL,
- `c_city` string NULL,
- `c_nation` string NULL,
- `c_region` string NULL,
- `c_phone` string NULL,
- `c_mktsegment` string NULL
- )
- DUPLICATE KEY(`c_custkey`)
- DISTRIBUTED BY HASH(`c_custkey`) BUCKETS 1
- PROPERTIES (
- "file_cache_ttl_seconds" = "3600"
- )
- """
-
- sql """
- insert into customer values
- (1, 'Customer#000000001', 'address1', 'city1', 'nation1', 'region1',
'phone1', 'segment1'),
- (2, 'Customer#000000002', 'address2', 'city2', 'nation2', 'region2',
'phone2', 'segment2'),
- (3, 'Customer#000000003', 'address3', 'city3', 'nation3', 'region3',
'phone3', 'segment3'),
- (4, 'Customer#000000004', 'address4', 'city4', 'nation4', 'region4',
'phone4', 'segment4'),
- (5, 'Customer#000000005', 'address5', 'city5', 'nation5', 'region5',
'phone5', 'segment5')
- """
- sql "sync"
-
- sql "select count(*) from customer"
-
- Thread.sleep(10000)
-
- def get_tablet_id = { String tbl_name ->
- def tablets = sql "show tablets from ${tbl_name}"
- assertEquals(tablets.size(), 1, "Should have exactly one tablet with
BUCKETS=1")
- return tablets[0][0] as Long
- }
-
- def tablet_id = get_tablet_id("customer")
- println "Tablet ID: ${tablet_id}"
-
- def desc_cache_info = sql "desc information_schema.file_cache_info"
- assertTrue(desc_cache_info.size() > 0, "desc
information_schema.file_cache_info should not be empty")
- assertEquals(desc_cache_info[0][0].toString().toUpperCase(), "HASH")
- assertEquals(desc_cache_info[1][0].toString().toUpperCase(), "OFFSET")
-
- def cache_info = sql "select * from information_schema.file_cache_info"
-
- assertTrue(cache_info.size() > 0, "file_cache_info should not be empty for
tablet_id ${tablet_id}")
-
- println "First query - File cache info for tablet_id ${tablet_id}:"
- cache_info.each { row ->
- println " ${row}"
- }
-
- def clearResults = []
- backendSockets.each { socket ->
- httpTest {
- endpoint ""
- uri socket + "/api/file_cache?op=clear&sync=true"
- op "get"
- check {respCode, body ->
- assertEquals(respCode, 200, "clear local cache fail, maybe you
can find something in respond: " + parseJson(body))
- clearResults.add(true)
+ def expectedRows = (1..5).collect { i ->
+ [i as Long, String.format('Customer#%09d', i),
"address${i}".toString(),
+ "city${i}".toString(), "nation${i}".toString(),
"region${i}".toString(),
+ "phone${i}".toString(), "segment${i}".toString()]
+ }
+ def readTable = {
+ sql "use @${clusterName}"
+ // Read and validate every column; result caches are disabled for
both reads.
+ def rows = sql "select * from ${tableName} order by c_custkey"
+ assertTrue(rows.every { it.size() == 8 && it.every { value ->
value != null } },
+ "Unexpected data result: ${rows}")
+ def normalized = rows.collect { row -> [row[0] as Long] +
row.drop(1).collect { it.toString() } }
+ assertEquals(expectedRows, normalized)
+ }
+ def getCache = {
+ sql "use @${clusterName}"
+ def rows = sql """select be_id, cache_path, tablet_id, `hash`,
`offset`, size, lower(type)
+ from information_schema.file_cache_info
+ where tablet_id = ${tabletId} and be_id in
(${backendIds.join(',')})"""
+ def blocks = [:]
+ rows.each { row ->
+ def key = [row[0] as Long, row[1].toString(), row[2] as Long,
+ row[3].toString(), row[4] as Long]
+ assertTrue(backendIds.contains(key[0]) && key[2] == tabletId,
"Unexpected cache owner: ${row}")
+ assertTrue(key[4] >= 0L && (row[5] as Long) > 0L, "Invalid
cache range: ${row}")
+ assertTrue(!blocks.containsKey(key), "Duplicate cache block:
${key}")
+ blocks[key] = [size: row[5] as Long, type: row[6]?.toString()]
+ }
+ blocks
+ }
+ def summarizeCache = { Map blocks ->
+ blocks.groupBy { key, value -> [key[0], key[2]] }.collectEntries {
owner, entries ->
+ [(owner): [block_count: entries.size(), bytes:
entries.values().sum(0L) { it.size },
+ bytes_by_type: entries.values().groupBy { it.type
}.collectEntries { type, values ->
+ [(type): values.sum(0L) { it.size }]
+ }]]
+ }
+ }
+ def waitForCache = { String phase, boolean expectPresent, long
timeoutMs ->
+ long startedMs = System.currentTimeMillis()
+ long deadlineMs = startedMs + timeoutMs
+ long stableSince = 0L
+ long lastLogMs = 0L
+ def previous = null
+ while (System.currentTimeMillis() < deadlineMs) {
+ def blocks = getCache()
+ boolean matches = expectPresent ? !blocks.isEmpty() :
blocks.isEmpty()
+ lastState = [phase: phase, elapsed_ms:
System.currentTimeMillis() - startedMs,
+ cache_by_be_tablet: summarizeCache(blocks),
blocks: blocks]
+ if (blocks != previous || System.currentTimeMillis() -
lastLogMs >= 30000L) {
+ logger.info("file_cache_info lifecycle:
cluster=${clusterName}, ${lastState}")
+ lastLogMs = System.currentTimeMillis()
+ }
+ if (!matches || blocks != previous) {
+ stableSince = System.currentTimeMillis()
+ }
+ previous = blocks
+ // A transient empty metadata snapshot is not a completed
clear.
+ if (matches && System.currentTimeMillis() - stableSince >=
3000L) {
+ logger.info("file_cache_info phase complete: ${lastState}")
+ return blocks
+ }
+ sleep(1000)
+ }
+ assertTrue(false, "Timeout waiting for ${phase}:
cluster=${clusterName}, tablet=${tabletId}, " +
+ "BEs=${backendIds}, clear_responses=${clearResponses},
last_state=${lastState}")
+ }
+ def waitForReaders = {
+ long deadlineMs = System.currentTimeMillis() + 60000L
+ long quietSince = System.currentTimeMillis()
+ while (System.currentTimeMillis() < deadlineMs) {
+ def active = sql_return_maparray("SHOW PROCESSLIST").findAll {
row ->
+ row.Db?.toString() == dbName &&
row.Info?.toString()?.contains(tableName) &&
+ !"${row.Command}".equalsIgnoreCase("Sleep")
+ }
+ lastState = [phase: "reader_drain", active_queries: active]
+ if (!active.isEmpty()) {
+ quietSince = System.currentTimeMillis()
+ } else if (System.currentTimeMillis() - quietSince >= 3000L) {
+ return
+ }
+ sleep(1000)
+ }
+ assertTrue(false, "Timeout waiting for table queries to finish:
${lastState}")
+ }
+ def clearCache = { long beId ->
+ def be = backends[beId]
+ String url =
"http://${be.Host}:${be.HttpPort}/api/file_cache?op=clear&sync=true";
+ long startedMs = System.currentTimeMillis()
+ clearResponses[beId] = [url: url]
+ logger.info("file_cache_info clear start: BE=${beId}, url=${url}")
+ def connection = new URL(url).openConnection()
Review Comment:
[P1] Open this request through `Http.openConnection`. In TLS runs the BE
endpoint is HTTPS and may require the configured client certificate, but the
literal `http://` plus `new URL(...).openConnection()` bypasses both. Because
this cache clear is asserted and exceptions are rethrown, the suite aborts
before establishing its cache baseline.
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy:
##########
@@ -3646,6 +3797,18 @@ class Suite implements GroovyInterceptable {
}
}
+ def scp_udf_file_to_all_fe = { udf_file_path ->
+ def udf_file = new File(udf_file_path).absoluteFile
+ assertTrue(udf_file.isFile(), "UDF file does not exist: ${udf_file}")
+ def fe_hosts = sql_return_maparray("SHOW FRONTENDS").collect { it.Host
}.unique()
+ assertTrue(!fe_hosts.isEmpty(), "No frontend found to copy UDF file
to")
+
+ fe_hosts.each { fe_host ->
Review Comment:
[P1] Skip distribution when the sole FE is local, as
`scp_udf_file_to_all_be` already does for the single-BE case. The standard
runner starts one local FE and keeps these artifacts in the same checkout, but
this loop now requires passwordless root SSH/SCP to localhost even though the
destination file is already present. Since both helpers assert command success,
every converted Java/Python UDF suite fails on ordinary single-node runners.
##########
regression-test/suites/cloud_p0/cache/ttl/test_ttl.groovy:
##########
@@ -15,190 +15,284 @@
// specific language governing permissions and limitations
// under the License.
-import org.codehaus.groovy.runtime.IOGroovyMethods
-
-suite("test_ttl") {
- def custoBeConfig = [
- enable_evict_file_cache_in_advance : false,
- file_cache_enter_disk_resource_limit_mode_percent : 99,
- file_cache_background_ttl_gc_interval_ms : 1000,
- file_cache_background_ttl_info_update_interval_ms : 1000,
- file_cache_background_tablet_id_flush_interval_ms : 1000
+suite("test_ttl", "nonConcurrent") {
+ def customBeConfig = [
+ enable_evict_file_cache_in_advance: false,
+ file_cache_enter_disk_resource_limit_mode_percent: 99,
+ file_cache_background_ttl_gc_interval_ms: 1000,
+ file_cache_background_ttl_info_update_interval_ms: 1000,
+ file_cache_background_tablet_id_flush_interval_ms: 1000
]
-
- setBeConfigTemporary(custoBeConfig) {
- def clusters = sql " SHOW CLUSTERS; "
- assertTrue(!clusters.isEmpty())
- def validCluster = clusters[0][0]
- sql """use @${validCluster};""";
- def ttlProperties = """ PROPERTIES("file_cache_ttl_seconds"="180") """
- String[][] backends = sql """ show backends """
- String backendId;
- def backendIdToBackendIP = [:]
- def backendIdToBackendHttpPort = [:]
- def backendIdToBackendBrpcPort = [:]
- for (String[] backend in backends) {
- if (backend[9].equals("true") &&
backend[19].contains("${validCluster}")) {
- backendIdToBackendIP.put(backend[0], backend[1])
- backendIdToBackendHttpPort.put(backend[0], backend[4])
- backendIdToBackendBrpcPort.put(backend[0], backend[5])
+ setBeConfigTemporary(customBeConfig) {
+ String tableName = "test_ttl_natural_expiration"
+ long ttlSeconds = 300L
+ int bucketCount = 4
+ def clusters = sql "SHOW CLUSTERS"
+ assertTrue(!clusters.isEmpty())
+ String clusterName = clusters[0][0].toString()
+ def backends = sql_return_maparray("SHOW BACKENDS").findAll { be ->
+ "${be.Alive}".equalsIgnoreCase("true") &&
+ parseJson(be.Tag.toString()).compute_group_name ==
clusterName
+ }.collectEntries { be -> [(be.BackendId as Long): be] }
+ assertTrue(!backends.isEmpty(), "No alive backends in ${clusterName}")
+ def backendIds = backends.keySet()
+ def tabletIds = []
+ def metadata = [:]
+ def lastState = [:]
+ def metricsBaseline = null
+ def originalQueryCache = sql("select @@enable_sql_cache,
@@enable_query_cache")[0].collect { value ->
+ String setting = value.toString().toLowerCase(Locale.ROOT)
+ assertTrue(setting in ["true", "false", "0", "1"], "Unexpected
cache setting: ${value}")
+ setting
}
- }
- assertEquals(backendIdToBackendIP.size(), 1)
-
- backendId = backendIdToBackendIP.keySet()[0]
- def url = backendIdToBackendIP.get(backendId) + ":" +
backendIdToBackendHttpPort.get(backendId) +
"""/api/file_cache?op=clear&sync=true"""
- logger.info(url)
- def clearFileCache = { check_func ->
- try {
- httpTest {
- endpoint ""
- uri url
- op "get"
- body ""
- check check_func
+ def serverTime = { sql("select unix_timestamp()")[0][0] as Long }
+ def waitUntil = { String stage, long deadlineMs, Closure ready ->
+ while (System.currentTimeMillis() < deadlineMs) {
+ if (ready()) {
+ return
+ }
+ sleep(1000)
}
- } catch (Exception e) {
- logger.error("Failed to clear file cache: ${e.message}")
- throw e
+ assertTrue(false, "Timeout waiting for ${stage}:
cluster=${clusterName}, " +
+ "tablets=${tabletIds}, BEs=${backendIds},
last_state=${lastState}")
}
- }
-
- sql new
File("""${context.file.parent}/../ddl/customer_ttl_delete.sql""").text
- def load_customer_once = { String table ->
- try {
- sql (new
File("""${context.file.parent}/../ddl/${table}.sql""").text + ttlProperties)
- sql """ alter table ${table} set ("disable_auto_compaction" =
"true") """ // no influence from compaction
- def totalRows = 200
- def batchSize = 100
- def commentSuffix = ' ' + ('X' * 50)
- for (int offset = 0; offset < totalRows; offset += batchSize) {
- def sb = new StringBuilder()
- int batchEnd = Math.min(totalRows, offset + batchSize)
- for (int idx = offset; idx < batchEnd; idx++) {
- def customerId = 10001 + idx
- def customerName = String.format('Customer#%09d',
customerId)
- sb.append("""INSERT INTO ${table} VALUES (
- ${customerId},
- '${customerName}',
- 'Address Line 1',
- 15,
- '123-456-7890',
- 12345.67,
- 'AUTOMOBILE',
- 'This is a test comment for the
customer.${commentSuffix}'
- );
- """)
- }
- sql sb.toString()
+ def getCache = {
+ sql "use @${clusterName}"
+ def blocks = [:]
+ def rows = sql """select be_id, cache_path, tablet_id, `hash`,
`offset`, size, lower(type)
+ from information_schema.file_cache_info
+ where tablet_id in (${tabletIds.join(',')}) and be_id in
(${backendIds.join(',')})"""
+ rows.each { row ->
+ def key = [row[0] as Long, row[1].toString(), row[2] as Long,
+ row[3].toString(), row[4] as Long]
+ assertTrue(backendIds.contains(key[0]) &&
tabletIds.contains(key[2]), "Unexpected cache owner: ${row}")
+ assertTrue(key[4] >= 0L && (row[5] as Long) > 0L, "Invalid
cache range: ${row}")
+ assertTrue(!blocks.containsKey(key), "Duplicate cache block:
${key}")
+ blocks[key] = [size: row[5] as Long, type: row[6]?.toString()]
}
- } catch (Exception e) {
- logger.error("Failed to load customer data: ${e.message}")
- throw e
+ blocks
}
- }
-
- def getMetricsMethod = { check_func ->
- try {
- httpTest {
- endpoint backendIdToBackendIP.get(backendId) + ":" +
backendIdToBackendBrpcPort.get(backendId)
- uri "/brpc_metrics"
- op "get"
- check check_func
+ def blockSizes = { Map blocks -> blocks.collectEntries { key, value ->
[(key): value.size] } }
+ def summarizeCache = { Map blocks ->
+ blocks.groupBy { key, value -> [key[0], key[2]] }.collectEntries {
owner, entries ->
+ [(owner): [block_count: entries.size(), bytes:
entries.values().sum(0L) { it.size },
+ by_type: (["ttl", "normal"] +
entries.values().collect { it.type }).unique().collectEntries { type ->
+ def values = entries.values().findAll { it.type
== type }
+ [(type): [block_count: values.size(), bytes:
values.sum(0L) { it.size }]]
+ }]]
}
- } catch (Exception e) {
- logger.error("Failed to get metrics: ${e.message}")
- throw e
}
- }
-
- def getTabletIds = { String tableName ->
- def tablets = sql "show tablets from ${tableName}"
- assertTrue(tablets.size() > 0, "No tablets found for table
${tableName}")
- tablets.collect { it[0] as Long }
- }
-
- def waitForFileCacheType = { List<Long> tabletIds, String expectedType,
long timeoutMs = 60000L, long intervalMs = 2000L ->
- long start = System.currentTimeMillis()
- while (System.currentTimeMillis() - start < timeoutMs) {
- boolean allMatch = true
- for (Long tabletId in tabletIds) {
- def rows = sql "select type from
information_schema.file_cache_info where tablet_id = ${tabletId}"
- if (rows.isEmpty()) {
- logger.warn("file_cache_info is empty for tablet
${tabletId} while waiting for ${expectedType}")
- allMatch = false
- break
+ def metricNames = ["file_cache_ttl_cache_size",
"file_cache_normal_queue_cache_size",
+ "file_cache_ttl_cache_lru_queue_size",
"file_cache_ttl_cache_lru_queue_element_count",
+ "file_cache_normal_queue_element_count",
"file_cache_ttl_cache_evict_size",
+ "file_cache_normal_queue_evict_size",
"file_cache_ttl_mgr_tablet_id_set_size"]
+ def getMetrics = {
+ // These include other tables and update asynchronously. Only
scoped cache data is asserted.
+ backends.collectEntries { id, be ->
+ def values = metricNames.collectEntries { [(it): null] }
+ try {
+ String text = new
URL("http://${be.Host}:${be.BrpcPort}/brpc_metrics";)
+ .getText(connectTimeout: 5000, readTimeout: 5000)
+ text.readLines().each { line ->
+ def fields = line.trim().split(/\s+/)
+ if (fields.size() == 2 && !fields[0].startsWith("#")) {
+ metricNames.each { name ->
+ if (fields[0] == name ||
fields[0].endsWith("_${name}")) {
+ values[name] = (values[name] ?: 0L) +
(fields[1] as Long)
+ }
+ }
+ }
+ }
+ } catch (Exception e) {
+ logger.warn("Cannot read diagnostic metrics for BE ${id}:
${e.message}")
}
- def mismatches = rows.findAll { row ->
!row[0]?.toString()?.equalsIgnoreCase(expectedType) }
- if (!mismatches.isEmpty()) {
- logger.info("tablet ${tabletId} has cache types
${rows.collect { it[0] }} while waiting for ${expectedType}")
- allMatch = false
- break
+ def missing = values.findAll { name, value -> value == null
}.keySet()
+ if (!missing.isEmpty()) {
+ logger.warn("Unavailable diagnostic metrics: BE=${id},
names=${missing}")
}
+ [(id): values]
}
- if (allMatch) {
- logger.info("All file cache entries for tablets ${tabletIds}
are ${expectedType}")
- return
+ }
+ def logMetrics = { String stage ->
+ def current = getMetrics()
+ def delta = backendIds.collectEntries { id ->
+ [(id): metricNames.collectEntries { name ->
+ [(name): metricsBaseline != null &&
metricsBaseline[id][name] != null && current[id][name] != null ?
+ current[id][name] - metricsBaseline[id][name] :
null]
+ }]
}
- sleep(intervalMs)
+ logger.info("TTL metrics: stage=${stage}, cluster=${clusterName},
by_be=${current}, " +
+ "delta_from_ttl_baseline=${delta}")
+ current
}
- assertTrue(false, "Timeout waiting for file_cache_info type
${expectedType} for tablets ${tabletIds}")
- }
-
- clearFileCache.call() {
- respCode, body -> {}
- }
- sleep(10000)
+ String address = "Address Line 1"
+ String phone = "123-456-7890"
+ String segment = "AUTOMOBILE"
+ String comment = "This is a test comment for the customer. " + ("X" *
50)
+ def customerIds = (10001..10200).toList()
+ def customerNames = customerIds.collect {
String.format('Customer#%09d', it) }
+ def normalize = { List values -> values.collect { new
BigDecimal(it.toString()).stripTrailingZeros() } }
+ def expectedData = normalize([customerIds.size(), customerIds.sum(0L),
customerNames.sum(0L) { it.length() },
+ 200L * address.length(), 200L * 15L, 200L * phone.length(),
+ 200G * 12345.67G, 200L * segment.length(), 200L *
comment.length()])
+ def scanTable = {
+ // Read every column; result caches and metadata-only COUNT cannot
satisfy this query.
+ def rows = sql """select count(*), sum(C_CUSTKEY),
sum(length(C_NAME)), sum(length(C_ADDRESS)),
+ sum(C_NATIONKEY), sum(length(rtrim(C_PHONE))), sum(C_ACCTBAL),
+ sum(length(rtrim(C_MKTSEGMENT))), sum(length(C_COMMENT)) from
${tableName}"""
+ assertTrue(rows.size() == 1 && rows[0].size() == 9 &&
rows[0].every { it != null },
+ "Unexpected data result: ${rows}")
+ assertEquals(expectedData, normalize(rows[0]))
+ }
+ Throwable failure = null
+ try {
+ sql "use @${clusterName}"
+ sql "set enable_sql_cache = false"
+ sql "set enable_query_cache = false"
+ logMetrics("before_create")
+ sql "drop table if exists ${tableName} force"
+ long prepareStartedMs = System.currentTimeMillis()
+ // Leave a minute for observing TTL before expiry, even on ASAN
and with delayed metadata flushes.
+ long prepareDeadlineMs = prepareStartedMs + (ttlSeconds - 60L) *
1000L
+ String ddl = new
File("${context.file.parent}/../ddl/customer_ttl.sql").text
+ .replace('customer_ttl', tableName).replace('BUCKETS 32',
"BUCKETS ${bucketCount}")
+ sql(ddl + """ PROPERTIES("file_cache_ttl_seconds"="${ttlSeconds}",
"disable_auto_compaction"="true")""")
+ tabletIds = sql_return_maparray("show tablets from ${tableName}")
+ .collect { it.TabletId as Long }.unique().sort()
+ assertTrue(tabletIds.size() == bucketCount, "Expected
${bucketCount} tablets: ${tabletIds}")
+ String values = customerIds.withIndex().collect { id, i ->
+ "(${id}, '${customerNames[i]}', '${address}', 15, '${phone}',
12345.67, '${segment}', '${comment}')"
+ }.join(",")
+ long insertStartedMs = System.currentTimeMillis()
+ sql "insert into ${tableName} values ${values}"
+ logger.info("TTL fixture: rows=200, tablets=${tabletIds},
insert_elapsed_ms=${System.currentTimeMillis() - insertStartedMs}")
+ scanTable()
- def tabletIds = []
- load_customer_once("customer_ttl")
- sleep(10000)
- tabletIds = getTabletIds.call("customer_ttl")
- waitForFileCacheType.call(tabletIds, "ttl")
- getMetricsMethod.call() {
- respCode, body ->
- assertEquals("${respCode}".toString(), "200")
- String out = "${body}".toString()
- def strs = out.split('\n')
- Boolean flag1 = false;
- long ttl_cache_size = 0;
- for (String line in strs) {
- if (flag1) break;
- if (line.contains("ttl_cache_size")) {
- if (line.startsWith("#")) {
- continue
+ // The full scan above verifies data; cloud header JSON does not
include rowsets.
+ waitUntil("tablet TTL metadata", prepareDeadlineMs) {
+ metadata = [:]
+ sql_return_maparray("show tablets from ${tableName}").each {
tablet ->
+ long id = tablet.TabletId as Long
+ long primary = tablet.PrimaryBackendId as Long
+ if (!(id in tabletIds) || !backends.containsKey(primary)
|| (tablet.BackendId as Long) != primary) {
+ return
+ }
+ def be = backends[primary]
+ try {
+ def meta = parseJson(new
URL("http://${be.Host}:${be.HttpPort}/api/meta/header/${id}";)
Review Comment:
[P1] Route this required metadata read through the TLS-aware HTTP helper.
With `enableTLS=true`, direct cleartext access to the BE fails; the catch then
retries until `prepareDeadlineMs`, so the suite cannot reach its expiry
assertions. `Http.openConnection` supplies both the HTTPS rewrite and
configured SSL context.
##########
regression-test/suites/fault_injection_p0/test_skip_index_compaction_fault_injection.groovy:
##########
@@ -156,13 +168,8 @@ suite("test_skip_index_compaction_fault_injection",
"nonConcurrent") {
has_update_be_config = true
check_config.call("inverted_index_compaction_enable", "true");
-
- try {
-
GetDebugPoint().enableDebugPointForAllBEs("Compaction::open_inverted_index_file_writer")
- run_test.call(tableName2)
- } finally {
-
GetDebugPoint().disableDebugPointForAllBEs("Compaction::open_inverted_index_file_writer")
- }
+ run_test.call(tableName1, "Compaction::open_inverted_index_file_reader")
Review Comment:
[P1] Define and create the table used by this reader-fault case. This suite
only declares and creates `tableName2`; `tableName1` appears nowhere else, so
Groovy raises `MissingPropertyException` here before either compaction scenario
runs. If both faults should share one table, reset or recreate `tableName2`
between them.
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy:
##########
@@ -324,6 +357,117 @@ class Suite implements GroovyInterceptable {
// }
// )
// }
+ /** Wait for continuous version coverage on each tablet's serving BE, with
lazy commit enabled. */
+ void syncAndWaitTabletVersion(Collection<Map> tablets, long version, int
timeoutSeconds = 60) {
+ Assertions.assertFalse(tablets.isEmpty(), "no tablets to synchronize")
+ List<Map> tabletList = tablets.toList()
+ Map<String, List<Map>> tabletGroups = tabletList.groupBy {
it.BackendId.toString() }
+ Map<String, BackendClientImpl> backendClients = [:]
+ Set<Integer> ready = [] as Set
+ Map<String, Object> lastStates = [:]
+ int pollCount = 0
+ try {
+ if (isCloudMode()) {
+ def backendById = sql_return_maparray("SHOW
BACKENDS").collectEntries {
+ [(it.BackendId.toString()): it]
+ }
+ tabletGroups.keySet().each { backendId ->
+ def backend = backendById[backendId]
+ Assertions.assertNotNull(backend,
+ "backend ${backendId} for tablets
${tabletGroups[backendId]*.TabletId} was not found")
+ backendClients[backendId] = new BackendClientImpl(
+ new TNetworkAddress(backend.Host.toString(),
backend.BePort as int),
+ backend.HttpPort as int)
+ }
+ }
+
+ awaitUntil(timeoutSeconds, 0.5) {
+ // The BE RPC is asynchronous. Retry it periodically in case
an earlier queued
+ // task observed no advancement while Meta Service was
finalizing lazy commit.
+ if (!backendClients.isEmpty() && pollCount++ % 10 == 0) {
+ tabletGroups.each { backendId, backendTablets ->
+ backendClients[backendId].client.syncLoadForTablets(
+ new TSyncLoadForTabletsRequest(
+ backendTablets.collect { it.TabletId
as long }))
+ }
+ }
+ tabletList.eachWithIndex { tablet, index ->
+ if (!ready.contains(index)) {
+ def status =
Http.GET(tablet.CompactionStatus.toString(), true, false)
Review Comment:
[P1] Pass the configured HTTP credentials to this poll and the sibling
`MetaUrl` read in `getRowsetMetaAtVersion`. The default `Http.GET` overload
hard-codes `root` with an empty password, but `/api/compaction/show` and
`/api/meta/header` are ADMIN-authenticated when `enable_all_http_auth=true`. A
cluster with the supported nonempty root password therefore gets 403 inside
these centralized readiness helpers. Use the overload that takes
`context.config.feHttpUser` and `feHttpPassword`.
##########
regression-test/suites/audit/test_audit_log_queue_time.groovy:
##########
@@ -16,108 +16,159 @@
// under the License.
suite("test_audit_log_queue_time", "nonConcurrent") {
-
- setGlobalVarTemporary([enable_audit_plugin: true], {
+ // Check admin privilege
+ try {
+ sql "set global enable_audit_plugin = true"
+ } catch (Exception e) {
+ log.warn("skip this case, because " + e.getMessage())
+ assertTrue(e.getMessage().toUpperCase().contains("ADMIN"))
+ return
+ }
def tableName = "audit_queue_time_test"
def wgName = "test_queue_time_wg"
- def testMarker = UUID.randomUUID().toString().substring(0, 8)
+ def maxConcurrency = 1
+ def queueTimeoutMs = 30000
+ def guaranteedQueueTimeMs = 1000
+ def blockerSleepTime = 10
+ def markerPrefix = UUID.randomUUID().toString().substring(0, 8)
+ def blockerMarker = "${markerPrefix}_blocker"
+ def queuedMarker = "${markerPrefix}_queued"
+ def threads = []
+ def queryErrors = Collections.synchronizedList([])
- // Cleanup environment
- sql "drop table if exists ${tableName}"
- sql "drop workload group if exists ${wgName}"
+ def getQueueState = {
+ def row = sql("show workload groups").find { it[1].toString() ==
wgName }
+ if (row == null) {
+ return null
+ }
+ return [
+ running: row[row.size() - 2] as int,
+ waiting: row[row.size() - 1] as int
+ ]
+ }
- // Create test table
- sql """
- CREATE TABLE `${tableName}` (
- `id` bigint,
- `name` varchar(32)
- ) ENGINE=OLAP
- DUPLICATE KEY(`id`)
- DISTRIBUTED BY HASH(`id`) BUCKETS 1
- PROPERTIES ("replication_allocation" = "tag.location.default: 1")
- """
+ def waitForQueueState = { int expectedRunning, int expectedWaiting ->
+ def state = null
+ for (int i = 0; i < 100; i++) {
+ state = getQueueState()
+ if (state != null
+ && state.running == expectedRunning
+ && state.waiting == expectedWaiting) {
+ return
+ }
+ sleep(100)
+ }
+ throw new RuntimeException("workload group ${wgName} did not reach "
+ + "running=${expectedRunning}, waiting=${expectedWaiting};
last state=${state}")
+ }
- sql "insert into ${tableName} values (1, 'test')"
+ try {
+ // Cleanup environment
+ sql "drop table if exists ${tableName}"
+ sql "drop workload group if exists ${wgName}"
+
+ // Create test table
+ sql """
+ CREATE TABLE `${tableName}` (
+ `id` bigint,
+ `name` varchar(32)
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1")
+ """
+
+ sql "insert into ${tableName} values (1, 'test')"
+
+ // max_concurrency=1 ensures that the second query enters the queue.
+ sql """
+ create workload group ${wgName}
+ properties (
+ 'max_concurrency' = '${maxConcurrency}',
+ 'max_queue_size' = '10',
+ 'queue_timeout' = '${queueTimeoutMs}'
+ )
+ """
+
+ // Wait for workload group to take effect.
+ Thread.sleep(5000)
+
+ // Truncate audit_log for easier testing.
+ sql "truncate table __internal_schema.audit_log"
+
+ // Occupy the only running slot before submitting the query whose
queue time is audited.
+ threads << Thread.start {
+ try {
+ sql "set workload_group=${wgName}"
+ sql """
+ select sleep(${blockerSleepTime}), '${blockerMarker}' as
marker
+ from ${tableName} limit 1
+ """
+ } catch (Throwable t) {
+ queryErrors.add(t)
+ }
+ }
+
+ waitForQueueState(maxConcurrency, 0)
- def maxConcurrency = 1
- // Create workload group: max_concurrency=1 ensures queries queue up
- sql """
- create workload group ${wgName}
- properties (
- 'max_concurrency' = '${maxConcurrency}',
- 'max_queue_size' = '10',
- 'queue_timeout' = '30000'
- )
- """
-
- // Wait for workload group to take effect
- Thread.sleep(5000)
-
- // Truncate audit_log for easier testing
- sql "truncate table __internal_schema.audit_log"
-
- // Submit concurrent queries with marker for later lookup
- def sqlSleepTime = 5
- def queuedSqlCnt = 1
- def threads = []
- for (int i = 0; i < maxConcurrency + queuedSqlCnt; i++) {
- def idx = i
threads << Thread.start {
try {
sql "set workload_group=${wgName}"
- // Use sleep function to simulate long query, ensuring
subsequent queries need to queue
sql """
- select sleep(${sqlSleepTime}), '${testMarker}_${idx}' as
marker
+ select id, '${queuedMarker}' as marker
from ${tableName} limit 1
"""
- } catch (Exception e) {
- log.warn("Query ${idx} failed: ${e.getMessage()}")
+ } catch (Throwable t) {
+ queryErrors.add(t)
}
}
- }
- // Wait for all queries to complete
- threads.each { it.join() }
-
- // Wait for audit log to flush
- Thread.sleep(5000)
- sql "call flush_audit_log()"
- Thread.sleep(5000)
-
- // Verify queue_time_ms column exists
- def schemaResult = sql "desc internal.__internal_schema.audit_log"
- def hasQueueTimeMs = schemaResult.any { it[0] == "queue_time_ms" }
- assertTrue(hasQueueTimeMs)
-
- // check result
- def retry = 10
- def query = """
- select query_id, queue_time_ms, stmt
- from __internal_schema.audit_log
- where stmt like '%${testMarker}%'
- and queue_time_ms > 0
- order by time
- """
- def auditResult = sql "${query}"
-
- while (auditResult.isEmpty()) {
- if (retry-- < 0) {
- throw new RuntimeException("It has retried a few but still failed,
you need to check it")
+ waitForQueueState(maxConcurrency, 1)
+ sleep(guaranteedQueueTimeMs)
+ waitForQueueState(maxConcurrency, 1)
+
+ threads.each { it.join() }
+ assertTrue(queryErrors.isEmpty(), "query failures: ${queryErrors}")
+
+ // Verify queue_time_ms column exists.
+ def schemaResult = sql "desc internal.__internal_schema.audit_log"
+ def hasQueueTimeMs = schemaResult.any { it[0] == "queue_time_ms" }
+ assertTrue(hasQueueTimeMs)
+
+ // Match only the queued query. Excluding audit_log statements
prevents the lookup
+ // queries themselves from matching the marker during retries.
+ def query = """
+ select query_id, queue_time_ms, stmt
+ from __internal_schema.audit_log
+ where stmt like '%${queuedMarker}%'
+ and stmt not like '%__internal_schema.audit_log%'
+ order by time
+ """
+ def auditResult = []
+ for (int retry = 0; retry < 10 && auditResult.isEmpty(); retry++) {
+ sql "call flush_audit_log()"
+ sleep(1000)
+ auditResult = sql "${query}"
}
- sql "call flush_audit_log()"
- sleep(3000)
- auditResult = sql "${query}"
- }
- auditResult.each { row ->
- assertTrue(row[1] >= sqlSleepTime * 1000)
+ assertFalse(auditResult.isEmpty(), "queued query was not found in
audit log")
+ logger.info("Queued query audit result: ${auditResult}")
+ auditResult.each { row ->
+ def queueTimeMs = row[1] as long
+ assertTrue(queueTimeMs >= guaranteedQueueTimeMs,
+ "queue_time_ms ${queueTimeMs} is less than guaranteed wait
${guaranteedQueueTimeMs}")
+ assertTrue(queueTimeMs < queueTimeoutMs,
+ "queue_time_ms ${queueTimeMs} reached queue timeout
${queueTimeoutMs}")
+ }
+ } finally {
+ threads.each { thread ->
+ if (thread.isAlive()) {
+ thread.join(blockerSleepTime * 1000 + 5000)
+ }
+ }
+ sql "drop table if exists ${tableName}"
+ sql "drop workload group if exists ${wgName}"
+ sql "set global enable_audit_plugin = false"
Review Comment:
[P1] Restore the original value rather than always disabling the plugin.
This suite previously used `setGlobalVarTemporary`, which snapshots `SHOW
GLOBAL VARIABLES` and restores it in `finally`; the replacement never reads the
origin. Pipelines such as `cloud_p1/conf/session_variables.sql` intentionally
start with `enable_audit_plugin=true`, so a successful run now turns auditing
off for every later suite.
##########
regression-test/suites/fault_injection_p0/cloud/test_cloud_mow_partial_update_retry.groovy:
##########
@@ -97,6 +137,9 @@ suite("test_cloud_mow_partial_update_retry",
"nonConcurrent") {
throw e
} finally {
GetDebugPoint().clearDebugPointsForAllBEs()
+ if (t1 != null && t1.isAlive()) {
Review Comment:
[P2] Verify this worker actually terminates after clearing the debug points.
`Thread.start` creates a non-daemon worker; if an earlier handshake or
assertion fails, this exceptional path waits 60 seconds but lets the suite
return even when the SQL thread remains alive, and it hides
`firstLoadException`. The companion stale-response suite asserts termination
here.
##########
regression-test/suites/schema_change/test_alter_table_column_with_delete_drop_column_dup_key.groovy:
##########
@@ -19,9 +19,19 @@ import org.awaitility.Awaitility
suite("test_alter_table_column_with_delete_drop_column_dup_key",
"schema_change") {
def tbName1 = "alter_table_column_dup_with_delete_drop_column_dup_key"
- def getJobState = { tableName ->
- def jobStateResult = sql """ SHOW ALTER TABLE COLUMN WHERE
IndexName='${tableName}' ORDER BY createtime DESC LIMIT 1 """
- return jobStateResult[0][9]
+ int maxTrySeconds = 1200
Review Comment:
[P1] Finish the wait-helper conversion for every consumer. This hunk removes
the only `max_try_secs`, `res`, and `getJobState` definitions in favor of
`maxTrySeconds`/`waitForColumnState`, but the later second-drop block still
uses all three old names. The suite therefore throws
`MissingPropertyException`; replace that remaining block with
`waitForColumnState(tbName1, "value3", false)`.
##########
regression-test/suites/cloud_p0/multi_cluster/test_apsaradb_internal_stage.groovy:
##########
@@ -94,13 +95,13 @@ suite("test_apsarad_internal_stage_copy_into") {
def tableName = "customer_apsaradb_internal_stage"
def uploadFile = { remoteFilePath, localFilePath ->
+ assertTrue(new File(localFilePath).isFile(), "Missing upload fixture:
${localFilePath}")
StringBuilder strBuilder = new StringBuilder()
- strBuilder.append("""curl -u """ + context.config.feCloudHttpUser +
":" + context.config.feCloudHttpPassword)
+ strBuilder.append("""curl -u """ + context.config.feHttpUser + ":" +
context.config.feHttpPassword)
strBuilder.append(""" -H fileName:""" + remoteFilePath)
strBuilder.append(""" -H host:""" + "private")
strBuilder.append(""" -T """ + localFilePath)
- def feHttpAddress = context.config.isDorisEnv ?
context.config.feHttpAddress : context.config.feCloudHttpAddress
- strBuilder.append(""" -L http://"""; + feHttpAddress +
"""/copy/upload""")
+ strBuilder.append(""" -L http://"""; + context.config.feHttpAddress +
"""/copy/upload""")
Review Comment:
[P1] Apply the suite's TLS curl helpers to this upload. The changed request
now targets `feHttpAddress`, but remains hard-coded to `http://` and supplies
none of `getDorisCurlTlsOptions()`. Against an mTLS FE, curl exits nonzero and
the asserted upload blocks the COPY test. Please use `getDorisHttpScheme()` and
the configured cert/key/CA options.
##########
regression-test/plugins/plugin_compaction.groovy:
##########
@@ -19,50 +19,51 @@ import org.apache.doris.regression.suite.Suite
import java.util.concurrent.TimeUnit
import org.awaitility.Awaitility;
-Suite.metaClass.be_get_compaction_status{ String ip, String port, String
tablet_id /* param */->
- return curl("GET",
String.format("http://%s:%s/api/compaction/run_status?tablet_id=%s";, ip, port,
tablet_id),
- null, 10, context.config.feHttpUser, context.config.feHttpPassword)
+Suite.metaClass.be_get_compaction_status{ String ip, String port, String
tablet_id,
+ Integer timeout_sec = 10, Integer
max_retries = 10 /* param */->
+ return delegate.curl("GET",
String.format("http://%s:%s/api/compaction/run_status?tablet_id=%s";, ip, port,
tablet_id),
+ null, timeout_sec, "", "", max_retries)
Review Comment:
[P1] Preserve the configured HTTP credentials in these compaction requests.
This call now explicitly passes an empty user/password, and the other
refactored helpers use the same empty defaults, so curl sends no Basic auth.
Local and cloud compaction endpoints are ADMIN-authenticated and return 401
when `enable_all_http_auth=true`; keep the new timeout/retry arguments but pass
`context.config.feHttpUser` and `feHttpPassword` as before.
##########
regression-test/suites/compaction/test_base_compaction_with_dup_key_max_file_size_limit.groovy:
##########
@@ -182,35 +143,33 @@
suite("test_base_compaction_with_dup_key_max_file_size_limit", "p2") {
// [0-3] 2G nooverlapping
// [4-4] 1G nooverlapping
// cp: 5
- trigger_and_wait_compaction(tableName, "cumulative")
+ trigger_and_wait_compaction(tableName, "cumulative",
compactionTimeoutSeconds)
+
+ // The rowset layout is complete. Enable the limit only for the
operation under test so it
+ // cannot interfere with construction of the large base rowset.
+ set_be_param("base_compaction_dup_key_max_file_size_mbytes", "512")
- // Due to the limit of
config::base_compaction_dup_key_max_file_size_mbytes(1G),
- // can not do base compaction, return E-808
+ // The first input rowset is now larger than the 512MB limit, so the
size gate filters it
+ // and manual base compaction must be rejected with E-808.
// rowsets:
// [0-3] 2G nooverlapping
// [4-4] 1G nooverlapping
// cp: 5
- String trigger_backend_host =
backendId_to_backendIP[trigger_backend_id]
- String trigger_backend_http_port =
backendId_to_backendHttpPort[trigger_backend_id]
- def baseCompactionResult = triggerCompaction(trigger_backend_host,
trigger_backend_http_port,
- "base", tablet_id)
- String lastBaseStatus = baseCompactionResult
- if (!baseCompactionResult.contains("E-808")) {
- // Manual compaction may return success after its 2s async wait
before
- // the background task records BE_NO_SUITABLE_VERSION.
- for (int i = 0; i < 300; i++) {
- lastBaseStatus = getBaseCompactionStatus(trigger_backend_host,
trigger_backend_http_port, tablet_id)
- if (lastBaseStatus.contains("E-808")) {
- break
- }
- sleep(1000)
- }
- }
- assertTrue(lastBaseStatus.contains("E-808"),
- "base compaction result does not contain E-808,
result=${baseCompactionResult}, lastBaseStatus=${lastBaseStatus}");
+ def (compactionCode, compactionOut, compactionErr) =
be_run_base_compaction(
+ backendId_to_backendIP[trigger_backend_id],
+ backendId_to_backendHttpPort[trigger_backend_id], tablet_id)
+ logger.info("Run expected-to-fail base compaction:
code=${compactionCode}, out=${compactionOut}, err=${compactionErr}")
+ assertEquals(0, compactionCode)
+ assertTrue(compactionOut.contains("E-808"), "Expected E-808, actual
response: ${compactionOut}")
Review Comment:
[P1] Keep the bounded `last base status` poll when the immediate response is
`Success`. In local mode this endpoint runs base compaction in a detached
thread, waits only two seconds, and deliberately returns success even though
rowset selection can publish E-808 later. The removed code handled that
contract; asserting E-808 only in this response makes the p2 suite
timing-dependent.
##########
regression-test/suites/inverted_index_p2/show_data/test_show_index_data_p2.groovy:
##########
@@ -361,29 +385,14 @@ suite("test_show_index_data_p2", "p2") {
// 1. load data
def executor = Executors.newFixedThreadPool(5)
- def loadFutures = (1..110).collect { i ->
- def fileName = "documents-" + i + ".json"
- executor.submit({
+ (1..110).each { i ->
+ executor.submit {
Review Comment:
[P1] Retain these futures and call `get()` on each before continuing.
`submit` stores worker exceptions in the returned `Future`, so discarding all
110 results makes failed stream-load assertions invisible; the code also
ignores a false return from `awaitTermination`, allowing later
compaction/schema-change checks to race workers still running.
##########
regression-test/suites/query_profile/s3_load_profile_test.groovy:
##########
@@ -17,20 +17,45 @@
import groovy.json.JsonSlurper
import org.apache.doris.regression.action.ProfileAction
+import org.apache.doris.regression.util.Http
def fetchProfile = { masterHTTPAddr, id ->
- def dst = 'http://' + masterHTTPAddr
- def conn = new URL(dst +
"/api/profile/text/?query_id=$id").openConnection()
- conn.setRequestMethod("GET")
- def encoding =
Base64.getEncoder().encodeToString((context.config.feHttpUser + ":" +
- (context.config.feHttpPassword == null ? "" :
context.config.feHttpPassword)).getBytes("UTF-8"))
- conn.setRequestProperty("Authorization", "Basic ${encoding}")
- return conn.getInputStream().getText()
+ def user = context.config.isCloudMode() ? context.config.feCloudHttpUser :
context.config.feHttpUser
Review Comment:
[P1] Authenticate these profile reads with `feHttpUser`/`feHttpPassword`
even in cloud mode. `masterHTTPAddr` comes from `SHOW FRONTENDS`, so these are
FE-local profile endpoints, not calls to the separate cloud HTTP endpoint. This
PR makes the same correction in `ProfileAction`; when the credential sets
differ, this branch receives 401 while locating or fetching the load profile.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
