github-actions[bot] commented on code in PR #67582:
URL: https://github.com/apache/doris/pull/67582#discussion_r3950212563
##########
fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java:
##########
@@ -1198,6 +1199,95 @@ static void validateRequiredQuerySelection(boolean
skipMissingVersion, int useFi
}
}
+ /** Immutable routing result for one expanded multi-key lookup tuple. */
+ public static final class PointQueryRoute {
+ private final long tabletId;
+ private final List<Long> candidateBackendIds;
+ private final boolean backendOrderBySelection;
+
+ private PointQueryRoute(long tabletId, List<Long> candidateBackendIds,
+ boolean backendOrderBySelection) {
+ this.tabletId = tabletId;
+ this.candidateBackendIds = candidateBackendIds;
+ this.backendOrderBySelection = backendOrderBySelection;
+ }
+
+ public long getTabletId() {
+ return tabletId;
+ }
+
+ public List<Long> getCandidateBackendIds() {
+ return candidateBackendIds;
+ }
+
+ public boolean isBackendOrderBySelection() {
+ return backendOrderBySelection;
+ }
+ }
+
+ /**
+ * Resolve one complete primary-key tuple for the multi-key point-query
coordinator. Routing is
+ * serial because the reusable scan node owns mutable pruning state. The
original predicates are
+ * restored before returning; only the copied route escapes this method.
+ *
+ * @return null when the tuple does not map to an existing non-empty
partition or selected bucket
+ */
+ public PointQueryRoute routePointQueryKeyTuple(List<LiteralExpr>
keyValues) throws UserException {
+ List<Column> keyColumns = olapTable.getBaseSchemaKeyColumns();
+ Preconditions.checkState(keyValues.size() == keyColumns.size(),
+ "Point-query key tuple has %s values but table requires %s",
+ keyValues.size(), keyColumns.size());
+
+ List<Expr> originalConjuncts = conjuncts;
+ List<Expr> routingConjuncts = new ArrayList<>(keyColumns.size());
+ for (int i = 0; i < keyColumns.size(); ++i) {
+ LiteralExpr value = keyValues.get(i);
+ Preconditions.checkState(value != null && !(value instanceof
NullLiteral),
+ "Multi-key point query cannot route a NULL key value");
+ SlotRef slot = Preconditions.checkNotNull(
+ findPointQueryKeySlot(originalConjuncts,
keyColumns.get(i).getName()),
+ "Missing point-query predicate for key column %s",
keyColumns.get(i).getName());
+ routingConjuncts.add(new
BinaryPredicate(BinaryPredicate.Operator.EQ, slot, value));
+ }
+
+ try {
+ conjuncts = routingConjuncts;
+ // Unlike the ordinary point-query path, multi-get evaluates this
mutable scan node
+ // repeatedly with different key tuples.
+ columnFilters.clear();
+ columnNameToRange.clear();
+ totalTabletsNum = 0;
+ selectedSplitNum = 0;
+ totalBytes = 0;
+ tabletBytes.clear();
+ lazyEvaluateRangeLocations();
+ if (scanTabletIds.isEmpty()) {
+ return null;
+ }
+ Preconditions.checkState(scanTabletIds.size() == 1,
+ "Multi-key point query must route to exactly one tablet");
+ if (scanBackendIds.isEmpty()) {
Review Comment:
**[P2] Preserve `skip_bad_tablet` for each routed key.**
`computeTabletInfo()` records the tablet before replica selection; when no
replica is queryable and `skip_bad_tablet=true`, `addScanRangeLocations()`
deliberately skips its range, leaving this method with a tablet ID but no
backend. Throwing here makes one bad tablet abort the whole multi-get, whereas
the existing single-key path returns an empty batch. Please return no route for
this skipped key while retaining the error when skipping is disabled, and add
an unavailable-tablet case.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java:
##########
@@ -1198,6 +1199,95 @@ static void validateRequiredQuerySelection(boolean
skipMissingVersion, int useFi
}
}
+ /** Immutable routing result for one expanded multi-key lookup tuple. */
+ public static final class PointQueryRoute {
+ private final long tabletId;
+ private final List<Long> candidateBackendIds;
+ private final boolean backendOrderBySelection;
+
+ private PointQueryRoute(long tabletId, List<Long> candidateBackendIds,
+ boolean backendOrderBySelection) {
+ this.tabletId = tabletId;
+ this.candidateBackendIds = candidateBackendIds;
+ this.backendOrderBySelection = backendOrderBySelection;
+ }
+
+ public long getTabletId() {
+ return tabletId;
+ }
+
+ public List<Long> getCandidateBackendIds() {
+ return candidateBackendIds;
+ }
+
+ public boolean isBackendOrderBySelection() {
+ return backendOrderBySelection;
+ }
+ }
+
+ /**
+ * Resolve one complete primary-key tuple for the multi-key point-query
coordinator. Routing is
+ * serial because the reusable scan node owns mutable pruning state. The
original predicates are
+ * restored before returning; only the copied route escapes this method.
+ *
+ * @return null when the tuple does not map to an existing non-empty
partition or selected bucket
+ */
+ public PointQueryRoute routePointQueryKeyTuple(List<LiteralExpr>
keyValues) throws UserException {
+ List<Column> keyColumns = olapTable.getBaseSchemaKeyColumns();
+ Preconditions.checkState(keyValues.size() == keyColumns.size(),
+ "Point-query key tuple has %s values but table requires %s",
+ keyValues.size(), keyColumns.size());
+
+ List<Expr> originalConjuncts = conjuncts;
+ List<Expr> routingConjuncts = new ArrayList<>(keyColumns.size());
+ for (int i = 0; i < keyColumns.size(); ++i) {
+ LiteralExpr value = keyValues.get(i);
+ Preconditions.checkState(value != null && !(value instanceof
NullLiteral),
+ "Multi-key point query cannot route a NULL key value");
+ SlotRef slot = Preconditions.checkNotNull(
+ findPointQueryKeySlot(originalConjuncts,
keyColumns.get(i).getName()),
+ "Missing point-query predicate for key column %s",
keyColumns.get(i).getName());
+ routingConjuncts.add(new
BinaryPredicate(BinaryPredicate.Operator.EQ, slot, value));
+ }
+
+ try {
+ conjuncts = routingConjuncts;
+ // Unlike the ordinary point-query path, multi-get evaluates this
mutable scan node
+ // repeatedly with different key tuples.
+ columnFilters.clear();
+ columnNameToRange.clear();
+ totalTabletsNum = 0;
+ selectedSplitNum = 0;
+ totalBytes = 0;
+ tabletBytes.clear();
+ lazyEvaluateRangeLocations();
Review Comment:
**[P1] Apply TABLESAMPLE only once for the statement.** This per-tuple route
calls `lazyEvaluateRangeLocations()`, which clears `sampleTabletIds` and
chooses fresh random seeks for a non-REPEATABLE sample. An `IN` lookup can
consequently return the union of several independent tablet samples, unlike
normal execution's single sample. Please exclude sampled scans from multi-get
eligibility or retain one sampled tablet set across all tuples, and add a
multi-bucket TABLESAMPLE case.
##########
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++);
+ if (!failedBackends.contains(backend.getId()) &&
SimpleScheduler.isAvailable(backend)) {
+ return backend;
+ }
+ }
+ return null;
+ }
+ }
+
+ private final class ResultAccumulator {
+ private final List<ByteBuffer> rows = new ArrayList<>();
+ private final TDeserializer deserializer;
+ private long resultBytes;
+
+ private ResultAccumulator() throws TException {
+ deserializer = new TDeserializer(new
TCustomProtocolFactory(maxMessageSize));
+ }
+
+ private void add(InternalService.PTabletKeyLookupResponse response)
throws TException {
+ if (response.hasEmptyBatch() && response.getEmptyBatch()) {
+ return;
+ }
+ if (!response.hasRowBatch() || response.getRowBatch().isEmpty()) {
+ throw new TException("No row batch or empty batch found in
point-query response");
+ }
+
+ TResultBatch batch = new TResultBatch();
+ try {
+ deserializer.deserialize(batch,
response.getRowBatch().toByteArray());
+ } catch (TException e) {
+ if (ResultReceiver.isMessageSizeExceeded(e)) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ throw e;
+ }
+ for (ByteBuffer row : batch.getRows()) {
+ resultBytes += row.remaining();
+ if (resultBytes > maxMessageSize) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ rows.add(row);
+ }
+ }
+
+ private RowBatch finish() {
+ RowBatch rowBatch = new RowBatch();
+ if (rows.isEmpty()) {
+ return rowBatch;
+ }
+ TResultBatch resultBatch = new TResultBatch();
+ resultBatch.setRows(rows);
+ resultBatch.setIsCompressed(false);
+ resultBatch.setPacketSeq(0);
+ rowBatch.setBatch(resultBatch);
+ return rowBatch;
+ }
+ }
+
+ public PointQueryMultiExecutor(ShortCircuitQueryContext context,
+ StatementContext statementContext, int maxMessageSize) {
+ super(context, maxMessageSize);
+ this.context = context;
+ this.statementContext = statementContext;
+ this.maxMessageSize = maxMessageSize;
+ }
+
+ public static void directExecuteShortCircuitQuery(StmtExecutor executor,
+ PreparedStatementContext preparedStmtCtx) throws Exception {
+ // Multi-get reads current bindings without mutating the cached IN
predicate.
+ executor.executeAndSendResult(false, false,
+ preparedStmtCtx.shortCircuitQueryContext.get().analzyedQuery,
+ executor.getContext().getMysqlChannel(), null, null);
+ }
+
+ @Override
+ public void setTimeout(long timeoutMs) {
+ this.timeoutMs = timeoutMs;
+ }
+
+ @Override
+ public RowBatch getNext() throws Exception {
+ try {
+ return getNextInternal();
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ invalidateCache();
+ throw e;
+ } finally {
+ cancelInFlightRpcs();
+ }
+ }
+
+ private RowBatch getNextInternal() throws Exception {
+ long deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ Map<Long, TabletTask> tasksByTablet = buildTabletTasks(deadlineNanos);
+ ResultAccumulator accumulator = new ResultAccumulator();
+ List<TabletTask> pending = new ArrayList<>(tasksByTablet.values());
+ while (!pending.isEmpty()) {
+ checkCancelledOrTimedOut(deadlineNanos);
+ pending = executeRound(pending, accumulator, deadlineNanos);
+ }
+ return accumulator.finish();
+ }
+
+ private Map<Long, TabletTask> buildTabletTasks(long deadlineNanos) throws
TException, UserException {
+ List<Column> keyColumns =
context.scanNode.getOlapTable().getBaseSchemaKeyColumns();
+ Map<String, Integer> keyIndexes = new HashMap<>(keyColumns.size());
+ for (int i = 0; i < keyColumns.size(); ++i) {
+ keyIndexes.put(normalizeColumnName(keyColumns.get(i).getName()),
i);
+ }
+
+ List<PlaceholderId> inPlaceholderIds =
statementContext.getPointQueryInPlaceholderIds();
+ SlotReference inSlot =
statementContext.getIdToComparisonSlot().get(inPlaceholderIds.get(0));
+ int inKeyIndex = keyIndexes.get(normalizeColumnName(
+ inSlot.getOriginalColumn().get().getName()));
+ LiteralExpr[] keyValues = new LiteralExpr[keyColumns.size()];
+ for (Map.Entry<PlaceholderId, SlotReference> entry
+ : statementContext.getIdToComparisonSlot().entrySet()) {
+ SlotReference slot = entry.getValue();
+ int keyIndex = keyIndexes.get(normalizeColumnName(
+ slot.getOriginalColumn().get().getName()));
+ // Eligibility checking guarantees exactly one placeholder for
each equality key.
+ if (keyIndex != inKeyIndex) {
+ keyValues[keyIndex] = ((Literal)
statementContext.getIdToPlaceholderRealExpr()
+ .get(entry.getKey())).toLegacyLiteral();
+ }
+ }
+
+ TSerializer serializer = new TSerializer();
+ ByteString[] serializedKeyValues = new ByteString[keyColumns.size()];
+ for (int i = 0; i < keyValues.length; ++i) {
+ if (i == inKeyIndex) {
+ continue;
+ }
+ if (keyValues[i] instanceof NullLiteral) {
+ return Collections.emptyMap();
+ }
+ keyValues[i] = normalizeKeyLiteral(keyColumns.get(i),
keyValues[i]);
+ serializedKeyValues[i] = serializeKeyLiteral(keyValues[i],
serializer);
+ }
+
+ Set<InternalService.KeyTuple> seenTuples = new HashSet<>();
+ Map<Long, TabletTask> tasksByTablet = new LinkedHashMap<>();
+ for (PlaceholderId placeholderId : inPlaceholderIds) {
+ checkCancelledOrTimedOut(deadlineNanos);
+ LiteralExpr inValue = ((Literal)
statementContext.getIdToPlaceholderRealExpr()
+ .get(placeholderId)).toLegacyLiteral();
+ if (inValue instanceof NullLiteral) {
+ continue;
+ }
+ keyValues[inKeyIndex] =
normalizeKeyLiteral(keyColumns.get(inKeyIndex), inValue);
+ serializedKeyValues[inKeyIndex] =
serializeKeyLiteral(keyValues[inKeyIndex], serializer);
+ InternalService.KeyTuple.Builder tupleBuilder =
InternalService.KeyTuple.newBuilder();
+ for (ByteString serializedValue : serializedKeyValues) {
+ tupleBuilder.addKeyColumnLiterals(serializedValue);
+ }
+ InternalService.KeyTuple keyTuple = tupleBuilder.build();
+ if (!seenTuples.add(keyTuple)) {
+ continue;
+ }
+ PointQueryRoute route =
context.scanNode.routePointQueryKeyTuple(Arrays.asList(keyValues));
+ if (route == null) {
+ continue;
+ }
+ TabletTask task = tasksByTablet.get(route.getTabletId());
+ if (task == null) {
+ List<Backend> candidates = selectPointQueryBackends(route);
+ if (candidates.isEmpty()) {
+ throw new UserException("Tablet " + route.getTabletId()
+ + " has no available backend for multi-key point
query");
+ }
+ task = new TabletTask(route.getTabletId(), candidates);
+ tasksByTablet.put(route.getTabletId(), task);
+ }
+ task.keyTuples.add(keyTuple);
+ }
+ return tasksByTablet;
+ }
+
+ private static LiteralExpr normalizeKeyLiteral(Column column, LiteralExpr
literalExpr)
+ throws TException {
+ Type columnType = column.getType();
+ if (columnType.equals(literalExpr.getType())
+ || columnType.matchesType(literalExpr.getType())) {
+ return literalExpr;
+ }
+ try {
+ return
LiteralExprUtils.createLiteral(literalExpr.getStringValue(), columnType);
+ } catch (org.apache.doris.common.AnalysisException e) {
+ throw new TException("Failed to re-type literal for key column "
+ + column.getName() + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static ByteString serializeKeyLiteral(LiteralExpr literalExpr,
TSerializer serializer)
+ throws TException {
+ TExpr thriftExpr = ExprToThriftVisitor.treeToThrift(literalExpr);
+ Preconditions.checkState(thriftExpr.getNodesSize() == 1,
+ "Expected a single TExprNode for point-query key literal, got
%s",
+ thriftExpr.getNodesSize());
+ TExprNode exprNode = thriftExpr.getNodes().get(0);
+ return ByteString.copyFrom(serializer.serialize(exprNode));
+ }
+
+ private static String normalizeColumnName(String columnName) {
+ return columnName.toLowerCase(Locale.ROOT);
+ }
+
+ private static List<Backend> selectPointQueryBackends(PointQueryRoute
route) {
+ List<Backend> candidates = new
ArrayList<>(route.getCandidateBackendIds().size());
+ for (Long backendId : route.getCandidateBackendIds()) {
+ Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
+ if (SimpleScheduler.isAvailable(backend)) {
+ candidates.add(backend);
+ }
+ }
+ if (!route.isBackendOrderBySelection()) {
+ Collections.shuffle(candidates);
+ }
+ return candidates;
+ }
+
+ private InternalService.PTabletKeyLookupRequest buildLookupRequest(
+ TabletTask task, boolean includeQueryContext) {
+ InternalService.PTabletKeyLookupRequest.Builder builder
+ = InternalService.PTabletKeyLookupRequest.newBuilder()
+ .setTabletId(task.tabletId)
+ .setIsBinaryRow(true)
+ .addAllKeyTuples(task.keyTuples);
Review Comment:
**[P1] Keep multi-get compatible with old BEs.** During a rolling upgrade
this request can reach an old replica because backend selection has no
capability gate. The base BE accepts repeated tuples but returns `NotSupported`
when the result mixes live and deleted rows, so a valid `IN` lookup fails after
retrying old replicas. Please gate batching on BE capability or fall back to
single-key/normal execution, and cover the mixed live/deleted upgrade case.
##########
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++);
+ if (!failedBackends.contains(backend.getId()) &&
SimpleScheduler.isAvailable(backend)) {
+ return backend;
+ }
+ }
+ return null;
+ }
+ }
+
+ private final class ResultAccumulator {
+ private final List<ByteBuffer> rows = new ArrayList<>();
+ private final TDeserializer deserializer;
+ private long resultBytes;
+
+ private ResultAccumulator() throws TException {
+ deserializer = new TDeserializer(new
TCustomProtocolFactory(maxMessageSize));
+ }
+
+ private void add(InternalService.PTabletKeyLookupResponse response)
throws TException {
+ if (response.hasEmptyBatch() && response.getEmptyBatch()) {
+ return;
+ }
+ if (!response.hasRowBatch() || response.getRowBatch().isEmpty()) {
+ throw new TException("No row batch or empty batch found in
point-query response");
+ }
+
+ TResultBatch batch = new TResultBatch();
+ try {
+ deserializer.deserialize(batch,
response.getRowBatch().toByteArray());
+ } catch (TException e) {
+ if (ResultReceiver.isMessageSizeExceeded(e)) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ throw e;
+ }
+ for (ByteBuffer row : batch.getRows()) {
+ resultBytes += row.remaining();
+ if (resultBytes > maxMessageSize) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ rows.add(row);
+ }
+ }
+
+ private RowBatch finish() {
+ RowBatch rowBatch = new RowBatch();
+ if (rows.isEmpty()) {
+ return rowBatch;
+ }
+ TResultBatch resultBatch = new TResultBatch();
+ resultBatch.setRows(rows);
+ resultBatch.setIsCompressed(false);
+ resultBatch.setPacketSeq(0);
+ rowBatch.setBatch(resultBatch);
+ return rowBatch;
+ }
+ }
+
+ public PointQueryMultiExecutor(ShortCircuitQueryContext context,
+ StatementContext statementContext, int maxMessageSize) {
+ super(context, maxMessageSize);
+ this.context = context;
+ this.statementContext = statementContext;
+ this.maxMessageSize = maxMessageSize;
+ }
+
+ public static void directExecuteShortCircuitQuery(StmtExecutor executor,
+ PreparedStatementContext preparedStmtCtx) throws Exception {
+ // Multi-get reads current bindings without mutating the cached IN
predicate.
+ executor.executeAndSendResult(false, false,
+ preparedStmtCtx.shortCircuitQueryContext.get().analzyedQuery,
+ executor.getContext().getMysqlChannel(), null, null);
+ }
+
+ @Override
+ public void setTimeout(long timeoutMs) {
+ this.timeoutMs = timeoutMs;
+ }
+
+ @Override
+ public RowBatch getNext() throws Exception {
+ try {
+ return getNextInternal();
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ invalidateCache();
+ throw e;
+ } finally {
+ cancelInFlightRpcs();
+ }
+ }
+
+ private RowBatch getNextInternal() throws Exception {
+ long deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ Map<Long, TabletTask> tasksByTablet = buildTabletTasks(deadlineNanos);
+ ResultAccumulator accumulator = new ResultAccumulator();
+ List<TabletTask> pending = new ArrayList<>(tasksByTablet.values());
+ while (!pending.isEmpty()) {
+ checkCancelledOrTimedOut(deadlineNanos);
+ pending = executeRound(pending, accumulator, deadlineNanos);
+ }
+ return accumulator.finish();
+ }
+
+ private Map<Long, TabletTask> buildTabletTasks(long deadlineNanos) throws
TException, UserException {
+ List<Column> keyColumns =
context.scanNode.getOlapTable().getBaseSchemaKeyColumns();
+ Map<String, Integer> keyIndexes = new HashMap<>(keyColumns.size());
+ for (int i = 0; i < keyColumns.size(); ++i) {
+ keyIndexes.put(normalizeColumnName(keyColumns.get(i).getName()),
i);
+ }
+
+ List<PlaceholderId> inPlaceholderIds =
statementContext.getPointQueryInPlaceholderIds();
+ SlotReference inSlot =
statementContext.getIdToComparisonSlot().get(inPlaceholderIds.get(0));
+ int inKeyIndex = keyIndexes.get(normalizeColumnName(
+ inSlot.getOriginalColumn().get().getName()));
+ LiteralExpr[] keyValues = new LiteralExpr[keyColumns.size()];
+ for (Map.Entry<PlaceholderId, SlotReference> entry
+ : statementContext.getIdToComparisonSlot().entrySet()) {
+ SlotReference slot = entry.getValue();
+ int keyIndex = keyIndexes.get(normalizeColumnName(
+ slot.getOriginalColumn().get().getName()));
+ // Eligibility checking guarantees exactly one placeholder for
each equality key.
+ if (keyIndex != inKeyIndex) {
+ keyValues[keyIndex] = ((Literal)
statementContext.getIdToPlaceholderRealExpr()
+ .get(entry.getKey())).toLegacyLiteral();
+ }
+ }
+
+ TSerializer serializer = new TSerializer();
+ ByteString[] serializedKeyValues = new ByteString[keyColumns.size()];
+ for (int i = 0; i < keyValues.length; ++i) {
+ if (i == inKeyIndex) {
+ continue;
+ }
+ if (keyValues[i] instanceof NullLiteral) {
+ return Collections.emptyMap();
+ }
+ keyValues[i] = normalizeKeyLiteral(keyColumns.get(i),
keyValues[i]);
+ serializedKeyValues[i] = serializeKeyLiteral(keyValues[i],
serializer);
+ }
+
+ Set<InternalService.KeyTuple> seenTuples = new HashSet<>();
+ Map<Long, TabletTask> tasksByTablet = new LinkedHashMap<>();
+ for (PlaceholderId placeholderId : inPlaceholderIds) {
+ checkCancelledOrTimedOut(deadlineNanos);
+ LiteralExpr inValue = ((Literal)
statementContext.getIdToPlaceholderRealExpr()
+ .get(placeholderId)).toLegacyLiteral();
+ if (inValue instanceof NullLiteral) {
+ continue;
+ }
+ keyValues[inKeyIndex] =
normalizeKeyLiteral(keyColumns.get(inKeyIndex), inValue);
+ serializedKeyValues[inKeyIndex] =
serializeKeyLiteral(keyValues[inKeyIndex], serializer);
+ InternalService.KeyTuple.Builder tupleBuilder =
InternalService.KeyTuple.newBuilder();
+ for (ByteString serializedValue : serializedKeyValues) {
+ tupleBuilder.addKeyColumnLiterals(serializedValue);
+ }
+ InternalService.KeyTuple keyTuple = tupleBuilder.build();
+ if (!seenTuples.add(keyTuple)) {
+ continue;
+ }
+ PointQueryRoute route =
context.scanNode.routePointQueryKeyTuple(Arrays.asList(keyValues));
+ if (route == null) {
+ continue;
+ }
+ TabletTask task = tasksByTablet.get(route.getTabletId());
+ if (task == null) {
+ List<Backend> candidates = selectPointQueryBackends(route);
+ if (candidates.isEmpty()) {
+ throw new UserException("Tablet " + route.getTabletId()
+ + " has no available backend for multi-key point
query");
+ }
+ task = new TabletTask(route.getTabletId(), candidates);
+ tasksByTablet.put(route.getTabletId(), task);
+ }
+ task.keyTuples.add(keyTuple);
+ }
+ return tasksByTablet;
+ }
+
+ private static LiteralExpr normalizeKeyLiteral(Column column, LiteralExpr
literalExpr)
+ throws TException {
+ Type columnType = column.getType();
+ if (columnType.equals(literalExpr.getType())
+ || columnType.matchesType(literalExpr.getType())) {
+ return literalExpr;
+ }
+ try {
+ return
LiteralExprUtils.createLiteral(literalExpr.getStringValue(), columnType);
+ } catch (org.apache.doris.common.AnalysisException e) {
+ throw new TException("Failed to re-type literal for key column "
+ + column.getName() + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static ByteString serializeKeyLiteral(LiteralExpr literalExpr,
TSerializer serializer)
+ throws TException {
+ TExpr thriftExpr = ExprToThriftVisitor.treeToThrift(literalExpr);
+ Preconditions.checkState(thriftExpr.getNodesSize() == 1,
+ "Expected a single TExprNode for point-query key literal, got
%s",
+ thriftExpr.getNodesSize());
+ TExprNode exprNode = thriftExpr.getNodes().get(0);
+ return ByteString.copyFrom(serializer.serialize(exprNode));
+ }
+
+ private static String normalizeColumnName(String columnName) {
+ return columnName.toLowerCase(Locale.ROOT);
+ }
+
+ private static List<Backend> selectPointQueryBackends(PointQueryRoute
route) {
+ List<Backend> candidates = new
ArrayList<>(route.getCandidateBackendIds().size());
+ for (Long backendId : route.getCandidateBackendIds()) {
+ Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
+ if (SimpleScheduler.isAvailable(backend)) {
+ candidates.add(backend);
+ }
+ }
+ if (!route.isBackendOrderBySelection()) {
+ Collections.shuffle(candidates);
+ }
+ return candidates;
+ }
+
+ private InternalService.PTabletKeyLookupRequest buildLookupRequest(
+ TabletTask task, boolean includeQueryContext) {
+ InternalService.PTabletKeyLookupRequest.Builder builder
Review Comment:
**[P1] Preserve one FE-visible version for every tablet task.** Normal scan
ranges carry the partition visible version, but this request drops it; the BE
then uses `_version == -1`, reads every installed rowset, and probes at
`INT32_MAX`. A publish can therefore become visible on only some tasks (or
before FE exposes it), producing a fractured statement result. Please retain
the routed partition version and cap each lookup to it, or disable multi-tablet
short circuit until that snapshot can be enforced.
##########
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++);
+ if (!failedBackends.contains(backend.getId()) &&
SimpleScheduler.isAvailable(backend)) {
+ return backend;
+ }
+ }
+ return null;
+ }
+ }
+
+ private final class ResultAccumulator {
+ private final List<ByteBuffer> rows = new ArrayList<>();
+ private final TDeserializer deserializer;
+ private long resultBytes;
+
+ private ResultAccumulator() throws TException {
+ deserializer = new TDeserializer(new
TCustomProtocolFactory(maxMessageSize));
+ }
+
+ private void add(InternalService.PTabletKeyLookupResponse response)
throws TException {
+ if (response.hasEmptyBatch() && response.getEmptyBatch()) {
+ return;
+ }
+ if (!response.hasRowBatch() || response.getRowBatch().isEmpty()) {
+ throw new TException("No row batch or empty batch found in
point-query response");
+ }
+
+ TResultBatch batch = new TResultBatch();
+ try {
+ deserializer.deserialize(batch,
response.getRowBatch().toByteArray());
+ } catch (TException e) {
+ if (ResultReceiver.isMessageSizeExceeded(e)) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ throw e;
+ }
+ for (ByteBuffer row : batch.getRows()) {
+ resultBytes += row.remaining();
Review Comment:
**[P2] Keep the receiver limit per response.**
`max_msg_size_of_result_receiver` is a per-message deserialization limit, and
each tablet response is already checked under it. Summing every response's row
bytes here applies the same limit to the whole multi-tablet result, so several
individually legal responses can fail only because their aggregate crosses the
threshold. Please remove this query-wide sum or use a separately named and
documented total-result bound, and cover multiple responses whose aggregate
exceeds the per-message limit.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1545,8 +1545,13 @@ public void executeAndSendResult(boolean isOutfileQuery,
boolean isSendFields,
// ExecuteCommand publishes this same context after a
successful first prepared execution.
statementContext.setShortCircuitQueryContext(shortCircuitQueryContext);
}
- coordBase = new PointQueryExecutor(shortCircuitQueryContext,
-
context.getSessionVariable().getMaxMsgSizeOfResultReceiver());
+ if (statementContext.isMultiKeyPointQuery()) {
+ coordBase = new
PointQueryMultiExecutor(shortCircuitQueryContext, statementContext,
Review Comment:
**[P2] Publish the multi executor to the cancellation path.** This instance
only lives in local `coordBase`, while `StmtExecutor.cancel()` reaches the
delegate, a `CancelableCommand`, or the `coord` field. Consequently client
disconnect, `KILL QUERY`, and the connection timeout checker cannot call this
executor's `cancel()`, leaving all fan-out RPCs alive until the point-query
deadline. Please register and clear a race-safe cancellation target around this
execution.
##########
be/src/service/point_query_executor.cpp:
##########
@@ -627,7 +627,7 @@ Status PointQueryExecutor::_lookup_row_data() {
}
}
// filter rows by delete sign
- if (_row_hits > 0 && _reusable->delete_sign_idx() != -1) {
+ if ((_row_hits > 0 || _result_block->rows() > 0) &&
_reusable->delete_sign_idx() != -1) {
Review Comment:
**[P1] Do not clear cached partial-row-store hits with no delete-sign
data.** When row cache is enabled for a table with partial `row_store_columns`,
a hit deserializes the stored user columns but skips missing-column completion,
leaving the delete-sign vector empty and `_row_hits == 0`. This new `rows() >
0` path then gets `filtered == total == 0` and clears every warmed live row;
two identical multi-gets return rows and then empty. Please populate the
missing sign for cached rows or bypass partial-row caching, and add a
cache-warm partial-row-store test.
##########
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++);
+ if (!failedBackends.contains(backend.getId()) &&
SimpleScheduler.isAvailable(backend)) {
+ return backend;
+ }
+ }
+ return null;
+ }
+ }
+
+ private final class ResultAccumulator {
+ private final List<ByteBuffer> rows = new ArrayList<>();
+ private final TDeserializer deserializer;
+ private long resultBytes;
+
+ private ResultAccumulator() throws TException {
+ deserializer = new TDeserializer(new
TCustomProtocolFactory(maxMessageSize));
+ }
+
+ private void add(InternalService.PTabletKeyLookupResponse response)
throws TException {
+ if (response.hasEmptyBatch() && response.getEmptyBatch()) {
+ return;
+ }
+ if (!response.hasRowBatch() || response.getRowBatch().isEmpty()) {
+ throw new TException("No row batch or empty batch found in
point-query response");
+ }
+
+ TResultBatch batch = new TResultBatch();
+ try {
+ deserializer.deserialize(batch,
response.getRowBatch().toByteArray());
+ } catch (TException e) {
+ if (ResultReceiver.isMessageSizeExceeded(e)) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ throw e;
+ }
+ for (ByteBuffer row : batch.getRows()) {
+ resultBytes += row.remaining();
+ if (resultBytes > maxMessageSize) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ rows.add(row);
+ }
+ }
+
+ private RowBatch finish() {
+ RowBatch rowBatch = new RowBatch();
+ if (rows.isEmpty()) {
+ return rowBatch;
+ }
+ TResultBatch resultBatch = new TResultBatch();
+ resultBatch.setRows(rows);
+ resultBatch.setIsCompressed(false);
+ resultBatch.setPacketSeq(0);
+ rowBatch.setBatch(resultBatch);
+ return rowBatch;
+ }
+ }
+
+ public PointQueryMultiExecutor(ShortCircuitQueryContext context,
+ StatementContext statementContext, int maxMessageSize) {
+ super(context, maxMessageSize);
+ this.context = context;
+ this.statementContext = statementContext;
+ this.maxMessageSize = maxMessageSize;
+ }
+
+ public static void directExecuteShortCircuitQuery(StmtExecutor executor,
+ PreparedStatementContext preparedStmtCtx) throws Exception {
+ // Multi-get reads current bindings without mutating the cached IN
predicate.
+ executor.executeAndSendResult(false, false,
+ preparedStmtCtx.shortCircuitQueryContext.get().analzyedQuery,
+ executor.getContext().getMysqlChannel(), null, null);
+ }
+
+ @Override
+ public void setTimeout(long timeoutMs) {
+ this.timeoutMs = timeoutMs;
+ }
+
+ @Override
+ public RowBatch getNext() throws Exception {
+ try {
+ return getNextInternal();
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ invalidateCache();
+ throw e;
+ } finally {
+ cancelInFlightRpcs();
+ }
+ }
+
+ private RowBatch getNextInternal() throws Exception {
+ long deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ Map<Long, TabletTask> tasksByTablet = buildTabletTasks(deadlineNanos);
+ ResultAccumulator accumulator = new ResultAccumulator();
+ List<TabletTask> pending = new ArrayList<>(tasksByTablet.values());
+ while (!pending.isEmpty()) {
+ checkCancelledOrTimedOut(deadlineNanos);
+ pending = executeRound(pending, accumulator, deadlineNanos);
+ }
+ return accumulator.finish();
+ }
+
+ private Map<Long, TabletTask> buildTabletTasks(long deadlineNanos) throws
TException, UserException {
+ List<Column> keyColumns =
context.scanNode.getOlapTable().getBaseSchemaKeyColumns();
+ Map<String, Integer> keyIndexes = new HashMap<>(keyColumns.size());
+ for (int i = 0; i < keyColumns.size(); ++i) {
+ keyIndexes.put(normalizeColumnName(keyColumns.get(i).getName()),
i);
+ }
+
+ List<PlaceholderId> inPlaceholderIds =
statementContext.getPointQueryInPlaceholderIds();
+ SlotReference inSlot =
statementContext.getIdToComparisonSlot().get(inPlaceholderIds.get(0));
+ int inKeyIndex = keyIndexes.get(normalizeColumnName(
+ inSlot.getOriginalColumn().get().getName()));
+ LiteralExpr[] keyValues = new LiteralExpr[keyColumns.size()];
+ for (Map.Entry<PlaceholderId, SlotReference> entry
+ : statementContext.getIdToComparisonSlot().entrySet()) {
+ SlotReference slot = entry.getValue();
+ int keyIndex = keyIndexes.get(normalizeColumnName(
+ slot.getOriginalColumn().get().getName()));
+ // Eligibility checking guarantees exactly one placeholder for
each equality key.
+ if (keyIndex != inKeyIndex) {
+ keyValues[keyIndex] = ((Literal)
statementContext.getIdToPlaceholderRealExpr()
+ .get(entry.getKey())).toLegacyLiteral();
+ }
+ }
+
+ TSerializer serializer = new TSerializer();
+ ByteString[] serializedKeyValues = new ByteString[keyColumns.size()];
+ for (int i = 0; i < keyValues.length; ++i) {
+ if (i == inKeyIndex) {
+ continue;
+ }
+ if (keyValues[i] instanceof NullLiteral) {
+ return Collections.emptyMap();
+ }
+ keyValues[i] = normalizeKeyLiteral(keyColumns.get(i),
keyValues[i]);
+ serializedKeyValues[i] = serializeKeyLiteral(keyValues[i],
serializer);
+ }
+
+ Set<InternalService.KeyTuple> seenTuples = new HashSet<>();
+ Map<Long, TabletTask> tasksByTablet = new LinkedHashMap<>();
+ for (PlaceholderId placeholderId : inPlaceholderIds) {
+ checkCancelledOrTimedOut(deadlineNanos);
+ LiteralExpr inValue = ((Literal)
statementContext.getIdToPlaceholderRealExpr()
+ .get(placeholderId)).toLegacyLiteral();
+ if (inValue instanceof NullLiteral) {
+ continue;
+ }
+ keyValues[inKeyIndex] =
normalizeKeyLiteral(keyColumns.get(inKeyIndex), inValue);
+ serializedKeyValues[inKeyIndex] =
serializeKeyLiteral(keyValues[inKeyIndex], serializer);
+ InternalService.KeyTuple.Builder tupleBuilder =
InternalService.KeyTuple.newBuilder();
+ for (ByteString serializedValue : serializedKeyValues) {
+ tupleBuilder.addKeyColumnLiterals(serializedValue);
+ }
+ InternalService.KeyTuple keyTuple = tupleBuilder.build();
+ if (!seenTuples.add(keyTuple)) {
+ continue;
+ }
+ PointQueryRoute route =
context.scanNode.routePointQueryKeyTuple(Arrays.asList(keyValues));
+ if (route == null) {
+ continue;
+ }
+ TabletTask task = tasksByTablet.get(route.getTabletId());
+ if (task == null) {
+ List<Backend> candidates = selectPointQueryBackends(route);
+ if (candidates.isEmpty()) {
+ throw new UserException("Tablet " + route.getTabletId()
+ + " has no available backend for multi-key point
query");
+ }
+ task = new TabletTask(route.getTabletId(), candidates);
+ tasksByTablet.put(route.getTabletId(), task);
+ }
+ task.keyTuples.add(keyTuple);
+ }
+ return tasksByTablet;
+ }
+
+ private static LiteralExpr normalizeKeyLiteral(Column column, LiteralExpr
literalExpr)
+ throws TException {
+ Type columnType = column.getType();
+ if (columnType.equals(literalExpr.getType())
+ || columnType.matchesType(literalExpr.getType())) {
+ return literalExpr;
+ }
+ try {
+ return
LiteralExprUtils.createLiteral(literalExpr.getStringValue(), columnType);
Review Comment:
**[P1] Replan when a cached binding needs different comparison coercion.**
COM_STMT_EXECUTE can supply new parameter types on a later call, but cache
reuse bypasses analysis and this code force-parses the value as the key type.
For an INT key, a cached statement rebound with DOUBLE `1.5` is a valid wider
comparison that should return no row; here it fails strict integer parsing
instead. Please retain the cache's binding/coercion contract and fall back to
normal analysis when it changes, with a second-execution type-change test.
##########
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++);
+ if (!failedBackends.contains(backend.getId()) &&
SimpleScheduler.isAvailable(backend)) {
+ return backend;
+ }
+ }
+ return null;
+ }
+ }
+
+ private final class ResultAccumulator {
+ private final List<ByteBuffer> rows = new ArrayList<>();
+ private final TDeserializer deserializer;
+ private long resultBytes;
+
+ private ResultAccumulator() throws TException {
+ deserializer = new TDeserializer(new
TCustomProtocolFactory(maxMessageSize));
+ }
+
+ private void add(InternalService.PTabletKeyLookupResponse response)
throws TException {
+ if (response.hasEmptyBatch() && response.getEmptyBatch()) {
+ return;
+ }
+ if (!response.hasRowBatch() || response.getRowBatch().isEmpty()) {
+ throw new TException("No row batch or empty batch found in
point-query response");
+ }
+
+ TResultBatch batch = new TResultBatch();
+ try {
+ deserializer.deserialize(batch,
response.getRowBatch().toByteArray());
+ } catch (TException e) {
+ if (ResultReceiver.isMessageSizeExceeded(e)) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ throw e;
+ }
+ for (ByteBuffer row : batch.getRows()) {
+ resultBytes += row.remaining();
+ if (resultBytes > maxMessageSize) {
+ throw new TException(
+ "MaxMessageSize reached, try increase
max_msg_size_of_result_receiver");
+ }
+ rows.add(row);
+ }
+ }
+
+ private RowBatch finish() {
+ RowBatch rowBatch = new RowBatch();
+ if (rows.isEmpty()) {
+ return rowBatch;
+ }
+ TResultBatch resultBatch = new TResultBatch();
+ resultBatch.setRows(rows);
+ resultBatch.setIsCompressed(false);
+ resultBatch.setPacketSeq(0);
+ rowBatch.setBatch(resultBatch);
+ return rowBatch;
+ }
+ }
+
+ public PointQueryMultiExecutor(ShortCircuitQueryContext context,
+ StatementContext statementContext, int maxMessageSize) {
+ super(context, maxMessageSize);
+ this.context = context;
+ this.statementContext = statementContext;
+ this.maxMessageSize = maxMessageSize;
+ }
+
+ public static void directExecuteShortCircuitQuery(StmtExecutor executor,
+ PreparedStatementContext preparedStmtCtx) throws Exception {
+ // Multi-get reads current bindings without mutating the cached IN
predicate.
+ executor.executeAndSendResult(false, false,
+ preparedStmtCtx.shortCircuitQueryContext.get().analzyedQuery,
+ executor.getContext().getMysqlChannel(), null, null);
+ }
+
+ @Override
+ public void setTimeout(long timeoutMs) {
+ this.timeoutMs = timeoutMs;
+ }
+
+ @Override
+ public RowBatch getNext() throws Exception {
+ try {
+ return getNextInternal();
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ invalidateCache();
+ throw e;
+ } finally {
+ cancelInFlightRpcs();
+ }
+ }
+
+ private RowBatch getNextInternal() throws Exception {
+ long deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ Map<Long, TabletTask> tasksByTablet = buildTabletTasks(deadlineNanos);
+ ResultAccumulator accumulator = new ResultAccumulator();
+ List<TabletTask> pending = new ArrayList<>(tasksByTablet.values());
+ while (!pending.isEmpty()) {
+ checkCancelledOrTimedOut(deadlineNanos);
+ pending = executeRound(pending, accumulator, deadlineNanos);
+ }
+ return accumulator.finish();
+ }
+
+ private Map<Long, TabletTask> buildTabletTasks(long deadlineNanos) throws
TException, UserException {
+ List<Column> keyColumns =
context.scanNode.getOlapTable().getBaseSchemaKeyColumns();
+ Map<String, Integer> keyIndexes = new HashMap<>(keyColumns.size());
+ for (int i = 0; i < keyColumns.size(); ++i) {
+ keyIndexes.put(normalizeColumnName(keyColumns.get(i).getName()),
i);
+ }
+
+ List<PlaceholderId> inPlaceholderIds =
statementContext.getPointQueryInPlaceholderIds();
+ SlotReference inSlot =
statementContext.getIdToComparisonSlot().get(inPlaceholderIds.get(0));
+ int inKeyIndex = keyIndexes.get(normalizeColumnName(
+ inSlot.getOriginalColumn().get().getName()));
+ LiteralExpr[] keyValues = new LiteralExpr[keyColumns.size()];
+ for (Map.Entry<PlaceholderId, SlotReference> entry
+ : statementContext.getIdToComparisonSlot().entrySet()) {
+ SlotReference slot = entry.getValue();
+ int keyIndex = keyIndexes.get(normalizeColumnName(
+ slot.getOriginalColumn().get().getName()));
+ // Eligibility checking guarantees exactly one placeholder for
each equality key.
+ if (keyIndex != inKeyIndex) {
+ keyValues[keyIndex] = ((Literal)
statementContext.getIdToPlaceholderRealExpr()
+ .get(entry.getKey())).toLegacyLiteral();
+ }
+ }
+
+ TSerializer serializer = new TSerializer();
+ ByteString[] serializedKeyValues = new ByteString[keyColumns.size()];
+ for (int i = 0; i < keyValues.length; ++i) {
+ if (i == inKeyIndex) {
+ continue;
+ }
+ if (keyValues[i] instanceof NullLiteral) {
+ return Collections.emptyMap();
+ }
+ keyValues[i] = normalizeKeyLiteral(keyColumns.get(i),
keyValues[i]);
+ serializedKeyValues[i] = serializeKeyLiteral(keyValues[i],
serializer);
+ }
+
+ Set<InternalService.KeyTuple> seenTuples = new HashSet<>();
+ Map<Long, TabletTask> tasksByTablet = new LinkedHashMap<>();
+ for (PlaceholderId placeholderId : inPlaceholderIds) {
+ checkCancelledOrTimedOut(deadlineNanos);
+ LiteralExpr inValue = ((Literal)
statementContext.getIdToPlaceholderRealExpr()
+ .get(placeholderId)).toLegacyLiteral();
+ if (inValue instanceof NullLiteral) {
+ continue;
+ }
+ keyValues[inKeyIndex] =
normalizeKeyLiteral(keyColumns.get(inKeyIndex), inValue);
+ serializedKeyValues[inKeyIndex] =
serializeKeyLiteral(keyValues[inKeyIndex], serializer);
+ InternalService.KeyTuple.Builder tupleBuilder =
InternalService.KeyTuple.newBuilder();
+ for (ByteString serializedValue : serializedKeyValues) {
+ tupleBuilder.addKeyColumnLiterals(serializedValue);
+ }
+ InternalService.KeyTuple keyTuple = tupleBuilder.build();
+ if (!seenTuples.add(keyTuple)) {
Review Comment:
**[P1] Deduplicate canonical storage keys, not serialized literals.** Equal
DECIMAL bindings such as `1.0` and `1.00` can serialize to different
`KeyTuple`s, so both pass this set; BE then parses both at the column scale,
encodes the same primary key twice, and emits the unique-key row twice. Please
canonicalize to the exact key-column representation before hashing (or
deduplicate encoded keys on BE) and cover equal BigDecimal values with
different scales.
--
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]
