This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 9c05b33a7b7 [test](regression) Improve BE cancel_query workload policy
regression coverage (#65145)
9c05b33a7b7 is described below
commit 9c05b33a7b7c7f11c1542b3d01f8e7bc3f7cfcf3
Author: Wen Zhenghu <[email protected]>
AuthorDate: Fri Jul 10 16:02:44 2026 +0800
[test](regression) Improve BE cancel_query workload policy regression
coverage (#65145)
This PR improves regression coverage for BE-side
workload policy cancel_query behavior. It adds dedicated regression
coverage for query_time, be_scan_rows, be_scan_bytes, and
query_be_memory_bytes, and adjusts the remote scan bytes workload policy
suite to preserve and restore the FE publish interval while using a
longer wait window for BE-side policy propagation. It also aligns the BE
cancel message assertions with the metric names returned by the BE
runtime.
Note: publish_topic_info_interval_ms is currently mutable in
configuration only. Updating it at runtime does not effectively change
the actual publish interval of the already started TopicPublisherThread,
so tests cannot rely on lowering this FE config to make BE policy
publication take effect immediately.
---
.../test_workload_policy_remote_scan_bytes.groovy | 127 ++++++++++++----
..._be_workload_policy_cancel_query_metrics.groovy | 162 +++++++++++++++++++++
2 files changed, 264 insertions(+), 25 deletions(-)
diff --git
a/regression-test/suites/external_table_p0/hive/test_workload_policy_remote_scan_bytes.groovy
b/regression-test/suites/external_table_p0/hive/test_workload_policy_remote_scan_bytes.groovy
index 3b6cb46f1b3..b790e421073 100644
---
a/regression-test/suites/external_table_p0/hive/test_workload_policy_remote_scan_bytes.groovy
+++
b/regression-test/suites/external_table_p0/hive/test_workload_policy_remote_scan_bytes.groovy
@@ -42,6 +42,98 @@ suite("test_workload_policy_remote_scan_bytes",
"p0,external") {
forComputeGroupStr = " for ${validCluster} "
}
+ // Wait for the default FE->BE publish cycle instead of mutating a shared
FE config.
+ def waitForBePolicyPublish = {
+ Thread.sleep(40000)
+ }
+
+ // Flush and poll audit rows through a separate connection so diagnostics
do not
+ // inherit the tested workload group or change the primary assertion
result.
+ def fetchAuditRowsForMarker = { queryMarker ->
+ return connect(context.config.jdbcUser, context.config.jdbcPassword,
context.config.jdbcUrl) {
+ sql """SET workload_group = ''"""
+ sql """SET enable_sql_cache = false"""
+ sql """CALL flush_audit_log()"""
+ int retry = 20
+ def auditRows = []
+ String auditSql = """
+ SELECT time, state, error_message, scan_bytes, scan_rows,
+ scan_bytes_from_remote_storage,
scan_bytes_from_local_storage,
+ workload_group, stmt
+ FROM __internal_schema.audit_log
+ WHERE stmt LIKE '%${queryMarker}%'
+ AND workload_group = '${workloadGroupName}'
+ ORDER BY time DESC
+ LIMIT 5
+ """
+ while (retry-- >= 0) {
+ auditRows = sql auditSql
+ if (!auditRows.isEmpty()) {
+ return auditRows
+ }
+ sleep(3000)
+ sql """CALL flush_audit_log()"""
+ }
+ return auditRows
+ }
+ }
+
+ def runRemoteScanPolicyCase = { String lineitemDb, boolean
enableFileScannerV2,
+ boolean requireCancellation ->
+ String scannerName = enableFileScannerV2 ? "FileScannerV2" :
"FileScanner"
+ String queryMarker = "remote_scan_${enableFileScannerV2 ? 'v2' :
'legacy'}_probe_"
+ + System.currentTimeMillis()
+ Throwable queryException = null
+
+ logger.info("Run remote scan bytes workload policy test item with
${scannerName}, "
+ + "requireCancellation=${requireCancellation}")
+ sql """SET workload_group = '${workloadGroupName}'"""
+ sql """SET enable_file_cache = false"""
+ sql """SET enable_sql_cache = false"""
+ sql """SET enable_file_scanner_v2 = ${enableFileScannerV2}"""
+ try {
+ sql """
+ SELECT /* ${queryMarker} */ SUM(SLEEP(1) + l_quantity)
+ FROM (
+ SELECT l_quantity
+ FROM ${catalogName}.${lineitemDb}.lineitem
+ LIMIT 10
+ ) s
+ """
+ } catch (Throwable t) {
+ queryException = t
+ }
+
+ String msg = queryException == null ? null :
queryException.getMessage()
+ boolean cancelledByExpectedPolicy = msg != null
+ && msg.contains("cancelled by workload policy: ${policyName}")
+ && msg.contains("scan_bytes_from_remote_storage")
+
+ if (queryException != null) {
+ assertTrue(cancelledByExpectedPolicy,
+ "unexpected ${scannerName} query failure: " + msg)
+ } else if (requireCancellation) {
+ def auditRows = fetchAuditRowsForMarker(queryMarker)
+ logger.info("Remote scan bytes workload policy audit rows for
${queryMarker}: " + auditRows)
+ assertTrue(queryException != null,
+ "${scannerName} query should be cancelled by remote scan
bytes workload policy")
+ } else {
+ def auditRows = fetchAuditRowsForMarker(queryMarker)
+ assertFalse(auditRows.isEmpty(),
+ "expected an audit row for the non-blocking ${scannerName}
validation")
+ logger.info("${scannerName} remote scan bytes cancellation is not
enforced yet; "
+ + "audit rows for ${queryMarker}: " + auditRows)
+ }
+
+ if (cancelledByExpectedPolicy) {
+ logger.info("Remote scan bytes workload policy cancel message for
${scannerName}: " + msg)
+ }
+ }
+
+ def scannerV2Variable = sql """SHOW VARIABLES LIKE
'enable_file_scanner_v2'"""
+ assertEquals(1, scannerV2Variable.size())
+ String originalEnableFileScannerV2 = scannerV2Variable[0][1].toString()
+
try {
sql """DROP WORKLOAD POLICY IF EXISTS ${policyName}"""
sql """DROP WORKLOAD POLICY IF EXISTS ${invalidPolicyName}"""
@@ -103,38 +195,23 @@ suite("test_workload_policy_remote_scan_bytes",
"p0,external") {
assertEquals(policyName, policy[0][0])
assertTrue(policy[0][1].toString().contains("be_scan_bytes_from_remote_storage
> 1"))
- // Wait for FE to publish the new BE-side policy before issuing the
query.
- Thread.sleep(15000)
+ waitForBePolicyPublish()
- Throwable queryException = null
- sql """SET workload_group = '${workloadGroupName}'"""
- sql """SET enable_file_cache = false"""
- sql """SET enable_sql_cache = false"""
- try {
- sql """
- SELECT SUM(SLEEP(1) + l_quantity)
- FROM (
- SELECT l_quantity
- FROM ${catalogName}.${lineitemDb}.lineitem
- LIMIT 10
- ) s
- """
- } catch (Throwable t) {
- queryException = t
- }
- assertTrue(queryException != null, "query should be cancelled by
remote scan bytes workload policy")
- String msg = queryException.getMessage()
- logger.info("Remote scan bytes workload policy cancel message: " + msg)
- assertTrue(msg != null && msg.contains("cancelled by workload policy:
${policyName}"),
- "unexpected cancel policy: " + msg)
- assertTrue(msg.contains("scan_bytes_from_remote_storage"),
- "remote scan bytes counter is missing from cancel message: " +
msg)
+ // Test item 1: the legacy FileScanner already publishes remote scan
bytes to the
+ // query-level IO context, so cancellation is required and fully
asserted.
+ runRemoteScanPolicyCase(lineitemDb, false, true)
+
+ // Test item 2: FileScannerV2 currently does not publish file scan
bytes to the
+ // query-level IO context. Keep this path as non-blocking coverage
until the BE-side
+ // FileScannerV2 accounting fix is merged, then change
requireCancellation to true.
+ runRemoteScanPolicyCase(lineitemDb, true, false)
} finally {
try {
sql """SET workload_group = ''"""
} catch (Throwable t) {
logger.info("ignore reset workload_group failure: " +
t.getMessage())
}
+ sql """SET enable_file_scanner_v2 = ${originalEnableFileScannerV2}"""
sql """DROP WORKLOAD POLICY IF EXISTS ${policyName}"""
sql """DROP WORKLOAD POLICY IF EXISTS ${invalidPolicyName}"""
sql """DROP WORKLOAD GROUP IF EXISTS ${workloadGroupName}
${forComputeGroupStr}"""
diff --git
a/regression-test/suites/workload_manager_p0/test_be_workload_policy_cancel_query_metrics.groovy
b/regression-test/suites/workload_manager_p0/test_be_workload_policy_cancel_query_metrics.groovy
new file mode 100644
index 00000000000..006b7a15416
--- /dev/null
+++
b/regression-test/suites/workload_manager_p0/test_be_workload_policy_cancel_query_metrics.groovy
@@ -0,0 +1,162 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_be_workload_policy_cancel_query_metrics") {
+ String dbName = context.config.getDbNameByFile(context.file)
+ String workloadGroupName = "test_be_cancel_query_metrics_wg"
+ String forComputeGroupStr = ""
+ String currentCgName = ""
+ if (isCloudMode()) {
+ def clusters = sql "SHOW CLUSTERS"
+ assertTrue(!clusters.isEmpty())
+ String validCluster = clusters[0][0]
+ currentCgName = "${validCluster}."
+ forComputeGroupStr = " for ${validCluster} "
+ }
+
+ // Wait for the default FE->BE publish cycle instead of mutating a shared
FE config.
+ def waitForBePolicyPublish = {
+ Thread.sleep(40000)
+ }
+
+ // Verify that the query was cancelled by the expected BE policy and that
+ // the cancellation message exposes the metric that triggered it.
+ def assertCancelledByPolicy = { querySql, policyName, metricName ->
+ Throwable queryException = null
+ try {
+ sql querySql
+ } catch (Throwable t) {
+ queryException = t
+ }
+ assertTrue(queryException != null, "query should be cancelled by
workload policy ${policyName}")
+ String msg = queryException.getMessage()
+ logger.info("BE workload policy cancel message for ${policyName}:
${msg}")
+ assertTrue(msg != null && msg.contains("cancelled by workload policy:
${policyName}"),
+ "unexpected cancel policy: " + msg)
+ assertTrue(msg.contains(metricName), "missing trigger metric
${metricName}: " + msg)
+ }
+
+ // Build isolated workload groups so all policies can be published once
and verified independently.
+ def policyCases = [
+ [
+ workloadGroupName: "${workloadGroupName}_time",
+ policyName : "test_be_cancel_query_time_policy",
+ conditionSql : "query_time > 1",
+ metricName : "query_time"
+ ],
+ [
+ workloadGroupName: "${workloadGroupName}_scan_rows",
+ policyName : "test_be_cancel_query_scan_rows_policy",
+ conditionSql : "be_scan_rows > 1",
+ metricName : "scan_rows"
+ ],
+ [
+ workloadGroupName: "${workloadGroupName}_scan_bytes",
+ policyName : "test_be_cancel_query_scan_bytes_policy",
+ conditionSql : "be_scan_bytes > 1",
+ metricName : "scan_bytes"
+ ],
+ [
+ workloadGroupName: "${workloadGroupName}_memory",
+ policyName : "test_be_cancel_query_memory_policy",
+ conditionSql : "query_be_memory_bytes > 1",
+ metricName : "query_memory"
+ ]
+ ]
+
+ try {
+ sql """CREATE DATABASE IF NOT EXISTS ${dbName}"""
+ sql """DROP TABLE IF EXISTS
${dbName}.test_be_workload_policy_cancel_query_metrics_tbl"""
+ sql """
+ CREATE TABLE
${dbName}.test_be_workload_policy_cancel_query_metrics_tbl (
+ id INT,
+ payload STRING
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num" = "1")
+ """
+ sql """
+ INSERT INTO
${dbName}.test_be_workload_policy_cancel_query_metrics_tbl
+ SELECT number, REPEAT('x', 4096)
+ FROM numbers("number" = "200")
+ """
+
+ // Use a scan-heavy query for time/rows/bytes and a sort-heavy query
for
+ // memory so the BE runtime counters can trip the dedicated policy.
+ String scanQuerySql = """
+ SELECT SUM(SLEEP(1) + id)
+ FROM (
+ SELECT id
+ FROM ${dbName}.test_be_workload_policy_cancel_query_metrics_tbl
+ ORDER BY id
+ LIMIT 10
+ ) s
+ """
+ String memoryQuerySql = """
+ SELECT SUM(SLEEP(1) + payload_len)
+ FROM (
+ SELECT LENGTH(payload) AS payload_len
+ FROM ${dbName}.test_be_workload_policy_cancel_query_metrics_tbl
+ ORDER BY payload
+ LIMIT 10
+ ) s
+ """
+
+ policyCases.each { policyCase ->
+ sql """DROP WORKLOAD POLICY IF EXISTS ${policyCase.policyName}"""
+ sql """DROP WORKLOAD GROUP IF EXISTS
${policyCase.workloadGroupName} ${forComputeGroupStr}"""
+ sql """
+ CREATE WORKLOAD GROUP ${policyCase.workloadGroupName}
${forComputeGroupStr}
+ PROPERTIES ('max_cpu_percent' = '100')
+ """
+ }
+
+ // Publish all BE-side policies together and reuse the same
propagation wait.
+ policyCases.each { policyCase ->
+ sql """
+ CREATE WORKLOAD POLICY ${policyCase.policyName}
+ CONDITIONS(${policyCase.conditionSql})
+ ACTIONS(cancel_query)
+ PROPERTIES(
+ 'priority' = '100',
+ 'workload_group' =
'${currentCgName}${policyCase.workloadGroupName}'
+ )
+ """
+ }
+
+ waitForBePolicyPublish()
+
+ // Switch workload groups between assertions so each query can only
match one policy.
+ policyCases.each { policyCase ->
+ sql """SET workload_group = '${policyCase.workloadGroupName}'"""
+ sql """SET enable_sql_cache = false"""
+ String querySql = policyCase.metricName == "query_memory" ?
memoryQuerySql : scanQuerySql
+ assertCancelledByPolicy(querySql, policyCase.policyName,
policyCase.metricName)
+ }
+ } finally {
+ try {
+ sql """SET workload_group = ''"""
+ } catch (Throwable t) {
+ logger.info("ignore reset workload_group failure: " +
t.getMessage())
+ }
+ policyCases.each { policyCase ->
+ sql """DROP WORKLOAD POLICY IF EXISTS ${policyCase.policyName}"""
+ sql """DROP WORKLOAD GROUP IF EXISTS
${policyCase.workloadGroupName} ${forComputeGroupStr}"""
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
