deardeng commented on code in PR #65173: URL: https://github.com/apache/doris/pull/65173#discussion_r3747154604
########## 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: The community implementation in this PR is strictly no-op and never produces or forwards a load-selection hint. The downstream enterprise provider already overrides getForwardedLoadSelectionHint() and reconstructs the preferred key and mode, so the current downstream implementation does not lose PREFER or REQUIRE semantics. That said, the public SPI contract is currently too permissive because another provider could omit this override and silently fall back to null. We will tighten this contract in a dedicated follow-up PR and synchronize the interface change between Apache Doris and the downstream implementation. This PR will keep the current no-op community contract and will not address the SPI hardening as part of its scope. ########## 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 pressure counter is still table-scoped while backend cache entries can become selection-specific when a non-community provider enables load selection. This path is not activated by the Apache Doris default provider because the community implementation is strictly no-op and does not create selection-specific group-commit cache entries. Therefore, this does not change the existing Apache Doris group-commit behavior. A complete fix needs to align the backend cache key and pressure-accounting identity, including the updateLoadData() path, which currently carries only the table id. We will handle that end-to-end in a dedicated follow-up PR and apply the corresponding fix to both upstream and downstream. This PR will remain focused on introducing the public extension framework with no-op community behavior. -- 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]
