morningman opened a new pull request, #68353:
URL: https://github.com/apache/doris/pull/68353
### 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 or retry.
- `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 (#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.
**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.closeOrReleaseResources()`
before dropping its context (per attempt, including the cancel path the job
drives).
- `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.
- `SuiteContext`: all three thread-local accessors run the finished-thread
sweep; `trackDorisConnection` and `closeLeftoverDorisConnections` are
serialized on the registry, and once the suite is over a late registration is
closed at once and refused with an `IllegalStateException` naming the suite and
the thread - nothing would ever close that connection, and the suite's verdict
is already in (thread failures are not reported to the listeners).
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`, ...
- `StreamingInsertTask.closeOrReleaseResources()` (from
`AbstractStreamingTask.execute()`'s per-attempt finally and
`StreamingInsertJob.clearRunningStreamTask`): closes
`ctx.getStatementContext()` before `ctx = null`.
- `Coordinator.stopScanNodes(List<ScanNode>)` (new, protected static): the
per-node loop; called from `Coordinator.close()` / `cancel()` and
`NereidsCoordinator.close()` / `cancel()`.
- `SuiteContext`: `getConnection` / `getMasterConnection` /
`getArrowFlightSqlConnection` → `closeConnectionsOfFinishedThreads()`;
`trackDorisConnection` and `closeLeftoverDorisConnections` synchronize on
`openedDorisConnections` and share the `dorisConnectionsClosed` flag.
- 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 ... StreamingInsertTask.closeOrReleaseResources() ..
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
- [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? -->
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` (releasing an attempt stops the scan
nodes its statement registered); the neighbouring
`ConnectorStatementScopeTest`, `ConnectProcessorForwardProtocolTest`,
`ConnectProcessorRetryTest`, `CoordinatorTest`, `NereidsCoordinatorTest`,
`OldCoordinatorTest`, `StreamingInsertTaskAuditTest`, `StmtExecutorTest`,
`MysqlConnectProcessorCursorFetchTest`, `ArrowFlightDeferralGateTest` pass
unchanged.
- Behavior changed:
- [ ] No.
- [x] Yes. <!-- Explain the behavior change -->
- 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 is refused a new connection (`IllegalStateException`) instead of opening
one nobody closes.
- 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 -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
