This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch tpch500
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/tpch500 by this push:
new 3992fc52436 [opt](task-assignment) use consistent hash as default task
assigner and cache the consistent hash ring (#28369)
3992fc52436 is described below
commit 3992fc5243635ccb2f386ca3af021d0817ba447f
Author: Mingyu Chen <[email protected]>
AuthorDate: Wed Dec 13 23:06:03 2023 +0800
[opt](task-assignment) use consistent hash as default task assigner and
cache the consistent hash ring (#28369)
1. Use consistent hash algo as the default assigner for file query scan node
A consistent assignment can better utilize the page cache of BE node.
2. Cache the consistent hash ring
Init a consistent hash ring is time-consuming because there a thousands
of virtual node need to be added.
So cache it for better performance
---
.../planner/external/FederationBackendPolicy.java | 59 +++++++++++++++++++++-
.../doris/planner/external/FileQueryScanNode.java | 9 ++--
2 files changed, 60 insertions(+), 8 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FederationBackendPolicy.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FederationBackendPolicy.java
index 9e23463235f..bde2fa8c191 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FederationBackendPolicy.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FederationBackendPolicy.java
@@ -30,6 +30,9 @@ import org.apache.doris.thrift.TFileRangeDesc;
import org.apache.doris.thrift.TScanRangeLocations;
import com.google.common.base.Preconditions;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -46,7 +49,9 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
public class FederationBackendPolicy {
@@ -59,6 +64,53 @@ public class FederationBackendPolicy {
private int nextBe = 0;
private boolean initialized = false;
+ // Create a ConsistentHash ring may be a time-consuming operation, so we
cache it.
+ private static LoadingCache<HashCacheKey,
ConsistentHash<TScanRangeLocations, Backend>> consistentHashCache;
+
+ static {
+ consistentHashCache = CacheBuilder.newBuilder().maximumSize(5)
+ .build(new CacheLoader<HashCacheKey,
ConsistentHash<TScanRangeLocations, Backend>>() {
+ @Override
+ public ConsistentHash<TScanRangeLocations, Backend>
load(HashCacheKey key) {
+ return new ConsistentHash<>(Hashing.murmur3_128(), new
ScanRangeHash(),
+ new BackendHash(), key.bes,
Config.virtual_node_number);
+ }
+ });
+ }
+
+ private static class HashCacheKey {
+ // sorted backend ids as key
+ private List<Long> beIds;
+ // backends is not part of key, just an attachment
+ private List<Backend> bes;
+
+ HashCacheKey(List<Backend> backends) {
+ this.bes = backends;
+ this.beIds = backends.stream().map(b ->
b.getId()).sorted().collect(Collectors.toList());
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof HashCacheKey)) {
+ return false;
+ }
+ return Objects.equals(beIds, ((HashCacheKey) obj).beIds);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(beIds);
+ }
+
+ @Override
+ public String toString() {
+ return "HashCache{" + "beIds=" + beIds + '}';
+ }
+ }
+
public void init() throws UserException {
if (!initialized) {
init(Collections.emptyList());
@@ -96,8 +148,11 @@ public class FederationBackendPolicy {
throw new UserException("No available backends");
}
backendMap.putAll(backends.stream().collect(Collectors.groupingBy(Backend::getHost)));
- consistentHash = new ConsistentHash<>(Hashing.murmur3_128(), new
ScanRangeHash(),
- new BackendHash(), backends, Config.virtual_node_number);
+ try {
+ consistentHash = consistentHashCache.get(new
HashCacheKey(backends));
+ } catch (ExecutionException e) {
+ throw new UserException("failed to get consistent hash", e);
+ }
}
public Backend getNextBe() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
index ada9f1fda61..ccf7373583f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
@@ -311,7 +311,6 @@ public abstract class FileQueryScanNode extends
FileScanNode {
params.setProperties(locationProperties);
}
- boolean enableSqlCache =
ConnectContext.get().getSessionVariable().enableFileCache;
boolean enableShortCircuitRead =
HdfsResource.enableShortCircuitRead(locationProperties);
List<String> pathPartitionKeys = getPathPartitionKeys();
for (Split split : inputSplits) {
@@ -369,14 +368,12 @@ public abstract class FileQueryScanNode extends
FileScanNode {
curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc);
TScanRangeLocation location = new TScanRangeLocation();
Backend selectedBackend;
- if (enableSqlCache) {
- // Use consistent hash to assign the same scan range into the
same backend among different queries
- selectedBackend =
backendPolicy.getNextConsistentBe(curLocations);
- } else if (enableShortCircuitRead) {
+ if (enableShortCircuitRead) {
// Try to find a local BE if enable hdfs short circuit read
selectedBackend =
backendPolicy.getNextLocalBe(Arrays.asList(fileSplit.getHosts()));
} else {
- selectedBackend = backendPolicy.getNextBe();
+ // Use consistent hash to assign the same scan range into the
same backend among different queries
+ selectedBackend =
backendPolicy.getNextConsistentBe(curLocations);
}
setLocationPropertiesIfNecessary(selectedBackend, locationType,
locationProperties);
location.setBackendId(selectedBackend.getId());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
