morningman commented on code in PR #69996:
URL: https://github.com/apache/doris/pull/69996#discussion_r4466352092
##########
be/src/service/arrow_flight/arrow_flight_batch_reader.cpp:
##########
@@ -103,6 +105,37 @@ void ArrowFlightBatchReaderBase::close(const Status&
reason) {
_statement->result_addr.port == config::brpc_port) {
env->result_mgr()->cancel_arrow_flight_query(_statement->query_id,
reason);
}
+ // A result endpoint can outlive its local fragment. Find the owning FE by
buffer ID,
+ // then use its query-wide cancellation route, including for older result
BEs.
+ for (const auto& [address, info] : env->get_running_frontends()) {
+ Status status;
+ // One retry: the FE keeps the route when an attempt fails (for
example a BE missed the
+ // FE's one-second deadline), and a cached connection to the FE may
have gone stale.
+ for (int attempt = 0; attempt < 2; ++attempt) {
+ FrontendServiceConnection client(env->frontend_client_cache(),
address, 2000, &status);
Review Comment:
Fixed in `152c56c38cc`. Agreed, and the cache worked against this call in
both directions. A reused transport keeps whatever its opener set: profile and
status reports open theirs with `thrift_rpc_timeout_ms` (60 s), and the
information_schema scanners open theirs to the master FE with the query's
execution timeout (900 s by default). The other way round, this call's 2 s
transports, and the 1 s ones that `reopen(1000)` made after any thrift error
(an older FE's unknown-method reply included), went back to the shared cache,
where `fetchSplitBatch`, whose FE side holds the call for up to
`fetch_splits_max_wait_time` (1 s by default), or a status report could take
them. An FE that has stopped also stays on every BE's list: the master sends it
with process UUID 0, which `get_running_frontends()` counts as running, so when
it sorts before the FE that planned the query, every early close on every BE
waits for it.
The cancellation now keeps FE connections of its own
(`ExecEnv::flight_cancel_frontend_client_cache`, at most two kept per FE), as
`MasterServerClient` does, so each transport it uses was opened with its 2 s
timeout and none reaches another caller. Acquiring a connection does not back
off (`max_retries = 0`), a transport is reopened with the same 2 s, and an FE
that cannot be connected to gets no second attempt, since neither reason for
the retry applies to it. An FE ahead of the planning one now costs an early
close about 0.1 s when it refuses connections, 3.1 s when it does not answer
the connect (`thrift_connect_timeout_seconds`), and 4 s when it accepts the
connection and does not answer. The local cleanup of #69945 still runs first.
The Cost paragraph of the description understated this wait and is corrected.
Tests: `CloseIgnoresLongTimeoutSharedTransport` leaves a 60 s transport to
an FE whose first answer stalls for 5 s in the shared cache and requires the
early close to finish within 4 s, through the retry;
`CloseMovesPastUnreachableFrontend` puts an FE that refuses connections first
and requires the close to reach the planning FE within 2 s. Before this change
they took 5.0 s and 14.7 s.
##########
fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlQueryCancellation.java:
##########
@@ -0,0 +1,187 @@
+// 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.arrowflight;
+
+import org.apache.doris.arrowflight.results.FlightSqlEndpointsLocation;
+import org.apache.doris.common.Status;
+import org.apache.doris.proto.InternalService.PCancelPlanFragmentResult;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.Coordinator;
+import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TStatus;
+import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TUniqueId;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import java.util.stream.Collectors;
+
+/**
+ * Routes the early close of a Flight result stream, which a BE reports
through cancelFlightQuery, to every BE of
+ * the stream's query. A route outlives the request and the session that
registered it: the client reads the
+ * results from the BEs, so it can still hold a stream of the query open after
its session has started another
+ * request or has been torn down, and close that stream early then. A route
goes when its cancellation succeeds,
+ * when its query's execution timeout (plus 5 s) has passed, and when {@link
#MAX_ROUTED_RESULTS} newer result ids
+ * push it out. Those go oldest first, as the oldest routes are the likeliest
to belong to streams that have
+ * ended; a query whose route was pushed out ends at its timeout, as every
Flight query did before routes existed.
+ */
+public final class FlightSqlQueryCancellation {
+ private static final Logger LOG =
LogManager.getLogger(FlightSqlQueryCancellation.class);
+ // Counted in result ids, which cost a few hundred bytes each, so the
registry stays within some tens of MB
+ // however many Flight queries are registered within one execution timeout.
+ static final int MAX_ROUTED_RESULTS = 100_000;
+ public static final FlightSqlQueryCancellation INSTANCE =
+ new FlightSqlQueryCancellation(System::nanoTime,
MAX_ROUTED_RESULTS);
+ private final LongSupplier nanoClock;
+ private final int capacity;
+ // In registration order, so the eldest entry belongs to the oldest route.
Guarded by this.
+ private final LinkedHashMap<TUniqueId, Route> results = new
LinkedHashMap<>();
+
+ FlightSqlQueryCancellation(LongSupplier nanoClock, int capacity) {
+ this.nanoClock = nanoClock;
+ this.capacity = capacity;
+ }
+
+ /**
+ * Registers the route of a Flight query whose results stay on the BEs,
from the result ids of the endpoints
+ * the session collected for it, before its coordinator is closed or
deferred. Nothing of the session is
+ * kept, so neither its next request nor its teardown drops the route.
+ */
+ public void register(ConnectContext context, Coordinator coord) {
+ register(coord.getQueryId(),
context.getFlightSqlEndpointsLocations().stream()
+
.map(FlightSqlEndpointsLocation::getFinstId).distinct().collect(Collectors.toList()),
+ coord.getBackendBrpcAddresses(),
coord.getQueryOptions().getExecutionTimeout());
+ }
+
+ public void register(TUniqueId queryId, List<TUniqueId> resultIds,
+ List<TNetworkAddress> backends, int timeoutSeconds) {
+ if (backends.isEmpty()) {
+ throw new IllegalArgumentException("Flight query has no
cancellation backends");
+ }
+ // Keep only cancellation addresses, not the coordinator, scan state,
or query queue slot.
+ // Ordinary Flight coordinators are unregistered when GetFlightInfo
returns.
+ Route route = new Route(queryId.deepCopy(),
+
resultIds.stream().map(TUniqueId::deepCopy).distinct().collect(Collectors.toList()),
+
backends.stream().map(TNetworkAddress::deepCopy).distinct().collect(Collectors.toList()),
+ nanoClock.getAsLong() + TimeUnit.SECONDS.toNanos(Math.max(0L,
timeoutSeconds) + 5));
+ synchronized (this) {
+ for (TUniqueId resultId : route.resultIds) {
+ results.put(resultId, route);
+ }
+ // Whole routes leave from the eldest end: the expired ones there,
then as many as the bound requires.
+ long now = nanoClock.getAsLong();
+ while (!results.isEmpty()) {
+ Route eldest = results.values().iterator().next();
+ if (results.size() <= capacity && !eldest.isExpired(now)) {
Review Comment:
Fixed in `1c9a89481ed`. Agreed. The result ids of a query already share one
route and its address list, but the bound counted only the ids, and in parallel
result mode a query has one. The registry now also bounds the BE addresses that
its routes hold, at 1,000,000 (about 30 MB), and drops whole routes, oldest
first, while either bound is exceeded; a route's addresses are released when
its last result id leaves, also when a newer registration takes the id over.
Under the old bound, a route held about 3 KB with 100 BEs and 29 KB with 1,000,
so a full registry held about 0.3 GB and 2.9 GB; now it stays within about 55
MB whatever the cluster size. Large clusters give up retention instead: with
1,000 BEs the registry keeps the routes of the last 1,000 queries, about 9 s of
them at 110 queries/s, and a query whose route was pushed out ends at its
timeout, as before this PR. Dropping a route when its streams reach EOF would
need the BEs to report every EOF to the FE as well, one more F
E call per endpoint and query; I left that out.
Tests: `boundCountsRetainedBackends` requires a registration whose addresses
exceed the bound to drop the oldest route as a whole, and
`reregisteredResultReleasesOldRouteBackends` requires a route whose result id
another registration took over to release its addresses once.
--
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]
