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 a0c5a870caa branch-4.1: [test](regression) Improve BE cancel_query
workload policy regression coverage #65145 (#65458)
a0c5a870caa is described below
commit a0c5a870caa40306779a59c947191759618c7f0e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Jul 13 15:11:56 2026 +0800
branch-4.1: [test](regression) Improve BE cancel_query workload policy
regression coverage #65145 (#65458)
Cherry-picked from #65145
Co-authored-by: Wen Zhenghu <[email protected]>
---
.../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]
