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 482d6d12d6f branch-4.1: [feat](audit log) add queue_time_ms in
audit_log (#66641)
482d6d12d6f is described below
commit 482d6d12d6f0c7b0f36bc2c9bd39dee1ecdaa508
Author: bobhan1 <[email protected]>
AuthorDate: Tue Aug 11 18:39:40 2026 +0800
branch-4.1: [feat](audit log) add queue_time_ms in audit_log (#66641)
### What problem does this PR solve?
Related PR: #60418
Problem Summary:
Backport #60418 to `branch-4.1` for CIR-21335.
The audit log's `query_time` includes workload-group queue time, so
users cannot distinguish queue delay from the remaining end-to-end query
latency. This PR adds `queue_time_ms` to `__internal_schema.audit_log`
and propagates it through the audit event and loader paths.
`branch-4.1` already contained the `getQueueTimeMs()` helper, so the
conflict resolution keeps that implementation and ports the missing
schema, event, loader, call-site, unit-test, and regression-test
changes.
### Release note
Add `queue_time_ms` to `__internal_schema.audit_log`.
### Validation
- `./run-fe-ut.sh --run org.apache.doris.qe.AuditEventProcessorTest`
- `./build.sh --be --fe -j100`
- `./run-regression-test.sh --run -d regression-test/suites/audit -s
test_audit_log_queue_time`
- `git diff --check upstream/branch-4.1...HEAD`
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
Co-authored-by: Fy <[email protected]>
---
.../org/apache/doris/catalog/InternalSchema.java | 2 +
.../java/org/apache/doris/plugin/AuditEvent.java | 7 ++
.../org/apache/doris/plugin/audit/AuditLoader.java | 1 +
.../java/org/apache/doris/qe/AuditLogHelper.java | 3 +
.../apache/doris/qe/AuditEventProcessorTest.java | 2 +
.../data/audit/test_audit_log_behavior.out | 1 +
.../suites/audit/test_audit_log_queue_time.groovy | 129 +++++++++++++++++++++
7 files changed, 145 insertions(+)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchema.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchema.java
index ac8f58d1fa7..59f29cf2d89 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchema.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchema.java
@@ -152,6 +152,8 @@ public class InternalSchema {
// execution info
AUDIT_SCHEMA.add(new ColumnDef("query_time",
TypeDef.create(PrimitiveType.BIGINT),
ColumnNullableType.NULLABLE));
+ AUDIT_SCHEMA.add(new ColumnDef("queue_time_ms",
+ TypeDef.create(PrimitiveType.BIGINT),
ColumnNullableType.NULLABLE));
AUDIT_SCHEMA.add(new ColumnDef("cpu_time_ms",
TypeDef.create(PrimitiveType.BIGINT),
ColumnNullableType.NULLABLE));
AUDIT_SCHEMA.add(new ColumnDef("peak_memory_bytes",
diff --git a/fe/fe-core/src/main/java/org/apache/doris/plugin/AuditEvent.java
b/fe/fe-core/src/main/java/org/apache/doris/plugin/AuditEvent.java
index 3b08dc382d4..c54c32f6a18 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/plugin/AuditEvent.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/plugin/AuditEvent.java
@@ -84,6 +84,8 @@ public class AuditEvent {
// execution info
@AuditField(value = "Time(ms)", colName = "query_time")
public long queryTime = -1;
+ @AuditField(value = "QueueTimeMs", colName = "queue_time_ms")
+ public long queueTimeMs = -1;
@AuditField(value = "CpuTimeMS", colName = "cpu_time_ms")
public long cpuTimeMs = -1;
@AuditField(value = "PeakMemoryBytes", colName = "peak_memory_bytes")
@@ -231,6 +233,11 @@ public class AuditEvent {
return this;
}
+ public AuditEventBuilder setQueueTimeMs(long queueTimeMs) {
+ auditEvent.queueTimeMs = queueTimeMs;
+ return this;
+ }
+
public AuditEventBuilder setScanBytes(long scanBytes) {
auditEvent.scanBytes = scanBytes;
return this;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java
b/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java
index 1d4aeddb3e7..bbc86646635 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java
@@ -168,6 +168,7 @@ public class AuditLoader extends Plugin implements
AuditPlugin {
// execution info
logBuffer.append(event.queryTime).append(AUDIT_TABLE_COL_SEPARATOR);
+ logBuffer.append(event.queueTimeMs).append(AUDIT_TABLE_COL_SEPARATOR);
logBuffer.append(event.cpuTimeMs).append(AUDIT_TABLE_COL_SEPARATOR);
logBuffer.append(event.peakMemoryBytes).append(AUDIT_TABLE_COL_SEPARATOR);
logBuffer.append(event.scanBytes).append(AUDIT_TABLE_COL_SEPARATOR);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java
index d6c992412c2..07681fc8b4a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java
@@ -196,6 +196,8 @@ public class AuditLogHelper {
}
String cluster = Config.isCloudMode() ? cloudCluster : "";
String stmtType = getStmtType(parsedStmt);
+ long queueTimeMs = getQueueTimeMs(ctx);
+
AuditEventBuilder auditEventBuilder = ctx.getAuditEventBuilder();
// ATTN: MUST reset, otherwise, the same AuditEventBuilder instance
will be used in the next query.
@@ -214,6 +216,7 @@ public class AuditLogHelper {
.setErrorMessage((ctx.getState().getErrorMessage() == null ?
"" :
ctx.getState().getErrorMessage().replace("\n", "
").replace("\t", " ")))
.setQueryTime(elapseMs)
+ .setQueueTimeMs(queueTimeMs)
.setCpuTimeMs(statistics == null ? 0 : statistics.getCpuMs())
.setPeakMemoryBytes(statistics == null ? 0 :
statistics.getMaxPeakMemoryBytes())
.setScanBytes(statistics == null ? 0 :
statistics.getScanBytes())
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/AuditEventProcessorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/AuditEventProcessorTest.java
index f80b485609c..884f9a5badf 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/AuditEventProcessorTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/AuditEventProcessorTest.java
@@ -58,6 +58,7 @@ public class AuditEventProcessorTest {
.setDb("db1")
.setState("EOF")
.setQueryTime(2000)
+ .setQueueTimeMs(2000)
.setScanBytes(100000)
.setScanRows(200000)
.setReturnRows(1)
@@ -68,6 +69,7 @@ public class AuditEventProcessorTest {
Assert.assertEquals("127.0.0.1", event.clientIp);
Assert.assertEquals(200000, event.scanRows);
Assert.assertEquals("SELECT", event.stmtType);
+ Assert.assertEquals(2000, event.queueTimeMs);
}
@Test
diff --git a/regression-test/data/audit/test_audit_log_behavior.out
b/regression-test/data/audit/test_audit_log_behavior.out
index ebf50840302..f0a40463e9a 100644
--- a/regression-test/data/audit/test_audit_log_behavior.out
+++ b/regression-test/data/audit/test_audit_log_behavior.out
@@ -11,6 +11,7 @@ state varchar(128) Yes false \N NONE
error_code int Yes false \N NONE
error_message text Yes false \N NONE
query_time bigint Yes false \N NONE
+queue_time_ms bigint Yes false \N NONE
cpu_time_ms bigint Yes false \N NONE
peak_memory_bytes bigint Yes false \N NONE
scan_bytes bigint Yes false \N NONE
diff --git a/regression-test/suites/audit/test_audit_log_queue_time.groovy
b/regression-test/suites/audit/test_audit_log_queue_time.groovy
new file mode 100644
index 00000000000..e40d1688c42
--- /dev/null
+++ b/regression-test/suites/audit/test_audit_log_queue_time.groovy
@@ -0,0 +1,129 @@
+// 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_audit_log_queue_time", "nonConcurrent") {
+ // 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)
+
+ // 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')"
+
+ 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
+ from ${tableName} limit 1
+ """
+ } catch (Exception e) {
+ log.warn("Query ${idx} failed: ${e.getMessage()}")
+ }
+ }
+ }
+
+ // 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")
+ }
+ sql "call flush_audit_log()"
+ sleep(3000)
+ auditResult = sql "${query}"
+ }
+
+ auditResult.each { row ->
+ assertTrue(row[1] >= sqlSleepTime * 1000)
+ }
+
+ assertTrue(auditResult.size() >= queuedSqlCnt)
+
+ // Cleanup
+ sql "drop table if exists ${tableName}"
+ sql "drop workload group if exists ${wgName}"
+ sql "set global enable_audit_plugin = false"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
