From 032ff57a16f1db95841522bd71806cc71143eb76 Mon Sep 17 00:00:00 2001 From: Dylan Date: Thu, 8 Oct 2026 17:40:35 +0800 Subject: [PATCH 01/12] [improvement](fe) Spread hot external splits across hash candidates ### What problem does this PR solve? Issue Number: N/A Related PR: #68726 Problem Summary: Repeated independent queries for the same remote external file split can choose the same backend under consistent hashing. Each policy starts with zero assigned weight and the original tie rule selects the same candidate, so increasing the existing candidate count does not remove the hotspot. Add an opt-in external_scan_consistent_hash_spread_num session variable. The default value of 1 preserves the existing scheduler. Values greater than 1 select stable, distinct hash candidates and choose the candidate with the least policy-assigned weight, breaking ties uniformly at random. Candidate counts are capped by eligible backends. Local preference and mandatory Host constraints remain in effect; global redistribution is disabled only in the new mode to preserve candidate and locality bounds. Each split executes once. In a local experiment using three backends on one shared physical host, a one-million-row, single-file workload at concurrency 32 improved pooled successful QPS by 2.4234x for COUNT and 2.4143x for SUM across three paired 60-second rounds with file caching disabled. All 12 measurements had zero errors. This does not establish a threefold guarantee or customer result. The round-robin and COUNT changes in the related PR remain separate. ### Release note Add an optional external_scan_consistent_hash_spread_num session setting to spread repeated external scan queries across consistent hash candidates. The default preserves existing scheduling. With file caching enabled, spreading can increase warmup reads and cache space on additional backends. ### Check List (For Author) - Test: Regression test / Unit Test / Manual test - Standard FE build and Checkstyle passed. - FederationBackendPolicySpreadTest: 19 tests passed. - FederationBackendPolicyTest: 9 existing tests passed. - VariableMgrTest#testExternalScanConsistentHashSpreadVariable passed. - Parameter and S3 TVF regression suites generated and verified through the standard regression entry point against the isolated personal cluster. - COUNT and SUM: 100 independent queries each, correct results, one split and one execution target per query, distributed across three backends. - Three paired cache-off capacity rounds for COUNT and SUM passed. - Cold file-cache cost sampling: 30 queries per mode passed; spreading warmed three backend copies and incurred three times the remote reads. - Behavior changed: Yes, only when the new setting is greater than 1 and the scan uses consistent hashing; default scheduling is unchanged. - Does this need documentation: Yes, included in docs/external-scan-consistent-hash-spread.md. --- docs/external-scan-consistent-hash-spread.md | 63 +++ .../datasource/scan/ExternalScanNode.java | 3 +- .../scan/FederationBackendPolicy.java | 49 ++- .../org/apache/doris/qe/SessionVariable.java | 17 + .../FederationBackendPolicySpreadTest.java | 377 ++++++++++++++++++ .../org/apache/doris/qe/VariableMgrTest.java | 23 ++ ...t_external_scan_consistent_hash_spread.out | 44 ++ ...l_scan_consistent_hash_spread_variable.out | 16 + ...xternal_scan_consistent_hash_spread.groovy | 82 ++++ ...can_consistent_hash_spread_variable.groovy | 44 ++ 10 files changed, 714 insertions(+), 4 deletions(-) create mode 100644 docs/external-scan-consistent-hash-spread.md create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java create mode 100644 regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out create mode 100644 regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out create mode 100644 regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy create mode 100644 regression-test/suites/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.groovy diff --git a/docs/external-scan-consistent-hash-spread.md b/docs/external-scan-consistent-hash-spread.md new file mode 100644 index 00000000000000..becd745477310b --- /dev/null +++ b/docs/external-scan-consistent-hash-spread.md @@ -0,0 +1,63 @@ +# Spread hot external scan splits across consistent hash candidates + +Repeated queries for the same remote file split can all choose the same backend +under consistent hashing. Each query starts with zero assigned weight, and the +original tie rule always chooses the same candidate. Increasing the original +candidate count alone does not remove this bias. + +## Enable the optional scheduling mode + +```sql +SET use_consistent_hash_for_external_scan = true; +SET external_scan_consistent_hash_spread_num = 3; +``` + +`external_scan_consistent_hash_spread_num` accepts integers from 1 through +2147483647. Its default is **1**, which preserves the original scheduler, +including its configured hash candidate count and split redistribution. It does +not force the original scheduler to use just one candidate. `UNSET VARIABLE +external_scan_consistent_hash_spread_num` restores the session default. + +Values greater than 1 enable spreading only when an external scan uses consistent +hashing. This occurs when either `enable_file_cache` or +`use_consistent_hash_for_external_scan` is enabled. The new variable alone does +not switch a round robin scan to consistent hashing, and it does not enable file +caching. Other consumers of the backend policy, including file load and schema +scans, retain their existing behavior. + +## Assignment and locality + +For a remote split without a preferred local backend, the scheduler takes the +first N distinct eligible backends from the existing consistent hash ring, capped +by the number of eligible backends in the query's compute group. It selects the +backend with the least weight already assigned by this policy. Ties are broken +uniformly at random. A fresh query can therefore choose a different backend for +the same split, while multiple splits and batches share their policy's weight +history. Each split is assigned for execution exactly once. + +The mode preserves local preference and mandatory Host constraints for splits +that cannot be accessed remotely. These constraints can choose a backend outside +the remote split's hash candidates. Different backends on the same Host remain +distinct hash candidates. + +Global split redistribution is disabled in this mode: it could otherwise move a +split outside its hash candidates or mandatory locality. Queries with many splits +balance within each split's candidates, rather than across every eligible backend. +Some split distributions can consequently be less balanced than with the original +global redistribution. The default mode retains that redistribution. + +## Capacity and cache tradeoffs + +Spreading independently planned queries can increase throughput when the original +hot backend is saturated and the other candidate backends have available capacity. +Random choice does not guarantee a particular assignment sequence or throughput +multiplier. Frontend planning, remote storage, network capacity and shared host +resources can still limit throughput. No CPU monitoring or shared query counter is +introduced. + +With file caching enabled, the same data can occupy cache space on multiple +backends and incur additional warmup reads. The variable bounds remote hash +candidates; it does not create cache replicas in advance. Locality takes precedence +over this bound. To measure scheduling changes separately from cache hits, keep +file caching disabled and enable consistent hashing explicitly on both sides of +a comparison. diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java index c378c285dc55b2..7b1b9fb3d1cf8d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java @@ -47,7 +47,8 @@ public abstract class ExternalScanNode extends ScanNode { protected final FederationBackendPolicy backendPolicy = (ConnectContext.get() != null && (ConnectContext.get().getSessionVariable().enableFileCache || ConnectContext.get().getSessionVariable().getUseConsistentHashForExternalScan())) - ? new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING) + ? new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING, + ConnectContext.get().getSessionVariable().getExternalScanConsistentHashSpreadNum()) : new FederationBackendPolicy(); public ExternalScanNode(PlanNodeId id, TupleDescriptor desc, String planNodeName, diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java index bb23481fd4de52..a57c1d9756239d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java @@ -67,6 +67,8 @@ import java.util.Random; import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ThreadLocalRandom; +import java.util.function.IntUnaryOperator; import java.util.stream.Collectors; public class FederationBackendPolicy { @@ -90,6 +92,8 @@ public Map getAssignedWeightPerBackend() { private NodeSelectionStrategy nodeSelectionStrategy; private boolean enableSplitsRedistribution = true; + private final int consistentHashSpreadNum; + private final IntUnaryOperator randomIndex; // Create a ConsistentHash ring may be a time-consuming operation, so we cache it. private static LoadingCache> consistentHashCache; @@ -142,7 +146,20 @@ public String toString() { } public FederationBackendPolicy(NodeSelectionStrategy nodeSelectionStrategy) { + this(nodeSelectionStrategy, 1); + } + + public FederationBackendPolicy(NodeSelectionStrategy nodeSelectionStrategy, int consistentHashSpreadNum) { + this(nodeSelectionStrategy, consistentHashSpreadNum, bound -> ThreadLocalRandom.current().nextInt(bound)); + } + + @VisibleForTesting + FederationBackendPolicy(NodeSelectionStrategy nodeSelectionStrategy, int consistentHashSpreadNum, + IntUnaryOperator randomIndex) { + Preconditions.checkArgument(consistentHashSpreadNum > 0, "consistentHashSpreadNum must be at least 1"); this.nodeSelectionStrategy = nodeSelectionStrategy; + this.consistentHashSpreadNum = consistentHashSpreadNum; + this.randomIndex = randomIndex; } public FederationBackendPolicy() { @@ -285,7 +302,8 @@ public Multimap computeScanRangeAssignment(List splits) t } case CONSISTENT_HASHING: { candidateNodes = consistentHash.getNode(split, - Config.split_assigner_min_consistent_hash_candidate_num); + isConsistentHashSpreadEnabled() ? Math.min(consistentHashSpreadNum, backends.size()) + : Config.split_assigner_min_consistent_hash_candidate_num); break; } default: { @@ -301,7 +319,8 @@ public Multimap computeScanRangeAssignment(List splits) t throw new UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG); } - Backend selectedBackend = chooseNodeForSplit(candidateNodes); + Backend selectedBackend = isConsistentHashSpreadEnabled() && split.isRemotelyAccessible() + ? chooseNodeForSpread(candidateNodes) : chooseNodeForSplit(candidateNodes); List alternativeBackends = new ArrayList<>(candidateNodes); alternativeBackends.remove(selectedBackend); split.setAlternativeHosts( @@ -311,7 +330,9 @@ public Multimap computeScanRangeAssignment(List splits) t assignedWeightPerBackend.get(selectedBackend) + split.getSplitWeight().getRawValue()); } - if (enableSplitsRedistribution) { + // Global redistribution can move a split outside its hash candidates or its locality constraints. + // The spread mode balances weights within each split's candidates during initial assignment instead. + if (enableSplitsRedistribution && !isConsistentHashSpreadEnabled()) { equateDistribution(assignment); } return assignment; @@ -484,6 +505,28 @@ private Backend chooseNodeForSplit(List candidateNodes) { return chosenNode; } + private boolean isConsistentHashSpreadEnabled() { + return nodeSelectionStrategy == NodeSelectionStrategy.CONSISTENT_HASHING && consistentHashSpreadNum > 1; + } + + private Backend chooseNodeForSpread(List candidateNodes) { + Backend chosenNode = null; + long minWeight = Long.MAX_VALUE; + int tiedNodes = 0; + for (Backend node : candidateNodes) { + long queuedWeight = assignedWeightPerBackend.get(node); + if (queuedWeight < minWeight) { + chosenNode = node; + minWeight = queuedWeight; + tiedNodes = 1; + } else if (queuedWeight == minWeight && randomIndex.applyAsInt(++tiedNodes) == 0) { + // Reservoir sampling gives every node with the minimum weight the same probability. + chosenNode = node; + } + } + return chosenNode; + } + public int numBackends() { return backends.size(); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index ddaec7b2ae95b0..4148b516a33ba0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -1486,6 +1486,15 @@ public enum IgnoreSplitType { description = "Use consistent hashing to split the appearance for external scan") public boolean useConsistentHashForExternalScan = false; + public static final String EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM = "external_scan_consistent_hash_spread_num"; + @VarAttrDef.VarAttr(name = EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, + checker = "checkExternalScanConsistentHashSpreadNum", needForward = true, + description = "Number of consistent hash candidates for spreading hot remote external splits. " + + "1 preserves the original scheduling; values above 1 enable random tie breaking within " + + "the lowest assigned weight candidates and disable global split redistribution. " + + "Requires file cache or use_consistent_hash_for_external_scan.") + public int externalScanConsistentHashSpreadNum = 1; + @VarAttrDef.VarAttr(name = PROFILE_LEVEL, fuzzy = false, needForward = true, setter = "setProfileLevel", checker = "checkProfileLevel", description = "The level of query profile, " @@ -6403,6 +6412,14 @@ public boolean getUseConsistentHashForExternalScan() { return useConsistentHashForExternalScan; } + public int getExternalScanConsistentHashSpreadNum() { + return externalScanConsistentHashSpreadNum; + } + + public void checkExternalScanConsistentHashSpreadNum(String value) throws Exception { + checkFieldValue(EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, 1, value); + } + public void setForceJniScanner(boolean force) { forceJniScanner = force; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java new file mode 100644 index 00000000000000..15a3d0eef13f21 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -0,0 +1,377 @@ +// 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.datasource.scan; + +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.analysis.TupleId; +import org.apache.doris.catalog.Env; +import org.apache.doris.common.Config; +import org.apache.doris.common.UserException; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.common.util.LocationPath; +import org.apache.doris.datasource.split.FileSplit; +import org.apache.doris.planner.PlanNodeId; +import org.apache.doris.planner.ScanContext; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.SessionVariable; +import org.apache.doris.resource.computegroup.ComputeGroup; +import org.apache.doris.spi.Split; +import org.apache.doris.system.Backend; + +import com.google.common.collect.Multimap; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.IntUnaryOperator; + +public class FederationBackendPolicySpreadTest { + private static final AtomicLong NEXT_BACKEND_ID = new AtomicLong(91000); + + private MockedStatic mockedEnv; + private ConnectContext previousContext; + private ConnectContext context; + private ComputeGroup computeGroup; + private List backends; + private int originalCandidateNum; + private boolean originalLocalScheduling; + + @BeforeEach + public void setUp() { + previousContext = ConnectContext.get(); + originalCandidateNum = Config.split_assigner_min_consistent_hash_candidate_num; + originalLocalScheduling = Config.split_assigner_optimized_local_scheduling; + Config.split_assigner_min_consistent_hash_candidate_num = 2; + Config.split_assigner_optimized_local_scheduling = true; + mockedEnv = Mockito.mockStatic(Env.class); + mockedEnv.when(Env::getCurrentEnv).thenReturn(Mockito.mock(Env.class)); + backends = new ArrayList<>(); + long firstId = NEXT_BACKEND_ID.getAndAdd(3); + for (int i = 0; i < 3; i++) { + Backend backend = new Backend(firstId + i, "192.0.2." + (i + 1), 9050); + backend.setAlive(true); + backends.add(backend); + } + computeGroup = Mockito.mock(ComputeGroup.class); + Mockito.when(computeGroup.getBackendList()).thenAnswer(invocation -> backends); + context = new ConnectContext(); + context.setComputeGroup(computeGroup); + context.setThreadLocalInfo(); + } + + @AfterEach + public void tearDown() { + Config.split_assigner_min_consistent_hash_candidate_num = originalCandidateNum; + Config.split_assigner_optimized_local_scheduling = originalLocalScheduling; + mockedEnv.close(); + ConnectContext.remove(); + if (previousContext != null) { + previousContext.setThreadLocalInfo(); + } + } + + private FileSplit split() { + return new FileSplit(LocationPath.of("s3://bucket/hot.csv"), 0, 1000, 1000, + 0, new String[0], Collections.emptyList()); + } + + private FederationBackendPolicy policy(int candidates, IntUnaryOperator random) throws UserException { + FederationBackendPolicy policy = new FederationBackendPolicy( + NodeSelectionStrategy.CONSISTENT_HASHING, candidates, random); + policy.init(); + return policy; + } + + private Backend assign(FederationBackendPolicy policy, Split split) throws UserException { + Multimap assignment = policy.computeScanRangeAssignment( + new ArrayList<>(Collections.singletonList(split))); + Assertions.assertEquals(1, assignment.size()); + Assertions.assertEquals(1, assignment.keySet().size()); + Assertions.assertSame(split, assignment.values().iterator().next()); + return assignment.keySet().iterator().next(); + } + + private static IntUnaryOperator selectCandidate(int index) { + // Replace the reservoir through the desired index, then keep that node. + return bound -> bound <= index + 1 ? 0 : bound - 1; + } + + @Test + public void testIndependentHotQueriesCanChooseEveryCandidateExactlyOnce() throws Exception { + Set selected = new HashSet<>(); + for (int index = 0; index < 3; index++) { + FederationBackendPolicy policy = policy(3, selectCandidate(index)); + FileSplit split = split(); + List candidates = policy.consistentHash.getNode(split, 3); + Backend target = assign(policy, split); + Assertions.assertEquals(candidates.get(index), target); + selected.add(target); + Assertions.assertEquals(2, split.getAlternativeHosts().size()); + } + Assertions.assertEquals(new HashSet<>(backends), selected); + } + + @Test + public void testTopTwoCandidateBoundarySurvivesMultipleBatches() throws Exception { + FederationBackendPolicy policy = policy(2, bound -> bound - 1); + Set candidates = new HashSet<>(policy.consistentHash.getNode(split(), 2)); + for (int batch = 0; batch < 2; batch++) { + List splits = new ArrayList<>(); + for (int i = 0; i < 20; i++) { + splits.add(split()); + } + Multimap assignment = policy.computeScanRangeAssignment(splits); + Assertions.assertEquals(20, assignment.size()); + Assertions.assertTrue(candidates.containsAll(assignment.keySet())); + for (Backend backend : candidates) { + Assertions.assertEquals(10, assignment.get(backend).size()); + Assertions.assertEquals((batch + 1) * 1000L, policy.getAssignedWeightPerBackend().get(backend)); + } + } + for (Backend backend : backends) { + if (!candidates.contains(backend)) { + Assertions.assertEquals(0L, policy.getAssignedWeightPerBackend().get(backend)); + } + } + } + + @Test + public void testMinimumWeightWinsBeforeRandomTieBreaking() throws Exception { + FederationBackendPolicy policy = policy(3, selectCandidate(2)); + List candidates = policy.consistentHash.getNode(split(), 3); + policy.getAssignedWeightPerBackend().put(candidates.get(0), 1000L); + policy.getAssignedWeightPerBackend().put(candidates.get(1), 0L); + policy.getAssignedWeightPerBackend().put(candidates.get(2), 100L); + Assertions.assertEquals(candidates.get(1), assign(policy, split())); + } + + @Test + public void testDefaultPreservesOriginalCandidateNumberAndTieBreaking() throws Exception { + FederationBackendPolicy original = new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING); + original.init(); + FederationBackendPolicy disabled = policy(1, bound -> { + throw new AssertionError("Disabled spread must not use randomness"); + }); + List candidates = disabled.consistentHash.getNode(split(), 2); + Assertions.assertEquals(candidates.get(1), assign(disabled, split())); + Assertions.assertEquals(candidates.get(1), assign(original, split())); + } + + @Test + public void testDefaultPreservesGlobalRedistribution() throws Exception { + FederationBackendPolicy disabled = policy(1, bound -> { + throw new AssertionError("Disabled spread must not use randomness"); + }); + List splits = new ArrayList<>(); + for (int i = 0; i < 30; i++) { + splits.add(split()); + } + Assertions.assertEquals(3, disabled.computeScanRangeAssignment(splits).keySet().size()); + } + + @Test + public void testCandidateCountIsCappedByEligibleBackends() throws Exception { + Set targets = new HashSet<>(); + for (int index = 0; index < 3; index++) { + targets.add(assign(policy(Integer.MAX_VALUE, selectCandidate(index)), split())); + } + Assertions.assertEquals(new HashSet<>(backends), targets); + } + + @Test + public void testSingleBackend() throws Exception { + backends = new ArrayList<>(Collections.singletonList(backends.get(0))); + Assertions.assertEquals(backends.get(0), assign(policy(3, bound -> bound - 1), split())); + } + + @Test + public void testSameHostBackendsRemainDistinctCandidates() throws Exception { + backends = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + Backend backend = new Backend(92000 + i, "192.0.2.20", 9050 + i); + backend.setAlive(true); + backends.add(backend); + } + Set targets = new HashSet<>(); + for (int index = 0; index < 3; index++) { + targets.add(assign(policy(3, selectCandidate(index)), split())); + } + Assertions.assertEquals(new HashSet<>(backends), targets); + } + + @Test + public void testUnavailableAndOtherComputeGroupBackendsAreExcluded() throws Exception { + Backend dead = backends.get(2); + dead.setAlive(false); + Backend otherGroup = backends.get(1); + backends = new ArrayList<>(Arrays.asList(backends.get(0), dead)); + FederationBackendPolicy policy = policy(3, bound -> bound - 1); + Assertions.assertEquals(1, policy.numBackends()); + Assertions.assertEquals(backends.get(0), assign(policy, split())); + Assertions.assertFalse(policy.getBackends().contains(otherGroup)); + Assertions.assertFalse(policy.getBackends().contains(dead)); + } + + @Test + public void testMembershipChangesRefreshCandidates() throws Exception { + Backend removed = backends.remove(2); + FederationBackendPolicy afterRemoval = policy(3, selectCandidate(1)); + Assertions.assertEquals(2, afterRemoval.numBackends()); + Assertions.assertFalse(afterRemoval.consistentHash.getNode(split(), 3).contains(removed)); + Assertions.assertTrue(backends.contains(assign(afterRemoval, split()))); + backends.add(removed); + FederationBackendPolicy afterAddition = policy(3, selectCandidate(2)); + Assertions.assertEquals(3, afterAddition.numBackends()); + Assertions.assertEquals(new HashSet<>(backends), + new HashSet<>(afterAddition.consistentHash.getNode(split(), 3))); + Assertions.assertTrue(backends.contains(assign(afterAddition, split()))); + } + + @Test + public void testLocalPreferenceCanChooseOutsideHashCandidates() throws Exception { + FederationBackendPolicy policy = policy(2, bound -> { + throw new AssertionError("Local selection must not use spread randomness"); + }); + FileSplit split = split(); + List candidates = policy.consistentHash.getNode(split, 2); + Backend local = backends.stream().filter(backend -> !candidates.contains(backend)).findFirst().get(); + split.setHosts(new String[] {local.getHost()}); + Assertions.assertEquals(local, assign(policy, split)); + } + + @Test + public void testNonRemoteSplitsRetainHostConstraintWithoutRedistribution() throws Exception { + for (boolean preferLocal : Arrays.asList(true, false)) { + Config.split_assigner_optimized_local_scheduling = preferLocal; + FederationBackendPolicy policy = policy(3, bound -> { + throw new AssertionError("Non-remote selection must not use spread randomness"); + }); + FileSplit split = new FileSplit(LocationPath.of("file:///hot.csv"), 0, 1000, 1000, + 0, new String[] {backends.get(0).getHost()}, Collections.emptyList()) { + @Override + public boolean isRemotelyAccessible() { + return false; + } + }; + List splits = new ArrayList<>(Collections.nCopies(12, split)); + Multimap assignment = policy.computeScanRangeAssignment(splits); + Assertions.assertEquals(12, assignment.size()); + Assertions.assertEquals(Collections.singleton(backends.get(0)), assignment.keySet()); + } + } + + @Test + public void testRemoteLocalityIsIgnoredWhenLocalPreferenceIsDisabled() throws Exception { + Config.split_assigner_optimized_local_scheduling = false; + FederationBackendPolicy policy = policy(2, selectCandidate(1)); + FileSplit split = split(); + List candidates = policy.consistentHash.getNode(split, 2); + Backend local = backends.stream().filter(backend -> !candidates.contains(backend)).findFirst().get(); + split.setHosts(new String[] {local.getHost()}); + Assertions.assertEquals(candidates.get(1), assign(policy, split)); + } + + @Test + public void testQueryAndLoadDisabledBackendsAreExcluded() throws Exception { + backends.get(1).setQueryDisabled(true); + backends.get(2).setLoadDisabled(true); + FederationBackendPolicy policy = policy(3, bound -> bound - 1); + Assertions.assertEquals(1, policy.numBackends()); + Assertions.assertEquals(backends.get(0), assign(policy, split())); + } + + @Test + public void testMissingRequiredHostFails() throws Exception { + FederationBackendPolicy policy = policy(3, bound -> bound - 1); + FileSplit split = new FileSplit(LocationPath.of("file:///hot.csv"), 0, 1000, 1000, + 0, new String[] {"192.0.2.200"}, Collections.emptyList()) { + @Override + public boolean isRemotelyAccessible() { + return false; + } + }; + Assertions.assertThrows(UserException.class, () -> assign(policy, split)); + } + + @Test + public void testNoEligibleBackendFails() { + backends.clear(); + Assertions.assertThrows(UserException.class, () -> policy(3, bound -> bound - 1)); + } + + @Test + public void testInvalidSpreadCountFails() { + Assertions.assertThrows(IllegalArgumentException.class, + () -> new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING, 0)); + } + + @Test + public void testNonHashStrategiesDoNotUseSpreadRandomness() throws Exception { + for (NodeSelectionStrategy strategy : Arrays.asList(NodeSelectionStrategy.ROUND_ROBIN, + NodeSelectionStrategy.RANDOM)) { + FederationBackendPolicy policy = new FederationBackendPolicy(strategy, 3, bound -> { + throw new AssertionError("Non-hash policy must not use spread randomness"); + }); + policy.init(); + Assertions.assertTrue(backends.contains(assign(policy, split()))); + } + } + + @Test + public void testExternalScanGateAndSessionIsolation() throws Exception { + SessionVariable session = context.getSessionVariable(); + session.externalScanConsistentHashSpreadNum = 3; + for (boolean cache : Arrays.asList(false, true)) { + for (boolean hash : Arrays.asList(false, true)) { + session.enableFileCache = cache; + session.useConsistentHashForExternalScan = hash; + FederationBackendPolicy policy = new TestExternalScanNode().backendPolicy; + Assertions.assertEquals(cache || hash ? NodeSelectionStrategy.CONSISTENT_HASHING + : NodeSelectionStrategy.ROUND_ROBIN, + Deencapsulation.getField(policy, "nodeSelectionStrategy")); + Assertions.assertEquals(cache || hash ? 3 : 1, + (int) Deencapsulation.getField(policy, "consistentHashSpreadNum")); + } + } + FederationBackendPolicy plain = new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING); + Assertions.assertEquals(1, (int) Deencapsulation.getField(plain, "consistentHashSpreadNum")); + ConnectContext.remove(); + FederationBackendPolicy noContext = new TestExternalScanNode().backendPolicy; + Assertions.assertEquals(1, (int) Deencapsulation.getField(noContext, "consistentHashSpreadNum")); + } + + private static class TestExternalScanNode extends ExternalScanNode { + TestExternalScanNode() { + super(new PlanNodeId(0), new TupleDescriptor(new TupleId(0)), "test", ScanContext.EMPTY, false); + } + + @Override + protected void createScanRangeLocations() { + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java index 2bd7c696de3ded..2a8b23fbe1da68 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java @@ -258,4 +258,27 @@ public void testAdaptiveBatchSizeRejectsZeroByteValues() { new StringLiteral("0")))); Assertions.assertTrue(blockSizeException.getMessage().contains("preferred_block_size_bytes")); } + + @Test + public void testExternalScanConsistentHashSpreadVariable() throws Exception { + SessionVariable var = new SessionVariable(); + Assertions.assertEquals(1, var.getExternalScanConsistentHashSpreadNum()); + for (String value : new String[] {"1", "2", "3", "2147483647"}) { + VariableMgr.setVar(var, new SetVar(SetType.SESSION, + SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, new StringLiteral(value))); + Assertions.assertEquals(Integer.parseInt(value), var.getExternalScanConsistentHashSpreadNum()); + } + for (String value : new String[] {"0", "-1", "2147483648", "abc"}) { + Assertions.assertThrows(DdlException.class, () -> VariableMgr.setVar(var, new SetVar(SetType.SESSION, + SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, new StringLiteral(value)))); + Assertions.assertEquals(Integer.MAX_VALUE, var.getExternalScanConsistentHashSpreadNum()); + } + SessionVariable forwarded = new SessionVariable(); + forwarded.setForwardedSessionVariables(var.getForwardVariables()); + Assertions.assertEquals(Integer.MAX_VALUE, forwarded.getExternalScanConsistentHashSpreadNum()); + VariableMgr.setVar(var, new SetVar(SetType.SESSION, + SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, null)); + Assertions.assertEquals(1, var.getExternalScanConsistentHashSpreadNum()); + } + } diff --git a/regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out b/regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out new file mode 100644 index 00000000000000..3f709e4d3fce8c --- /dev/null +++ b/regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out @@ -0,0 +1,44 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !remote -- +1 alpha +2 \N +3 gamma + +-- !remote -- +1 alpha +2 \N +3 gamma + +-- !remote -- +1 alpha +2 \N +3 gamma + +-- !remote -- +1 alpha +2 \N +3 gamma + +-- !cache -- +1 alpha +2 \N +3 gamma + +-- !round_robin -- +1 alpha +2 \N +3 gamma + +-- !multiple_files -- +1 alpha +1 alpha +2 \N +2 \N +3 gamma +3 gamma + +-- !unset -- +1 alpha +2 \N +3 gamma + diff --git a/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out b/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out new file mode 100644 index 00000000000000..bd5f809d8bda42 --- /dev/null +++ b/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out @@ -0,0 +1,16 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !default -- +external_scan_consistent_hash_spread_num 1 1 0 + +-- !enabled -- +external_scan_consistent_hash_spread_num 3 1 1 + +-- !maximum -- +external_scan_consistent_hash_spread_num 2147483647 1 1 + +-- !after_errors -- +external_scan_consistent_hash_spread_num 2147483647 1 1 + +-- !unset -- +external_scan_consistent_hash_spread_num 1 1 0 + diff --git a/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy new file mode 100644 index 00000000000000..b451e0cb9a379c --- /dev/null +++ b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy @@ -0,0 +1,82 @@ +// 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. + +suite("test_external_scan_consistent_hash_spread", "p0,external") { + String ak = getS3AK() + String sk = getS3SK() + String endpoint = getS3Endpoint() + String region = getS3Region() + String bucket = getS3BucketName() + String pathStyle = getS3Provider().equalsIgnoreCase("S3") ? "true" : "false" + String prefix = "s3://${bucket}/test_external_scan_consistent_hash_spread/${UUID.randomUUID()}/part_" + + sql "drop table if exists test_external_scan_consistent_hash_spread" + sql """ + create table test_external_scan_consistent_hash_spread (id int, value string) + distributed by hash(id) buckets 1 + properties ("replication_num" = "1") + """ + sql "insert into test_external_scan_consistent_hash_spread values (1, 'alpha'), (2, null), (3, 'gamma')" + sql """ + select id, value from test_external_scan_consistent_hash_spread order by id + into outfile "${prefix}single_" format as parquet + properties ( + "s3.endpoint" = "${endpoint}", "s3.region" = "${region}", + "s3.access_key" = "${ak}", "s3.secret_key" = "${sk}", + "s3.path_style_access" = "${pathStyle}" + ) + """ + + def remoteQuery = { pattern -> + """select id, value from s3( + "uri" = "${pattern}", "format" = "parquet", + "s3.endpoint" = "${endpoint}", "s3.region" = "${region}", + "s3.access_key" = "${ak}", "s3.secret_key" = "${sk}", + "use_path_style" = "${pathStyle}" + ) order by id""" + } + // Disable result caching so every setting exercises an external scan. + sql "set enable_sql_cache = false" + sql "set enable_query_cache = false" + sql "set enable_file_cache = false" + sql "set use_consistent_hash_for_external_scan = true" + for (int candidates : [1, 2, 3, 2147483647]) { + sql "set external_scan_consistent_hash_spread_num = ${candidates}" + order_qt_remote remoteQuery("${prefix}single_*.parquet") + } + + sql "set external_scan_consistent_hash_spread_num = 3" + sql "set enable_file_cache = true" + sql "set use_consistent_hash_for_external_scan = false" + order_qt_cache remoteQuery("${prefix}single_*.parquet") + sql "set enable_file_cache = false" + order_qt_round_robin remoteQuery("${prefix}single_*.parquet") + + sql """ + select id, value from test_external_scan_consistent_hash_spread order by id + into outfile "${prefix}second_" format as parquet + properties ( + "s3.endpoint" = "${endpoint}", "s3.region" = "${region}", + "s3.access_key" = "${ak}", "s3.secret_key" = "${sk}", + "s3.path_style_access" = "${pathStyle}" + ) + """ + sql "set use_consistent_hash_for_external_scan = true" + order_qt_multiple_files remoteQuery("${prefix}*.parquet") + sql "unset variable external_scan_consistent_hash_spread_num" + order_qt_unset remoteQuery("${prefix}single_*.parquet") +} diff --git a/regression-test/suites/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.groovy b/regression-test/suites/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.groovy new file mode 100644 index 00000000000000..d30c759ef0df0d --- /dev/null +++ b/regression-test/suites/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.groovy @@ -0,0 +1,44 @@ +// 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. + +suite("test_external_scan_consistent_hash_spread_variable", "p0") { + order_qt_default "show session variables like 'external_scan_consistent_hash_spread_num'" + sql "set external_scan_consistent_hash_spread_num = 3" + order_qt_enabled "show session variables like 'external_scan_consistent_hash_spread_num'" + sql "set external_scan_consistent_hash_spread_num = 2147483647" + order_qt_maximum "show session variables like 'external_scan_consistent_hash_spread_num'" + + test { + sql "set external_scan_consistent_hash_spread_num = 0" + exception "greater than or equal 1" + } + test { + sql "set external_scan_consistent_hash_spread_num = -1" + exception "greater than or equal 1" + } + test { + sql "set external_scan_consistent_hash_spread_num = 2147483648" + exception "2147483648" + } + test { + sql "set external_scan_consistent_hash_spread_num = 'abc'" + exception "abc" + } + order_qt_after_errors "show session variables like 'external_scan_consistent_hash_spread_num'" + sql "unset variable external_scan_consistent_hash_spread_num" + order_qt_unset "show session variables like 'external_scan_consistent_hash_spread_num'" +} From 0ada7c2418d9d4dae71cf0621792b6e2cfce5caf Mon Sep 17 00:00:00 2001 From: Dylan Date: Thu, 8 Oct 2026 22:43:44 +0800 Subject: [PATCH 02/12] [improvement](fe) Automatically balance remote external scan splits ### What problem does this PR solve? Issue Number: N/A Related PR: #68726 Problem Summary: Independently planned queries for one remote split can keep selecting one backend. Enable automatic external scan spreading by default with value 0, using all eligible backends in the current compute group and minimum assigned weight with random ties. Preserve locality and single split execution. Value 1 restores original scheduling; explicit N retains bounded hash candidates. Other backend policy consumers remain unchanged. In a shared-host three-BE cache/hash-off SUM experiment at concurrency 64, three completed alternating pairs increased QPS from 13.867 to 40.506 (2.9211x). The first attempt at pair three had a gRPC keepalive failure in original mode; its one-error 6.10 QPS group is retained separately and the complete pair was repeated after recovery. This is not customer or physical three-host validation. No COUNT, DOP or CPU-routing optimization is added. ### Release note Remote external scans automatically spread across eligible backends by default. Set external_scan_consistent_hash_spread_num to 1 to restore original scheduling. Automatic file-cache scans can increase cache copies and warmup reads. Persisted global settings are retained on upgrade. ### Check List (For Author) - Test: Unit Test / Regression test / Manual test - Standard FE build and Checkstyle; 32 focused FE tests passed. - Two targeted regression suites framework-generated and verified. - 400 correct SUM query profiles verified one split and one target. - Three completed paired C64 SUM measurements; interruption disclosed above. - Behavior changed: Yes, default automatic external scheduling with explicit rollback. - Does this need documentation: Yes, docs/external-scan-consistent-hash-spread.md included. --- docs/external-scan-consistent-hash-spread.md | 48 +++++++---- .../datasource/scan/ExternalScanNode.java | 23 ++++-- .../scan/FederationBackendPolicy.java | 17 ++-- .../org/apache/doris/qe/SessionVariable.java | 12 +-- .../FederationBackendPolicySpreadTest.java | 82 ++++++++++++++++++- .../org/apache/doris/qe/VariableMgrTest.java | 8 +- ...t_external_scan_consistent_hash_spread.out | 15 ++++ ...l_scan_consistent_hash_spread_variable.out | 16 ++-- ...xternal_scan_consistent_hash_spread.groovy | 8 +- ...can_consistent_hash_spread_variable.groovy | 10 +-- 10 files changed, 186 insertions(+), 53 deletions(-) diff --git a/docs/external-scan-consistent-hash-spread.md b/docs/external-scan-consistent-hash-spread.md index becd745477310b..2f19ce3b033dac 100644 --- a/docs/external-scan-consistent-hash-spread.md +++ b/docs/external-scan-consistent-hash-spread.md @@ -5,31 +5,42 @@ under consistent hashing. Each query starts with zero assigned weight, and the original tie rule always chooses the same candidate. Increasing the original candidate count alone does not remove this bias. -## Enable the optional scheduling mode +## Automatic scheduling by default + +`external_scan_consistent_hash_spread_num` accepts integers from 0 through +2147483647. Its default is **0**. Remote external scans automatically spread +across all eligible backends in the query's compute group, regardless of the +number of backends. No manual hash setting is required. + +| Value | Behavior | +| --- | --- | +| 0 (default) | Balance across all eligible backends, using the existing hash ring when file cache or consistent hashing is enabled, otherwise the random strategy. | +| 1 | Preserve original scheduling, configured candidate counts and global redistribution. | +| N > 1 | Spread within at most N hash candidates when file cache or consistent hashing is enabled; otherwise preserve original round robin scheduling. | ```sql +-- Restore original behavior. +SET external_scan_consistent_hash_spread_num = 1; +-- Limit hash candidates while retaining spreading. SET use_consistent_hash_for_external_scan = true; SET external_scan_consistent_hash_spread_num = 3; +-- Restore automatic scheduling. +UNSET VARIABLE external_scan_consistent_hash_spread_num; ``` -`external_scan_consistent_hash_spread_num` accepts integers from 1 through -2147483647. Its default is **1**, which preserves the original scheduler, -including its configured hash candidate count and split redistribution. It does -not force the original scheduler to use just one candidate. `UNSET VARIABLE -external_scan_consistent_hash_spread_num` restores the session default. +Existing persisted global values are retained on upgrade. If an upgraded cluster +still has a global value of 1, an administrator can set the global value to 0 +for new sessions; existing sessions can use `UNSET VARIABLE` to restore the +compiled default or set their session value explicitly. -Values greater than 1 enable spreading only when an external scan uses consistent -hashing. This occurs when either `enable_file_cache` or -`use_consistent_hash_for_external_scan` is enabled. The new variable alone does -not switch a round robin scan to consistent hashing, and it does not enable file -caching. Other consumers of the backend policy, including file load and schema -scans, retain their existing behavior. +The variable does not enable file caching. Other consumers of the backend policy, +including file load and schema scans, retain their existing behavior. ## Assignment and locality -For a remote split without a preferred local backend, the scheduler takes the -first N distinct eligible backends from the existing consistent hash ring, capped -by the number of eligible backends in the query's compute group. It selects the +For a remote split without a preferred local backend, automatic mode considers +all eligible backends. Explicit N uses the first N distinct eligible backends +from the existing consistent hash ring, capped by the compute group size. It selects the backend with the least weight already assigned by this policy. Ties are broken uniformly at random. A fresh query can therefore choose a different backend for the same split, while multiple splits and batches share their policy's weight @@ -44,7 +55,7 @@ Global split redistribution is disabled in this mode: it could otherwise move a split outside its hash candidates or mandatory locality. Queries with many splits balance within each split's candidates, rather than across every eligible backend. Some split distributions can consequently be less balanced than with the original -global redistribution. The default mode retains that redistribution. +global redistribution. Setting the variable to 1 retains that redistribution. ## Capacity and cache tradeoffs @@ -55,8 +66,9 @@ multiplier. Frontend planning, remote storage, network capacity and shared host resources can still limit throughput. No CPU monitoring or shared query counter is introduced. -With file caching enabled, the same data can occupy cache space on multiple -backends and incur additional warmup reads. The variable bounds remote hash +With file caching enabled, automatic mode can store the same data on every +eligible backend and incur additional warmup reads. Use 1 to restore the original +cache placement or explicit N to cap hash candidates. The variable bounds remote hash candidates; it does not create cache replicas in advance. Locality takes precedence over this bound. To measure scheduling changes separately from cache hits, keep file caching disabled and enable consistent hashing explicitly on both sides of diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java index 7b1b9fb3d1cf8d..20269cd67c1c8a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java @@ -44,12 +44,23 @@ public abstract class ExternalScanNode extends ScanNode { // set to false means this scan node does not need to check column priv. protected boolean needCheckColumnPriv; - protected final FederationBackendPolicy backendPolicy = (ConnectContext.get() != null - && (ConnectContext.get().getSessionVariable().enableFileCache - || ConnectContext.get().getSessionVariable().getUseConsistentHashForExternalScan())) - ? new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING, - ConnectContext.get().getSessionVariable().getExternalScanConsistentHashSpreadNum()) - : new FederationBackendPolicy(); + protected final FederationBackendPolicy backendPolicy = createBackendPolicy(); + + private static FederationBackendPolicy createBackendPolicy() { + ConnectContext context = ConnectContext.get(); + if (context == null) { + return new FederationBackendPolicy(); + } + int spreadNum = context.getSessionVariable().getExternalScanConsistentHashSpreadNum(); + if (context.getSessionVariable().enableFileCache + || context.getSessionVariable().getUseConsistentHashForExternalScan()) { + return new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING, spreadNum); + } + if (spreadNum == 0) { + return new FederationBackendPolicy(NodeSelectionStrategy.RANDOM, spreadNum); + } + return new FederationBackendPolicy(); + } public ExternalScanNode(PlanNodeId id, TupleDescriptor desc, String planNodeName, ScanContext scanContext, boolean needCheckColumnPriv) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java index a57c1d9756239d..ffe8a32d901ff0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java @@ -156,7 +156,7 @@ public FederationBackendPolicy(NodeSelectionStrategy nodeSelectionStrategy, int @VisibleForTesting FederationBackendPolicy(NodeSelectionStrategy nodeSelectionStrategy, int consistentHashSpreadNum, IntUnaryOperator randomIndex) { - Preconditions.checkArgument(consistentHashSpreadNum > 0, "consistentHashSpreadNum must be at least 1"); + Preconditions.checkArgument(consistentHashSpreadNum >= 0, "consistentHashSpreadNum must be at least 0"); this.nodeSelectionStrategy = nodeSelectionStrategy; this.consistentHashSpreadNum = consistentHashSpreadNum; this.randomIndex = randomIndex; @@ -297,12 +297,14 @@ public Multimap computeScanRangeAssignment(List splits) t } case RANDOM: { randomCandidates.reset(); - candidateNodes = selectNodes(Config.split_assigner_min_random_candidate_num, randomCandidates); + candidateNodes = consistentHashSpreadNum == 0 ? backends + : selectNodes(Config.split_assigner_min_random_candidate_num, randomCandidates); break; } case CONSISTENT_HASHING: { candidateNodes = consistentHash.getNode(split, - isConsistentHashSpreadEnabled() ? Math.min(consistentHashSpreadNum, backends.size()) + consistentHashSpreadNum == 0 ? backends.size() + : isSpreadEnabled() ? Math.min(consistentHashSpreadNum, backends.size()) : Config.split_assigner_min_consistent_hash_candidate_num); break; } @@ -319,7 +321,7 @@ public Multimap computeScanRangeAssignment(List splits) t throw new UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG); } - Backend selectedBackend = isConsistentHashSpreadEnabled() && split.isRemotelyAccessible() + Backend selectedBackend = isSpreadEnabled() && split.isRemotelyAccessible() ? chooseNodeForSpread(candidateNodes) : chooseNodeForSplit(candidateNodes); List alternativeBackends = new ArrayList<>(candidateNodes); alternativeBackends.remove(selectedBackend); @@ -332,7 +334,7 @@ public Multimap computeScanRangeAssignment(List splits) t // Global redistribution can move a split outside its hash candidates or its locality constraints. // The spread mode balances weights within each split's candidates during initial assignment instead. - if (enableSplitsRedistribution && !isConsistentHashSpreadEnabled()) { + if (enableSplitsRedistribution && !isSpreadEnabled()) { equateDistribution(assignment); } return assignment; @@ -505,8 +507,9 @@ private Backend chooseNodeForSplit(List candidateNodes) { return chosenNode; } - private boolean isConsistentHashSpreadEnabled() { - return nodeSelectionStrategy == NodeSelectionStrategy.CONSISTENT_HASHING && consistentHashSpreadNum > 1; + private boolean isSpreadEnabled() { + return (nodeSelectionStrategy == NodeSelectionStrategy.CONSISTENT_HASHING && consistentHashSpreadNum != 1) + || (nodeSelectionStrategy == NodeSelectionStrategy.RANDOM && consistentHashSpreadNum == 0); } private Backend chooseNodeForSpread(List candidateNodes) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index 4148b516a33ba0..5c3f8bddfbcbbb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -1489,11 +1489,11 @@ public enum IgnoreSplitType { public static final String EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM = "external_scan_consistent_hash_spread_num"; @VarAttrDef.VarAttr(name = EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, checker = "checkExternalScanConsistentHashSpreadNum", needForward = true, - description = "Number of consistent hash candidates for spreading hot remote external splits. " - + "1 preserves the original scheduling; values above 1 enable random tie breaking within " - + "the lowest assigned weight candidates and disable global split redistribution. " - + "Requires file cache or use_consistent_hash_for_external_scan.") - public int externalScanConsistentHashSpreadNum = 1; + description = "0 automatically spreads remote external splits across eligible backends (default). " + + "1 preserves the original scheduling. Values above 1 bound consistent hash candidates " + + "when file cache or use_consistent_hash_for_external_scan is enabled. " + + "Spreading balances assigned weights with random ties and disables global redistribution.") + public int externalScanConsistentHashSpreadNum = 0; @VarAttrDef.VarAttr(name = PROFILE_LEVEL, fuzzy = false, needForward = true, setter = "setProfileLevel", checker = "checkProfileLevel", @@ -6417,7 +6417,7 @@ public int getExternalScanConsistentHashSpreadNum() { } public void checkExternalScanConsistentHashSpreadNum(String value) throws Exception { - checkFieldValue(EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, 1, value); + checkFieldValue(EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, 0, value); } public void setForceJniScanner(boolean force) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java index 15a3d0eef13f21..af84ecd8a1ddd3 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -327,7 +327,7 @@ public void testNoEligibleBackendFails() { @Test public void testInvalidSpreadCountFails() { Assertions.assertThrows(IllegalArgumentException.class, - () -> new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING, 0)); + () -> new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING, -1)); } @Test @@ -365,6 +365,86 @@ public void testExternalScanGateAndSessionIsolation() throws Exception { Assertions.assertEquals(1, (int) Deencapsulation.getField(noContext, "consistentHashSpreadNum")); } + @Test + public void testAutomaticModeSupportsAnyBackendCountAndBatches() throws Exception { + for (int count : Arrays.asList(1, 2, 3, 5)) { + backends = new ArrayList<>(); + for (int i = 0; i < count; i++) { + Backend backend = new Backend(NEXT_BACKEND_ID.getAndIncrement(), "192.0.2." + (i + 1), 9050); + backend.setAlive(true); + backends.add(backend); + } + for (NodeSelectionStrategy strategy : Arrays.asList(NodeSelectionStrategy.RANDOM, + NodeSelectionStrategy.CONSISTENT_HASHING)) { + Set targets = new HashSet<>(); + for (int index = 0; index < count; index++) { + FederationBackendPolicy policy = new FederationBackendPolicy(strategy, 0, selectCandidate(index)); + policy.init(); + targets.add(assign(policy, split())); + } + Assertions.assertEquals(new HashSet<>(backends), targets); + FederationBackendPolicy batched = new FederationBackendPolicy(strategy, 0, bound -> bound - 1); + batched.init(); + for (int batch = 0; batch < 2; batch++) { + for (int i = 0; i < count; i++) { + assign(batched, split()); + } + for (Backend backend : backends) { + Assertions.assertEquals((batch + 1) * 100L, + batched.getAssignedWeightPerBackend().get(backend)); + } + } + } + } + } + + @Test + public void testAutomaticModeRetainsEligibilityAndMandatoryLocality() throws Exception { + Backend unavailable = backends.get(2); + unavailable.setAlive(false); + Backend otherGroup = backends.remove(1); + for (NodeSelectionStrategy strategy : Arrays.asList(NodeSelectionStrategy.RANDOM, + NodeSelectionStrategy.CONSISTENT_HASHING)) { + FederationBackendPolicy policy = new FederationBackendPolicy(strategy, 0, bound -> bound - 1); + policy.init(); + Assertions.assertEquals(1, policy.numBackends()); + Assertions.assertEquals(backends.get(0), assign(policy, split())); + Assertions.assertFalse(policy.getBackends().contains(otherGroup)); + FileSplit local = new FileSplit(LocationPath.of("file:///hot.csv"), 0, 1000, 1000, + 0, new String[] {backends.get(0).getHost()}, Collections.emptyList()) { + @Override + public boolean isRemotelyAccessible() { + return false; + } + }; + Assertions.assertEquals(backends.get(0), assign(policy, local)); + local.setHosts(new String[] {otherGroup.getHost()}); + Assertions.assertThrows(UserException.class, () -> assign(policy, local)); + } + } + + @Test + public void testAutomaticExternalScanStrategyAndLegacyOptOut() { + SessionVariable session = context.getSessionVariable(); + Assertions.assertEquals(0, session.getExternalScanConsistentHashSpreadNum()); + for (boolean cache : Arrays.asList(false, true)) { + for (boolean hash : Arrays.asList(false, true)) { + session.enableFileCache = cache; + session.useConsistentHashForExternalScan = hash; + FederationBackendPolicy policy = new TestExternalScanNode().backendPolicy; + Assertions.assertEquals(cache || hash ? NodeSelectionStrategy.CONSISTENT_HASHING + : NodeSelectionStrategy.RANDOM, + Deencapsulation.getField(policy, "nodeSelectionStrategy")); + Assertions.assertEquals(0, (int) Deencapsulation.getField(policy, "consistentHashSpreadNum")); + } + } + session.enableFileCache = false; + session.useConsistentHashForExternalScan = false; + session.externalScanConsistentHashSpreadNum = 1; + Assertions.assertEquals(NodeSelectionStrategy.ROUND_ROBIN, + Deencapsulation.getField(new TestExternalScanNode().backendPolicy, "nodeSelectionStrategy")); + } + private static class TestExternalScanNode extends ExternalScanNode { TestExternalScanNode() { super(new PlanNodeId(0), new TupleDescriptor(new TupleId(0)), "test", ScanContext.EMPTY, false); diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java index 2a8b23fbe1da68..fe8d27b183b424 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java @@ -262,13 +262,13 @@ public void testAdaptiveBatchSizeRejectsZeroByteValues() { @Test public void testExternalScanConsistentHashSpreadVariable() throws Exception { SessionVariable var = new SessionVariable(); - Assertions.assertEquals(1, var.getExternalScanConsistentHashSpreadNum()); - for (String value : new String[] {"1", "2", "3", "2147483647"}) { + Assertions.assertEquals(0, var.getExternalScanConsistentHashSpreadNum()); + for (String value : new String[] {"0", "1", "2", "3", "2147483647"}) { VariableMgr.setVar(var, new SetVar(SetType.SESSION, SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, new StringLiteral(value))); Assertions.assertEquals(Integer.parseInt(value), var.getExternalScanConsistentHashSpreadNum()); } - for (String value : new String[] {"0", "-1", "2147483648", "abc"}) { + for (String value : new String[] {"-1", "2147483648", "abc"}) { Assertions.assertThrows(DdlException.class, () -> VariableMgr.setVar(var, new SetVar(SetType.SESSION, SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, new StringLiteral(value)))); Assertions.assertEquals(Integer.MAX_VALUE, var.getExternalScanConsistentHashSpreadNum()); @@ -278,7 +278,7 @@ public void testExternalScanConsistentHashSpreadVariable() throws Exception { Assertions.assertEquals(Integer.MAX_VALUE, forwarded.getExternalScanConsistentHashSpreadNum()); VariableMgr.setVar(var, new SetVar(SetType.SESSION, SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, null)); - Assertions.assertEquals(1, var.getExternalScanConsistentHashSpreadNum()); + Assertions.assertEquals(0, var.getExternalScanConsistentHashSpreadNum()); } } diff --git a/regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out b/regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out index 3f709e4d3fce8c..672d7229fffc2e 100644 --- a/regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out +++ b/regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out @@ -1,4 +1,19 @@ -- This file is automatically generated. You should know what you did if you want to edit this +-- !automatic -- +1 alpha +2 \N +3 gamma + +-- !automatic_cache -- +1 alpha +2 \N +3 gamma + +-- !remote -- +1 alpha +2 \N +3 gamma + -- !remote -- 1 alpha 2 \N diff --git a/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out b/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out index bd5f809d8bda42..62fdf47c1e79c7 100644 --- a/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out +++ b/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out @@ -1,16 +1,22 @@ -- This file is automatically generated. You should know what you did if you want to edit this -- !default -- -external_scan_consistent_hash_spread_num 1 1 0 +external_scan_consistent_hash_spread_num 0 0 0 -- !enabled -- -external_scan_consistent_hash_spread_num 3 1 1 +external_scan_consistent_hash_spread_num 3 0 1 -- !maximum -- -external_scan_consistent_hash_spread_num 2147483647 1 1 +external_scan_consistent_hash_spread_num 2147483647 0 1 + +-- !automatic -- +external_scan_consistent_hash_spread_num 0 0 0 + +-- !legacy -- +external_scan_consistent_hash_spread_num 1 0 1 -- !after_errors -- -external_scan_consistent_hash_spread_num 2147483647 1 1 +external_scan_consistent_hash_spread_num 1 0 1 -- !unset -- -external_scan_consistent_hash_spread_num 1 1 0 +external_scan_consistent_hash_spread_num 0 0 0 diff --git a/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy index b451e0cb9a379c..ca16e2ee20d715 100644 --- a/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy +++ b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy @@ -53,8 +53,14 @@ suite("test_external_scan_consistent_hash_spread", "p0,external") { sql "set enable_sql_cache = false" sql "set enable_query_cache = false" sql "set enable_file_cache = false" + sql "set use_consistent_hash_for_external_scan = false" + sql "unset variable external_scan_consistent_hash_spread_num" + order_qt_automatic remoteQuery("${prefix}single_*.parquet") + sql "set enable_file_cache = true" + order_qt_automatic_cache remoteQuery("${prefix}single_*.parquet") + sql "set enable_file_cache = false" sql "set use_consistent_hash_for_external_scan = true" - for (int candidates : [1, 2, 3, 2147483647]) { + for (int candidates : [0, 1, 2, 3, 2147483647]) { sql "set external_scan_consistent_hash_spread_num = ${candidates}" order_qt_remote remoteQuery("${prefix}single_*.parquet") } diff --git a/regression-test/suites/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.groovy b/regression-test/suites/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.groovy index d30c759ef0df0d..6a608af1adf994 100644 --- a/regression-test/suites/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.groovy +++ b/regression-test/suites/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.groovy @@ -22,13 +22,13 @@ suite("test_external_scan_consistent_hash_spread_variable", "p0") { sql "set external_scan_consistent_hash_spread_num = 2147483647" order_qt_maximum "show session variables like 'external_scan_consistent_hash_spread_num'" - test { - sql "set external_scan_consistent_hash_spread_num = 0" - exception "greater than or equal 1" - } + sql "set external_scan_consistent_hash_spread_num = 0" + order_qt_automatic "show session variables like 'external_scan_consistent_hash_spread_num'" + sql "set external_scan_consistent_hash_spread_num = 1" + order_qt_legacy "show session variables like 'external_scan_consistent_hash_spread_num'" test { sql "set external_scan_consistent_hash_spread_num = -1" - exception "greater than or equal 1" + exception "greater than or equal 0" } test { sql "set external_scan_consistent_hash_spread_num = 2147483648" From 7e4fb1c549d514a81fc8e5a6e53ab30bed2dd16c Mon Sep 17 00:00:00 2001 From: Dylan Date: Thu, 8 Oct 2026 23:10:10 +0800 Subject: [PATCH 03/12] [improvement](fe) Avoid unused alternative hosts for spread scans ### What problem does this PR solve? Issue Number: N/A Related PR: #68726 Problem Summary: Spread scheduling disables global split redistribution, but still builds and retains alternative Host lists for every split. The only consumer of these lists is redistribution. Automatic mode considers all eligible backends, so these unused lists retain O(split count * backend count) Host references and allocate unnecessary temporary candidate copies. Build alternative Host lists only for the original scheduling mode. Preserve candidate selection, assigned weights, locality and split execution unchanged. Add tests for multiple automatic hash/random splits without retained lists and an original-mode control that still retains its alternative Host. ### Release note None ### Check List (For Author) - Test: Unit Test / Regression test - 33 focused FE unit tests passed (23 spread policy, 9 original policy, 1 variable). - Standard FE build and Checkstyle passed. - Parameter and external scan regression suites verified on the rebuilt FE. - Behavior changed: No, remove unused allocation without changing scheduling. - Does this need documentation: No, scheduling and parameter semantics are unchanged. --- .../scan/FederationBackendPolicy.java | 11 ++++++---- .../FederationBackendPolicySpreadTest.java | 22 ++++++++++++++++++- 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java index ffe8a32d901ff0..5348889ef74a19 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java @@ -323,10 +323,13 @@ public Multimap computeScanRangeAssignment(List splits) t Backend selectedBackend = isSpreadEnabled() && split.isRemotelyAccessible() ? chooseNodeForSpread(candidateNodes) : chooseNodeForSplit(candidateNodes); - List alternativeBackends = new ArrayList<>(candidateNodes); - alternativeBackends.remove(selectedBackend); - split.setAlternativeHosts( - alternativeBackends.stream().map(each -> each.getHost()).collect(Collectors.toList())); + // Alternative hosts are used only by global redistribution, which spread mode disables. + if (!isSpreadEnabled()) { + List alternativeBackends = new ArrayList<>(candidateNodes); + alternativeBackends.remove(selectedBackend); + split.setAlternativeHosts( + alternativeBackends.stream().map(each -> each.getHost()).collect(Collectors.toList())); + } assignment.put(selectedBackend, split); assignedWeightPerBackend.put(selectedBackend, assignedWeightPerBackend.get(selectedBackend) + split.getSplitWeight().getRawValue()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java index af84ecd8a1ddd3..55b931267f3044 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -131,7 +131,7 @@ public void testIndependentHotQueriesCanChooseEveryCandidateExactlyOnce() throws Backend target = assign(policy, split); Assertions.assertEquals(candidates.get(index), target); selected.add(target); - Assertions.assertEquals(2, split.getAlternativeHosts().size()); + Assertions.assertNull(split.getAlternativeHosts()); } Assertions.assertEquals(new HashSet<>(backends), selected); } @@ -423,6 +423,26 @@ public boolean isRemotelyAccessible() { } } + @Test + public void testSpreadDoesNotRetainUnusedAlternativeHosts() throws Exception { + for (NodeSelectionStrategy strategy : Arrays.asList(NodeSelectionStrategy.RANDOM, + NodeSelectionStrategy.CONSISTENT_HASHING)) { + FederationBackendPolicy policy = new FederationBackendPolicy(strategy, 0, bound -> bound - 1); + policy.init(); + List splits = new ArrayList<>(); + for (int i = 0; i < 30; i++) { + splits.add(split()); + } + Assertions.assertEquals(30, policy.computeScanRangeAssignment(splits).size()); + for (Split split : splits) { + Assertions.assertNull(split.getAlternativeHosts()); + } + } + FileSplit legacySplit = split(); + assign(policy(1, bound -> bound - 1), legacySplit); + Assertions.assertEquals(1, legacySplit.getAlternativeHosts().size()); + } + @Test public void testAutomaticExternalScanStrategyAndLegacyOptOut() { SessionVariable session = context.getSessionVariable(); From 813826a7febf0ea08fd99104017247da106e9673 Mon Sep 17 00:00:00 2001 From: Dylan Date: Fri, 9 Oct 2026 11:39:40 +0800 Subject: [PATCH 04/12] [improvement](fe) Preserve original external scan scheduling by default ### What problem does this PR solve? Issue Number: N/A Related PR: #68726 Problem Summary: Keep the external scan spreading option at 1 by default so existing scheduling, candidate counts and redistribution are preserved. Explain that 1 is a mode rather than a single-backend limit. Automatic spreading remains available with an explicit 0. Update default-value and strategy tests, explicitly enable automatic mode in the external regression, and remove user documentation from the source repository in favor of the separate website documentation change. ### Release note Preserve original external scan scheduling by default. Enable automatic spreading explicitly with external_scan_consistent_hash_spread_num = 0, or bound hash candidates with a value greater than 1. ### Check List (For Author) - Test: 33 focused FE unit tests; standard clean FE build and Checkstyle; isolated parameter and external scan regressions. Variable output generated by the regression framework. - Behavior changed: Yes; restore the default from 0 to 1 compared with the previous PR revision. Explicit spreading algorithms are unchanged. - Does this need documentation: Yes; English and Chinese website documentation prepared separately, not published yet. --- docs/external-scan-consistent-hash-spread.md | 75 ------------------- .../org/apache/doris/qe/SessionVariable.java | 8 +- .../FederationBackendPolicySpreadTest.java | 16 +++- .../org/apache/doris/qe/VariableMgrTest.java | 4 +- ...l_scan_consistent_hash_spread_variable.out | 14 ++-- ...xternal_scan_consistent_hash_spread.groovy | 2 +- 6 files changed, 29 insertions(+), 90 deletions(-) delete mode 100644 docs/external-scan-consistent-hash-spread.md diff --git a/docs/external-scan-consistent-hash-spread.md b/docs/external-scan-consistent-hash-spread.md deleted file mode 100644 index 2f19ce3b033dac..00000000000000 --- a/docs/external-scan-consistent-hash-spread.md +++ /dev/null @@ -1,75 +0,0 @@ -# Spread hot external scan splits across consistent hash candidates - -Repeated queries for the same remote file split can all choose the same backend -under consistent hashing. Each query starts with zero assigned weight, and the -original tie rule always chooses the same candidate. Increasing the original -candidate count alone does not remove this bias. - -## Automatic scheduling by default - -`external_scan_consistent_hash_spread_num` accepts integers from 0 through -2147483647. Its default is **0**. Remote external scans automatically spread -across all eligible backends in the query's compute group, regardless of the -number of backends. No manual hash setting is required. - -| Value | Behavior | -| --- | --- | -| 0 (default) | Balance across all eligible backends, using the existing hash ring when file cache or consistent hashing is enabled, otherwise the random strategy. | -| 1 | Preserve original scheduling, configured candidate counts and global redistribution. | -| N > 1 | Spread within at most N hash candidates when file cache or consistent hashing is enabled; otherwise preserve original round robin scheduling. | - -```sql --- Restore original behavior. -SET external_scan_consistent_hash_spread_num = 1; --- Limit hash candidates while retaining spreading. -SET use_consistent_hash_for_external_scan = true; -SET external_scan_consistent_hash_spread_num = 3; --- Restore automatic scheduling. -UNSET VARIABLE external_scan_consistent_hash_spread_num; -``` - -Existing persisted global values are retained on upgrade. If an upgraded cluster -still has a global value of 1, an administrator can set the global value to 0 -for new sessions; existing sessions can use `UNSET VARIABLE` to restore the -compiled default or set their session value explicitly. - -The variable does not enable file caching. Other consumers of the backend policy, -including file load and schema scans, retain their existing behavior. - -## Assignment and locality - -For a remote split without a preferred local backend, automatic mode considers -all eligible backends. Explicit N uses the first N distinct eligible backends -from the existing consistent hash ring, capped by the compute group size. It selects the -backend with the least weight already assigned by this policy. Ties are broken -uniformly at random. A fresh query can therefore choose a different backend for -the same split, while multiple splits and batches share their policy's weight -history. Each split is assigned for execution exactly once. - -The mode preserves local preference and mandatory Host constraints for splits -that cannot be accessed remotely. These constraints can choose a backend outside -the remote split's hash candidates. Different backends on the same Host remain -distinct hash candidates. - -Global split redistribution is disabled in this mode: it could otherwise move a -split outside its hash candidates or mandatory locality. Queries with many splits -balance within each split's candidates, rather than across every eligible backend. -Some split distributions can consequently be less balanced than with the original -global redistribution. Setting the variable to 1 retains that redistribution. - -## Capacity and cache tradeoffs - -Spreading independently planned queries can increase throughput when the original -hot backend is saturated and the other candidate backends have available capacity. -Random choice does not guarantee a particular assignment sequence or throughput -multiplier. Frontend planning, remote storage, network capacity and shared host -resources can still limit throughput. No CPU monitoring or shared query counter is -introduced. - -With file caching enabled, automatic mode can store the same data on every -eligible backend and incur additional warmup reads. Use 1 to restore the original -cache placement or explicit N to cap hash candidates. The variable bounds remote hash -candidates; it does not create cache replicas in advance. Locality takes precedence -over this bound. To measure scheduling changes separately from cache hits, keep -file caching disabled and enable consistent hashing explicitly on both sides of -a comparison. diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index 5c3f8bddfbcbbb..fc26fd0a34b92c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -1489,11 +1489,13 @@ public enum IgnoreSplitType { public static final String EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM = "external_scan_consistent_hash_spread_num"; @VarAttrDef.VarAttr(name = EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, checker = "checkExternalScanConsistentHashSpreadNum", needForward = true, - description = "0 automatically spreads remote external splits across eligible backends (default). " - + "1 preserves the original scheduling. Values above 1 bound consistent hash candidates " + description = "1 preserves the original scheduling, candidate counts and redistribution (default). " + + "It does not limit scans to one backend. " + + "0 enables automatic spreading across all eligible backends in the compute group. " + + "Values above 1 bound consistent hash candidates " + "when file cache or use_consistent_hash_for_external_scan is enabled. " + "Spreading balances assigned weights with random ties and disables global redistribution.") - public int externalScanConsistentHashSpreadNum = 0; + public int externalScanConsistentHashSpreadNum = 1; @VarAttrDef.VarAttr(name = PROFILE_LEVEL, fuzzy = false, needForward = true, setter = "setProfileLevel", checker = "checkProfileLevel", diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java index 55b931267f3044..cccb0740ddc85f 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -444,9 +444,21 @@ public void testSpreadDoesNotRetainUnusedAlternativeHosts() throws Exception { } @Test - public void testAutomaticExternalScanStrategyAndLegacyOptOut() { + public void testLegacyDefaultAndAutomaticExternalScanStrategy() { SessionVariable session = context.getSessionVariable(); - Assertions.assertEquals(0, session.getExternalScanConsistentHashSpreadNum()); + Assertions.assertEquals(1, session.getExternalScanConsistentHashSpreadNum()); + for (boolean cache : Arrays.asList(false, true)) { + for (boolean hash : Arrays.asList(false, true)) { + session.enableFileCache = cache; + session.useConsistentHashForExternalScan = hash; + FederationBackendPolicy policy = new TestExternalScanNode().backendPolicy; + Assertions.assertEquals(cache || hash ? NodeSelectionStrategy.CONSISTENT_HASHING + : NodeSelectionStrategy.ROUND_ROBIN, + Deencapsulation.getField(policy, "nodeSelectionStrategy")); + Assertions.assertEquals(1, (int) Deencapsulation.getField(policy, "consistentHashSpreadNum")); + } + } + session.externalScanConsistentHashSpreadNum = 0; for (boolean cache : Arrays.asList(false, true)) { for (boolean hash : Arrays.asList(false, true)) { session.enableFileCache = cache; diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java index fe8d27b183b424..5f2001925f3e32 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java @@ -262,7 +262,7 @@ public void testAdaptiveBatchSizeRejectsZeroByteValues() { @Test public void testExternalScanConsistentHashSpreadVariable() throws Exception { SessionVariable var = new SessionVariable(); - Assertions.assertEquals(0, var.getExternalScanConsistentHashSpreadNum()); + Assertions.assertEquals(1, var.getExternalScanConsistentHashSpreadNum()); for (String value : new String[] {"0", "1", "2", "3", "2147483647"}) { VariableMgr.setVar(var, new SetVar(SetType.SESSION, SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, new StringLiteral(value))); @@ -278,7 +278,7 @@ public void testExternalScanConsistentHashSpreadVariable() throws Exception { Assertions.assertEquals(Integer.MAX_VALUE, forwarded.getExternalScanConsistentHashSpreadNum()); VariableMgr.setVar(var, new SetVar(SetType.SESSION, SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, null)); - Assertions.assertEquals(0, var.getExternalScanConsistentHashSpreadNum()); + Assertions.assertEquals(1, var.getExternalScanConsistentHashSpreadNum()); } } diff --git a/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out b/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out index 62fdf47c1e79c7..663ad55a92027d 100644 --- a/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out +++ b/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out @@ -1,22 +1,22 @@ -- This file is automatically generated. You should know what you did if you want to edit this -- !default -- -external_scan_consistent_hash_spread_num 0 0 0 +external_scan_consistent_hash_spread_num 1 1 0 -- !enabled -- -external_scan_consistent_hash_spread_num 3 0 1 +external_scan_consistent_hash_spread_num 3 1 1 -- !maximum -- -external_scan_consistent_hash_spread_num 2147483647 0 1 +external_scan_consistent_hash_spread_num 2147483647 1 1 -- !automatic -- -external_scan_consistent_hash_spread_num 0 0 0 +external_scan_consistent_hash_spread_num 0 1 1 -- !legacy -- -external_scan_consistent_hash_spread_num 1 0 1 +external_scan_consistent_hash_spread_num 1 1 0 -- !after_errors -- -external_scan_consistent_hash_spread_num 1 0 1 +external_scan_consistent_hash_spread_num 1 1 0 -- !unset -- -external_scan_consistent_hash_spread_num 0 0 0 +external_scan_consistent_hash_spread_num 1 1 0 diff --git a/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy index ca16e2ee20d715..ec1316f9e09d76 100644 --- a/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy +++ b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy @@ -54,7 +54,7 @@ suite("test_external_scan_consistent_hash_spread", "p0,external") { sql "set enable_query_cache = false" sql "set enable_file_cache = false" sql "set use_consistent_hash_for_external_scan = false" - sql "unset variable external_scan_consistent_hash_spread_num" + sql "set external_scan_consistent_hash_spread_num = 0" order_qt_automatic remoteQuery("${prefix}single_*.parquet") sql "set enable_file_cache = true" order_qt_automatic_cache remoteQuery("${prefix}single_*.parquet") From fe83abb6008f5440a464ecc32e0cc78963e7e2ef Mon Sep 17 00:00:00 2001 From: Dylan Date: Sat, 10 Oct 2026 01:35:22 +0800 Subject: [PATCH 05/12] [improvement](fe) Add stable identities for external scan splits ### What problem does this PR solve? Problem Summary: External scan scheduling can hash pathless connector ranges to the same key, and repeated locality hints can keep remote splits on one backend. Add connector-provided split identities and balance spread candidates by assigned weight while preserving the default scheduling behavior. ### Release note None ### Check List (For Author) - Test: Connector SPI surface and metadata tests; focused FE UT attempted but blocked by the read-only Maven repository. - Behavior changed: Yes (opt-in external scan spread scheduling). - Does this need documentation: No --- .../doris/connector/adbc/AdbcScanRange.java | 6 +++++ .../connector/adbc/AdbcScanRangeTest.java | 16 +++++++++++++ .../doris/connector/fluss/FlussLakeRange.java | 6 +++++ .../doris/connector/fluss/FlussScanRange.java | 10 ++++++++ .../connector/fluss/FlussScanRangeTest.java | 2 ++ .../connector/paimon/PaimonScanRange.java | 5 ++++ .../spi/scan/ConnectorScanRange.java | 8 +++++++ .../spi/ConnectorPluginSurfaceTest.java | 6 ++--- .../doris/connector/trino/TrinoScanRange.java | 5 ++++ fe/fe-connector/pom.xml | 2 +- .../doris/source/RemoteDorisSplit.java | 5 ++++ .../scan/FederationBackendPolicy.java | 23 ++++++++++++++----- .../datasource/split/PluginDrivenSplit.java | 17 ++++++++++++++ .../org/apache/doris/qe/SessionVariable.java | 1 + .../main/java/org/apache/doris/spi/Split.java | 2 +- .../FederationBackendPolicySpreadTest.java | 17 ++++++++++++++ .../PluginDrivenSplitPartitionValuesTest.java | 22 ++++++++++++++++++ .../org/apache/doris/qe/VariableMgrTest.java | 6 +++++ 18 files changed, 148 insertions(+), 11 deletions(-) diff --git a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcScanRange.java b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcScanRange.java index a52d0b307faca4..5b1a2efa73047f 100644 --- a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcScanRange.java +++ b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcScanRange.java @@ -88,6 +88,12 @@ public String getTableFormatType() { return TABLE_FORMAT_TYPE; } + @Override + public String getSplitIdentity() { + return properties.getOrDefault(PARAM_PARTITION_DESCRIPTOR, + properties.getOrDefault(PARAM_QUERY_SQL, TABLE_FORMAT_TYPE)); + } + @Override public Map getProperties() { return properties; diff --git a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcScanRangeTest.java b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcScanRangeTest.java index c25c2d0f4f2056..394e76ac1bcfc9 100644 --- a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcScanRangeTest.java +++ b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcScanRangeTest.java @@ -106,6 +106,22 @@ void carriesAPartitionDescriptorInsteadOfAStatement() { Assertions.assertFalse(params.containsKey("query_sql")); } + @Test + void usesPartitionDescriptorAsSchedulingIdentity() { + AdbcScanRange first = new AdbcScanRange.Builder() + .driverPath("/opt/doris/plugins/adbc_drivers/libadbc_driver_flightsql.so") + .uri("grpc://remote:9090") + .partitionDescriptor("partition-1") + .build(); + AdbcScanRange second = new AdbcScanRange.Builder() + .driverPath("/opt/doris/plugins/adbc_drivers/libadbc_driver_flightsql.so") + .uri("grpc://remote:9090") + .partitionDescriptor("partition-2") + .build(); + + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + } + @Test void refusesToCarryBothKindsOfWorkOrNeither() { // The two are alternatives, and BE rejects a range that says both or neither. Failing while diff --git a/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java b/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java index 549d7fb5e48b2a..dfd5d09297c1ae 100644 --- a/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java +++ b/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java @@ -178,6 +178,12 @@ public Optional getPath() { return inner.getPath(); } + @Override + public String getSplitIdentity() { + return inner.getSplitIdentity() + "\u0001" + + (tail == null ? RANGE_TYPE_LAKE : tail.encode()); + } + @Override public long getStart() { return inner.getStart(); diff --git a/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussScanRange.java b/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussScanRange.java index 0f720b91863607..009a7acb303de1 100644 --- a/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussScanRange.java +++ b/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussScanRange.java @@ -177,6 +177,16 @@ public String getTableFormatType() { return TABLE_FORMAT; } + @Override + public String getSplitIdentity() { + return properties.getOrDefault(PROP_RANGE_TYPE, "") + "\u0001" + + properties.getOrDefault(PROP_PARTITION_ID, "") + "\u0001" + + properties.getOrDefault(PROP_BUCKET_ID, "") + "\u0001" + + properties.getOrDefault(PROP_LOG_START_OFFSET, "") + "\u0001" + + properties.getOrDefault(PROP_LOG_STOP_OFFSET, "") + "\u0001" + + properties.getOrDefault(PROP_KV_SNAPSHOT_ID, ""); + } + @Override public Map getProperties() { return properties; diff --git a/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussScanRangeTest.java b/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussScanRangeTest.java index 99b4a058a99a60..8089638b3da60e 100644 --- a/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussScanRangeTest.java +++ b/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussScanRangeTest.java @@ -50,6 +50,7 @@ public class FlussScanRangeTest { @Test public void logRangeCarriesOffsetsAndNothingElse() { FlussScanRange range = FlussScanRange.log(FlussScanRange.Partition.NONE, 3, 10L, 42L); + FlussScanRange otherRange = FlussScanRange.log(FlussScanRange.Partition.NONE, 4, 10L, 42L); Map expected = new LinkedHashMap<>(); expected.put("fluss.range_type", "LOG"); @@ -58,6 +59,7 @@ public void logRangeCarriesOffsetsAndNothingElse() { expected.put("fluss.log_stop_offset", "42"); Assertions.assertEquals(expected, range.getProperties()); Assertions.assertEquals(FlussScanRange.RangeType.LOG, range.getRangeType()); + Assertions.assertNotEquals(range.getSplitIdentity(), otherRange.getSplitIdentity()); } @Test diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanRange.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanRange.java index 77c0d073659bfa..ecb351d61c1fd7 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanRange.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanRange.java @@ -125,6 +125,11 @@ public Optional getPath() { return Optional.ofNullable(path); } + @Override + public String getSplitIdentity() { + return path != null ? path : properties.getOrDefault("paimon.split", "paimon"); + } + @Override public long getStart() { return start; diff --git a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java index 6778bd3ffcba8d..0510f739860467 100644 --- a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java +++ b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java @@ -46,6 +46,14 @@ default Optional getPath() { return Optional.empty(); } + /** + * Returns a stable identity used to assign connector ranges to external-scan backends. Connectors whose JNI + * ranges do not have a path should override this with their serialized split identity. + */ + default String getSplitIdentity() { + return getPath().orElse(""); + } + /** Returns the byte offset to start reading from. */ default long getStart() { return 0; diff --git a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java index a2b20ae311d973..f79f3d07d509af 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java +++ b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java @@ -87,9 +87,9 @@ public void connectorApiMajorTracksTheRecordedSurfaceChange() throws IOException Assertions.assertNotNull(in, "missing connector plugin API version resource"); version.load(in); } - // Major 12 adds the SUPPORTS_FIELD_ID_ACCESS_PATH and SUPPORTS_SYS_TABLE_NESTED_COLUMN_PRUNE - // capabilities: a plugin naming either constant cannot link against an older FE. - Assertions.assertEquals("12.0", version.getProperty("api.version")); + // Major 13 adds ConnectorScanRange#getSplitIdentity: plugins compiled against the previous + // connector surface cannot link against an FE that exposes the new scheduling identity hook. + Assertions.assertEquals("13.0", version.getProperty("api.version")); } /** Root entry points plus provider/handle types returned to connector plugins. */ diff --git a/fe/fe-connector/fe-connector-trino/src/main/java/org/apache/doris/connector/trino/TrinoScanRange.java b/fe/fe-connector/fe-connector-trino/src/main/java/org/apache/doris/connector/trino/TrinoScanRange.java index 0f4d363b09052f..ebac65fa310690 100644 --- a/fe/fe-connector/fe-connector-trino/src/main/java/org/apache/doris/connector/trino/TrinoScanRange.java +++ b/fe/fe-connector/fe-connector-trino/src/main/java/org/apache/doris/connector/trino/TrinoScanRange.java @@ -79,6 +79,11 @@ public String getTableFormatType() { return "trino_connector"; } + @Override + public String getSplitIdentity() { + return properties.getOrDefault(KEY_SPLIT, "trino_connector"); + } + @Override public Map getProperties() { return properties; diff --git a/fe/fe-connector/pom.xml b/fe/fe-connector/pom.xml index 121868e3dd830f..9bf53c1050bfa3 100644 --- a/fe/fe-connector/pom.xml +++ b/fe/fe-connector/pom.xml @@ -55,7 +55,7 @@ under the License. of the latter two means bumping this property as well (and fe-extension-spi means bumping all five families). --> - 12.0 + 13.0 diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java index 3cccf8821f9257..e056cf9ba871b9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java @@ -43,6 +43,11 @@ public String getLocation() { return location; } + @Override + public String getSplitIdentity() { + return location; + } + public TableFormatType getTableFormatType() { return tableFormatType; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java index 5348889ef74a19..2c045fbf6e1a1a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java @@ -269,7 +269,15 @@ public Multimap computeScanRangeAssignment(List splits) t Optional chosenNode = candidateNodes.stream() .min(Comparator.comparingLong(ownerNode -> assignedWeightPerBackend.get(ownerNode))); - if (chosenNode.isPresent()) { + // A host hint is a local preference for remotely readable splits. In spread mode, do not + // let a repeated hint monopolize one backend while other eligible backends are idle. + boolean localNodeIsWithinGlobalMinimum = chosenNode.isPresent(); + if (localNodeIsWithinGlobalMinimum && isSpreadEnabled()) { + long minimumAssignedWeight = Collections.min(assignedWeightPerBackend.values()); + localNodeIsWithinGlobalMinimum = assignedWeightPerBackend.get(chosenNode.get()) + <= minimumAssignedWeight; + } + if (localNodeIsWithinGlobalMinimum) { Backend selectedBackend = chosenNode.get(); assignment.put(selectedBackend, split); assignedWeightPerBackend.put(selectedBackend, @@ -302,10 +310,13 @@ public Multimap computeScanRangeAssignment(List splits) t break; } case CONSISTENT_HASHING: { - candidateNodes = consistentHash.getNode(split, - consistentHashSpreadNum == 0 ? backends.size() - : isSpreadEnabled() ? Math.min(consistentHashSpreadNum, backends.size()) - : Config.split_assigner_min_consistent_hash_candidate_num); + if (consistentHashSpreadNum == 0) { + candidateNodes = backends; + } else { + candidateNodes = consistentHash.getNode(split, + isSpreadEnabled() ? Math.min(consistentHashSpreadNum, backends.size()) + : Config.split_assigner_min_consistent_hash_candidate_num); + } break; } default: { @@ -551,7 +562,7 @@ public void funnel(Backend backend, PrimitiveSink primitiveSink) { private static class SplitHash implements Funnel { @Override public void funnel(Split split, PrimitiveSink primitiveSink) { - primitiveSink.putBytes(split.getConsistentHashString().getBytes(StandardCharsets.UTF_8)); + primitiveSink.putBytes(split.getSplitIdentity().getBytes(StandardCharsets.UTF_8)); primitiveSink.putLong(split.getStart()); primitiveSink.putLong(split.getLength()); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/PluginDrivenSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/PluginDrivenSplit.java index 2504b09adf0b9f..85a8c2b2c270a7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/PluginDrivenSplit.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/PluginDrivenSplit.java @@ -35,6 +35,7 @@ public class PluginDrivenSplit extends FileSplit { private final ConnectorScanRange connectorScanRange; + private final String splitIdentity; public PluginDrivenSplit(ConnectorScanRange scanRange) { super(buildPath(scanRange), @@ -45,6 +46,7 @@ public PluginDrivenSplit(ConnectorScanRange scanRange) { scanRange.getHosts().toArray(new String[0]), buildPartitionValues(scanRange)); this.connectorScanRange = scanRange; + this.splitIdentity = buildSplitIdentity(scanRange); // FIX-A1: thread the connector's proportional split weight into the FileSplit scheduling fields so // FederationBackendPolicy distributes by size (legacy parity) instead of uniform standard() weight. // Set ONLY when the connector provides BOTH a weight (>= 0; 0 is a real weight) and a positive @@ -73,6 +75,21 @@ public String getPathString() { return connectorScanRange.getPath().orElse("connector://virtual"); } + /** + * Returns a stable connector supplied identity for consistent-hash scheduling. JNI ranges often have no + * file path or byte range, so include their immutable connector metadata instead of collapsing all of them + * onto the same virtual path. + */ + @Override + public String getSplitIdentity() { + return splitIdentity; + } + + private static String buildSplitIdentity(ConnectorScanRange scanRange) { + String identity = scanRange.getSplitIdentity(); + return identity.isEmpty() ? scanRange.getTableFormatType() : identity; + } + private static LocationPath buildPath(ConnectorScanRange scanRange) { String pathStr = scanRange.getPath().orElse("connector://virtual"); return LocationPath.of(pathStr); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index fc26fd0a34b92c..9e6a59c623ee8a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -1483,6 +1483,7 @@ public enum IgnoreSplitType { public static final String USE_CONSISTENT_HASHING_FOR_EXTERNAL_SCAN = "use_consistent_hash_for_external_scan"; @VarAttrDef.VarAttr(name = USE_CONSISTENT_HASHING_FOR_EXTERNAL_SCAN, + needForward = true, description = "Use consistent hashing to split the appearance for external scan") public boolean useConsistentHashForExternalScan = false; diff --git a/fe/fe-core/src/main/java/org/apache/doris/spi/Split.java b/fe/fe-core/src/main/java/org/apache/doris/spi/Split.java index 48b6acd68e39c2..a5aa55186a237d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/spi/Split.java +++ b/fe/fe-core/src/main/java/org/apache/doris/spi/Split.java @@ -48,7 +48,7 @@ default boolean isRemotelyAccessible() { void setAlternativeHosts(List alternativeHosts); - default String getConsistentHashString() { + default String getSplitIdentity() { return getPathString(); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java index cccb0740ddc85f..08f3f2620b4420 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -443,6 +443,23 @@ public void testSpreadDoesNotRetainUnusedAlternativeHosts() throws Exception { Assertions.assertEquals(1, legacySplit.getAlternativeHosts().size()); } + @Test + public void testRepeatedRemoteHostHintDoesNotMonopolizeSpreadMode() throws Exception { + FederationBackendPolicy policy = policy(3, bound -> 0); + List splits = new ArrayList<>(); + String localHost = backends.get(0).getHost(); + for (int i = 0; i < 30; i++) { + splits.add(new FileSplit(LocationPath.of("s3://bucket/file-" + i), 0, 1000, 1000, + 0, new String[] {localHost}, Collections.emptyList())); + } + + Multimap assignment = policy.computeScanRangeAssignment(splits); + + Assertions.assertTrue(assignment.keySet().size() > 1); + Assertions.assertTrue(assignment.containsKey(backends.get(0)), + "the first local preference should remain eligible in spread mode"); + } + @Test public void testLegacyDefaultAndAutomaticExternalScanStrategy() { SessionVariable session = context.getSessionVariable(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/PluginDrivenSplitPartitionValuesTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/PluginDrivenSplitPartitionValuesTest.java index 2deb928f853a7e..b17ecfc0ad99cc 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/PluginDrivenSplitPartitionValuesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/PluginDrivenSplitPartitionValuesTest.java @@ -93,4 +93,26 @@ public void nonEmptyMapYieldsValuesInOrder() { PluginDrivenSplit split = new PluginDrivenSplit(range(parts, true)); Assertions.assertEquals(java.util.Arrays.asList("1", "2"), split.getPartitionValues()); } + + @Test + public void connectorMetadataParticipatesInConsistentHashIdentity() { + PluginDrivenSplit first = new PluginDrivenSplit(rangeWithProperty("paimon.split", "split-1")); + PluginDrivenSplit second = new PluginDrivenSplit(rangeWithProperty("paimon.split", "split-2")); + + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + } + + private static ConnectorScanRange rangeWithProperty(String key, String value) { + return new ConnectorScanRange() { + @Override + public Map getProperties() { + return Collections.singletonMap(key, value); + } + + @Override + public String getSplitIdentity() { + return value; + } + }; + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java index 5f2001925f3e32..b79b5097660374 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java @@ -262,6 +262,12 @@ public void testAdaptiveBatchSizeRejectsZeroByteValues() { @Test public void testExternalScanConsistentHashSpreadVariable() throws Exception { SessionVariable var = new SessionVariable(); + VariableMgr.setVar(var, new SetVar(SetType.SESSION, + SessionVariable.USE_CONSISTENT_HASHING_FOR_EXTERNAL_SCAN, new StringLiteral("true"))); + SessionVariable hashForwarded = new SessionVariable(); + hashForwarded.setForwardedSessionVariables(var.getForwardVariables()); + Assertions.assertTrue(hashForwarded.getUseConsistentHashForExternalScan()); + Assertions.assertEquals(1, var.getExternalScanConsistentHashSpreadNum()); for (String value : new String[] {"0", "1", "2", "3", "2147483647"}) { VariableMgr.setVar(var, new SetVar(SetType.SESSION, From 886cc62894c82256cef04a145c9593c5c5003a01 Mon Sep 17 00:00:00 2001 From: Dylan Date: Sat, 10 Oct 2026 01:42:30 +0800 Subject: [PATCH 06/12] [fix](fe) Avoid full-ring scans for spread candidates ### What problem does this PR solve? Problem Summary: Full-ring construction is unnecessary when spread mode already allows every eligible backend, and the external scan regression fixture used an unrecognized path-style property. Use the filtered backend list directly and align both OUTFILE exports with the recognized property. ### Release note None ### Check List (For Author) - Test: git diff --check; regression fixture review. - Behavior changed: No. - Does this need documentation: No --- .../doris/datasource/scan/FederationBackendPolicy.java | 8 +++++--- .../tvf/test_external_scan_consistent_hash_spread.groovy | 4 ++-- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java index 2c045fbf6e1a1a..4f86929565b0ba 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java @@ -313,9 +313,11 @@ public Multimap computeScanRangeAssignment(List splits) t if (consistentHashSpreadNum == 0) { candidateNodes = backends; } else { - candidateNodes = consistentHash.getNode(split, - isSpreadEnabled() ? Math.min(consistentHashSpreadNum, backends.size()) - : Config.split_assigner_min_consistent_hash_candidate_num); + int candidateCount = isSpreadEnabled() + ? Math.min(consistentHashSpreadNum, backends.size()) + : Config.split_assigner_min_consistent_hash_candidate_num; + candidateNodes = candidateCount >= backends.size() ? backends + : consistentHash.getNode(split, candidateCount); } break; } diff --git a/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy index ec1316f9e09d76..bdf76c6208ca9f 100644 --- a/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy +++ b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy @@ -37,7 +37,7 @@ suite("test_external_scan_consistent_hash_spread", "p0,external") { properties ( "s3.endpoint" = "${endpoint}", "s3.region" = "${region}", "s3.access_key" = "${ak}", "s3.secret_key" = "${sk}", - "s3.path_style_access" = "${pathStyle}" + "use_path_style" = "${pathStyle}" ) """ @@ -78,7 +78,7 @@ suite("test_external_scan_consistent_hash_spread", "p0,external") { properties ( "s3.endpoint" = "${endpoint}", "s3.region" = "${region}", "s3.access_key" = "${ak}", "s3.secret_key" = "${sk}", - "s3.path_style_access" = "${pathStyle}" + "use_path_style" = "${pathStyle}" ) """ sql "set use_consistent_hash_for_external_scan = true" From bda2d26826f412b279213fcac80cfa9421d52905 Mon Sep 17 00:00:00 2001 From: Dylan Date: Sat, 10 Oct 2026 10:07:15 +0800 Subject: [PATCH 07/12] [test](fe) Fix external scan scheduling test setup --- .../datasource/scan/FederationBackendPolicySpreadTest.java | 4 ++-- .../scan/PluginDrivenScanNodeScanProviderSelectionTest.java | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java index 08f3f2620b4420..daf63688ea65cd 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -117,8 +117,8 @@ private Backend assign(FederationBackendPolicy policy, Split split) throws UserE } private static IntUnaryOperator selectCandidate(int index) { - // Replace the reservoir through the desired index, then keep that node. - return bound -> bound <= index + 1 ? 0 : bound - 1; + // Select only when the reservoir reaches the desired candidate index. + return bound -> bound == index + 1 ? 0 : bound - 1; } @Test diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java index 6345c869de1cda..575d23915bc44d 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java @@ -64,6 +64,7 @@ public void mixedNativeRangesKeepParquetScannerContract() { TFileScanRangeParams params = new TFileScanRangeParams(); Deencapsulation.setField(node, "params", params); ConnectorScanRange scanRange = Mockito.mock(ConnectorScanRange.class); + Mockito.when(scanRange.getSplitIdentity()).thenReturn(""); PluginDrivenSplit split = new PluginDrivenSplit(scanRange); TFileRangeDesc range = new TFileRangeDesc(); range.setFormatType(TFileFormatType.FORMAT_ORC); From b1bada874d3cf810382269f4f98ee71f36e0075b Mon Sep 17 00:00:00 2001 From: Dylan Date: Sat, 10 Oct 2026 18:15:31 +0800 Subject: [PATCH 08/12] [test](fe) Avoid assuming hash candidate order --- .../scan/FederationBackendPolicySpreadTest.java | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java index daf63688ea65cd..213a2cae7a77b8 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -122,18 +122,14 @@ private static IntUnaryOperator selectCandidate(int index) { } @Test - public void testIndependentHotQueriesCanChooseEveryCandidateExactlyOnce() throws Exception { - Set selected = new HashSet<>(); + public void testIndependentHotQueriesChooseOnlyHashCandidates() throws Exception { for (int index = 0; index < 3; index++) { FederationBackendPolicy policy = policy(3, selectCandidate(index)); FileSplit split = split(); - List candidates = policy.consistentHash.getNode(split, 3); Backend target = assign(policy, split); - Assertions.assertEquals(candidates.get(index), target); - selected.add(target); + Assertions.assertTrue(policy.consistentHash.getNode(split, 3).contains(target)); Assertions.assertNull(split.getAlternativeHosts()); } - Assertions.assertEquals(new HashSet<>(backends), selected); } @Test @@ -210,18 +206,19 @@ public void testSingleBackend() throws Exception { } @Test - public void testSameHostBackendsRemainDistinctCandidates() throws Exception { + public void testSameHostBackendsRemainEligibleCandidates() throws Exception { backends = new ArrayList<>(); for (int i = 0; i < 3; i++) { Backend backend = new Backend(92000 + i, "192.0.2.20", 9050 + i); backend.setAlive(true); backends.add(backend); } - Set targets = new HashSet<>(); for (int index = 0; index < 3; index++) { - targets.add(assign(policy(3, selectCandidate(index)), split())); + FederationBackendPolicy policy = policy(3, selectCandidate(index)); + FileSplit split = split(); + Backend target = assign(policy, split); + Assertions.assertTrue(policy.consistentHash.getNode(split, 3).contains(target)); } - Assertions.assertEquals(new HashSet<>(backends), targets); } @Test From 24fc97c21d82df52826f3337843d96353a160c7d Mon Sep 17 00:00:00 2001 From: Dylan Date: Sun, 11 Oct 2026 11:57:33 +0800 Subject: [PATCH 09/12] [test](fe) Match result sink in group by rewrite tests --- .../rules/rewrite/EliminateGroupByKeyByUniformTest.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/EliminateGroupByKeyByUniformTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/EliminateGroupByKeyByUniformTest.java index d5748699acc869..41bc87035aa38c 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/EliminateGroupByKeyByUniformTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/EliminateGroupByKeyByUniformTest.java @@ -135,8 +135,8 @@ void testLeftJoinOnConditionNotRewrite() { .analyze("select t1.b,t2.b from eli_gbk_by_uniform_t t1 left join eli_gbk_by_uniform_t t2 on t1.b=t2.b and t1.b=100 group by t1.b,t2.b,t2.c;") .rewrite() .printlnTree() - .matches(logicalAggregate().when(agg -> agg.getGroupByExpressions().size() == 2 - && agg.getGroupByExpressions().get(0).toSql().equals("b"))); + .matches(logicalResultSink(logicalAggregate().when(agg -> agg.getGroupByExpressions().size() == 2 + && agg.getGroupByExpressions().get(0).toSql().equals("b")))); } @Test @@ -145,8 +145,8 @@ void testLeftJoinWhereConditionRewrite() { .analyze("select t1.b,t2.b from eli_gbk_by_uniform_t t1 left join eli_gbk_by_uniform_t t2 on t1.b=t2.b where t1.b=100 group by t1.b,t2.b,t2.c;") .rewrite() .printlnTree() - .matches(logicalAggregate().when(agg -> agg.getGroupByExpressions().size() == 1 - && agg.getGroupByExpressions().get(0).toSql().equals("c"))); + .matches(logicalResultSink(logicalAggregate().when(agg -> agg.getGroupByExpressions().size() == 1 + && agg.getGroupByExpressions().get(0).toSql().equals("c")))); } @Test From 22d6ebafe16dfdfaa76e25357a106028deab5037 Mon Sep 17 00:00:00 2001 From: Dylan Date: Mon, 12 Oct 2026 14:26:23 +0800 Subject: [PATCH 10/12] [fix](fe) Address split scheduling review findings --- .../doris/connector/fluss/FlussLakeRange.java | 5 +- .../connector/fluss/FlussLakeRangeTest.java | 11 ++++ .../spi/ConnectorPluginSurfaceTest.java | 7 +-- .../resources/connector-plugin-surface.txt | 18 ++++++ fe/fe-connector/pom.xml | 2 +- .../scan/FederationBackendPolicy.java | 63 ++++++++++++++----- 6 files changed, 84 insertions(+), 22 deletions(-) diff --git a/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java b/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java index dfd5d09297c1ae..47699d4e2d40d6 100644 --- a/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java +++ b/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java @@ -180,8 +180,9 @@ public Optional getPath() { @Override public String getSplitIdentity() { - return inner.getSplitIdentity() + "\u0001" - + (tail == null ? RANGE_TYPE_LAKE : tail.encode()); + // The lake file is the cacheable unit. The suppressing log tail changes as the table grows, + // but must not remap the unchanged file to another backend. + return inner.getSplitIdentity(); } @Override diff --git a/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussLakeRangeTest.java b/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussLakeRangeTest.java index 2494adc6f6546e..31e7a824a26f2b 100644 --- a/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussLakeRangeTest.java +++ b/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussLakeRangeTest.java @@ -174,6 +174,17 @@ public void theFormatNameIsTheScansOwn() { Assertions.assertEquals("fluss", FlussLakeRange.plain(new LakeSplit()).getTableFormatType()); } + @Test + public void changingTailDoesNotChangeLakeFileIdentity() { + ConnectorScanRange lake = new LakeSplit(); + FlussLakeRange first = FlussLakeRange.suppressed(lake, + new FlussLakeRange.Tail(DT_20260101, 2, 100L, 105L)); + FlussLakeRange second = FlussLakeRange.suppressed(lake, + new FlussLakeRange.Tail(DT_20260101, 2, 100L, 205L)); + + Assertions.assertEquals(first.getSplitIdentity(), second.getSplitIdentity()); + } + /** * The payload is both halves, in their own thrift fields: whatever the sibling wrote, untouched, plus * the fluss descriptor. Asserted whole rather than by spot check — this is an untyped string map that diff --git a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java index fe025b288b9ee6..f76a5378caccee 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java +++ b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java @@ -87,10 +87,8 @@ public void connectorApiMajorTracksTheRecordedSurfaceChange() throws IOException Assertions.assertNotNull(in, "missing connector plugin API version resource"); version.load(in); } - // Major 13 carries textual write-input semantics on ConnectorColumn and adds - // ConnectorScanRange#getSplitIdentity: plugins compiled against the previous connector - // surface cannot link against an FE that exposes either new API hook. - Assertions.assertEquals("13.0", version.getProperty("api.version")); + // Major 14 adds ConnectorScanRange#getSplitIdentity to the connector plugin surface. + Assertions.assertEquals("14.0", version.getProperty("api.version")); } /** Root entry points plus provider/handle types returned to connector plugins. */ @@ -105,6 +103,7 @@ public void connectorApiMajorTracksTheRecordedSurfaceChange() throws IOException org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot.class, org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot.Builder.class, ConnectorScanPlanProvider.class, + org.apache.doris.connector.spi.scan.ConnectorScanRange.class, ConnectorWriteHandle.class, ConnectorChangelogMode.class, ConnectorRowLevelDmlRequest.class, diff --git a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt index f0ae0af27d158c..7776066bcf1c31 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt +++ b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt @@ -168,6 +168,24 @@ org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#supportsSystemTabl org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#supportsSystemTableTimeTravel():boolean org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#supportsTableSample():boolean org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#usesHiveParquetInt96TimeZone():boolean +org.apache.doris.connector.spi.scan.ConnectorScanRange#getFileFormat():java.lang.String +org.apache.doris.connector.spi.scan.ConnectorScanRange#getFileSize():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getHosts():java.util.List +org.apache.doris.connector.spi.scan.ConnectorScanRange#getLength():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getModificationTime():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getPartitionValues():java.util.Map +org.apache.doris.connector.spi.scan.ConnectorScanRange#getPath():java.util.Optional +org.apache.doris.connector.spi.scan.ConnectorScanRange#getProperties():java.util.Map +org.apache.doris.connector.spi.scan.ConnectorScanRange#getPushDownRowCount():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getSelfSplitWeight():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getSplitIdentity():java.lang.String +org.apache.doris.connector.spi.scan.ConnectorScanRange#getStart():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getTableFormatType():java.lang.String +org.apache.doris.connector.spi.scan.ConnectorScanRange#getTargetSplitSize():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#isNativeReadRange():boolean +org.apache.doris.connector.spi.scan.ConnectorScanRange#isPartitionBearing():boolean +org.apache.doris.connector.spi.scan.ConnectorScanRange#isSingleUse():boolean +org.apache.doris.connector.spi.scan.ConnectorScanRange#populateRangeParams(org.apache.doris.thrift.TTableFormatFileDesc,org.apache.doris.thrift.TFileRangeDesc):void org.apache.doris.connector.spi.scan.ScanNodePropertyKeys#field:FILE_FORMAT_TYPE:java.lang.String=file_format_type org.apache.doris.connector.spi.scan.ScanNodePropertyKeys#field:LOCATION_PREFIX:java.lang.String=location. org.apache.doris.connector.spi.scan.ScanNodePropertyKeys#field:PATH_PARTITION_KEYS:java.lang.String=path_partition_keys diff --git a/fe/fe-connector/pom.xml b/fe/fe-connector/pom.xml index 9bf53c1050bfa3..913f9ba889077e 100644 --- a/fe/fe-connector/pom.xml +++ b/fe/fe-connector/pom.xml @@ -55,7 +55,7 @@ under the License. of the latter two means bumping this property as well (and fe-extension-spi means bumping all five families). --> - 13.0 + 14.0 diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java index 4f86929565b0ba..f0cd1429dce1f6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java @@ -62,10 +62,12 @@ import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.NavigableMap; import java.util.Objects; import java.util.Optional; import java.util.Random; import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ThreadLocalRandom; import java.util.function.IntUnaryOperator; @@ -84,6 +86,7 @@ public Map getAssignedWeightPerBackend() { } private Map assignedWeightPerBackend = Maps.newHashMap(); + private final NavigableMap assignedWeightCounts = new TreeMap<>(); protected ConsistentHash consistentHash; @@ -208,8 +211,10 @@ public void init(BeSelectionPolicy policy) throws UserException { throw new UserException("No available backends for compute group: " + computeGroup.toString()); } } + assignedWeightPerBackend.clear(); + assignedWeightCounts.clear(); for (Backend backend : backends) { - assignedWeightPerBackend.put(backend, 0L); + setAssignedWeight(backend, 0L); } backendMap.putAll(backends.stream().collect(Collectors.groupingBy(Backend::getHost))); @@ -256,6 +261,7 @@ public Multimap computeScanRangeAssignment(List splits) t for (List backendList : backendMap.values()) { backends.addAll(backendList); } + rebuildAssignedWeightCounts(); ResettableRandomizedIterator randomCandidates = new ResettableRandomizedIterator<>(backends); // optimizedLocalScheduling enables prioritized assignment of splits to local nodes when splits contain @@ -273,15 +279,14 @@ public Multimap computeScanRangeAssignment(List splits) t // let a repeated hint monopolize one backend while other eligible backends are idle. boolean localNodeIsWithinGlobalMinimum = chosenNode.isPresent(); if (localNodeIsWithinGlobalMinimum && isSpreadEnabled()) { - long minimumAssignedWeight = Collections.min(assignedWeightPerBackend.values()); + long minimumAssignedWeight = assignedWeightCounts.firstKey(); localNodeIsWithinGlobalMinimum = assignedWeightPerBackend.get(chosenNode.get()) <= minimumAssignedWeight; } if (localNodeIsWithinGlobalMinimum) { Backend selectedBackend = chosenNode.get(); assignment.put(selectedBackend, split); - assignedWeightPerBackend.put(selectedBackend, - assignedWeightPerBackend.get(selectedBackend) + split.getSplitWeight().getRawValue()); + addAssignedWeight(selectedBackend, split.getSplitWeight().getRawValue()); continue; } } @@ -313,11 +318,14 @@ public Multimap computeScanRangeAssignment(List splits) t if (consistentHashSpreadNum == 0) { candidateNodes = backends; } else { - int candidateCount = isSpreadEnabled() - ? Math.min(consistentHashSpreadNum, backends.size()) - : Config.split_assigner_min_consistent_hash_candidate_num; - candidateNodes = candidateCount >= backends.size() ? backends - : consistentHash.getNode(split, candidateCount); + if (isSpreadEnabled()) { + int candidateCount = Math.min(consistentHashSpreadNum, backends.size()); + candidateNodes = candidateCount >= backends.size() ? backends + : consistentHash.getNode(split, candidateCount); + } else { + candidateNodes = consistentHash.getNode( + split, Config.split_assigner_min_consistent_hash_candidate_num); + } } break; } @@ -344,8 +352,7 @@ public Multimap computeScanRangeAssignment(List splits) t alternativeBackends.stream().map(each -> each.getHost()).collect(Collectors.toList())); } assignment.put(selectedBackend, split); - assignedWeightPerBackend.put(selectedBackend, - assignedWeightPerBackend.get(selectedBackend) + split.getSplitWeight().getRawValue()); + addAssignedWeight(selectedBackend, split.getSplitWeight().getRawValue()); } // Global redistribution can move a split outside its hash candidates or its locality constraints. @@ -409,10 +416,8 @@ private void equateDistribution(ListMultimap assignment) { // move split from max to min Split redistributedSplit = redistributeSplit(assignment, maxNode, minNode); - assignedWeightPerBackend.put(maxNode, - assignedWeightPerBackend.get(maxNode) - redistributedSplit.getSplitWeight().getRawValue()); - assignedWeightPerBackend.put(minNode, Math.addExact( - assignedWeightPerBackend.get(minNode), redistributedSplit.getSplitWeight().getRawValue())); + addAssignedWeight(maxNode, -redistributedSplit.getSplitWeight().getRawValue()); + addAssignedWeight(minNode, redistributedSplit.getSplitWeight().getRawValue()); // add max back into maxNodes only if it still has assignments if (assignment.containsKey(maxNode)) { @@ -528,6 +533,34 @@ private boolean isSpreadEnabled() { || (nodeSelectionStrategy == NodeSelectionStrategy.RANDOM && consistentHashSpreadNum == 0); } + private void setAssignedWeight(Backend backend, long weight) { + Long previousWeight = assignedWeightPerBackend.put(backend, weight); + if (previousWeight != null) { + decrementWeightCount(previousWeight); + } + assignedWeightCounts.merge(weight, 1, Integer::sum); + } + + private void addAssignedWeight(Backend backend, long delta) { + setAssignedWeight(backend, Math.addExact(assignedWeightPerBackend.get(backend), delta)); + } + + private void decrementWeightCount(long weight) { + int count = assignedWeightCounts.get(weight); + if (count == 1) { + assignedWeightCounts.remove(weight); + } else { + assignedWeightCounts.put(weight, count - 1); + } + } + + private void rebuildAssignedWeightCounts() { + assignedWeightCounts.clear(); + for (Long weight : assignedWeightPerBackend.values()) { + assignedWeightCounts.merge(weight, 1, Integer::sum); + } + } + private Backend chooseNodeForSpread(List candidateNodes) { Backend chosenNode = null; long minWeight = Long.MAX_VALUE; From 267ca8854726fcfd18e87e46a921bc1ae6621d9f Mon Sep 17 00:00:00 2001 From: Dylan Date: Tue, 13 Oct 2026 13:49:22 +0800 Subject: [PATCH 11/12] [fix](fe) Distinguish synthetic scan tasks and retain incremental weights ### What problem does this PR solve? Related PR: #68791 Problem Summary: MaxCompute ranges from different read sessions and Iceberg JNI metadata tasks can share synthetic paths and coordinates. Bounded consistent-hash scheduling then confines distinct work to the same backend candidates. Include the existing read-session or serialized-task identity through the connector hook while preserving native Iceberg file identities. Remove the redundant per-batch weight histogram rebuild because production weight updates already maintain it incrementally. Cover identity separation, ten-backend bounded candidate assignment and histogram consistency across singleton batches. ### Release note Synthetic MaxCompute and Iceberg JNI metadata scan ranges use work-specific consistent-hash identities. This can change their backend placement, including legacy hash mode. Native Iceberg file identities are unchanged. ### Check List (For Author) - Test: Unit Test / Manual test - 70 targeted FE and connector unit tests passed; standard build.sh --fe and Checkstyle passed; git diff --check passed. - Behavior changed: Yes, distinguish synthetic work for hash placement and remove redundant histogram rebuilds. - Does this need documentation: No, existing session-variable semantics are unchanged. --- .../connector/iceberg/IcebergScanRange.java | 6 ++ .../iceberg/IcebergScanRangeTest.java | 20 ++++++ .../maxcompute/MaxComputeScanRange.java | 6 ++ .../maxcompute/MaxComputeScanRangeTest.java | 18 +++++ .../scan/FederationBackendPolicy.java | 8 --- .../FederationBackendPolicySpreadTest.java | 65 ++++++++++++++++++- 6 files changed, 112 insertions(+), 11 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java index 9d8fd8312738ae..a22ac6341226c2 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java @@ -156,6 +156,12 @@ private IcebergScanRange(Builder builder) { this.positionDeleteContentSizeInBytes = builder.positionDeleteContentSizeInBytes; } + @Override + public String getSplitIdentity() { + // JNI metadata tasks share a dummy path; native ranges retain their actual file identity. + return serializedSplit != null ? serializedSplit : ConnectorScanRange.super.getSplitIdentity(); + } + @Override public Optional getPath() { return Optional.ofNullable(path); diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanRangeTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanRangeTest.java index 2902283d70d957..e657951100b4f6 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanRangeTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanRangeTest.java @@ -42,6 +42,26 @@ */ public class IcebergScanRangeTest { + @Test + public void metadataTasksUseTaskIdentityAndNativeRangesKeepFileIdentity() { + IcebergScanRange first = new IcebergScanRange.Builder().path("/dummyPath") + .serializedSplit("task-a").build(); + IcebergScanRange second = new IcebergScanRange.Builder().path("/dummyPath") + .serializedSplit("task-b").build(); + IcebergScanRange sameWork = new IcebergScanRange.Builder().path("/dummyPath") + .serializedSplit("task-a").build(); + Assertions.assertEquals(first.getPath(), second.getPath()); + Assertions.assertEquals(first.getStart(), second.getStart()); + Assertions.assertEquals(first.getLength(), second.getLength()); + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + Assertions.assertEquals(first.getSplitIdentity(), sameWork.getSplitIdentity()); + IcebergScanRange nativeRange = new IcebergScanRange.Builder().path("s3://bucket/file.parquet").build(); + Assertions.assertEquals("s3://bucket/file.parquet", nativeRange.getSplitIdentity()); + IcebergScanRange nativeDeletes = new IcebergScanRange.Builder().path("s3://bucket/delete.parquet") + .positionDeleteSysTableSplit(1, TFileFormatType.FORMAT_PARQUET, "s3://bucket/delete.parquet").build(); + Assertions.assertEquals("s3://bucket/delete.parquet", nativeDeletes.getSplitIdentity()); + } + @Test public void builderProducesFileScanRangeWithFileFields() { IcebergScanRange range = new IcebergScanRange.Builder() diff --git a/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputeScanRange.java b/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputeScanRange.java index 7e00f3b19875f6..f74d2e67980bfd 100644 --- a/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputeScanRange.java +++ b/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputeScanRange.java @@ -74,6 +74,12 @@ public Optional getPath() { return Optional.of("/row_offset"); } + @Override + public String getSplitIdentity() { + // Offsets and split indices are relative to the ODPS read session, not a physical file. + return getPath().get() + "/" + properties.get(PROP_SESSION_ID); + } + @Override public long getStart() { return start; diff --git a/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputeScanRangeTest.java b/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputeScanRangeTest.java index 3ce6e9b34bd9e3..55b9e1edfb6aef 100644 --- a/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputeScanRangeTest.java +++ b/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputeScanRangeTest.java @@ -63,6 +63,24 @@ */ public class MaxComputeScanRangeTest { + @Test + public void equalCoordinatesFromDifferentSessionsHaveDifferentIdentities() { + for (String type : new String[] {MaxComputeScanRange.SPLIT_TYPE_ROW_OFFSET, + MaxComputeScanRange.SPLIT_TYPE_BYTE_SIZE}) { + MaxComputeScanRange first = MaxComputeScanRange.builder().splitType(type) + .sessionId("session-a").start(0).length(1000).build(); + MaxComputeScanRange second = MaxComputeScanRange.builder().splitType(type) + .sessionId("session-b").start(0).length(1000).build(); + MaxComputeScanRange sameWork = MaxComputeScanRange.builder().splitType(type) + .sessionId("session-a").start(0).length(1000).build(); + Assertions.assertEquals(first.getPath(), second.getPath()); + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + Assertions.assertEquals(first.getSplitIdentity(), sameWork.getSplitIdentity()); + Assertions.assertEquals(0, first.getStart()); + Assertions.assertEquals(1000, first.getLength()); + } + } + private static final long SPLIT_BYTE_SIZE = 268435456L; // ODPS default byte-size split @Test diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java index f0cd1429dce1f6..3fa4f87ca7c315 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java @@ -261,7 +261,6 @@ public Multimap computeScanRangeAssignment(List splits) t for (List backendList : backendMap.values()) { backends.addAll(backendList); } - rebuildAssignedWeightCounts(); ResettableRandomizedIterator randomCandidates = new ResettableRandomizedIterator<>(backends); // optimizedLocalScheduling enables prioritized assignment of splits to local nodes when splits contain @@ -554,13 +553,6 @@ private void decrementWeightCount(long weight) { } } - private void rebuildAssignedWeightCounts() { - assignedWeightCounts.clear(); - for (Long weight : assignedWeightPerBackend.values()) { - assignedWeightCounts.merge(weight, 1, Integer::sum); - } - } - private Backend chooseNodeForSpread(List candidateNodes) { Backend chosenNode = null; long minWeight = Long.MAX_VALUE; diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java index 213a2cae7a77b8..53a94f9aaea8dc 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -24,7 +24,9 @@ import org.apache.doris.common.UserException; import org.apache.doris.common.jmockit.Deencapsulation; import org.apache.doris.common.util.LocationPath; +import org.apache.doris.connector.spi.scan.ConnectorScanRange; import org.apache.doris.datasource.split.FileSplit; +import org.apache.doris.datasource.split.PluginDrivenSplit; import org.apache.doris.planner.PlanNodeId; import org.apache.doris.planner.ScanContext; import org.apache.doris.qe.ConnectContext; @@ -46,7 +48,10 @@ import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Optional; import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.atomic.AtomicLong; import java.util.function.IntUnaryOperator; @@ -156,13 +161,67 @@ public void testTopTwoCandidateBoundarySurvivesMultipleBatches() throws Exceptio } } + @Test + public void testSyntheticPathsDoNotConfineDistinctWorkToOneCandidateSet() throws Exception { + backends = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + Backend backend = new Backend(NEXT_BACKEND_ID.getAndIncrement(), "192.0.2." + (i + 1), 9050); + backend.setAlive(true); + backends.add(backend); + } + for (String path : Arrays.asList("/row_offset", "/dummyPath")) { + FederationBackendPolicy policy = policy(2, bound -> 0); + Set targets = new HashSet<>(); + for (int i = 0; i < 100; i++) { + String identity = path + "/work-" + i; + PluginDrivenSplit split = new PluginDrivenSplit(new ConnectorScanRange() { + @Override + public Map getProperties() { + return Collections.emptyMap(); + } + + @Override + public Optional getPath() { + return Optional.of(path); + } + + @Override + public String getSplitIdentity() { + return identity; + } + }); + List candidates = policy.consistentHash.getNode(split, 2); + Backend selected = assign(policy, split); + Assertions.assertTrue(candidates.contains(selected)); + targets.add(selected); + } + Assertions.assertTrue(targets.size() > 2, + "Distinct work with equal synthetic paths must escape a single two-BE candidate set"); + } + } + + @Test + public void testWeightHistogramRemainsIncrementalAcrossSingletonBatches() throws Exception { + for (int spread : Arrays.asList(0, 1, 2)) { + FederationBackendPolicy policy = policy(spread, bound -> 0); + for (int batch = 0; batch < 30; batch++) { + assign(policy, split()); + Map expected = new TreeMap<>(); + for (Long weight : policy.getAssignedWeightPerBackend().values()) { + expected.merge(weight, 1, Integer::sum); + } + Assertions.assertEquals(expected, Deencapsulation.getField(policy, "assignedWeightCounts")); + } + } + } + @Test public void testMinimumWeightWinsBeforeRandomTieBreaking() throws Exception { FederationBackendPolicy policy = policy(3, selectCandidate(2)); List candidates = policy.consistentHash.getNode(split(), 3); - policy.getAssignedWeightPerBackend().put(candidates.get(0), 1000L); - policy.getAssignedWeightPerBackend().put(candidates.get(1), 0L); - policy.getAssignedWeightPerBackend().put(candidates.get(2), 100L); + Deencapsulation.invoke(policy, "setAssignedWeight", candidates.get(0), 1000L); + Deencapsulation.invoke(policy, "setAssignedWeight", candidates.get(1), 0L); + Deencapsulation.invoke(policy, "setAssignedWeight", candidates.get(2), 100L); Assertions.assertEquals(candidates.get(1), assign(policy, split())); } From 5d584fce62774c7f02ea7454dc33a8b8a555d4cb Mon Sep 17 00:00:00 2001 From: Dylan Date: Tue, 13 Oct 2026 14:17:06 +0800 Subject: [PATCH 12/12] [fix](fe) Include Flight tickets in remote scan identity ### What problem does this PR solve? Related PR: #68791 Problem Summary: Remote Doris result streams can advertise the same proxy URI with different Flight tickets. Hashing only that URI restricts distinct streams to the same bounded backend candidate set. Capture an immutable identity from the URI and the remaining ticket bytes when the split is created, without consuming the execution ByteBuffer. Add tests for distinct work, stable identity, heap/direct/read-only buffer state, and bounded assignment across eight backends. ### Release note Remote Doris consistent-hash placement distinguishes different result-stream tickets at the same endpoint. ### Check List (For Author) - Test: Unit Test / Manual test - 82 targeted FE/connector unit tests passed; standard clean FE build with Checkstyle passed; git diff --check passed. - Behavior changed: Yes, hash placement includes the result-stream ticket; execution parameters are unchanged. - Does this need documentation: No, existing session-variable semantics are unchanged. --- .../doris/source/RemoteDorisSplit.java | 9 ++- .../doris/source/RemoteDorisSplitTest.java | 72 +++++++++++++++++++ .../FederationBackendPolicySpreadTest.java | 25 +++++++ 3 files changed, 105 insertions(+), 1 deletion(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/doris/source/RemoteDorisSplitTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java index e056cf9ba871b9..58047dec8a9528 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java @@ -22,16 +22,23 @@ import org.apache.doris.datasource.split.FileSplit; import java.nio.ByteBuffer; +import java.util.Base64; public class RemoteDorisSplit extends FileSplit { private static final LocationPath DUMMY_PATH = LocationPath.of("/dummyPath"); private final String location; private final ByteBuffer ticket; + private final String splitIdentity; public RemoteDorisSplit(String location, ByteBuffer ticket) { super(DUMMY_PATH, 0, 0, 0, 0, null, null); this.location = location; this.ticket = ticket; + // Capture the work identity without consuming the buffer used to populate the execution range. + ByteBuffer identityTicket = ticket.asReadOnlyBuffer(); + byte[] ticketBytes = new byte[identityTicket.remaining()]; + identityTicket.get(ticketBytes); + this.splitIdentity = location.length() + ":" + location + Base64.getEncoder().encodeToString(ticketBytes); this.tableFormatType = TableFormatType.REMOTE_DORIS; } @@ -45,7 +52,7 @@ public String getLocation() { @Override public String getSplitIdentity() { - return location; + return splitIdentity; } public TableFormatType getTableFormatType() { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/source/RemoteDorisSplitTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/source/RemoteDorisSplitTest.java new file mode 100644 index 00000000000000..4d15b9e04c677a --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/source/RemoteDorisSplitTest.java @@ -0,0 +1,72 @@ +// 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.datasource.doris.source; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; + +public class RemoteDorisSplitTest { + @Test + public void sameLocationWithDifferentTicketsHasDifferentIdentity() { + RemoteDorisSplit first = new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 2})); + RemoteDorisSplit second = new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 3})); + RemoteDorisSplit sameWork = new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 2})); + RemoteDorisSplit otherLocation = new RemoteDorisSplit("grpc://other:8815", ByteBuffer.wrap(new byte[] {1, 2})); + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + Assertions.assertEquals(first.getSplitIdentity(), sameWork.getSplitIdentity()); + Assertions.assertNotEquals(first.getSplitIdentity(), otherLocation.getSplitIdentity()); + } + + @Test + public void identityCapturePreservesExecutionBufferAndRemainsImmutable() { + ByteBuffer ticket = ByteBuffer.wrap(new byte[] {9, 1, 2, 8}); + ticket.position(1); + ticket.limit(3); + ticket.mark(); + RemoteDorisSplit split = new RemoteDorisSplit("grpc://proxy:8815", ticket); + String identity = split.getSplitIdentity(); + Assertions.assertSame(ticket, split.getTicket()); + Assertions.assertEquals(1, ticket.position()); + Assertions.assertEquals(3, ticket.limit()); + Assertions.assertEquals(identity, + new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 2})).getSplitIdentity()); + ticket.reset(); + ticket.put(1, (byte) 7); + ticket.position(2); + Assertions.assertEquals(identity, split.getSplitIdentity()); + Assertions.assertEquals(2, ticket.position()); + Assertions.assertEquals(3, ticket.limit()); + } + + @Test + public void directReadOnlyTicketUsesTheRemainingBytes() { + ByteBuffer direct = ByteBuffer.allocateDirect(4); + direct.put(new byte[] {9, 1, 2, 8}); + direct.flip(); + direct.position(1); + direct.limit(3); + ByteBuffer ticket = direct.asReadOnlyBuffer(); + RemoteDorisSplit split = new RemoteDorisSplit("grpc://proxy:8815", ticket); + Assertions.assertEquals(1, ticket.position()); + Assertions.assertEquals(3, ticket.limit()); + Assertions.assertEquals(split.getSplitIdentity(), + new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 2})).getSplitIdentity()); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java index 53a94f9aaea8dc..7d3cb9422f2d86 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java @@ -25,6 +25,7 @@ import org.apache.doris.common.jmockit.Deencapsulation; import org.apache.doris.common.util.LocationPath; import org.apache.doris.connector.spi.scan.ConnectorScanRange; +import org.apache.doris.datasource.doris.source.RemoteDorisSplit; import org.apache.doris.datasource.split.FileSplit; import org.apache.doris.datasource.split.PluginDrivenSplit; import org.apache.doris.planner.PlanNodeId; @@ -43,6 +44,7 @@ import org.mockito.MockedStatic; import org.mockito.Mockito; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -200,6 +202,29 @@ public String getSplitIdentity() { } } + @Test + public void testRemoteDorisTicketsAtSameProxyUseDistinctCandidateSets() throws Exception { + backends = new ArrayList<>(); + for (int i = 0; i < 8; i++) { + Backend backend = new Backend(NEXT_BACKEND_ID.getAndIncrement(), "192.0.2." + (i + 1), 9050); + backend.setAlive(true); + backends.add(backend); + } + FederationBackendPolicy policy = policy(2, bound -> 0); + Set targets = new HashSet<>(); + for (int i = 0; i < 100; i++) { + ByteBuffer ticket = ByteBuffer.wrap(new byte[] {(byte) i}); + RemoteDorisSplit split = new RemoteDorisSplit("grpc://proxy:8815", ticket); + List candidates = policy.consistentHash.getNode(split, 2); + Backend selected = assign(policy, split); + Assertions.assertTrue(candidates.contains(selected)); + Assertions.assertEquals(0, ticket.position()); + targets.add(selected); + } + Assertions.assertTrue(targets.size() > 2, + "Different Flight tickets at the same proxy must escape a single candidate pair"); + } + @Test public void testWeightHistogramRemainsIncrementalAcrossSingletonBatches() throws Exception { for (int spread : Arrays.asList(0, 1, 2)) {