github-actions[bot] commented on code in PR #69996: URL: https://github.com/apache/doris/pull/69996#discussion_r4463117156
########## 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: [P2] Include backend address snapshots in the registry budget `results.size()` counts distinct result IDs, but in the default parallel-result mode every endpoint of a query uses the same query ID, so registration consumes one capacity unit while the `Route` deep-copies every participating BE address. Normal EOF never unregisters the route. For example, on a 1000-BE cluster, about 111 completed distributed Flight queries/sec during the default 900-second timeout can leave roughly 100000 routes retaining 100 million address objects plus their list references. That can consume gigabytes of FE heap despite the claimed tens-of-MB bound and despite all streams having finished. A distributed aggregate with a single result endpoint has the same mismatch. Account for retained backend addresses in the eviction weight, or otherwise bound/share those snapshots, while preserving whole-route eviction and the session-independent lifetime. ########## 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: [P2] Bound the actual FE call on stream teardown The 2000 ms argument is applied only when `ClientCacheHelper::_create_client` creates a transport; a cache hit keeps its previous socket timeout. The shared `frontend_client_cache` is also used by profile reporting with the default 60000 ms timeout (`runtime_query_statistics_mgr.cpp`), so an aborted DoGet can inherit a 60-second wait on an unresponsive FE before this loop reaches the FE that owns the route. `ClientConnection` also defaults to three acquisition attempts with 1+2+4-second backoffs inside each outer attempt, adding at least 14 seconds even on immediate connection refusal. These waits now run in `Close()` and the reader destructor, occupying the stream thread and delaying cancellation of the remote producers/buffers. Give this path an actual connection/RPC deadline, including reused transports and acquisition retries, or perform the reporting outside stream teardown; cover a long-timeout cached connection and an unreachable FE in the tests. -- 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]
