Skip to content

Commit 4a9956e

Browse files
zyp-Vzhangyipeng.0818
zyp-V
and
zhangyipeng.0818
authored
[opt](point query) Optimize backend path collection for point queries (#67151)
### What problem does this PR solve? Problem Summary: Point-query planning built alive disk path sets for every backend even though tablet pruning leaves only a small set of replica backends. This made planning work grow with cluster size. Build the full map for regular scans and collect paths only for the selected tablet replica backends for point queries. Use the current backend snapshot API so the backport remains compatible with cloud and remote OLAP tables. ### Release note Reduce point-query planning overhead in clusters with many backends. Co-authored-by: zhangyipeng.0818 <zhangyipeng.0818@bytedance.com>
1 parent fea01e6 commit 4a9956e

2 files changed

Lines changed: 90 additions & 13 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java‎

Lines changed: 43 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1007,22 +1007,14 @@ private void computeTabletInfo() throws UserException {
10071007
*/
10081008
Preconditions.checkState(scanBackendIds.isEmpty());
10091009
Preconditions.checkState(scanTabletIds.isEmpty());
1010-
Map<Long, Set<Long>> backendAlivePathHashs = Maps.newHashMap();
1011-
for (Backend backend : olapTable.getAllBackendsByAllCluster().values()) {
1012-
Set<Long> hashSet = Sets.newLinkedHashSet();
1013-
for (DiskInfo diskInfo : backend.getDisks().values()) {
1014-
if (diskInfo.isAlive()) {
1015-
hashSet.add(diskInfo.getPathHash());
1016-
}
1017-
}
1018-
backendAlivePathHashs.put(backend.getId(), hashSet);
1019-
}
1020-
10211010
ConnectContext connectContext = ConnectContext.get();
10221011
boolean isNereids = connectContext != null && connectContext.getState().isNereids();
10231012
boolean isPointQuery = connectContext != null
10241013
&& connectContext.getStatementContext() != null
10251014
&& connectContext.getStatementContext().isShortCircuitQuery();
1015+
ImmutableMap<Long, Backend> allBackends = olapTable.getAllBackendsByAllCluster();
1016+
Map<Long, Set<Long>> backendAlivePathHashes = isPointQuery
1017+
? null : getBackendAlivePathHashes(allBackends.values());
10261018
for (Long partitionId : selectedPartitionIds) {
10271019
final Partition partition = olapTable.getPartition(partitionId);
10281020
final MaterializedIndex selectedTable = olapTable.getPartitionIndex(partition, selectedIndexId);
@@ -1070,7 +1062,7 @@ private void computeTabletInfo() throws UserException {
10701062
scanTabletIds.addAll(allTabletIds);
10711063
}
10721064

1073-
if (!isPointQuery()) {
1065+
if (!isPointQuery) {
10741066
int bucketNum = partition.getDistributionInfo().getBucketNum();
10751067
for (int i = 0; i < allTabletIds.size(); i++) {
10761068
tabletId2BucketInfo.put(allTabletIds.get(i), encodeBucketInfo(i, bucketNum));
@@ -1079,8 +1071,46 @@ private void computeTabletInfo() throws UserException {
10791071

10801072
totalTabletsNum += selectedTable.getTablets().size();
10811073
selectedSplitNum += tablets.size();
1082-
addScanRangeLocations(partition, tablets, backendAlivePathHashs);
1074+
Map<Long, Set<Long>> currentBackendAlivePathHashes = isPointQuery
1075+
? getBackendAlivePathHashes(allBackends, tablets) : backendAlivePathHashes;
1076+
addScanRangeLocations(partition, tablets, currentBackendAlivePathHashes);
1077+
}
1078+
}
1079+
1080+
private static Map<Long, Set<Long>> getBackendAlivePathHashes(Collection<Backend> backends) {
1081+
Map<Long, Set<Long>> backendAlivePathHashes = Maps.newHashMap();
1082+
for (Backend backend : backends) {
1083+
backendAlivePathHashes.put(backend.getId(), getBackendAlivePathHashes(backend));
1084+
}
1085+
return backendAlivePathHashes;
1086+
}
1087+
1088+
@VisibleForTesting
1089+
static Map<Long, Set<Long>> getBackendAlivePathHashes(
1090+
Map<Long, Backend> backends, List<Tablet> tablets) {
1091+
Map<Long, Set<Long>> backendAlivePathHashes = Maps.newHashMap();
1092+
for (Tablet tablet : tablets) {
1093+
for (Replica replica : tablet.getReplicas()) {
1094+
long backendId = replica.getBackendIdWithoutException();
1095+
Backend backend = backends.get(backendId);
1096+
if (backend != null) {
1097+
backendAlivePathHashes.computeIfAbsent(
1098+
backendId, id -> getBackendAlivePathHashes(backend));
1099+
}
1100+
}
1101+
}
1102+
return backendAlivePathHashes;
1103+
}
1104+
1105+
private static Set<Long> getBackendAlivePathHashes(Backend backend) {
1106+
Map<String, DiskInfo> disks = backend.getDisks();
1107+
Set<Long> alivePathHashes = Sets.newHashSetWithExpectedSize(disks.size());
1108+
for (DiskInfo diskInfo : disks.values()) {
1109+
if (diskInfo.isAlive()) {
1110+
alivePathHashes.add(diskInfo.getPathHash());
1111+
}
10831112
}
1113+
return alivePathHashes;
10841114
}
10851115

10861116
/**

‎fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeTest.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,23 +28,30 @@
2828
import org.apache.doris.analysis.TupleDescriptor;
2929
import org.apache.doris.analysis.TupleId;
3030
import org.apache.doris.catalog.Column;
31+
import org.apache.doris.catalog.DiskInfo;
32+
import org.apache.doris.catalog.LocalReplica;
33+
import org.apache.doris.catalog.LocalTablet;
3134
import org.apache.doris.catalog.OlapTable;
3235
import org.apache.doris.catalog.Partition;
3336
import org.apache.doris.catalog.PartitionKey;
3437
import org.apache.doris.catalog.PrimitiveType;
3538
import org.apache.doris.catalog.RangePartitionInfo;
3639
import org.apache.doris.catalog.RangePartitionItem;
40+
import org.apache.doris.catalog.Replica.ReplicaState;
41+
import org.apache.doris.catalog.Tablet;
3742
import org.apache.doris.catalog.info.TableNameInfo;
3843
import org.apache.doris.common.AnalysisException;
3944
import org.apache.doris.common.Config;
4045
import org.apache.doris.common.util.DebugPointUtil;
4146
import org.apache.doris.datasource.InternalCatalog;
47+
import org.apache.doris.system.Backend;
4248
import org.apache.doris.thrift.TOlapScanNode;
4349
import org.apache.doris.thrift.TPaloScanRange;
4450
import org.apache.doris.thrift.TPartitionBoundary;
4551
import org.apache.doris.thrift.TScanRange;
4652
import org.apache.doris.thrift.TScanRangeLocations;
4753

54+
import com.google.common.collect.ImmutableMap;
4855
import com.google.common.collect.Lists;
4956
import com.google.common.collect.Maps;
5057
import com.google.common.collect.Range;
@@ -59,6 +66,7 @@
5966
import java.util.HashMap;
6067
import java.util.List;
6168
import java.util.Map;
69+
import java.util.Set;
6270
import java.util.stream.Collectors;
6371

6472
public class OlapScanNodeTest {
@@ -356,6 +364,45 @@ private Map<Long, Long> getBucketInfo(OlapScanNode scanNode) throws Exception {
356364
return (Map<Long, Long>) bucketInfoField.get(scanNode);
357365
}
358366

367+
@Test
368+
public void testPointQueryBackendAlivePathsOnlyUseSelectedTabletBackends() {
369+
Backend firstBackend = backendWithDisks(1L, 11L, 12L);
370+
Backend secondBackend = backendWithDisks(2L, 21L, 22L);
371+
Backend unrelatedBackend = backendWithDisks(3L, 31L, 32L);
372+
Map<Long, Backend> backends = ImmutableMap.of(
373+
firstBackend.getId(), firstBackend,
374+
secondBackend.getId(), secondBackend,
375+
unrelatedBackend.getId(), unrelatedBackend);
376+
377+
LocalTablet selectedTablet = new LocalTablet(10L);
378+
selectedTablet.addReplica(new LocalReplica(101L, firstBackend.getId(), 0, ReplicaState.NORMAL), true);
379+
selectedTablet.addReplica(new LocalReplica(102L, secondBackend.getId(), 0, ReplicaState.NORMAL), true);
380+
selectedTablet.addReplica(new LocalReplica(103L, 4L, 0, ReplicaState.NORMAL), true);
381+
382+
Map<Long, Set<Long>> alivePathHashes = OlapScanNode.getBackendAlivePathHashes(
383+
backends, Lists.<Tablet>newArrayList(selectedTablet));
384+
385+
Assert.assertEquals(2, alivePathHashes.size());
386+
Assert.assertEquals(Collections.singleton(11L), alivePathHashes.get(firstBackend.getId()));
387+
Assert.assertEquals(Collections.singleton(21L), alivePathHashes.get(secondBackend.getId()));
388+
Assert.assertFalse(alivePathHashes.containsKey(unrelatedBackend.getId()));
389+
Assert.assertFalse(alivePathHashes.containsKey(4L));
390+
}
391+
392+
private Backend backendWithDisks(long backendId, long alivePathHash, long offlinePathHash) {
393+
DiskInfo aliveDisk = new DiskInfo("/alive-" + backendId);
394+
aliveDisk.setPathHash(alivePathHash);
395+
DiskInfo offlineDisk = new DiskInfo("/offline-" + backendId);
396+
offlineDisk.setPathHash(offlinePathHash);
397+
offlineDisk.setState(DiskInfo.DiskState.OFFLINE);
398+
399+
Backend backend = new Backend(backendId, "127.0.0." + backendId, 9050);
400+
backend.setDisks(ImmutableMap.of(
401+
aliveDisk.getRootPath(), aliveDisk,
402+
offlineDisk.getRootPath(), offlineDisk));
403+
return backend;
404+
}
405+
359406
private Partition mockPartition(String name) {
360407
Partition partition = Mockito.mock(Partition.class);
361408
Mockito.when(partition.getName()).thenReturn(name);

0 commit comments

Comments
 (0)