github-actions[bot] commented on code in PR #67582: URL: https://github.com/apache/doris/pull/67582#discussion_r3956213856
########## fe/fe-core/src/main/java/org/apache/doris/qe/PointQueryMultiExecutor.java: ########## @@ -0,0 +1,506 @@ +// 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.qe; + +import org.apache.doris.analysis.ExprToThriftVisitor; +import org.apache.doris.analysis.LiteralExpr; +import org.apache.doris.analysis.LiteralExprUtils; +import org.apache.doris.analysis.NullLiteral; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.Type; +import org.apache.doris.common.Config; +import org.apache.doris.common.Status; +import org.apache.doris.common.UserException; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.plans.PlaceholderId; +import org.apache.doris.planner.OlapScanNode.PointQueryRoute; +import org.apache.doris.proto.InternalService; +import org.apache.doris.rpc.BackendServiceProxy; +import org.apache.doris.rpc.RpcException; +import org.apache.doris.rpc.TCustomProtocolFactory; +import org.apache.doris.system.Backend; +import org.apache.doris.thrift.TExpr; +import org.apache.doris.thrift.TExprNode; +import org.apache.doris.thrift.TResultBatch; +import org.apache.doris.thrift.TStatusCode; + +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import com.google.protobuf.ByteString; +import org.apache.thrift.TDeserializer; +import org.apache.thrift.TException; +import org.apache.thrift.TSerializer; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +/** + * Query-level coordinator for the supported single-column IN point query. Key tuples routed to + * the same tablet are merged into one ordinary tablet_fetch_data request. RPCs are concurrent across + * BEs but serial on each BE, because requests with the same UUID share a reusable execution context + * there. Execution uses lightweight requests when enabled and resends a full request on a cold cache. + * Cached executions require parameters bound with types compatible with their key columns; + * parameter type changes requiring different comparison semantics are not supported. + */ +public class PointQueryMultiExecutor extends PointQueryExecutor { + private final ShortCircuitQueryContext context; + private final StatementContext statementContext; + private final int maxMessageSize; + private final Set<Future<?>> currentRpcFutures = Collections.synchronizedSet(new HashSet<>()); + private final Set<Long> failedBackends = new HashSet<>(); + private long timeoutMs = Config.point_query_timeout_ms; + private volatile boolean cancelled; + + private static final class TabletTask { + private final long tabletId; + private final List<Backend> candidateBackends; + private final List<InternalService.KeyTuple> keyTuples = new ArrayList<>(); + private Backend backend; + private InternalService.PTabletKeyLookupRequest request; + private Future<InternalService.PTabletKeyLookupResponse> future; + private int attemptCount; + private String lastFailure; + + private TabletTask(long tabletId, List<Backend> candidateBackends) { + this.tabletId = tabletId; + this.candidateBackends = candidateBackends; + } + + private Backend nextBackend(Set<Long> failedBackends) { + int maxAttempts = Math.max(1, + Math.min(Config.max_point_query_retry_time, candidateBackends.size())); + while (attemptCount < maxAttempts) { + Backend backend = candidateBackends.get(attemptCount++); Review Comment: [P2] Do not charge globally skipped backends to this tablet's retry budget. `executeRound()` preassigns every task and increments `attemptCount` here before any wave runs. If two tasks choose B1, the first task can transport-fail and add B1 to `failedBackends`; the second task's later wave is then skipped without an RPC, but it has already spent one of the default two attempts. If its next backend B2 also fails, B3 is never considered even though the task issued only one RPC. Please track candidate position separately from actual dispatch attempts (or refund a pre-dispatch skip) and add a deterministic same-BE wave retry test. -- 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]
