github-actions[bot] commented on code in PR #68353:
URL: https://github.com/apache/doris/pull/68353#discussion_r4234044795
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java:
##########
@@ -94,13 +91,45 @@ protected void doInitialize() throws UserException {
source = new RemoteDorisSource(desc);
}
+ // Builds the query the scan runs on the remote frontend from its slots
and conjuncts, final by now: EXPLAIN
+ // shows it, startSplit runs it.
+ @Override
+ protected void convertPredicate() {
+ createColumns();
+ createFilters();
+ }
+
@Override
- public List<Split> getSplits(int numBackends) throws UserException {
- List<Pair<String, ByteBuffer>> locationAndTicketList = executeQuery();
+ public boolean isBatchMode() {
Review Comment:
[P2] Limit remote Doris batch scans to useful backends. With `isBatchMode()`
always true, `FileQueryScanNode` creates a split-source scan range on every
eligible local BE. A remote query that returns one endpoint now starts scan
work on every BE and sends empty split-fetch RPCs from the extras; previously
that endpoint was assigned to one BE. A selective lookup therefore pays startup
and RPC cost proportional to the local cluster size. Bound the participating
backend set using a useful concurrency estimate while retaining dispatch-time
remote execution.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisFlightSession.java:
##########
@@ -80,23 +82,31 @@ private RemoteDorisFlightSession(Pair<String, Integer>
hostAndPort, BufferAlloca
}
/**
- * 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.
+ * Opens a session on the remote frontend at {@code hostAndPort} with the
catalog's credentials, waiting at most
+ * {@code timeoutSec} for the handshake. Nothing is left behind when this
fails: a handshake that was refused or
+ * timed out 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 {
+ static RemoteDorisFlightSession open(Pair<String, Integer> hostAndPort,
String user, String password,
+ int timeoutSec) 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()) {
+ // What FlightClient#authenticateBasicToken does, with a deadline:
it waits for the handshake without
+ // one. The scan opens the session while the coordinator
dispatches the plan, holding the query's
+ // admission slot, and a remote frontend that accepts the
connection but never answers would hold both
+ // for good: neither KILL nor the query's timeout interrupts the
wait.
+ ClientIncomingAuthHeaderMiddleware.Factory bearerToken =
+ new ClientIncomingAuthHeaderMiddleware.Factory(new
ClientBearerHeaderHandler());
+ flightClient = FlightClient.builder(allocator, new
Location(uri)).intercept(bearerToken).build();
+ flightClient.handshake(new CredentialCallOption(new
BasicAuthCredentialWriter(user, password)),
+ CallOptions.timeout(timeoutSec, TimeUnit.SECONDS));
Review Comment:
[P2] Clean up sessions created before a handshake deadline. The remote
`FlightSessionsInConnectPool.openSession()` registers a connection before its
bearer token is returned. If this new deadline expires after registration but
before the bearer header arrives, the catch closes only local resources; this
client cannot send `CloseSession` without that token. Each retry can leave
another remote connection slot occupied until `wait_timeout`, potentially
exhausting the remote user's limit. Add server-side cleanup for an abandoned
handshake and test delayed delivery after registration.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java:
##########
@@ -94,13 +91,45 @@ protected void doInitialize() throws UserException {
source = new RemoteDorisSource(desc);
}
+ // Builds the query the scan runs on the remote frontend from its slots
and conjuncts, final by now: EXPLAIN
+ // shows it, startSplit runs it.
+ @Override
+ protected void convertPredicate() {
+ createColumns();
+ createFilters();
+ }
+
@Override
- public List<Split> getSplits(int numBackends) throws UserException {
- List<Pair<String, ByteBuffer>> locationAndTicketList = executeQuery();
+ public boolean isBatchMode() {
+ return true;
+ }
- return locationAndTicketList.stream()
- .map(locationAndTicket -> new
RemoteDorisSplit(locationAndTicket.first, locationAndTicket.second))
- .collect(Collectors.toList());
+ // The params of the scan depend on none of its splits, and producing one
means running the remote query.
+ @Override
+ protected boolean needsSampleSplit() {
+ return false;
+ }
+
+ // Unknown until the remote frontend has run the query: each endpoint of
its result is a split.
+ @Override
+ public int numApproximateSplits() {
+ return 0;
+ }
+
+ /**
+ * Runs the query on the remote frontend and hands the endpoints of its
result to the backends as the splits.
+ * The split assignment calls this when the coordinator dispatches the
plan. The Flight SQL session the query
+ * runs in goes to the split assignment, which ends it when the
coordinator stops this scan: see
+ * {@link RemoteDorisFlightSession} for why it lives that long.
+ */
+ @Override
+ public void startSplit(int numBackends) throws UserException {
+ List<Split> splits = new ArrayList<>();
+ for (Pair<String, ByteBuffer> locationAndTicket : executeQuery()) {
Review Comment:
[P2] Bound dispatch-time Flight calls by the local query deadline.
`Coordinator.exec()` acquires a running workload-group token before
`startSplit()` calls `executeQuery()` synchronously. If the remote FE stalls,
local `query_timeout` or KILL cancels the scan but does not interrupt this
call; the separate catalog Flight timeout (15 seconds by default, configurable
longer) controls its return, and `Coordinator.close()` releases the running
token afterward. With `maxConcurrency=1`, later queries stay queued after the
first query is canceled. Use the remaining local deadline or a cancellable RPC
so cancellation releases admission promptly.
--
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]
