[fix](remote-doris) Release a remote Doris scan's Flight SQL session on the statement paths #68338 missed

[morningman]

What problem does this PR solve?

Related PR: #68338 (follow-up: the review of that PR found the paths below), #68101 / #68266 (why a leaked Flight SQL session costs the catalog user's connection quota on the remote FE)

Problem Summary:

Context. A Remote Doris catalog with use_arrow_flight = true reads another Doris cluster over Arrow Flight SQL: RemoteDorisScanNode opens a Flight SQL session on a remote FE while the plan is translated (getSplits), runs the query there, and hands the endpoints to the local BE. #68338 made that session end with the local query: the coordinator's close() / cancel() stops the scan node, which sends CloseSession; and because the session is opened before any coordinator exists and not every plan gets one, the node also registers itself with the StatementContext, whose close() stops what is still registered - the fallback for a statement that fails between planning and dispatch, or a plan probed and discarded.

That fallback assumes that whoever ran the statement ends it with StatementContext.close(). ConnectProcessor does (the per-statement finally of a COM_QUERY, and of a forwarded request on the master), TaskProcessor and MTMVTask do. Three owners of a statement do not, and one loop can skip the stop.

1. The problem, and what it cost

• A direct COM_STMT_EXECUTE (MysqlConnectProcessor.handleExecute) catches a failure and only finalizes the response; it never closes the per-execution StatementContext that ExecuteCommand allocates, and the next execution's nextStatementContext() only resets the connector scope before dropping it. A prepared query over a Remote Doris table that fails after planning (a SQL block rule on the scan, checkBlockRulesByScan runs after plan()) leaves its session on the remote FE until wait_timeout, once per execution. The forwarded COM_STMT_EXECUTE was already covered.
• A statement run under AutoCloseConnectContext - an EXPORT's SELECT INTO OUTFILE (external tables can be exported), an ANALYZE's statistics query, and a dozen other internal owners - ends with ConnectContext.clear(), which nulls the StatementContext without closing it. The same post-plan failure leaves the session behind.
• A streaming insert task (StreamingInsertTask) is run by the streaming scheduler, not by TaskProcessor: before() runs initPlan(..., false) once only to rewrite the TVF - a full plan, whose scan nodes open their sessions and whose coordinator is built and never executed or closed - and then plans the rewritten command again; closeOrReleaseResources() only nulls fields. A streaming INSERT ... SELECT that joins the TVF with a Remote Doris table leaks one session per attempt, success, retry or cancel. (The first cut of this PR closed the statement from closeOrReleaseResources(), which the review found insufficient: a STOP of the job only cancels the task - AbstractJob.updateJobStatus runs cancelAllTasks(false), and execute()'s finally skips the cleanup of a canceled attempt - and a PAUSE runs that cleanup on the job's control thread, possibly while before() is still planning: before the plan has registered its scan nodes, and while the planner holds the target table's read lock, which a StatementContext.close() from another thread would try to release.)
• Coordinator.close() and NereidsCoordinator.close() wrapped the whole scan-node loop in one try, and both cancel() loops had none. SplitAssignment.stop() rethrows the failure of its asynchronous split generation, so a batch-mode external scan planned before a Remote Doris scan could end the loop and skip it. On the normal path the statement's fallback (which is per node) still stopped it; on the deferred path - an Arrow Flight SQL client query, whose scan nodes are handed over to the coordinator kept alive for DoGet - the coordinator is the only owner, and the session had nobody left to close it. A cancel() that threw there also never sent its cancel RPCs.
• Regression framework ([fix](remote-doris) End a remote Doris scan's Flight SQL session with the query; stop the regression framework leaking connections #68338's second half): the sweep that closes the connections of a suite's finished threads ran only from getConnection(), so a suite in the arrow_flight_sql group (whose sql goes through getArrowFlightSqlConnection()) or one using master_sql never ran it and kept every finished thread's connection until teardown; and the teardown drain (snapshot + clear) was not serialized with registration, so a thread still running at the suite's end could register a connection between the two and have it dropped unclosed. The first cut of this PR refused such a thread a connection (IllegalStateException), and P0 build 1054369 showed what that costs: test_active_queries and test_backend_active_tasks return at once and leave a daemon thread polling their system table for five minutes on purpose, so their threads died on their first statement - and the exception was reported as a failure of test_partial_update_insert_schema_change, an unrelated suite that happened to be inside Awaitility.await(). Awaitility by default installs every await() as the JVM's default uncaught-exception handler and rethrows, from the awaiting thread, whatever any thread threw uncaught meanwhile; with ten suites in parallel that is the wrong suite by construction, and since the handler in place is whichever suite entered an await() last, sometimes no suite at all (the sibling thread's exception the same second went to stderr).

2. What this PR does, and why it helps

• Each of the three owners ends its statement with StatementContext.close(), the way ConnectProcessor does: MysqlConnectProcessor.handleExecute in a finally per execution (idempotent: an execution that ran its coordinator has nothing left to release; a forwarded execution is closed twice, the second a no-op), AutoCloseConnectContext.close() before clear() (the planner releases its table locks at the end of plan(), the connector scope close is close-once, and only a Remote Doris scan registers a node, so an owner whose statement scanned none sees a no-op), and StreamingInsertTask.endStatement(), from AbstractStreamingTask.execute()'s per-attempt finally on the task thread, for every outcome of the attempt - a canceled one included, since nothing else ends it then. The statement's reference lives in a field only the task thread touches; closeOrReleaseResources(), which the control thread runs on PAUSE, drops fields only, so no thread ever closes a statement another thread is still planning.
• Coordinator.stopScanNodes stops the nodes one by one, a failure logged per node, and both coordinators' close() and cancel() use it. A cancel() now always reaches its cancel RPCs.
• Regression framework: all three thread-local accessors run the finished-thread sweep. The connections a suite's threads open are recorded in an OpenedDorisConnections table (connection → thread), and each lives exactly as long as its thread: registration and the suite-end drain are serialized on the table; the drain closes the connections of threads that have finished and hands those of threads still running - with a warning naming the suite and the threads - to OpenedDorisConnections.STRAY, the table shared by all suites, where a registration after the drain goes too; every statement of any suite closes the strays whose thread has finished, and the end of the run closes what is left. A thread that outlives its suite thus keeps working, as those two suites intend, and its connection is closed within one statement of its end instead of never (or, in the first cut, of failing it). And Awaitility.doNotCatchUncaughtExceptionsByDefault() next to pollInSameThread(), with UncaughtThreadFailures installed as the JVM's default uncaught-exception handler in its place: a thread a suite constructs inherits the suite's collector (an InheritableThreadLocal set on the suite's thread, through any depth of threads started from it), the failure is recorded there, and Suite.doLazyCheck() throws the first one once the body has returned - that is, after every thread the suite joined has ended. So a raw Thread a suite starts and join()s (83 suites do; join() never rethrows) now fails its own suite, deterministically, instead of whichever suite happened to be awaiting, or none. A failure after the verdict (a thread left running past the suite's end) or on a thread no suite started fails nothing and is logged.

What it buys: the invariant #68338 introduced - a Remote Doris scan's session on the remote FE lives exactly as long as the local statement, whoever ran it - now holds for prepared statements, EXPORT/ANALYZE-style internal statements and streaming insert jobs, and survives a scan node whose stop() throws.

3. The classes, and how they call each other

• MysqlConnectProcessor.handleExecute (direct COM_STMT_EXECUTE): executor.execute() → ExecuteCommand.run → PreparedStatementContext.nextStatementContext() (a fresh StatementContext per execution) → plan / dispatch; the new finally closes ctx.getStatementContext(). PreparedStatementContext.nextStatementContext() keeps its connector-scope reset (it also drops the pinned writer schemas, which close() keeps); its comment is updated.
• AutoCloseConnectContext.close(): StatementContext.close() → ConnectContext.clear() → ConnectContext.remove(). Users: ExportTaskExecutor, the statistics tasks (StatisticsUtil.buildConnectContext), InternalSchemaInitializer, the cloud load/restore tasks, StreamingJobUtils, ...
• AbstractStreamingTask.endStatement() (new hook, no-op by default), called unconditionally from execute()'s per-attempt finally, before the closeOrReleaseResources() a canceled attempt skips. StreamingInsertTask.endStatement(): closes the StatementContext before() recorded in attemptStatement (task thread only) and clears it. StreamingInsertTask.closeOrReleaseResources() (task thread for a finished attempt; the control thread via StreamingInsertJob.clearRunningStreamTask for a paused one): drops fields, as before.
• Coordinator.stopScanNodes(List<ScanNode>) (new, protected static): the per-node loop; called from Coordinator.close() / cancel() and NereidsCoordinator.close() / cancel().
• OpenedDorisConnections (new): add (false once drained), remove, closeThoseOfFinishedThreads, drain, closeAll; STRAY is the instance shared by all suites. SuiteContext: getConnection / getMasterConnection / getArrowFlightSqlConnection → closeConnectionsOfFinishedThreads() (sweeps the suite's table and STRAY); trackDorisConnection → add, or STRAY.add after the drain; close() → closeLeftoverDorisConnections() → drain(). RegressionTest.initGroovyEnv: Awaitility.doNotCatchUncaughtExceptionsByDefault() + UncaughtThreadFailures.installAsDefaultHandler(); RegressionTest.main: STRAY.closeAll() after the last run.
• UncaughtThreadFailures (new): OWNER (the InheritableThreadLocal), add (false once the verdict is taken), takeAll, installAsDefaultHandler / uncaughtException. ScriptContext.createAndRunSuite sets OWNER to suite.threadFailures on the suite's thread for the run; Suite.doLazyCheck() → takeAll(), and throws the first.
• Untouched: RemoteDorisScanNode / RemoteDorisFlightSession (the session, stop(), the registration), StatementContext.close() itself, the deferral gate.

owner of the statement                             how the statement ends
COM_QUERY .......... ConnectProcessor.handleQuery ......... finally: StatementContext.close()   (already)
forwarded request .. ConnectProcessor.proxyExecute ........ finally: StatementContext.close()   (already)
job task ........... TaskProcessor.runTask ............... finally: StatementContext.close()   (already)
COM_STMT_EXECUTE ... MysqlConnectProcessor.handleExecute .. finally: StatementContext.close()   (this PR)
EXPORT / ANALYZE ... AutoCloseConnectContext.close() ...... StatementContext.close(), then clear()  (this PR)
streaming insert ... AbstractStreamingTask.execute() finally, every outcome .. endStatement(): StatementContext.close()  (this PR)
                                                          |
                                                          '-> stopScanNodesLeftBehind() -> RemoteDorisScanNode.stop() -> CloseSession
coordinator ........ Coordinator.close() / cancel() ... stopScanNodes(): per-node try/catch  (this PR)

Release note

None

Check List (For Author)

• Test

  • Regression test
  • 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

  Unit tests: RemoteDorisScanNodeTest (+2: a coordinator closes the session of the scan after one whose stop() throws; a statement run under AutoCloseConnectContext ends with the block), MysqlConnectProcessorExecuteCloseTest (a COM_STMT_EXECUTE failing after planning stops the scan node its plan registered), StreamingInsertTaskStatementCloseTest (ending an attempt stops the scan nodes its plan registered; an attempt canceled by a STOP still ends its statement from execute(); a PAUSE during planning leaves the final close to the task thread, which stops the node registered after the control thread's cleanup too); the neighbouring ConnectorStatementScopeTest, ConnectProcessorForwardProtocolTest, ConnectProcessorRetryTest, CoordinatorTest, NereidsCoordinatorTest, OldCoordinatorTest, StreamingInsertTaskAuditTest, StmtExecutorTest, MysqlConnectProcessorCursorFetchTest, ArrowFlightDeferralGateTest pass unchanged. Framework: OpenedDorisConnectionsTest (a connection is closed once the thread that opened it has finished; a drain hands everything over and refuses later registrations; closeAll closes the connections of running threads too; a close that throws does not stop the others), UncaughtThreadFailuresTest (a failure on a thread the suite started, or on a thread started from it, is the suite's; one after the verdict is not recorded; a thread no suite started fails nothing).

  Manual test: two scratch suites against a local cluster, run with -parallel 2 -suiteParallel 2 - one returns at once and leaves a daemon thread that runs five statements after a 2s sleep (like test_active_queries), the other spends 12s inside Awaitility.await().until { sql "select 1"; ... }. With the framework of the first cut the awaiting suite fails with Suite stray_daemon is over, but thread Thread-2 it left running still asks for a connection, as in P0 build 1054369; with this one both pass, the daemon's five statements run, and its connection is closed 100ms after the thread ends (Closed 1 connection(s) of finished threads (threads that outlived their suite)). With the collector (run again with -suiteParallel 4), the daemon variant whose statement fails 2s after its suite returned is logged as Uncaught exception in thread Thread-5, started by suite stray_daemon whose verdict is already taken; it fails no suite, and fails nothing - the awaiting neighbour passes; two more scratch suites start a raw Thread, join() it and return: the one whose thread fails a statement now fails, with Thread Thread-3 of suite joined_worker_fails died with an uncaught exception, and the one whose thread (and a thread started from it) runs fine passes.

• Behavior changed:

  • No.
  • Yes.
    • A prepared statement's execution (COM_STMT_EXECUTE) closes its StatementContext when it ends, like a COM_QUERY does; an EXPORT / ANALYZE / other AutoCloseConnectContext statement and a streaming insert attempt do the same.
    • A scan node whose stop() throws no longer keeps a coordinator from stopping the nodes after it, nor a cancel() from sending its cancel RPCs.
    • Regression framework: a suite thread left running past the suite's end keeps its connection until it finishes (closed by the next statement of any suite, or at the end of the run) instead of holding it until the client JVM collects it; a thread's uncaught exception fails the suite that started the thread (once that suite's body returns), no longer whichever suite is inside Awaitility.await() at that moment; a thread left running past its suite's end fails nothing.

• Does this need documentation?

  • No.
  • Yes.

Check List (For Reviewer who merge this PR)

• Confirm the release note
• Confirm test cases
• Confirm document
• Add branch pick label

[hello-stephen]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?

[morningman]

run buildall

[morningman]

run buildall

[hello-stephen]

TPC-H: Total hot run time: 27542 ms

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit aefca1782c94df7961e830e4348658204176d5c5, data reload: false

------ Round 1 ----------------------------------
============================================
q1	17729	3866	3868	3866
q2	2148	372	302	302
q3	10134	1417	805	805
q4	4688	485	350	350
q5	7479	825	556	556
q6	181	167	137	137
q7	747	784	587	587
q8	9333	1428	1571	1428
q9	5572	4186	4169	4169
q10	6827	1324	1035	1035
q11	437	275	252	252
q12	636	413	299	299
q13	18047	2608	1987	1987
q14	268	261	248	248
q15	q16	735	724	670	670
q17	1805	1091	1006	1006
q18	6556	5618	5546	5546
q19	1179	1169	993	993
q20	483	402	261	261
q21	5151	2908	2746	2746
q22	418	356	299	299
Total cold run time: 100553 ms
Total hot run time: 27542 ms

----- Round 2, with runtime_filter_mode=off -----
============================================
q1	4161	4068	4073	4068
q2	708	580	527	527
q3	4475	4854	4364	4364
q4	2227	2328	1462	1462
q5	4197	4124	4113	4113
q6	228	174	128	128
q7	1741	1627	1404	1404
q8	2198	2304	2212	2212
q9	7553	7670	7525	7525
q10	3746	3782	3161	3161
q11	578	395	369	369
q12	777	741	524	524
q13	2410	2859	2143	2143
q14	302	310	279	279
q15	q16	706	717	635	635
q17	7718	7261	7137	7137
q18	12182	11126	11827	11126
q19	1185	1043	1077	1043
q20	2255	2214	1938	1938
q21	5412	4343	4552	4343
q22	541	480	400	400
Total cold run time: 65300 ms
Total hot run time: 58901 ms

[hello-stephen]

TPC-DS: Total hot run time: 152352 ms

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit aefca1782c94df7961e830e4348658204176d5c5, data reload: false

query5	4336	602	479	479
query6	442	216	202	202
query7	4808	512	304	304
query8	319	182	165	165
query9	8815	3985	3972	3972
query10	442	308	252	252
query11	5804	3586	3262	3262
query12	155	93	89	89
query13	1260	604	420	420
query14	6524	4487	4196	4196
query14_1	4016	3950	3916	3916
query15	199	192	177	177
query16	955	418	405	405
query17	872	656	516	516
query18	2413	462	320	320
query19	192	173	140	140
query20	82	79	80	79
query21	218	133	114	114
query22	13002	12902	12862	12862
query23	13945	13021	12486	12486
query23_1	12361	12561	12445	12445
query24	7286	1178	666	666
query24_1	663	687	685	685
query25	528	417	345	345
query26	1247	306	155	155
query27	2735	541	320	320
query28	4541	1949	2006	1949
query29	1575	699	503	503
query30	301	217	184	184
query31	902	745	627	627
query32	165	98	85	85
query33	511	302	234	234
query34	1183	1119	626	626
query35	718	739	624	624
query36	789	774	725	725
query37	140	100	86	86
query38	1810	1765	1709	1709
query39	693	703	675	675
query39_1	649	660	667	660
query40	220	119	100	100
query41	66	64	62	62
query42	92	92	92	92
query43	333	342	296	296
query44	1344	703	696	696
query45	182	179	164	164
query46	1065	1159	719	719
query47	1483	1478	1400	1400
query48	373	411	287	287
query49	591	397	300	300
query50	946	338	263	263
query51	10592	10507	10337	10337
query52	87	86	75	75
query53	243	257	174	174
query54	241	196	189	189
query55	79	72	72	72
query56	222	232	215	215
query57	1408	1460	1365	1365
query58	285	260	253	253
query59	2003	2086	1863	1863
query60	274	243	228	228
query61	149	148	138	138
query62	394	321	263	263
query63	215	172	178	172
query64	2762	970	790	790
query65	3455	3410	3388	3388
query66	1830	428	298	298
query67	20509	20117	19828	19828
query68	3318	1584	851	851
query69	423	301	261	261
query70	903	817	832	817
query71	287	245	218	218
query72	2848	2667	2381	2381
query73	807	784	425	425
query74	4636	4473	4291	4291
query75	2331	2294	1935	1935
query76	2430	1111	721	721
query77	369	408	311	311
query78	9182	9093	8521	8521
query79	1183	1216	760	760
query80	542	475	390	390
query81	539	318	279	279
query82	312	160	128	128
query83	315	225	209	209
query84	322	151	123	123
query85	933	525	377	377
query86	328	248	215	215
query87	2005	1981	1829	1829
query88	3631	2764	2759	2759
query89	352	279	248	248
query90	1800	187	174	174
query91	173	157	127	127
query92	101	82	79	79
query93	1339	1375	838	838
query94	529	331	292	292
query95	663	374	444	374
query96	1120	777	328	328
query97	2412	2427	2305	2305
query98	158	150	152	150
query99	714	726	617	617
Total cold run time: 235598 ms
Total hot run time: 152352 ms

[hello-stephen]

ClickBench: Total hot run time: 24.15 s

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/clickbench-tools
ClickBench test result on commit aefca1782c94df7961e830e4348658204176d5c5, data reload: false

query1	0.00	0.00	0.01
query2	0.09	0.05	0.05
query3	0.26	0.14	0.13
query4	1.61	0.14	0.14
query5	0.25	0.22	0.22
query6	1.15	0.99	0.95
query7	0.04	0.01	0.01
query8	0.06	0.04	0.04
query9	0.39	0.34	0.34
query10	0.55	0.59	0.58
query11	0.21	0.16	0.14
query12	0.18	0.15	0.15
query13	0.48	0.47	0.48
query14	0.96	0.97	0.96
query15	0.62	0.56	0.61
query16	0.32	0.31	0.32
query17	1.06	1.06	1.13
query18	0.22	0.21	0.20
query19	2.03	1.96	2.00
query20	0.02	0.02	0.01
query21	15.49	0.20	0.14
query22	4.88	0.05	0.05
query23	16.14	0.29	0.12
query24	2.96	0.43	0.33
query25	0.10	0.05	0.04
query26	0.74	0.22	0.14
query27	0.05	0.05	0.03
query28	3.61	0.78	0.34
query29	12.54	4.01	3.21
query30	0.28	0.15	0.15
query31	2.78	0.56	0.32
query32	3.23	0.59	0.49
query33	3.14	3.26	3.21
query34	15.66	3.93	3.30
query35	3.23	3.20	3.25
query36	0.55	0.42	0.42
query37	0.09	0.06	0.06
query38	0.05	0.04	0.04
query39	0.04	0.03	0.03
query40	0.17	0.15	0.16
query41	0.09	0.03	0.03
query42	0.03	0.04	0.04
query43	0.04	0.04	0.03
Total cold run time: 96.39 s
Total hot run time: 24.15 s

[morningman]

run buildall

[hello-stephen]

TPC-H: Total hot run time: 27350 ms

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit 82309eb1ae55c1f055d88c477586ed96827fd88f, data reload: false

------ Round 1 ----------------------------------
============================================
q1	17587	3745	3771	3745
q2	2181	380	297	297
q3	10094	1409	783	783
q4	4687	475	349	349
q5	7482	834	560	560
q6	181	170	134	134
q7	750	782	616	616
q8	9311	1395	1495	1395
q9	5516	4174	4178	4174
q10	6829	1339	1026	1026
q11	433	268	251	251
q12	636	423	295	295
q13	18101	2611	1986	1986
q14	276	265	234	234
q15	q16	738	724	662	662
q17	1773	1132	1005	1005
q18	6465	5669	5536	5536
q19	1310	1187	1045	1045
q20	471	397	268	268
q21	5785	2895	2696	2696
q22	435	348	293	293
Total cold run time: 101041 ms
Total hot run time: 27350 ms

----- Round 2, with runtime_filter_mode=off -----
============================================
q1	4132	4017	4000	4000
q2	710	563	513	513
q3	4444	4833	4292	4292
q4	2217	2293	1430	1430
q5	4227	4093	4091	4091
q6	228	175	125	125
q7	1713	1631	1410	1410
q8	2158	2116	2268	2116
q9	7613	7553	7581	7553
q10	3759	3935	3157	3157
q11	540	409	370	370
q12	753	726	534	534
q13	2425	2835	2150	2150
q14	299	304	265	265
q15	q16	687	716	641	641
q17	7726	7145	7010	7010
q18	11862	11102	11859	11102
q19	1270	1056	1052	1052
q20	2243	2218	1963	1963
q21	5257	4398	4500	4398
q22	524	470	410	410
Total cold run time: 64787 ms
Total hot run time: 58582 ms

[hello-stephen]

TPC-DS: Total hot run time: 152869 ms

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit 82309eb1ae55c1f055d88c477586ed96827fd88f, data reload: false

query5	4328	604	460	460
query6	422	224	198	198
query7	4809	540	297	297
query8	321	177	162	162
query9	8819	3956	3961	3956
query10	443	313	263	263
query11	5831	3567	3245	3245
query12	143	88	87	87
query13	1244	603	448	448
query14	6543	4520	4261	4261
query14_1	3955	3939	3921	3921
query15	203	194	185	185
query16	982	457	421	421
query17	904	668	534	534
query18	2449	464	336	336
query19	197	180	150	150
query20	84	81	80	80
query21	225	130	112	112
query22	13036	13033	14059	13033
query23	14604	13538	12862	12862
query23_1	12863	12606	12512	12512
query24	7260	1109	671	671
query24_1	697	667	686	667
query25	557	447	369	369
query26	1249	300	168	168
query27	2761	555	332	332
query28	4555	1972	1943	1943
query29	1673	676	490	490
query30	308	217	171	171
query31	896	747	629	629
query32	150	86	90	86
query33	490	308	237	237
query34	1194	1062	604	604
query35	710	736	625	625
query36	789	788	675	675
query37	151	99	94	94
query38	1821	1758	1673	1673
query39	707	673	686	673
query39_1	651	644	644	644
query40	214	118	96	96
query41	63	60	59	59
query42	93	88	89	88
query43	327	335	294	294
query44	1329	702	698	698
query45	176	164	160	160
query46	1045	1154	722	722
query47	1480	1482	1400	1400
query48	404	406	297	297
query49	570	405	291	291
query50	912	335	246	246
query51	10676	10635	10301	10301
query52	87	85	73	73
query53	234	248	177	177
query54	264	232	178	178
query55	75	75	68	68
query56	237	213	213	213
query57	1358	1378	1391	1378
query58	319	258	249	249
query59	1965	2055	1826	1826
query60	285	242	224	224
query61	146	146	145	145
query62	397	321	260	260
query63	213	179	181	179
query64	2824	1014	819	819
query65	3466	3395	3385	3385
query66	1808	410	299	299
query67	20058	20215	20043	20043
query68	3006	1501	953	953
query69	409	301	250	250
query70	916	802	798	798
query71	299	229	204	204
query72	2701	2530	2270	2270
query73	856	767	424	424
query74	4629	4489	4274	4274
query75	2295	2267	1920	1920
query76	2291	1087	718	718
query77	355	382	290	290
query78	9101	9023	8488	8488
query79	1320	1131	731	731
query80	913	450	357	357
query81	581	320	270	270
query82	867	153	125	125
query83	296	222	195	195
query84	310	147	113	113
query85	900	500	378	378
query86	386	245	233	233
query87	1986	1977	1812	1812
query88	3611	2725	2697	2697
query89	351	287	245	245
query90	1855	181	178	178
query91	169	155	130	130
query92	106	89	89	89
query93	1443	1407	856	856
query94	612	343	313	313
query95	650	370	417	370
query96	1039	775	342	342
query97	2466	2437	2331	2331
query98	160	155	141	141
query99	734	711	612	612
Total cold run time: 237301 ms
Total hot run time: 152869 ms

[hello-stephen]

ClickBench: Total hot run time: 24.19 s

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/clickbench-tools
ClickBench test result on commit 82309eb1ae55c1f055d88c477586ed96827fd88f, data reload: false

query1	0.01	0.01	0.01
query2	0.09	0.05	0.05
query3	0.27	0.14	0.14
query4	1.61	0.14	0.14
query5	0.25	0.23	0.23
query6	1.16	0.94	0.94
query7	0.04	0.01	0.00
query8	0.05	0.04	0.04
query9	0.41	0.36	0.36
query10	0.58	0.60	0.58
query11	0.20	0.16	0.15
query12	0.18	0.15	0.15
query13	0.46	0.48	0.49
query14	0.96	0.97	0.95
query15	0.62	0.59	0.60
query16	0.31	0.32	0.33
query17	1.12	1.11	1.12
query18	0.22	0.20	0.21
query19	1.99	1.94	1.89
query20	0.02	0.01	0.02
query21	15.49	0.20	0.14
query22	4.99	0.05	0.05
query23	16.12	0.31	0.12
query24	3.00	0.41	0.32
query25	0.12	0.05	0.04
query26	0.72	0.20	0.14
query27	0.04	0.03	0.04
query28	3.53	0.83	0.34
query29	12.49	4.08	3.20
query30	0.29	0.15	0.16
query31	2.77	0.57	0.31
query32	3.23	0.59	0.48
query33	3.13	3.23	3.23
query34	15.55	3.97	3.28
query35	3.21	3.26	3.23
query36	0.56	0.44	0.43
query37	0.09	0.07	0.06
query38	0.06	0.04	0.04
query39	0.03	0.03	0.03
query40	0.17	0.14	0.14
query41	0.08	0.03	0.03
query42	0.04	0.03	0.03
query43	0.05	0.04	0.03
Total cold run time: 96.31 s
Total hot run time: 24.19 s

[morningman]

/review

[github-actions]

Request changes: two P1 correctness gaps remain.

Findings

1. Streaming-task cancellation does not give the new statement cleanup a post-quiescence owner. STOPPED skips it, and PAUSED can run it before planning finishes, so the pre-rewrite remote Doris Flight SQL session can still leak until remote wait_timeout.
2. The process-wide Awaitility setting removes uncaught raw-worker failures from every suite verdict before the framework has owner-local propagation, allowing product failures to turn green.

Critical checkpoint conclusions

• Goal and proof: The PR closes remote Doris sessions on several previously missed statement paths, but does not accomplish that goal for canceled streaming attempts. The direct cleanup test does not exercise STOPPED or cancellation racing planning.
• Scope and clarity: The FE cleanup changes are focused; the Awaitility switch is process-wide and therefore much broader than the connection-ownership fix.
• Concurrency: The streaming scheduler task thread races the PAUSE/STOP control thread over ctx and stmtExecutor; the job lock does not quiesce before(). Regression suites also use raw worker threads; join() provides ordering but no throwable propagation. Connection-map add/drain/remove synchronization otherwise survived the interleaving review, with heavy closes outside its monitor. No new lock-order or deadlock defect was found.
• Lifecycle: Prepared execution closes the correct statement generation, Arrow-deferred scans are handed to the coordinator, AutoClose owners finish synchronous use before teardown, and legacy/Nereids coordinators now attempt every scan-node stop. The remaining broken lifecycle is the canceled streaming attempt described inline. No static-initialization issue applies.
• Configuration and compatibility: No configuration, wire/storage format, symbol, rolling-upgrade, persistence, transaction, data-write, or FE-BE variable-passing change is introduced.
• Parallel and conditional paths: Direct/forwarded prepared execution, normal/Arrow ownership, AutoClose callers, legacy/Nereids coordinators, success/failure/retry, PAUSED/STOPPED, and regression suite/STRAY connection ownership were checked. Per-node cleanup isolation and connection-transfer conditions are sound; the canceled-task cleanup suppression is not sound with late planning.
• Tests and results: Added tests cover positive direct cleanup, coordinator failure isolation, prepared execution, and connection tracking, but omit the two cancellation paths and owner-local raw-worker failure propagation. This was a static-only review: the review prompt prohibited builds/tests, so author/CI claims were not treated as independent validation and no local test result is claimed.
• Observability: Existing logs are adequate to diagnose cleanup failures, but logging cannot replace deterministic session close or a failing suite verdict.
• Performance: No substantiated CPU, memory, or complexity regression was found in the per-node stop or connection tracking changes.
• User focus: No additional focus was supplied; the entire 13-file change set and relevant upstream/downstream call chains were reviewed.

All other candidates were either duplicates or dismissed with concrete code evidence. Overall state: changes requested pending fixes for both P1 findings and negative coverage for their failure paths.

[github-actions]

[P1] Close canceled attempts only after planning has quiesced

This cleanup is not guaranteed to run at a terminal task boundary. STOPPED only calls cancelAllTasks(false), which marks the task CANCELED; StreamingInsertJob.updateJobStatus force-cleans only PAUSED, while AbstractStreamingTask.execute() skips its own cleanup for every canceled task, so this line is never reached. There is also a PAUSED race before stmtExecutor is assigned: the control thread can close and clear the statement while baseCommand.initPlan(...) is still running, after which a RemoteDorisScanNode can register into the already-drained StatementContext and the canceled task again skips a final close. In both cases the pre-rewrite plan has no coordinator to close that Flight SQL session, leaving it until remote wait_timeout. Please make the task thread perform one final close after before()/run() can no longer register nodes for every terminal status, and cover STOPPED plus cancel-during-planning.

[github-actions]

[P1] Preserve a suite verdict for raw worker failures

This avoids cross-suite misattribution by making uncaught raw-worker failures unable to fail any suite. The tree still has many Thread.start/new Thread workers that run SQL or assertions and whose parents only call join(); join() never rethrows, and the framework installs no per-suite uncaught-error collector. Suite.thread() can propagate through its future, but those existing raw-thread callers do not use it. The manual result in this PR also explicitly shows the new behavior: a failing daemon only prints to stderr and fails no suite. Please add owner-local propagation (for example, migrate these workers to futures plus get(), or add a per-suite collector checked after joins) before disabling the only process-level signal, so product regressions cannot turn green.