This is an automated email from the ASF dual-hosted git repository.
englefly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c8eeb86e6a4 [fix](scan) Use query-available check for scan replica
selection (#66964)
c8eeb86e6a4 is described below
commit c8eeb86e6a47ca6ee36c4a1cb21b44b9ec22b4c5
Author: minghong <[email protected]>
AuthorDate: Fri Aug 21 14:32:40 2026 +0800
[fix](scan) Use query-available check for scan replica selection (#66964)
### What problem does this PR solve?
Problem Summary:
A backend that is alive but not query available (query-disabled or
shutting down) can still enter the scan candidate set built by
OlapScanNode.addScanRangeLocations(), which only requires isAlive() and
isMixNode(). When a tablet ends up with a single such candidate (single
replica table, or fixed/cooldown replica narrowing), the
LoadBalanceScanWorkerSelector selects it as the actual scan worker
without any availability check (the replicaLocations.size()==1 branch).
Meanwhile the MaterializationNode snapshots the query-available backend
set when the physical plan is translated
(NereidsPlanner.splitFragments), i.e. earlier and with a stricter
filter. The assigned scan worker then writes its own backend_id into
every rowid; when the materialization operator parses the rowid and
looks the backend_id up in its rpc_struct_map built from the earlier
snapshot, the id is missing and the query fails with
"MaterializationSinkOperatorX failed to find rpc_struct".
This change makes both places require the same predicate
Backend.isQueryAvailable() when selecting a scan replica host:
- OlapScanNode.addScanRangeLocations() requires isQueryAvailable() when
building the scan candidate set (fixed replica fallback, cooldown
replica narrowing and the replica loop), so a query-disabled but alive
backend no longer enters the candidates; a tablet with no
query-available replica now fails immediately with "has no queryable
replicas".
- LoadBalanceScanWorkerSelector.selectScanReplicaAndMinWorkloadWorker()
checks isQueryAvailable() in the single-replica branch and throws "No
available workers" if the only replica is not query available; the
multi-replica branch uses the same predicate.
---
.../worker/LoadBalanceScanWorkerSelector.java | 5 ++-
.../org/apache/doris/planner/OlapScanNode.java | 12 +++---
...anceScanWorkerSelectorBackendSelectionTest.java | 26 +++++++++++-
.../OlapScanNodeBackendSelectionConfigTest.java | 48 ++++++++++++++++++++++
4 files changed, 83 insertions(+), 8 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelector.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelector.java
index 2502879011f..f2d53a94375 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelector.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelector.java
@@ -263,6 +263,9 @@ public class LoadBalanceScanWorkerSelector implements
ScanWorkerSelector {
if (replicaLocations.size() == 1) {
TScanRangeLocation replicaLocation = replicaLocations.get(0);
DistributedPlanWorker worker = workerManager.getWorker(catalogId,
replicaLocation.getBackendId());
+ if (!((BackendWorker) worker).getBackend().isQueryAvailable()) {
+ throw new AnalysisException("No available workers");
+ }
ScanRanges scanRanges = new ScanRanges();
TScanRangeParams scanReplicaParams =
ScanWorkerSelector.buildScanReplicaParams(tabletLocation,
replicaLocation);
@@ -302,7 +305,7 @@ public class LoadBalanceScanWorkerSelector implements
ScanWorkerSelector {
for (TScanRangeLocation replicaLocation : replicaLocations) {
DistributedPlanWorker worker = workerManager.getWorker(catalogId,
replicaLocation.getBackendId());
- if (!worker.available()) {
+ if (!((BackendWorker) worker).getBackend().isQueryAvailable()) {
continue;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index ee7331cc3df..b7fb3388e87 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -654,9 +654,9 @@ public class OlapScanNode extends ScanNode {
}
Backend backend = allBackends.get(beId);
// If the fixed replica is bad, then not clear the
replicas using random replica
- if (backend == null || !backend.isAlive()) {
+ if (backend == null || !backend.isQueryAvailable()) {
if (LOG.isDebugEnabled()) {
- LOG.debug("backend {} not exists or is not alive
for replica {}", beId,
+ LOG.debug("backend {} not exists or is not query
available for replica {}", beId,
replica.getId());
}
Collections.shuffle(replicas);
@@ -685,7 +685,7 @@ public class OlapScanNode extends ScanNode {
if (replicaOptional.isPresent()) {
Replica replica = replicaOptional.get();
Backend backend =
allBackends.get(replica.getBackendIdWithoutException());
- if (backend != null && backend.isAlive()) {
+ if (backend != null && backend.isQueryAvailable()) {
replicas.clear();
replicas.add(replica);
}
@@ -720,14 +720,14 @@ public class OlapScanNode extends ScanNode {
clusterException = true;
continue;
}
- if (backend == null || !backend.isAlive()) {
+ if (backend == null || !backend.isQueryAvailable()) {
if (LOG.isDebugEnabled()) {
- LOG.debug("backend {} not exists or is not alive for
replica {}", backendId,
+ LOG.debug("backend {} not exists or is not query
available for replica {}", backendId,
replica.getId());
}
String err = "replica " + replica.getId() + "'s backend "
+ backendId
+ (backend != null ? " with tag " +
backend.getLocationTag() : "")
- + " does not exist or not alive";
+ + " does not exist or is not query available";
errs.add(err);
continue;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelectorBackendSelectionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelectorBackendSelectionTest.java
index 68c99e8ccfd..5629b9da18e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelectorBackendSelectionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelectorBackendSelectionTest.java
@@ -22,6 +22,7 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.NereidsException;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.resource.BackendSelection;
import org.apache.doris.resource.BackendSelectionManager;
@@ -178,13 +179,36 @@ class LoadBalanceScanWorkerSelectorBackendSelectionTest {
}
@Test
- void testSingleReplicaSelectionPreservesUnavailableWorkerBehavior() throws
Exception {
+ void testSingleReplicaSelectionRejectsUnavailableWorker() {
Backend unavailable = backend(1L, "preferred");
unavailable.setAlive(false);
DistributedPlanWorkerManager workerManager =
workerManager(unavailable);
LoadBalanceScanWorkerSelector selector = new
LoadBalanceScanWorkerSelector(
workerManager, new ConnectContext(), false);
+ Assertions.assertThrows(AnalysisException.class,
+ () -> select(selector, locations(location(1L)), 100L));
+ }
+
+ @Test
+ void testSingleReplicaSelectionRejectsQueryDisabledWorker() {
+ Backend queryDisabled = backend(1L, "preferred");
+ queryDisabled.setQueryDisabled(true);
+ DistributedPlanWorkerManager workerManager =
workerManager(queryDisabled);
+ LoadBalanceScanWorkerSelector selector = new
LoadBalanceScanWorkerSelector(
+ workerManager, new ConnectContext(), false);
+
+ Assertions.assertThrows(AnalysisException.class,
+ () -> select(selector, locations(location(1L)), 100L));
+ }
+
+ @Test
+ void testSingleReplicaSelectionKeepsAvailableWorker() throws Exception {
+ Backend available = backend(1L, "preferred");
+ DistributedPlanWorkerManager workerManager = workerManager(available);
+ LoadBalanceScanWorkerSelector selector = new
LoadBalanceScanWorkerSelector(
+ workerManager, new ConnectContext(), false);
+
WorkerScanRanges selected = select(selector, locations(location(1L)),
100L);
Assertions.assertEquals(1L, selected.worker.id());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeBackendSelectionConfigTest.java
b/fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeBackendSelectionConfigTest.java
index 62c8996e0ce..df0d2a0de2c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeBackendSelectionConfigTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeBackendSelectionConfigTest.java
@@ -288,6 +288,54 @@ class OlapScanNodeBackendSelectionConfigTest {
}
}
+ @Test
+ void testQueryDisabledBackendExcludedFromScanCandidates() throws Exception
{
+ Replica replica = replica(1, 10);
+ Backend disabledBackend = backend(1, "group_a");
+ disabledBackend.setQueryDisabled(true);
+ Backend otherBackend = backend(2, "group_b");
+
+ Tablet tablet = Mockito.mock(Tablet.class);
+ Mockito.when(tablet.getId()).thenReturn(20L);
+ Mockito.when(tablet.getQueryableReplicas(
+ Mockito.eq(10L), Mockito.anyMap(), Mockito.eq(false)))
+ .thenReturn(new ArrayList<>(ImmutableList.of(replica)));
+ Mockito.when(tablet.getCooldownReplicaId()).thenReturn(-1L);
+
+ UserException ex = Assertions.assertThrows(UserException.class,
+ () -> createScanRanges(tablet, disabledBackend, otherBackend));
+ Assertions.assertTrue(ex.getMessage().contains("has no queryable
replicas"));
+ }
+
+ @Test
+ void testCooldownReplicaOnQueryDisabledBackendIsNotNarrowed() throws
Exception {
+ Replica preferredReplica = replica(1, 10);
+ Replica cooldownReplica = replica(2, 10);
+ Backend preferredBackend = backend(1, "group_a");
+ Backend cooldownBackend = backend(2, "group_b");
+ cooldownBackend.setQueryDisabled(true);
+
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableCooldownReplicaAffinity = true;
+ ConnectContext context = Mockito.mock(ConnectContext.class);
+ QueryState queryState = Mockito.mock(QueryState.class);
+ Mockito.when(context.getSessionVariable()).thenReturn(sessionVariable);
+ Mockito.when(context.getState()).thenReturn(queryState);
+ Mockito.when(context.getStatementContext()).thenReturn(new
StatementContext());
+ Mockito.when(context.getComputeGroupSafely()).thenReturn(null);
+
+ try (MockedStatic<ConnectContext> mockedContext =
Mockito.mockStatic(ConnectContext.class)) {
+ mockedContext.when(ConnectContext::get).thenReturn(context);
+
+ List<TScanRangeLocations> scanRanges = createScanRanges(
+ preferredReplica, cooldownReplica, preferredBackend,
cooldownBackend);
+
+ Assertions.assertEquals(ImmutableList.of(1L),
+ scanRanges.get(0).getLocations().stream()
+ .map(location ->
location.getBackendId()).collect(Collectors.toList()));
+ }
+ }
+
private List<TScanRangeLocations> createScanRanges(Replica
preferredReplica, Replica cooldownReplica,
Backend preferredBackend, Backend cooldownBackend) throws
Exception {
Tablet tablet = Mockito.mock(Tablet.class);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
