github-actions[bot] commented on code in PR #65173: URL: https://github.com/apache/doris/pull/65173#discussion_r3581214242
########## fe/fe-core/src/main/java/org/apache/doris/resource/BackendSelectionPolicy.java: ########## @@ -0,0 +1,142 @@ +// 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.resource; + +import org.apache.doris.catalog.Replica; +import org.apache.doris.common.UserException; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.system.Backend; + +import java.util.List; +import java.util.function.Function; + +/** + * SPI for optional backend selection hints. + * <p> + * Implementations are discovered through {@link java.util.ServiceLoader}. Default methods preserve + * the existing backend selection behavior. Candidate-ordering methods must not mutate their input list: + * implementations that change the order must return a new list, while implementations that keep the order + * may return the input list unchanged. + */ +public interface BackendSelectionPolicy { + + /** Observability classification of how a repair clone source was selected. */ + enum RepairSourceSelectionResult { + DISABLED, + PREFERRED_HIT, + FALLBACK_NO_PREFERRED, + FALLBACK_PREFERRED_UNAVAILABLE, + FALLBACK_SLOT_FULL + } + + /** Whether repair-clone source selection is active. */ + default boolean isRepairSourceSelectionEnabled() { + return false; + } + + /** + * Reorder healthy clone-source candidates while preserving the caller's existing ordering inside + * each implementation-defined tier. Implementations must not drop candidates. + */ + default List<Replica> orderRepairSourceCandidates(List<Replica> healthyCandidates, long destBackendId) { + return healthyCandidates; + } + + /** + * Classify which implementation-defined tier the finally chosen source fell into, for observability. + */ + default RepairSourceSelectionResult classifyRepairSource(long chosenSrcBackendId, long destBackendId, + List<Replica> allReplicas, List<Replica> healthyCandidates) { + return RepairSourceSelectionResult.DISABLED; + } + + default BackendSelection.SelectionHint getQuerySelectionHint(ConnectContext context) { + return BackendSelection.SelectionHint.noSelection(); + } + + default boolean hasQuerySelectionPreference(BackendSelection.SelectionHint hint) { + return false; + } + + /** + * Classify query selection after the kernel has applied availability and access filters. + * Implementations must not mutate the candidates and must return a non-null result. + */ + default <T> BackendSelection.QuerySelectionResult classifyQuerySelection( + BackendSelection.SelectionHint hint, List<T> candidates, Function<T, Tag> beTagOf) { + return BackendSelection.QuerySelectionResult.DISABLED; + } + + /** Whether this provider implements required query and load candidate partitioning. */ + default boolean supportsRequiredSelection() { + return false; + } + + /** + * Partition query candidates for {@link BackendSelection.Mode#REQUIRE}. The two lists must contain every + * input candidate exactly once using the original instances. The kernel schedules only preferred candidates. + */ + default <T> BackendSelection.CandidateSelection<T> partitionRequiredQueryCandidates( + BackendSelection.SelectionHint hint, List<T> candidates, Function<T, Tag> beTagOf) throws UserException { + throw new UserException("BackendSelectionPolicy does not support required backend selection"); + } + + /** + * Optionally reorder query scan candidates before the existing scheduler chooses a backend. This + * is only a placement hint: callers may still apply their normal load-balancing policy after this + * method. Implementations must not drop candidates. + */ + default <T> List<T> orderQueryCandidates(BackendSelection.SelectionHint hint, List<T> candidates, + Function<T, Tag> beTagOf) throws UserException { + return candidates; + } + + default boolean isLoadSelectionEnabled(ConnectContext context) { + return false; + } + + /** + * Reorder load candidates without changing the candidate set. Implementations must return every input + * candidate exactly once and must not add candidates. + */ + default List<Backend> orderLoadCandidates(BackendSelection.SelectionHint hint, + List<Backend> candidates) throws UserException { + return candidates; + } + + /** + * Partition load candidates for {@link BackendSelection.Mode#REQUIRE}. The two lists must contain every + * input candidate exactly once using the original instances. The kernel schedules only preferred candidates. + */ + default BackendSelection.CandidateSelection<Backend> partitionRequiredLoadCandidates( + BackendSelection.SelectionHint hint, List<Backend> candidates) throws UserException { + throw new UserException("BackendSelectionPolicy does not support required backend selection"); + } + + default boolean hasLoadSelectionPreference(BackendSelection.SelectionHint hint) { + return false; + } + + default BackendSelection.SelectionHint getLoadSelectionHint(ConnectContext context) { + return null; + } + + default BackendSelection.SelectionHint getForwardedLoadSelectionHint(String preferredKey, String mode) { + return null; Review Comment: Forwarded group-commit requests can drop the selection hint unless every provider implements this extra hook. The direct-master path passes the resolved `SelectionHint` object into `selectBackendForGroupCommitInternal()`, where `PREFER` and `REQUIRE` are enforced. A follower now serializes only `loadSelectionPreferredKey`/`loadSelectionMode`, and the master reconstructs the hint only through `getForwardedLoadSelectionHint()`. Since the default is `null`, an existing provider that already implements normal load selection but not this forwarding-specific method will make follower-forwarded group commit use the no-selection path, including losing `REQUIRE` semantics. Please preserve the generic serialized key/mode in the kernel default path, or make this forwarding method an explicit required contract and add a forwarded group-commit test for a normal provider that does not override it. ########## fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java: ########## @@ -343,51 +372,82 @@ private long selectBackendForLocalGroupCommitInternal(long tableId) throws LoadE } @Nullable - private Long getCachedBackend(String cluster, long tableId) { + private Long getCachedCloudBackend(String cacheKey, String cluster, long tableId) { + return getCachedBackend(cacheKey, cluster, tableId); + } + + @Nullable + private Long getCachedLocalBackend(String cacheKey, long tableId) { + return getCachedBackend(cacheKey, null, tableId); + } + + @Nullable + private Long getCachedBackend(String cacheKey, @Nullable String cloudCluster, long tableId) { OlapTable table = (OlapTable) Env.getCurrentEnv().getInternalCatalog().getTableByTableId(tableId); - if (tableToBeMap.containsKey(encode(cluster, tableId))) { + // There are multiple threads getting cached backends for the same table. + // Maybe one thread removes the tableId from the tableToBeMap. + // Another thread gets the same tableId but can not find this tableId. + // So another thread needs to get the random backend. + Long backendId = tableToBeMap.getIfPresent(cacheKey); + if (backendId != null) { if (tableToPressureMap.get(tableId) == null) { return null; } else if (tableToPressureMap.get(tableId).get() < table.getGroupCommitDataBytes()) { - // There are multiple threads getting cached backends for the same table. - // Maybe one thread removes the tableId from the tableToBeMap. - // Another thread gets the same tableId but can not find this tableId. - // So another thread needs to get the random backend. - Long backendId = tableToBeMap.get(encode(cluster, tableId)); - if (backendId == null) { - return null; - } Backend backend = Env.getCurrentSystemInfo().getBackend(backendId); - if (isBackendAvailable(backend, cluster)) { + if (isBackendAvailable(backend, cloudCluster)) { return backend.getId(); } else { - tableToBeMap.remove(encode(cluster, tableId)); + tableToBeMap.invalidate(cacheKey); } } else { - tableToBeMap.remove(encode(cluster, tableId)); + tableToBeMap.invalidate(cacheKey); } } return null; } - private boolean isBackendAvailable(Backend backend, String cluster) { + private boolean isBackendAvailable(Backend backend, @Nullable String cloudCluster) { if (backend == null || !backend.isAlive() || backend.isDecommissioned() || backend.isDecommissioning() || !backend.isLoadAvailable()) { return false; } if (!Config.isCloudMode()) { return true; } - return cluster == null || cluster.equals(backend.getCloudClusterName()); + return cloudCluster == null || cloudCluster.equals(backend.getCloudClusterName()); } @Nullable - private Long getRandomBackend(String cluster, long tableId, List<Backend> backends) { + private Long getRandomCloudBackend(String cacheKey, String cluster, long tableId, List<Backend> backends) + throws LoadException { OlapTable table = (OlapTable) Env.getCurrentEnv().getInternalCatalog().getTableByTableId(tableId); Collections.shuffle(backends); - for (Backend backend : backends) { - if (isBackendAvailable(backend, cluster)) { - tableToBeMap.put(encode(cluster, tableId), backend.getId()); + return selectAvailableBackend(cacheKey, cluster, tableId, table, backends); + } + + @Nullable + private Long getRandomLocalBackend(String cacheKey, long tableId, List<Backend> backends, + @Nullable BackendSelection.SelectionHint selectionHint, boolean hasLoadSelectionPreference) + throws LoadException { + OlapTable table = (OlapTable) Env.getCurrentEnv().getInternalCatalog().getTableByTableId(tableId); + Collections.shuffle(backends); + List<Backend> orderedBackends; + try { + orderedBackends = hasLoadSelectionPreference + ? BackendSelectionService.orderLoadCandidates(selectionHint, backends) + : backends; + } catch (UserException e) { + throw new LoadException(e.getMessage()); + } + return selectAvailableBackend(cacheKey, null, tableId, table, orderedBackends); + } + + @Nullable + private Long selectAvailableBackend(String cacheKey, @Nullable String cloudCluster, long tableId, OlapTable table, + List<Backend> orderedBackends) { + for (Backend backend : orderedBackends) { + if (isBackendAvailable(backend, cloudCluster)) { + tableToBeMap.put(cacheKey, backend.getId()); tableToPressureMap.put(tableId, Review Comment: The backend cache is now keyed by the effective selection hint, but the pressure counter is still keyed only by table id. That means separate cache entries for the same table can reset each other's pressure. For example, a `g_a` request can fill backend A past `groupCommitDataBytes`; before the next `g_a` lookup invalidates it, a `g_b` miss reaches this cache write for backend B and the next line replaces `tableToPressureMap[tableId]` with a fresh counter. The next `g_a` request then sees the shared counter below threshold and keeps reusing backend A even though A already exceeded the rebalance limit. Please key the pressure counter with the same effective cache key, or otherwise carry the selected cache key through `updateLoadData()` so pressure is accumulated for the backend choice that received the load. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelector.java: ########## @@ -258,6 +267,7 @@ private WorkerScanRanges selectScanReplicaAndMinWorkloadWorker( replicaLocations = Lists.newArrayList(replicaLocations); Collections.sort(replicaLocations); } + replicaLocations = orderLoadReplicas(replicaLocations, catalogId); Review Comment: This applies the load-selection order, but the final choice below still minimizes `WorkerWorkload`, so `PREFER` can route to an unpreferred idle backend even when the preferred backend is available. For Nereids insert/select loads, `NereidsPlanner` enables this selector whenever the plan contains a `PhysicalOlapTableSink`, `orderLoadReplicas()` rebuilds `[preferred, fallback]`, and then the loop at lines 277-289 picks whichever worker has less assigned work. That is weaker than the other load-selection callers, which choose the first available ordered backend after `orderLoadCandidates()`, so a strict/prefer-local load policy can be silently ignored on this Nereids path. Please either choose the first available policy-ordered worker when load selection has a preference (with `REQUIRE` staying a hard filter), or document and test that Nereids load selection is only a workload tie-breaker. -- 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]
