github-actions[bot] commented on code in PR #68338:
URL: https://github.com/apache/doris/pull/68338#discussion_r4065229643
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1667,20 +1692,26 @@ public void executeAndSendResult(boolean
isOutfileQuery, boolean isSendFields,
if (!context.isReturnResultFromLocal()) {
profile.getSummaryProfile().setTempStartTime();
- // The client pulls the results from the BE later (Arrow
Flight SQL's DoGet). Only an
- // external-table scan in batch mode still needs the
coordinator after this point:
- // the BE fetches its splits lazily from the split source the
coordinator holds, so
- // closing the coordinator here would release that source too
early and break DoGet
- // (#62259). Such a coordinator is closed later by
ConnectContext: on the session's
- // next query, on teardown, or by the idle reaper in
checkTimeout. The trade-off is
- // that its query queue slot and query registration stay held
until then. Every
- // other query closes its coordinator in the finally block
below and releases both
- // right away, the BE buffering its results independently of
the coordinator
- // (#67503). A short-circuit point query is the one case with
a different coordBase,
- // and it cannot reach here: it has no Arrow result on either
side, so
+ // The client pulls the results from the BE later (Arrow
Flight SQL's DoGet). Only a
+ // scan the BE keeps depending on the FE for still needs the
coordinator after this
+ // point: an external-table scan in batch mode fetches its
splits lazily from the
+ // split source the coordinator holds (#62259), and a remote
Doris scan keeps the
+ // Flight SQL session open on the other frontend whose query
the BE reads
+ // (RemoteDorisScanNode); closing the coordinator here would
release either too early
+ // and break DoGet. Such a coordinator is closed later by
ConnectContext: on the
+ // session's next query, on teardown, or by the idle reaper in
checkTimeout. The
+ // trade-off is that its query queue slot and query
registration stay held until
+ // then. Every other query closes its coordinator in the
finally block below and
+ // releases both right away, the BE buffering its results
independently of the
+ // coordinator (#67503). A short-circuit point query is the
one case with a different
+ // coordBase, and it cannot reach here: it has no Arrow result
on either side, so
// LogicalResultSinkToShortCircuitPointQuery keeps a Flight
session on the normal
// execution path
(ProtocolAdapter.supportsShortCircuitPointQuery, #67368).
- if (coordBase == coord && coord.hasBatchSplitSource()) {
+ if (coordBase == coord && coord.mustOutliveDispatch()) {
+ // The coordinator outlives this statement, and with it
what its scan nodes hold
+ // for the BE: the statement's own end must not stop them
(StatementContext.close
+ // is the fallback for a plan no coordinator owns), the
coordinator's close does.
+
statementContext.handOverScanNodesToDeferredCoordinator(planner.getScanNodes());
Review Comment:
[P1] This handoff removes every planner scan node from the statement
fallback, but the deferred coordinator is not guaranteed to attempt every stop.
Coordinator.close() and NereidsCoordinator.close() wrap the whole loop in one
try; an earlier batch file scan can throw from SplitAssignment.stop() after an
asynchronous split failure, so a following RemoteDorisScanNode is skipped and
its session now has no remaining owner. Please isolate failures per scan node
(including the analogous cancel loops) and cover a deferred mixed-scan query
where the first stop throws.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisFlightSession.java:
##########
@@ -0,0 +1,153 @@
+// 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.
+
+package org.apache.doris.datasource.doris.source;
+
+import org.apache.doris.common.Pair;
+import org.apache.doris.common.UserException;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.arrow.flight.CallOptions;
+import org.apache.arrow.flight.CloseSessionRequest;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.Location;
+import org.apache.arrow.flight.grpc.CredentialCallOption;
+import org.apache.arrow.flight.sql.FlightSqlClient;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.Closeable;
+import java.net.URI;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * The Flight SQL session a {@link RemoteDorisScanNode} opens on a remote
Doris frontend for one
+ * scan, and ends with a CloseSession once the scan is over.
+ *
+ * <p>The handshake ({@code authenticateBasicToken}) opens a session on the
remote frontend: a
+ * connection in its pool, counted against {@code qe_max_connection}, the
Arrow Flight SQL sub-quota
+ * and the catalog user's {@code max_user_connections}, that only a
CloseSession, a KILL or
+ * {@code wait_timeout} ends. Closing the gRPC channel does not. A session
opened per scan and never
+ * closed therefore stays for hours and, one scan at a time, exhausts the
catalog user's connection
+ * quota on the remote frontend - refusing that user's MySQL connections there
as well.
+ *
+ * <p>The session outlives GetFlightInfo on purpose: the query it ran serves
the BE's DoGet of the
+ * endpoints, and the remote frontend cancels whatever a closed session was
still running - the
+ * query itself, when the remote table is an external table scanned in batch
mode and the query is
+ * therefore deferred there. So {@link #close()} is called from {@link
RemoteDorisScanNode#stop()},
+ * when the coordinator of the local query closes, and that coordinator is
kept alive until the BE
+ * has finished scanning ({@link
RemoteDorisScanNode#coordinatorMustOutliveDispatch()}).
+ */
+class RemoteDorisFlightSession implements Closeable {
+ private static final Logger LOG =
LogManager.getLogger(RemoteDorisFlightSession.class);
+
+ // A bound on the CloseSession round trip. close() runs on the local
query's teardown path,
+ // which must not hang on a remote frontend that has stopped answering;
the session is then left
+ // to the remote frontend's wait_timeout, as every session was before this
class existed.
+ @VisibleForTesting
+ static final int CLOSE_SESSION_TIMEOUT_SECONDS = 5;
+
+ private final Pair<String, Integer> hostAndPort;
+ private final BufferAllocator allocator;
+ private final FlightSqlClient client;
+ private final CredentialCallOption credential;
+ private boolean closed = false;
+
+ private RemoteDorisFlightSession(Pair<String, Integer> hostAndPort,
BufferAllocator allocator,
+ FlightSqlClient client, CredentialCallOption credential) {
+ this.hostAndPort = hostAndPort;
+ this.allocator = allocator;
+ this.client = client;
+ this.credential = credential;
+ }
+
+ /**
+ * Opens a session on the remote frontend at {@code hostAndPort} with the
catalog's credentials.
+ * Nothing is left behind when this fails: a handshake that was refused
opened no session, and
+ * the channel and allocator are released before the exception propagates.
+ */
+ static RemoteDorisFlightSession open(Pair<String, Integer> hostAndPort,
String user, String password)
+ throws Exception {
+ BufferAllocator allocator = new RootAllocator();
+ FlightClient flightClient = null;
+ try {
+ URI uri = new URI("grpc", null, hostAndPort.first,
hostAndPort.second, null, null, null);
+ flightClient = FlightClient.builder(allocator, new
Location(uri)).build();
+ Optional<CredentialCallOption> credential =
flightClient.authenticateBasicToken(user, password);
+ if (!credential.isPresent()) {
+ throw new UserException("Authenticates with a username and
password failure");
+ }
+ return new RemoteDorisFlightSession(hostAndPort, allocator, new
FlightSqlClient(flightClient),
+ credential.get());
+ } catch (Throwable t) {
+ closeQuietly(flightClient, allocator, hostAndPort);
+ throw t;
+ }
+ }
+
+ /** Runs {@code sql} on the remote frontend; the endpoints of the result
are where the BE reads it. */
+ FlightInfo execute(String sql, int timeoutSec) {
+ return client.execute(sql, credential, CallOptions.timeout(timeoutSec,
TimeUnit.SECONDS));
+ }
+
+ Pair<String, Integer> getHostAndPort() {
+ return hostAndPort;
+ }
+
+ /**
+ * Ends the session on the remote frontend (CloseSession), then releases
the channel and the
+ * allocator. Never throws: this runs on the local query's teardown path,
and a session the
+ * remote frontend could not be told to close is only left to its
wait_timeout. Idempotent.
+ */
+ @Override
+ public synchronized void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ try {
+ client.closeSession(new CloseSessionRequest(), credential,
+ CallOptions.timeout(CLOSE_SESSION_TIMEOUT_SECONDS,
TimeUnit.SECONDS));
Review Comment:
[P1] This deadline bounds one session, not query teardown. A plan can
contain many RemoteDorisScanNodes, and coordinator cancel/close stops them
synchronously and sequentially; timeout cancellation also runs on the single
connect-scheduler checker before backend cancellation. If the remote FE stops
answering, one query can therefore monopolize that checker for roughly 5
seconds per scan and delay all other query/idle timeouts for minutes. Please
give cleanup a query-wide bound or detach/close these sessions through bounded
parallel cleanup so cancellation does not scale as N * 5s.
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/RegressionTest.groovy:
##########
@@ -162,6 +163,19 @@ class RegressionTest {
static void initGroovyEnv(Config config) {
log.info("parallel = ${config.parallel}, suiteParallel =
${config.suiteParallel}, actionParallel = ${config.actionParallel}")
+ // Evaluate every Awaitility condition on the thread that awaits it,
as Suite.awaitUntil already
+ // does. A suite's connections are ThreadLocal to the thread that
opened them (see
+ // SuiteContext.getConnection), so a `sql` inside
`Awaitility.await()...until { }` on Awaitility's
+ // own polling thread opened a fresh connection that nothing closed
when that thread died with
+ // the await(): one connection leaked on the frontend per await(),
held until the client JVM
+ // garbage-collected it, and enough of them at once reach the user's
max_user_connections.
+ // Polled on the suite thread, the condition reuses the suite's
connection. The trade-off: an
+ // atMost() no longer bounds a condition that blocks - the poll runs
to completion before the
+ // bound is checked. A condition that runs statements is bounded by
their timeouts (the
+ // frontend's query_timeout, the framework's socketTimeout); a
condition that waits on
+ // anything else has to bound that wait itself, as SuiteCluster does
for its doris-compose
+ // subprocesses.
+ Awaitility.pollInSameThread()
Review Comment:
[P1] This process-wide setting makes atMost unable to stop a blocking
predicate, but existing non-SQL predicates have not all been bounded. For
example the 3-second awaits in the partial-update fault-injection suites call
be_get_compaction_status, whose curl helper can perform ten 10-second attempts
with 5-second sleeps. A failed BE can now turn that 3-second check into a
multi-minute stall. Please scope same-thread polling to the SQL helpers that
need ThreadLocal reuse, or independently cap every blocking predicate before
changing the global default.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -1107,6 +1164,9 @@ protected void finalize() throws Throwable {
@Override
public void close() {
releasePlannerResources();
+ // After the table locks: stopping a remote Doris scan's node sends a
CloseSession to the other
Review Comment:
[P1] This fallback is also never closed by streaming INSERT tasks.
StreamingInsertTask.before() first calls baseCommand.initPlan(...) only to
rewrite the one TVF; a supported TVF join with a remote-Doris table opens a
planning-only Flight session there. The rewritten command is then
planned/executed again, while StreamingTaskScheduler invokes the task directly
and its per-attempt cleanup only nulls fields. With no TaskProcessor close,
every success or retry can strand the first session until wait_timeout. Please
close/remove the task StatementContext on every attempt and cover a TVF join
with remote Doris.
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteContext.groovy:
##########
@@ -148,17 +156,71 @@ class SuiteContext implements Closeable {
// jdbc:mysql
Connection getConnection() {
+ closeConnectionsOfFinishedThreads()
def threadConnInfo = threadLocalConn.get()
if (threadConnInfo == null) {
threadConnInfo = new ConnectionInfo()
- threadConnInfo.conn = getConnectionByDbName(dbName)
+ threadConnInfo.conn =
trackDorisConnection(getConnectionByDbName(dbName))
threadConnInfo.username = config.jdbcUser
threadConnInfo.password = config.jdbcPassword
threadLocalConn.set(threadConnInfo)
}
return threadConnInfo.conn
}
+ private Connection trackDorisConnection(Connection conn) {
+ openedDorisConnections.put(conn, Thread.currentThread())
+ return conn
+ }
+
+ // Closes a connection one of the thread-local accessors opened, and
forgets it (see openedDorisConnections).
+ void closeDorisConnection(Connection conn, String what) {
+ openedDorisConnections.remove(conn)
+ closeQuietly(conn, what)
+ }
+
+ private static void closeQuietly(Connection conn, String what) {
+ try {
+ conn.close()
+ } catch (Throwable t) {
+ log.warn("Close ${what} failed".toString(), t)
+ }
+ }
+
+ // A thread the suite created itself took its thread-local connection to
the grave: nothing on that
+ // thread runs closeThreadLocal() once it has finished. Called on every
statement, this closes those
+ // connections, so a suite that starts a thread per step (Thread.start {
streamLoad ... }; join) holds
+ // at most the connections of the threads still running, not one per step
until the suite ends.
+ private void closeConnectionsOfFinishedThreads() {
+ int closed = 0
+ for (Map.Entry<Connection, Thread> entry :
openedDorisConnections.entrySet()) {
+ if (!entry.value.isAlive() &&
openedDorisConnections.remove(entry.key, entry.value)) {
+ closeQuietly(entry.key, "connection of finished thread
${entry.value.name}".toString())
+ closed++
+ }
+ }
+ if (closed > 0) {
+ log.info("Closed ${closed} connection(s) opened on threads of
suite ${suiteName} that have finished"
+ .toString())
+ }
+ }
+
+ // The connections still open once the suite is over, whichever thread
opened them (see
+ // openedDorisConnections). The warning names the suite: a `sql` on a
thread the suite created
+ // itself and left running (an Executors pool it never shut down) is what
leaves them behind.
+ private void closeLeftoverDorisConnections() {
+ List<Connection> leftover = new
ArrayList<>(openedDorisConnections.keySet())
Review Comment:
[P1] The snapshot and clear are not atomic with trackDorisConnection(). A
still-running suite worker can register between them, in which case clear drops
the connection without closing it, or just after clear, in which case it
remains after the only teardown drain. Please serialize registration with
teardown and mark the context closing so a late registration is closed/rejected
immediately; a barrier-based race test would make this deterministic.
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteContext.groovy:
##########
@@ -204,7 +266,7 @@ class SuiteContext implements Closeable {
def threadConnInfo = threadArrowFlightSqlConn.get()
if (threadConnInfo == null) {
threadConnInfo = new ConnectionInfo()
- threadConnInfo.conn =
config.getConnectionByArrowFlightSqlDbName(dbName)
+ threadConnInfo.conn =
trackDorisConnection(config.getConnectionByArrowFlightSqlDbName(dbName))
Review Comment:
[P1] Please run the finished-thread scavenger from this accessor too (and
from getMasterConnection, or centralize it in tracking). Arrow-group suites
route every sql call here, while closeConnectionsOfFinishedThreads() is invoked
only by getConnection(). A long Arrow suite that repeatedly creates raw worker
threads therefore keeps every finished thread's Flight connection until suite
teardown and can still hit the frontend connection quota that this tracker is
intended to prevent.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -1107,6 +1164,9 @@ protected void finalize() throws Throwable {
@Override
public void close() {
releasePlannerResources();
+ // After the table locks: stopping a remote Doris scan's node sends a
CloseSession to the other
+ // frontend, which must not be waited for under a lock.
+ stopScanNodesLeftBehind();
Review Comment:
[P1] External-table EXPORT also loses this fallback. ExportTaskExecutor
installs the ExportJob adapter's StatementContext inside
AutoCloseConnectContext; a remote-Doris EXPORT can open its Flight session
during planning and then fail at the post-plan scan block-rule check before any
coordinator owns it. AutoCloseConnectContext.close() calls
ConnectContext.clear(), which nulls the StatementContext without closing it,
and then removes the thread local. Please close the installed context before
clear/remove (and audit the other auto-close query owners), with a failed
remote-Doris EXPORT test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -1093,6 +1108,48 @@ public synchronized void releasePlannerResources() {
}
}
+ /**
+ * Registers a scan node whose {@link ScanNode#stop()} must have run by
the time this statement
+ * ends: the coordinator of the statement runs it when it closes, and
{@link #close()} runs it
+ * for a plan that never got a coordinator or whose coordinator nobody
closed (see
+ * {@link #scanNodesToStopAtClose}).
+ */
+ public void stopScanNodeAtClose(ScanNode scanNode) {
Review Comment:
[P1] Please close this fallback on every direct COM_STMT_EXECUTE as well. A
prepared remote-Doris query can open its Flight session during planning and
then fail before coordinator construction (for example at the post-plan scan
block-rule check). MysqlConnectProcessor catches that failure and only
finalizes the response; it never calls StatementContext.close(), and
nextStatementContext() resets only connector scope before dropping the old
context. Re-executing that prepared statement therefore leaks one remote
session each time until wait_timeout. Either add a per-execute finally or fully
close the outgoing prepared context, with coverage for this pre-coordinator
failure.
--
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]
