From b42f935cb109749d23472e7527063508fde5bec8 Mon Sep 17 00:00:00 2001 From: yujun Date: Fri, 17 Jul 2026 12:15:09 +0800 Subject: [PATCH] [test](fe) Cover internal query audit failure in FE unit test (#65696) The previous regression coverage for internal query audit failures depended on querying `__internal_schema.audit_log` after a fault-injection failure. That path is timing-sensitive because audit persistence is asynchronous, so the regression can fail even when `StmtExecutor.executeInternalQuery()` already reports the error audit event correctly. This PR replaces that flaky coverage with a deterministic FE unit test. The new test forces a planning failure in `executeInternalQuery()` and verifies that the submitted `AuditEvent` is marked as `ERR` and carries the failure details. The unstable fault-injection regression case is removed. --- .../qe/StmtExecutorInternalQueryTest.java | 48 ++++++++- ...st_audit_log_internal_query_failure.groovy | 98 ------------------- 2 files changed, 45 insertions(+), 101 deletions(-) delete mode 100644 regression-test/suites/fault_injection_p0/test_audit_log_internal_query_failure.groovy diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorInternalQueryTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorInternalQueryTest.java index 528d37895a809c..35df59c61e63c5 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorInternalQueryTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorInternalQueryTest.java @@ -18,13 +18,19 @@ package org.apache.doris.qe; import org.apache.doris.analysis.StatementBase; +import org.apache.doris.catalog.Env; import org.apache.doris.common.ErrorCode; +import org.apache.doris.common.jmockit.Deencapsulation; import org.apache.doris.nereids.NereidsPlanner; +import org.apache.doris.plugin.AuditEvent; +import org.apache.doris.resource.workloadschedpolicy.WorkloadRuntimeStatusMgr; import mockit.Mock; import mockit.MockUp; import org.junit.Assert; import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; public class StmtExecutorInternalQueryTest { @Test @@ -46,9 +52,6 @@ public void plan(StatementBase queryStmt, org.apache.doris.thrift.TQueryOptions @Test public void testExecuteInternalQuerySetsErrorStateOnFailure() { - // Regression test for CIR-20019: when the internal SQL execution throws, - // ConnectContext state must be set to ERR so AuditLogHelper records the failure - // instead of misleadingly logging State=OK with empty error message. ConnectContext ctx = new ConnectContext(); ctx.setThreadLocalInfo(); StmtExecutor executor = new StmtExecutor(ctx, "select * from table1"); @@ -73,4 +76,43 @@ public void plan(StatementBase queryStmt, org.apache.doris.thrift.TQueryOptions ConnectContext.remove(); } } + + @Test + public void testExecuteInternalQuerySubmitsErrorAuditEventOnFailure() { + ConnectContext ctx = new ConnectContext(); + ctx.setThreadLocalInfo(); + StmtExecutor executor = new StmtExecutor(ctx, "select * from table1"); + Env env = Env.getCurrentEnv(); + WorkloadRuntimeStatusMgr originalWorkloadRuntimeStatusMgr = env.getWorkloadRuntimeStatusMgr(); + WorkloadRuntimeStatusMgr workloadRuntimeStatusMgr = Mockito.mock(WorkloadRuntimeStatusMgr.class); + ArgumentCaptor auditEventCaptor = ArgumentCaptor.forClass(AuditEvent.class); + + Deencapsulation.setField(env, "workloadRuntimeStatusMgr", workloadRuntimeStatusMgr); + new MockUp() { + @Mock + public void plan(StatementBase queryStmt, org.apache.doris.thrift.TQueryOptions queryOptions) { + throw new RuntimeException("mock plan failure"); + } + }; + try { + Assert.assertThrows(RuntimeException.class, executor::executeInternalQuery); + Mockito.verify(workloadRuntimeStatusMgr).submitFinishQueryToAudit(auditEventCaptor.capture()); + } finally { + Deencapsulation.setField(env, "workloadRuntimeStatusMgr", originalWorkloadRuntimeStatusMgr); + ConnectContext.remove(); + } + + AuditEvent auditEvent = auditEventCaptor.getValue(); + Assert.assertEquals(AuditEvent.EventType.AFTER_QUERY, auditEvent.type); + Assert.assertEquals("ERR", auditEvent.state); + Assert.assertEquals(ErrorCode.ERR_INTERNAL_ERROR.getCode(), auditEvent.errorCode); + Assert.assertNotNull(auditEvent.errorMessage); + Assert.assertTrue("error message should mention root cause, got: " + auditEvent.errorMessage, + auditEvent.errorMessage.contains("mock plan failure")); + Assert.assertTrue("audit event should be marked as internal", auditEvent.isInternal); + Assert.assertTrue("audit event should be marked as query", auditEvent.isQuery); + Assert.assertTrue("audit event should be marked as nereids", auditEvent.isNereids); + Assert.assertEquals("select * from table1", auditEvent.stmt); + Assert.assertEquals("a8ec30e5ad0820f8c5bd16a82a4491ca", auditEvent.sqlHash); + } } diff --git a/regression-test/suites/fault_injection_p0/test_audit_log_internal_query_failure.groovy b/regression-test/suites/fault_injection_p0/test_audit_log_internal_query_failure.groovy deleted file mode 100644 index 233187e582e1dd..00000000000000 --- a/regression-test/suites/fault_injection_p0/test_audit_log_internal_query_failure.groovy +++ /dev/null @@ -1,98 +0,0 @@ -// 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. - -// Regression test for CIR-20019: when an internal query fails (for example -// the column-statistics gathering SQL that ANALYZE issues against a user -// table), the audit log entry must record state=ERR with a descriptive -// error_message instead of the previous misleading state=OK / return_rows=0. -suite('test_audit_log_internal_query_failure', 'nonConcurrent') { - def tbl = 'test_audit_log_internal_query_failure_t1' - - setGlobalVarTemporary([enable_audit_plugin: true], { - try { - sql "drop table if exists ${tbl}" - sql """ - create table ${tbl} (k int, v int) - duplicate key(k) - distributed by hash(k) buckets 1 - properties('replication_num'='1') - """ - sql "insert into ${tbl} values(1,10),(2,20),(3,30)" - - // Limit the injected IO error to this table's tablet so concurrent - // reads on system tables (such as audit_log itself) are unaffected. - def tabletRows = sql_return_maparray "show tablets from ${tbl}" - assertFalse(tabletRows.isEmpty(), "expected at least one tablet for ${tbl}") - def tabletId = tabletRows[0].TabletId - - // Capture the FE-side current timestamp to filter audit rows so - // stale entries from previous runs do not satisfy the assertion. - def startTime = (sql_return_maparray "select now() as ts")[0].ts.toString() - - GetDebugPoint().clearDebugPointsForAllBEs() - try { - GetDebugPoint().enableDebugPointForAllBEs( - "LocalFileReader::read_at_impl.io_error", - [ sub_path: "/${tabletId}/" ]) - - test { - sql "analyze table ${tbl} with sync" - exception "IO_ERROR" - } - } finally { - GetDebugPoint().clearDebugPointsForAllBEs() - } - - def currentDb = (sql_return_maparray "select database() as db")[0].db.toString() - def fullTableName = "internal.${currentDb}.${tbl}" - - // Force a flush so the failed internal query is queryable from - // __internal_schema.audit_log. - // The failed gather SQL reads from our user table and runs as an - // internal query; it must show up with state=ERR. Filter by start - // time to avoid matching stale entries from previous runs. - // Use queried_tables_and_views instead of stmt because stmt can be - // truncated by audit_plugin_max_sql_length when running in CI. - def query = """select state, error_code, error_message - from __internal_schema.audit_log - where is_internal = 1 - and array_contains(queried_tables_and_views, '${fullTableName}') - and state = 'ERR' - and `time` >= '${startTime}' - order by `time` desc limit 1""" - def res = [] - int retry = 60 - while (res.isEmpty() && retry-- > 0) { - sql "call flush_audit_log()" - sleep(2000) - res = sql_return_maparray "${query}" - } - assertFalse(res.isEmpty(), - "expected an audit_log entry with state=ERR for the failed gather query") - assertEquals('ERR', res[0].state.toString()) - assertNotEquals('0', res[0].error_code.toString()) - assertNotNull(res[0].error_message) - assertTrue(!res[0].error_message.toString().isEmpty(), - "audit_log error_message should not be empty, got: ${res[0].error_message}") - } finally { - try { - sql "drop table if exists ${tbl}" - } catch (Throwable ignored) { - } - } - }) -}