This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new c2aaed99db8 branch-4.2: [fix](scan) Use query-available check for scan
replica selection #66964 (#70025)
c2aaed99db8 is described below
commit c2aaed99db8355529f2924eda68e3598fb46e7b3
Author: morrySnow <[email protected]>
AuthorDate: Tue Nov 3 17:46:53 2026 +0800
branch-4.2: [fix](scan) Use query-available check for scan replica
selection #66964 (#70025)
picked from master #66964
Co-authored-by: minghong <[email protected]>
---
.../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 dca544ae857..39c47088588 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 b371c6db270..36b9601fbad 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
@@ -635,9 +635,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);
@@ -666,7 +666,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);
}
@@ -701,14 +701,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]
