This is an automated email from the ASF dual-hosted git repository.
CalvinKirs 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 25968c92bfe [test](fe) Stabilize scan termination and retry profile
assertions (#69894)
25968c92bfe is described below
commit 25968c92bfeb5b659788cf01e9311d57f1220d98
Author: 924060929 <[email protected]>
AuthorDate: Mon Nov 2 09:45:38 2026 +0800
[test](fe) Stabilize scan termination and retry profile assertions (#69894)
### What problem does this PR solve?
Issue Number: N/A
Problem Summary:
Two FE unit tests depend on asynchronous work without owning its
complete lifecycle:
-
`SplitProducerTest.readThatOutlivesItsStatementLoadsNothingIntoTheClosedScope`
waits for the source-close signal and immediately checks the producer
completion count. `SplitProducer.finish` invokes the completion callback
after closing the source, so the assertion can run before that callback.
Wait for the assignment termination future, which completes after both
cleanup and the callback.
-
`StmtExecutorTest.testSuccessfulRetryRendersMergedProfileAndActualRows`
registers its profile with the global background `ProfileManager` while
also writing and reading the profile synchronously. Use a real manager
whose daemon is not started, scoped to this test, so automatic spill,
memory release and archival cannot interfere with the temporary storage
lifecycle. Preserve the real persistence path and all merged-profile and
actual-row assertions, and include the readback content in the failing
assertion.
The tests originated in #69718 and #68658 respectively. These test-only
changes are independent of the Coordinator refactoring in #69830. The
exact CI interleaving was not reproduced locally; both original tests
also passed local repeated execution. No production behavior changes.
### Release note
None
### Check List (For Author)
- Test: On this independent master-based branch, `./run-fe-ut.sh --run
'SplitProducerTest,StmtExecutorTest'` passed all 81 cases (26 + 55).
Before extraction, both target cases also passed 50 repetitions each
(179 executions across both classes); FE Checkstyle passed with 0
violations.
- Behavior changed: No
- Does this need documentation: No
---
.../doris/datasource/split/SplitProducerTest.java | 1 +
.../java/org/apache/doris/qe/StmtExecutorTest.java | 151 +++++++++++----------
2 files changed, 79 insertions(+), 73 deletions(-)
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitProducerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitProducerTest.java
index 2e7c65c065c..2e847e8a05a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitProducerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitProducerTest.java
@@ -674,6 +674,7 @@ class SplitProducerTest {
fixture.assignment.stop();
scope.closeAll();
release.countDown();
+ fixture.assignment.getTerminationFuture().get(5, TimeUnit.SECONDS);
Assertions.assertTrue(source.closed.await(5, TimeUnit.SECONDS));
Assertions.assertEquals(0, loads.get(), "the closed scope refuses
the late read's load");
Assertions.assertEquals(1, source.closeCalls.get());
diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java
index 7344b9cae5c..8f4bd5f9455 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java
@@ -649,81 +649,86 @@ public class StmtExecutorTest extends TestWithFeService {
@Test
@ResourceLock("global")
public void testSuccessfulRetryRendersMergedProfileAndActualRows(@TempDir
Path profileDirectory) throws Exception {
- ProfileManager manager = ProfileManager.getInstance();
- connectContext.getSessionVariable().enableProfile = true;
- connectContext.getSessionVariable().autoProfileThresholdMs = 0;
- TUniqueId firstQueryId = new TUniqueId(0x22040L, 5L);
- TUniqueId retryQueryId = new TUniqueId(0x22040L, 6L);
- connectContext.setQueryId(firstQueryId);
- connectContext.setStartTime();
- StmtExecutor executor = new StmtExecutor(connectContext,
- analyzeAndGetStmtByNereids("select 1", connectContext));
-
executor.getSummaryProfile().setQueryBeginTime(System.currentTimeMillis());
- Profile profile = executor.getProfile();
- Statistics statistics = new Statistics(0, Collections.emptyMap());
- PhysicalEmptyRelation physicalPlan = new PhysicalEmptyRelation(new
RelationId(0), Collections.emptyList(),
- Optional.empty(), null, PhysicalProperties.ANY, statistics);
- int planNodeId = physicalPlan.getId();
- profile.setPhysicalPlan(physicalPlan);
- ExecutionProfile failedExecution = new ExecutionProfile(firstQueryId,
Collections.singletonList(0));
- ExecutionProfile successfulExecution = new
ExecutionProfile(retryQueryId, Collections.singletonList(0));
- TQueryOptions queryOptions = new TQueryOptions();
- queryOptions.enable_profile = true;
-
- try {
- Coordinator firstCoord = Mockito.mock(Coordinator.class);
-
Mockito.when(firstCoord.getQueryOptions()).thenReturn(queryOptions);
-
Mockito.when(firstCoord.getExecutionProfile()).thenReturn(failedExecution);
- profile.addExecutionProfile(failedExecution);
- QeProcessorImpl.INSTANCE.registerQuery(firstQueryId, new
QeProcessorImpl.QueryInfo(firstCoord));
- Assertions.assertTrue(failedExecution.updateProfile(
- createBackendRowsProfile(firstQueryId, planNodeId, 1000),
- new TNetworkAddress("127.0.0.1", 9060), true).ok());
- executor.updateProfile(false);
- executor.finalizeQuery(true);
+ // Exercise storage synchronously: the global daemon can spill and
release this completed
+ // profile while the test is writing or reading it from its temporary
directory.
+ ProfileManager manager = new ProfileManager() { };
+ try (MockedStatic<ProfileManager> managers =
Mockito.mockStatic(ProfileManager.class)) {
+ managers.when(ProfileManager::getInstance).thenReturn(manager);
+ connectContext.getSessionVariable().enableProfile = true;
+ connectContext.getSessionVariable().autoProfileThresholdMs = 0;
+ TUniqueId firstQueryId = new TUniqueId(0x22040L, 5L);
+ TUniqueId retryQueryId = new TUniqueId(0x22040L, 6L);
+ connectContext.setQueryId(firstQueryId);
+ connectContext.setStartTime();
+ StmtExecutor executor = new StmtExecutor(connectContext,
+ analyzeAndGetStmtByNereids("select 1", connectContext));
+
executor.getSummaryProfile().setQueryBeginTime(System.currentTimeMillis());
+ Profile profile = executor.getProfile();
+ Statistics statistics = new Statistics(0, Collections.emptyMap());
+ PhysicalEmptyRelation physicalPlan = new PhysicalEmptyRelation(new
RelationId(0), Collections.emptyList(),
+ Optional.empty(), null, PhysicalProperties.ANY,
statistics);
+ int planNodeId = physicalPlan.getId();
+ profile.setPhysicalPlan(physicalPlan);
+ ExecutionProfile failedExecution = new
ExecutionProfile(firstQueryId, Collections.singletonList(0));
+ ExecutionProfile successfulExecution = new
ExecutionProfile(retryQueryId, Collections.singletonList(0));
+ TQueryOptions queryOptions = new TQueryOptions();
+ queryOptions.enable_profile = true;
- connectContext.setQueryId(retryQueryId);
- Coordinator retryCoord = Mockito.mock(Coordinator.class);
-
Mockito.when(retryCoord.getQueryOptions()).thenReturn(queryOptions);
-
Mockito.when(retryCoord.getExecutionProfile()).thenReturn(successfulExecution);
- Mockito.when(retryCoord.getExecStatus()).thenReturn(Status.OK);
- executor.setCoord(retryCoord);
- profile.addExecutionProfile(successfulExecution);
- QeProcessorImpl.INSTANCE.registerQuery(retryQueryId, new
QeProcessorImpl.QueryInfo(retryCoord));
- successfulExecution.addFragmentBackend(0, 1L);
- successfulExecution.addFragmentBackend(0, 2L);
- Assertions.assertTrue(successfulExecution.updateProfile(
- createBackendRowsProfile(retryQueryId, planNodeId, 17),
- new TNetworkAddress("127.0.0.1", 9060), true).ok());
- Assertions.assertTrue(successfulExecution.updateProfile(
- createBackendRowsProfile(retryQueryId, planNodeId, 25),
- new TNetworkAddress("127.0.0.2", 9060), true).ok());
- Assertions.assertTrue(successfulExecution.isCompleted());
- executor.updateProfile(false);
- executor.finalizeQuery();
-
Assertions.assertNotNull(manager.findProfileElementObject(profile.getId()));
-
- String profileText = profile.getProfileByLevel();
- Assertions.assertTrue(profileText.contains("MergedProfile:"));
- Assertions.assertTrue(profileText.contains("Pipeline
0(instance_num=2)"));
- Assertions.assertTrue(profileText.contains("RowsProduced: sum
42"));
- Assertions.assertFalse(profileText.contains("build merged simple
profile failed"));
- Assertions.assertFalse(profileText.contains("DetailProfile(" +
DebugUtil.printId(firstQueryId) + ")"));
- Assertions.assertTrue(profileText.contains("DetailProfile(" +
DebugUtil.printId(retryQueryId) + ")"));
- Assertions.assertEquals(42L,
profile.rowsProducedMap.get(String.valueOf(planNodeId)));
- Assertions.assertEquals(42L, statistics.getActualRowCount());
+ try {
+ Coordinator firstCoord = Mockito.mock(Coordinator.class);
+
Mockito.when(firstCoord.getQueryOptions()).thenReturn(queryOptions);
+
Mockito.when(firstCoord.getExecutionProfile()).thenReturn(failedExecution);
+ profile.addExecutionProfile(failedExecution);
+ QeProcessorImpl.INSTANCE.registerQuery(firstQueryId, new
QeProcessorImpl.QueryInfo(firstCoord));
+ Assertions.assertTrue(failedExecution.updateProfile(
+ createBackendRowsProfile(firstQueryId, planNodeId,
1000),
+ new TNetworkAddress("127.0.0.1", 9060), true).ok());
+ executor.updateProfile(false);
+ executor.finalizeQuery(true);
+
+ connectContext.setQueryId(retryQueryId);
+ Coordinator retryCoord = Mockito.mock(Coordinator.class);
+
Mockito.when(retryCoord.getQueryOptions()).thenReturn(queryOptions);
+
Mockito.when(retryCoord.getExecutionProfile()).thenReturn(successfulExecution);
+ Mockito.when(retryCoord.getExecStatus()).thenReturn(Status.OK);
+ executor.setCoord(retryCoord);
+ profile.addExecutionProfile(successfulExecution);
+ QeProcessorImpl.INSTANCE.registerQuery(retryQueryId, new
QeProcessorImpl.QueryInfo(retryCoord));
+ successfulExecution.addFragmentBackend(0, 1L);
+ successfulExecution.addFragmentBackend(0, 2L);
+ Assertions.assertTrue(successfulExecution.updateProfile(
+ createBackendRowsProfile(retryQueryId, planNodeId, 17),
+ new TNetworkAddress("127.0.0.1", 9060), true).ok());
+ Assertions.assertTrue(successfulExecution.updateProfile(
+ createBackendRowsProfile(retryQueryId, planNodeId, 25),
+ new TNetworkAddress("127.0.0.2", 9060), true).ok());
+ Assertions.assertTrue(successfulExecution.isCompleted());
+ executor.updateProfile(false);
+ executor.finalizeQuery();
+
Assertions.assertNotNull(manager.findProfileElementObject(profile.getId()));
- profile.writeToStorage(profileDirectory.toString());
- Profile storedProfile =
Profile.read(profile.getProfileStoragePath());
- Assertions.assertNotNull(storedProfile);
- String storedText = storedProfile.getProfileByLevel();
- Assertions.assertTrue(storedText.contains("MergedProfile:"));
- Assertions.assertTrue(storedText.contains("RowsProduced: sum 42"));
- Assertions.assertFalse(storedText.contains("DetailProfile(" +
DebugUtil.printId(firstQueryId) + ")"));
- } finally {
- QeProcessorImpl.INSTANCE.unregisterQuery(firstQueryId);
- QeProcessorImpl.INSTANCE.unregisterQuery(retryQueryId);
- manager.removeProfile(profile);
+ String profileText = profile.getProfileByLevel();
+ Assertions.assertTrue(profileText.contains("MergedProfile:"));
+ Assertions.assertTrue(profileText.contains("Pipeline
0(instance_num=2)"));
+ Assertions.assertTrue(profileText.contains("RowsProduced: sum
42"));
+ Assertions.assertFalse(profileText.contains("build merged
simple profile failed"));
+ Assertions.assertFalse(profileText.contains("DetailProfile(" +
DebugUtil.printId(firstQueryId) + ")"));
+ Assertions.assertTrue(profileText.contains("DetailProfile(" +
DebugUtil.printId(retryQueryId) + ")"));
+ Assertions.assertEquals(42L,
profile.rowsProducedMap.get(String.valueOf(planNodeId)));
+ Assertions.assertEquals(42L, statistics.getActualRowCount());
+
+ profile.writeToStorage(profileDirectory.toString());
+ Profile storedProfile =
Profile.read(profile.getProfileStoragePath());
+ Assertions.assertNotNull(storedProfile);
+ String storedText = storedProfile.getProfileByLevel();
+ Assertions.assertTrue(storedText.contains("MergedProfile:"),
storedText);
+ Assertions.assertTrue(storedText.contains("RowsProduced: sum
42"));
+ Assertions.assertFalse(storedText.contains("DetailProfile(" +
DebugUtil.printId(firstQueryId) + ")"));
+ } finally {
+ QeProcessorImpl.INSTANCE.unregisterQuery(firstQueryId);
+ QeProcessorImpl.INSTANCE.unregisterQuery(retryQueryId);
+ manager.removeProfile(profile);
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
