This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 846f932b5ef [improvement](fe) Add opt-in spreading for remote external
scan splits (#68791)
846f932b5ef is described below
commit 846f932b5ef8cd418e785d35293345350b72f0e5
Author: Oliveira <[email protected]>
AuthorDate: Thu Oct 15 09:51:12 2026 +0800
[improvement](fe) Add opt-in spreading for remote external scan splits
(#68791)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #68726
Problem Summary:
Repeated independently planned queries for a single remotely accessible
external file split can select the same backend under consistent
hashing. The candidate order is stable, each query starts with zero
assigned weight, and the existing minimum-weight tie rule repeatedly
selects the same candidate. Increasing the configured candidate count
alone does not remove this hotspot.
Add the session variable `external_scan_consistent_hash_spread_num` with
default **1**, preserving existing scheduling:
| Value | Behavior |
| --- | --- |
| `1` (default) | Preserve the original strategy, configured candidate
counts, tie rule and global redistribution. This is a mode, not a
one-backend limit. |
| `0` | Enable automatic spreading across all eligible BEs in the
query's compute group. Use the hash strategy when file cache or
consistent hashing is enabled; otherwise use the random strategy. |
| `N > 1` | Spread within at most N distinct hash candidates when file
cache or consistent hashing is enabled; otherwise retain original
round-robin scheduling. |
In spread mode, select the candidate with the least query-local assigned
split weight and break ties uniformly at random. Assign each split once,
retain weight history across batches, and honor compute-group
eligibility, local preference and mandatory Host constraints. Candidate
counts are capped by the eligible BE count; the implementation is not
specific to three BEs.
Disable global split redistribution only in spread mode so it cannot
move a split outside its candidate or locality constraints. Do not
allocate unused alternative-host lists in this mode. With many splits,
balancing is limited to each split's candidates and can be less even
than the original global redistribution.
This PR adds no live CPU monitoring, shared query counter, COUNT
pushdown, DOP adjustment or BE cache protocol. It is independent of
#68726 and does not include that PR's commits. The existing round-robin
path remains available; when #68726 is merged, its round-robin fix also
remains in effect for original-mode callers.
The variable does not enable file caching. With caching enabled,
automatic mode can warm copies on every eligible BE and increase remote
warmup reads and cache storage. Persisted global values survive
upgrades; restoring the compiled default to 1 does not reset an already
saved global 0. `UNSET VARIABLE` restores the compiled session default
1.
Manual performance evidence, collected on scheduling revision
`7e4fb1c549d` before this final default-value adjustment: the same
independent FE binary, three unchanged BEs on one shared physical host
with distinct logical Hosts and disjoint nonexclusive four-core
affinities; one million-row single-file CSV SUM; file cache and
consistent hashing enabled; all candidate caches warmed; concurrency 64;
10-second warmup and 60-second measurement window; three alternating
pairs changing only `spread=1` versus `spread=0`.
| Mode | Mean successful QPS |
| --- | ---: |
| Original (`1`) | 13.3944 |
| Automatic (`0`) | 39.2778 |
The pooled ratio was **2.9324x**, with paired ratios 2.8782x–2.9603x.
All six groups had zero query errors. QPS counts correct completions
within the fixed window; late completions are excluded, all workers are
joined, and latency includes the tail of queries started within the
window. The original hot BE used 97.5%–98.7% of its four-core capacity;
automatic mode used 87.2%–97.7% across the three BEs. Whole-invocation
remote and peer read increments were zero.
This evidence applies to the measured warm-cache shared-host workload.
It is not a new benchmark of final revision `813826a7feb`, a cold-start
result, a strict >=3x guarantee, a three-physical-host acceptance result
or customer validation. The final adjustment changes the default and
documentation, not the explicit spread-selection algorithm; spreading
must now be explicitly enabled to seek this benefit. No throughput from
the related COUNT optimization is attributed to this PR.
### Release note
Add opt-in spreading for remote external file scan splits.
`external_scan_consistent_hash_spread_num` defaults to 1 and preserves
original scheduling. Set it to 0 to spread across all eligible BEs, or a
value above 1 to bound hash candidates when file cache or consistent
hashing is enabled. Spreading can increase cache copies and warmup
costs. Persisted global values are retained on upgrade.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
Validation performed on the final revision:
- Standard `run-fe-ut.sh --run
'FederationBackendPolicySpreadTest,VariableMgrTest#testExternalScanConsistentHashSpreadVariable,FederationBackendPolicyTest'`:
33 tests passed, zero failed/error/skipped. Coverage includes 1/2/3/5
BEs, random ties, multiple batches, locality, bounds, forwarding,
original-mode compatibility, default 1 and explicit 0 strategy
selection.
- Standard `build.sh --fe --clean -j16` and integrated Checkstyle
passed. An earlier production build failed with incomplete class/type
errors; its evidence was retained and the clean build succeeded without
unrelated source edits.
- Standard `run-regression-test.sh` with an isolated personal
configuration:
- `-g p0 -d query_p0/ddl/set_and_unset_commands -s
test_external_scan_consistent_hash_spread_variable`; output generated by
the framework using `-forceGenOut`, then verified normally.
- `-g external -d external_table_p0/tvf -s
test_external_scan_consistent_hash_spread`; explicitly enables 0 for
automatic-path coverage.
- Both suites passed with zero failed/fatal/skipped suites. The fixture
is generated by the suite in personal MinIO.
- Isolated FE/BE health checks passed; BE process IDs and binary hashes
were unchanged. No environment configuration or work log is included.
- Separate manual warm-cache experiment on revision `7e4fb1c549d`,
described above. Each mode was also certified by 100 query-owned
profiles: correct result, one source split/range, one positive execution
target, one million rows read, local-cache reads and no remote reads.
Original mode selected one BE; automatic mode selected all three.
- Standards and specification reviews of the implementation and final
default adjustment found no remaining actionable issues.
- Targeted local validation does not claim full TeamCity success on the
updated head. No BE source changed, so no new BE build or BE UT run was
required for this adjustment.
- Behavior changed:
- [ ] No.
- [x] Yes, when explicitly enabled. Default 1 preserves the existing
strategy. Values 0 and N>1 enable random minimum-weight ties and disable
global redistribution in their applicable paths; caching can require
more copies.
- Does this need documentation?
- [ ] No.
- [x] Yes. https://github.com/apache/doris-website/pull/4199
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
fe/be-java-extensions/paimon-s3-shade/pom.xml | 21 +
.../apache/doris/connector/adbc/AdbcScanRange.java | 6 +
.../doris/connector/adbc/AdbcScanRangeTest.java | 16 +
.../doris/connector/fluss/FlussLakeRange.java | 7 +
.../doris/connector/fluss/FlussScanRange.java | 10 +
.../doris/connector/fluss/FlussLakeRangeTest.java | 11 +
.../doris/connector/fluss/FlussScanRangeTest.java | 2 +
.../doris/connector/iceberg/IcebergScanRange.java | 6 +
.../connector/iceberg/IcebergScanRangeTest.java | 20 +
.../connector/maxcompute/MaxComputeScanRange.java | 6 +
.../maxcompute/MaxComputeScanRangeTest.java | 18 +
.../doris/connector/paimon/PaimonScanRange.java | 5 +
.../connector/spi/scan/ConnectorScanRange.java | 8 +
.../connector/spi/ConnectorPluginSurfaceTest.java | 5 +-
.../test/resources/connector-plugin-surface.txt | 18 +
.../doris/connector/trino/TrinoScanRange.java | 5 +
fe/fe-connector/pom.xml | 2 +-
.../datasource/doris/source/RemoteDorisSplit.java | 12 +
.../doris/datasource/scan/ExternalScanNode.java | 22 +-
.../datasource/scan/FederationBackendPolicy.java | 127 ++++-
.../doris/datasource/split/PluginDrivenSplit.java | 17 +
.../java/org/apache/doris/qe/SessionVariable.java | 20 +
.../src/main/java/org/apache/doris/spi/Split.java | 2 +-
.../doris/source/RemoteDorisSplitTest.java | 72 +++
.../scan/FederationBackendPolicySpreadTest.java | 599 +++++++++++++++++++++
...ginDrivenScanNodeScanProviderSelectionTest.java | 1 +
.../PluginDrivenSplitPartitionValuesTest.java | 22 +
.../org/apache/doris/http/DorisHttpTestCase.java | 30 +-
.../org/apache/doris/http/ForwardToMasterTest.java | 8 +
.../rewrite/EliminateGroupByKeyByUniformTest.java | 15 +-
.../java/org/apache/doris/qe/VariableMgrTest.java | 29 +
.../test_external_scan_consistent_hash_spread.out | 59 ++
...ternal_scan_consistent_hash_spread_variable.out | 22 +
...est_external_scan_consistent_hash_spread.groovy | 88 +++
...nal_scan_consistent_hash_spread_variable.groovy | 44 ++
35 files changed, 1301 insertions(+), 54 deletions(-)
diff --git a/fe/be-java-extensions/paimon-s3-shade/pom.xml
b/fe/be-java-extensions/paimon-s3-shade/pom.xml
index 02a80fc2bd3..427e4f66b3a 100644
--- a/fe/be-java-extensions/paimon-s3-shade/pom.xml
+++ b/fe/be-java-extensions/paimon-s3-shade/pom.xml
@@ -58,6 +58,27 @@ under the License.
</dependency>
</dependencies>
+ <profiles>
+ <profile>
+ <id>coverage</id>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- Retain embedded multi-release classes in
the runtime jar, but do not
+ analyze their duplicate class names in
the coverage report. -->
+ <exclude>**/META-INF/versions/**</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
<build>
<finalName>paimon-s3-shade</finalName>
<directory>${project.basedir}/target/</directory>
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 9dfc77fd01f..731565ee2a4 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 class AdbcScanRange implements ConnectorScanRange {
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<String, String> 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 b2db701e28f..be8b51b34c5 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 @@ class AdbcScanRangeTest {
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 partitionsCanBeReadOnlyOnceWhileStatementsRunOnEveryRead() {
// A partition is a ticket for a result stream the source produced
once, and reading it drains the
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 549d7fb5e48..47699d4e2d4 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,13 @@ public class FlussLakeRange implements ConnectorScanRange {
return inner.getPath();
}
+ @Override
+ public String getSplitIdentity() {
+ // 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
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 0f720b91863..009a7acb303 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 class FlussScanRange implements ConnectorScanRange {
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<String, String> getProperties() {
return properties;
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 2494adc6f65..31e7a824a26 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 class FlussLakeRangeTest {
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-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 99b4a058a99..8089638b3da 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<String, String> expected = new LinkedHashMap<>();
expected.put("fluss.range_type", "LOG");
@@ -58,6 +59,7 @@ public class FlussScanRangeTest {
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-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 9d8fd831273..a22ac634122 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 @@ public class IcebergScanRange implements
ConnectorScanRange {
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<String> 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 2902283d70d..e657951100b 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 @@ import java.util.Optional;
*/
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 7e00f3b1987..f74d2e67980 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 class MaxComputeScanRange implements
ConnectorScanRange {
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 3ce6e9b34bd..55b9e1edfb6 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 @@ import java.util.List;
*/
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-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 77c0d073659..ecb351d61c1 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 class PaimonScanRange implements ConnectorScanRange
{
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 572a78ba1e0..a4ac22e52c0 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 @@ public interface ConnectorScanRange extends Serializable {
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 fa459f99076..f76a5378cac 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,8 +87,8 @@ public class ConnectorPluginSurfaceTest {
Assertions.assertNotNull(in, "missing connector plugin API version
resource");
version.load(in);
}
- // Major 13 carries textual write-input semantics on ConnectorColumn.
- 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. */
@@ -103,6 +103,7 @@ public class ConnectorPluginSurfaceTest {
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 3f9e7f2703d..134d018f5c2 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
@@ -169,6 +169,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/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 0f4d363b090..ebac65fa310 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 class TrinoScanRange implements ConnectorScanRange {
return "trino_connector";
}
+ @Override
+ public String getSplitIdentity() {
+ return properties.getOrDefault(KEY_SPLIT, "trino_connector");
+ }
+
@Override
public Map<String, String> getProperties() {
return properties;
diff --git a/fe/fe-connector/pom.xml b/fe/fe-connector/pom.xml
index 9bf53c1050b..913f9ba8890 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).
-->
- <connector.plugin.api.version>13.0</connector.plugin.api.version>
+ <connector.plugin.api.version>14.0</connector.plugin.api.version>
</properties>
<modules>
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 3cccf8821f9..58047dec8a9 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.scan.TableFormatType;
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;
}
@@ -43,6 +50,11 @@ public class RemoteDorisSplit extends FileSplit {
return location;
}
+ @Override
+ public String getSplitIdentity() {
+ return splitIdentity;
+ }
+
public TableFormatType getTableFormatType() {
return tableFormatType;
}
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 c378c285dc5..20269cd67c1 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,11 +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)
- : 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 4b4572b2ce7..b5080ef3433 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,11 +62,15 @@ import java.util.Iterator;
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;
import java.util.stream.Collectors;
public class FederationBackendPolicy {
@@ -82,6 +86,7 @@ public class FederationBackendPolicy {
}
private Map<Backend, Long> assignedWeightPerBackend = Maps.newHashMap();
+ private final NavigableMap<Long, Integer> assignedWeightCounts = new
TreeMap<>();
protected ConsistentHash<Split, Backend> consistentHash;
@@ -90,6 +95,8 @@ public class FederationBackendPolicy {
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<HashCacheKey, ConsistentHash<Split, Backend>>
consistentHashCache;
@@ -142,7 +149,20 @@ public class FederationBackendPolicy {
}
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 0");
this.nodeSelectionStrategy = nodeSelectionStrategy;
+ this.consistentHashSpreadNum = consistentHashSpreadNum;
+ this.randomIndex = randomIndex;
}
public FederationBackendPolicy() {
@@ -191,10 +211,12 @@ public class FederationBackendPolicy {
throw new UserException("No available backends for compute
group: " + computeGroup.toString());
}
}
+ assignedWeightPerBackend.clear();
+ assignedWeightCounts.clear();
backendMap.clear();
backendMap.putAll(backends.stream().collect(Collectors.groupingBy(Backend::getHost)));
for (Backend backend : backends) {
- assignedWeightPerBackend.put(backend, 0L);
+ setAssignedWeight(backend, 0L);
}
try {
@@ -259,11 +281,18 @@ public class FederationBackendPolicy {
Optional<Backend> 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 =
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;
}
}
@@ -287,12 +316,23 @@ public class FederationBackendPolicy {
}
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,
-
Config.split_assigner_min_consistent_hash_candidate_num);
+ if (consistentHashSpreadNum == 0) {
+ candidateNodes = backends;
+ } else {
+ 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;
}
default: {
@@ -308,17 +348,22 @@ public class FederationBackendPolicy {
throw new
UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG);
}
- Backend selectedBackend = chooseNodeForSplit(candidateNodes);
- List<Backend> alternativeBackends = new
ArrayList<>(candidateNodes);
- alternativeBackends.remove(selectedBackend);
- split.setAlternativeHosts(
- alternativeBackends.stream().map(each ->
each.getHost()).collect(Collectors.toList()));
+ Backend selectedBackend = isSpreadEnabled() &&
split.isRemotelyAccessible()
+ ? chooseNodeForSpread(candidateNodes) :
chooseNodeForSplit(candidateNodes);
+ // Alternative hosts are used only by global redistribution, which
spread mode disables.
+ if (!isSpreadEnabled()) {
+ List<Backend> 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());
+ addAssignedWeight(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 && !isSpreadEnabled()) {
equateDistribution(assignment);
}
return assignment;
@@ -375,10 +420,8 @@ public class FederationBackendPolicy {
// 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)) {
@@ -489,6 +532,50 @@ public class FederationBackendPolicy {
return chosenNode;
}
+ private boolean isSpreadEnabled() {
+ return (nodeSelectionStrategy ==
NodeSelectionStrategy.CONSISTENT_HASHING && consistentHashSpreadNum != 1)
+ || (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 Backend chooseNodeForSpread(List<Backend> 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();
}
@@ -507,7 +594,7 @@ public class FederationBackendPolicy {
private static class SplitHash implements Funnel<Split> {
@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 2504b09adf0..85a8c2b2c27 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 @@ import java.util.Map;
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 class PluginDrivenSplit extends FileSplit {
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 class PluginDrivenSplit extends FileSplit {
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 3f04a2b4ac2..405d76e0908 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
@@ -1491,9 +1491,21 @@ public class SessionVariable implements Serializable,
Writable {
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;
+ 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 = "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 = 1;
+
@VarAttrDef.VarAttr(name = PROFILE_LEVEL, fuzzy = false, needForward =
true,
setter = "setProfileLevel", checker = "checkProfileLevel",
description = "The level of query profile, "
@@ -6487,6 +6499,14 @@ public class SessionVariable implements Serializable,
Writable {
return useConsistentHashForExternalScan;
}
+ public int getExternalScanConsistentHashSpreadNum() {
+ return externalScanConsistentHashSpreadNum;
+ }
+
+ public void checkExternalScanConsistentHashSpreadNum(String value) throws
Exception {
+ checkFieldValue(EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM, 0, value);
+ }
+
public void setForceJniScanner(boolean force) {
forceJniScanner = force;
}
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 48b6acd68e3..a5aa55186a2 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 @@ public interface Split {
void setAlternativeHosts(List<String> alternativeHosts);
- default String getConsistentHashString() {
+ default String getSplitIdentity() {
return getPathString();
}
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 00000000000..4d15b9e04c6
--- /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
new file mode 100644
index 00000000000..ea1b2c683d6
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java
@@ -0,0 +1,599 @@
+// 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.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;
+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.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+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;
+
+public class FederationBackendPolicySpreadTest {
+ private static final AtomicLong NEXT_BACKEND_ID = new AtomicLong(91000);
+
+ private MockedStatic<Env> mockedEnv;
+ private ConnectContext previousContext;
+ private ConnectContext context;
+ private ComputeGroup computeGroup;
+ private List<Backend> 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<Backend, Split> 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) {
+ // Select only when the reservoir reaches the desired candidate index.
+ return bound -> bound == index + 1 ? 0 : bound - 1;
+ }
+
+ private FederationBackendPolicy
policySelectingBackend(NodeSelectionStrategy strategy, int spread,
+ Backend target) throws UserException {
+ int[] targetIndex = new int[1];
+ FederationBackendPolicy policy = new FederationBackendPolicy(strategy,
spread,
+ bound -> bound == targetIndex[0] + 1 ? 0 : bound - 1);
+ policy.init();
+ // Initialization randomizes the order. Resolve the target in this
policy's actual candidate list.
+ targetIndex[0] = new ArrayList<>(policy.getBackends()).indexOf(target);
+ Assertions.assertTrue(targetIndex[0] >= 0);
+ return policy;
+ }
+
+ @Test
+ public void testIndependentHotQueriesChooseOnlyHashCandidates() throws
Exception {
+ for (int index = 0; index < 3; index++) {
+ FederationBackendPolicy policy = policy(3, selectCandidate(index));
+ FileSplit split = split();
+ Backend target = assign(policy, split);
+ Assertions.assertTrue(policy.consistentHash.getNode(split,
3).contains(target));
+ Assertions.assertNull(split.getAlternativeHosts());
+ }
+ }
+
+ @Test
+ public void testTopTwoCandidateBoundarySurvivesMultipleBatches() throws
Exception {
+ FederationBackendPolicy policy = policy(2, bound -> bound - 1);
+ Set<Backend> candidates = new
HashSet<>(policy.consistentHash.getNode(split(), 2));
+ for (int batch = 0; batch < 2; batch++) {
+ List<Split> splits = new ArrayList<>();
+ for (int i = 0; i < 20; i++) {
+ splits.add(split());
+ }
+ Multimap<Backend, Split> 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 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<Backend> targets = new HashSet<>();
+ for (int i = 0; i < 100; i++) {
+ String identity = path + "/work-" + i;
+ PluginDrivenSplit split = new PluginDrivenSplit(new
ConnectorScanRange() {
+ @Override
+ public Map<String, String> getProperties() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Optional<String> getPath() {
+ return Optional.of(path);
+ }
+
+ @Override
+ public String getSplitIdentity() {
+ return identity;
+ }
+ });
+ List<Backend> 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 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<Backend> 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<Backend> 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)) {
+ FederationBackendPolicy policy = policy(spread, bound -> 0);
+ for (int batch = 0; batch < 30; batch++) {
+ assign(policy, split());
+ Map<Long, Integer> 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<Backend> candidates = policy.consistentHash.getNode(split(), 3);
+ 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()));
+ }
+
+ @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<Backend> 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<Split> 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<Backend> targets = new HashSet<>();
+ for (int index = 0; index < 3; index++) {
+
targets.add(assign(policySelectingBackend(NodeSelectionStrategy.CONSISTENT_HASHING,
+ Integer.MAX_VALUE, backends.get(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 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);
+ }
+ for (int index = 0; index < 3; index++) {
+ FederationBackendPolicy policy = policy(3, selectCandidate(index));
+ FileSplit split = split();
+ Backend target = assign(policy, split);
+ Assertions.assertTrue(policy.consistentHash.getNode(split,
3).contains(target));
+ }
+ }
+
+ @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<Backend> 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<Split> splits = new ArrayList<>(Collections.nCopies(12,
split));
+ Multimap<Backend, Split> 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<Backend> 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, -1));
+ }
+
+ @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"));
+ }
+
+ @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<Backend> targets = new HashSet<>();
+ for (int index = 0; index < count; index++) {
+ FederationBackendPolicy policy =
policySelectingBackend(strategy, 0, backends.get(index));
+ 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 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<Split> 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 testRepeatedRemoteHostHintDoesNotMonopolizeSpreadMode() throws
Exception {
+ FederationBackendPolicy policy = policy(3, bound -> 0);
+ List<Split> 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<Backend, Split> 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();
+ 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;
+ 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);
+ }
+
+ @Override
+ protected void createScanRangeLocations() {
+ }
+ }
+}
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 99b07330767..f061e5c5a79 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 class PluginDrivenScanNodeScanProviderSelectionTest {
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);
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 2deb928f853..b17ecfc0ad9 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 class PluginDrivenSplitPartitionValuesTest {
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<String, String> getProperties() {
+ return Collections.singletonMap(key, value);
+ }
+
+ @Override
+ public String getSplitIdentity() {
+ return value;
+ }
+ };
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java
b/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java
index 17d0e7c83a7..4c06874d8d0 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java
@@ -67,12 +67,12 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.mockito.Mockito;
+import
org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
-import java.net.ServerSocket;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
@@ -237,37 +237,25 @@ public abstract class DorisHttpTestCase {
@BeforeAll
public static void initHttpServer() throws IllegalArgException,
InterruptedException {
- ServerSocket socket = null;
- try {
- socket = new ServerSocket(0);
- socket.setReuseAddress(true);
- HTTP_PORT = socket.getLocalPort();
- URI = "http://localhost:"; + HTTP_PORT + "/api/" + DB_NAME + "/" +
TABLE_NAME;
- CloudURI = "http://localhost:"; + HTTP_PORT;
- } catch (Exception e) {
- throw new IllegalStateException("Could not find a free TCP/IP port
to start HTTP Server on");
- } finally {
- if (socket != null) {
- try {
- socket.close();
- } catch (Exception e) {
- // CHECKSTYLE IGNORE THIS LINE
- }
- }
- }
-
FeConstants.runningUnitTest = true;
httpServer = new HttpServer();
- httpServer.setPort(HTTP_PORT);
+ // Let the actual server reserve its ephemeral port instead of closing
a temporary reservation.
+ httpServer.setPort(0);
httpServer.setMaxHttpPostSize(100 * 1024 * 1024);
httpServer.setAcceptors(2);
httpServer.setSelectors(4);
httpServer.setWorkers(0);
httpServer.start();
+ ServletWebServerApplicationContext applicationContext =
+ Deencapsulation.getField(httpServer, "applicationContext");
+ HTTP_PORT = applicationContext.getWebServer().getPort();
+ URI = "http://localhost:"; + HTTP_PORT + "/api/" + DB_NAME + "/" +
TABLE_NAME;
+ CloudURI = "http://localhost:"; + HTTP_PORT;
}
@AfterAll
public static void afterClass() {
+ httpServer.shutdown();
File file = new File(DORIS_HOME);
file.delete();
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/http/ForwardToMasterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/http/ForwardToMasterTest.java
index 83e12e73d97..386b30da2ec 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/http/ForwardToMasterTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/http/ForwardToMasterTest.java
@@ -41,6 +41,7 @@ public class ForwardToMasterTest extends DorisHttpTestCase {
.get()
.addHeader("Authorization", rootAuth)
.addHeader("forward_master_ut_test", "true")
+ .addHeader("Accept", "application/json")
.url(url)
.build();
Response response = networkClient.newCall(request).execute();
@@ -78,6 +79,7 @@ public class ForwardToMasterTest extends DorisHttpTestCase {
.addHeader("Content-Type", "application/json")
.addHeader("Authorization", rootAuth)
.addHeader("forward_master_ut_test", "true")
+ .addHeader("Accept", "application/json")
.post(RequestBody.create(jsonBody,
MediaType.parse("application/json")))
.build();
Response response = networkClient.newCall(request).execute();
@@ -91,6 +93,7 @@ public class ForwardToMasterTest extends DorisHttpTestCase {
.get()
.addHeader("Authorization", rootAuth)
.addHeader("forward_master_ut_test", "true")
+ .addHeader("Accept", "application/json")
.url(url)
.build();
Response response = networkClient.newCall(request).execute();
@@ -129,6 +132,7 @@ public class ForwardToMasterTest extends DorisHttpTestCase {
.addHeader("Content-Type", "application/json")
.addHeader("Authorization", rootAuth)
.addHeader("forward_master_ut_test", "true")
+ .addHeader("Accept", "application/json")
.post(RequestBody.create(jsonBody,
MediaType.parse("application/json")))
.build();
Response response = networkClient.newCall(request).execute();
@@ -141,6 +145,7 @@ public class ForwardToMasterTest extends DorisHttpTestCase {
Request request = new Request.Builder()
.get()
.addHeader("forward_master_ut_test", "true")
+ .addHeader("Accept", "application/json")
.addHeader("Authorization", rootAuth)
.url(url)
.build();
@@ -174,6 +179,7 @@ public class ForwardToMasterTest extends DorisHttpTestCase {
.url(url)
.addHeader("Authorization", rootAuth)
.addHeader("forward_master_ut_test", "true")
+ .addHeader("Accept", "application/json")
.post(emptyBody)
.build();
Response response = networkClient.newCall(request).execute();
@@ -199,6 +205,7 @@ public class ForwardToMasterTest extends DorisHttpTestCase {
.url(url)
.post(requestBody)
.addHeader("forward_master_ut_test", "true")
+ .addHeader("Accept", "application/json")
.addHeader("Authorization", rootAuth)
.build();
Response response = networkClient.newCall(request).execute();
@@ -219,6 +226,7 @@ public class ForwardToMasterTest extends DorisHttpTestCase {
.get()
.addHeader("Authorization", rootAuth)
.addHeader("forward_master_ut_test", "true")
+ .addHeader("Accept", "application/json")
.url(url)
.build();
Response response = networkClient.newCall(request).execute();
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 d5748699acc..d937a89d649 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
@@ -17,12 +17,15 @@
package org.apache.doris.nereids.rules.rewrite;
+import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.util.MemoPatternMatchSupported;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.utframe.TestWithFeService;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+
public class EliminateGroupByKeyByUniformTest extends TestWithFeService
implements MemoPatternMatchSupported {
@Override
protected void runBeforeAll() throws Exception {
@@ -131,22 +134,26 @@ public class EliminateGroupByKeyByUniformTest extends
TestWithFeService implemen
@Test
void testLeftJoinOnConditionNotRewrite() {
+ // An ON predicate does not filter the preserved side. The nullable
side can introduce NULL,
+ // so its non-null constant is not a uniform output value either; all
three keys must remain.
PlanChecker.from(connectContext)
.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(logicalAggregate().when(agg ->
agg.getGroupByExpressions().stream()
+
.map(Expression::toSql).toList().equals(Arrays.asList("b", "b", "c"))));
}
@Test
void testLeftJoinWhereConditionRewrite() {
+ // WHERE makes the preserved-side key constant. Keep the nullable-side
key because unmatched
+ // rows can still contain NULL instead of the constant inferred for
the right input.
PlanChecker.from(connectContext)
.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(logicalAggregate().when(agg ->
agg.getGroupByExpressions().stream()
+
.map(Expression::toSql).toList().equals(Arrays.asList("b", "c"))));
}
@Test
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 f584d59572a..5f3477e7b98 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
@@ -295,4 +295,33 @@ public class VariableMgrTest extends TestWithFeService {
new StringLiteral("0"))));
Assertions.assertTrue(blockSizeException.getMessage().contains("preferred_block_size_bytes"));
}
+
+ @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,
+ SessionVariable.EXTERNAL_SCAN_CONSISTENT_HASH_SPREAD_NUM,
new StringLiteral(value)));
+ Assertions.assertEquals(Integer.parseInt(value),
var.getExternalScanConsistentHashSpreadNum());
+ }
+ 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());
+ }
+ 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 00000000000..672d7229fff
--- /dev/null
+++
b/regression-test/data/external_table_p0/tvf/test_external_scan_consistent_hash_spread.out
@@ -0,0 +1,59 @@
+-- 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
+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 00000000000..663ad55a920
--- /dev/null
+++
b/regression-test/data/query_p0/ddl/set_and_unset_commands/test_external_scan_consistent_hash_spread_variable.out
@@ -0,0 +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
+
+-- !enabled --
+external_scan_consistent_hash_spread_num 3 1 1
+
+-- !maximum --
+external_scan_consistent_hash_spread_num 2147483647 1 1
+
+-- !automatic --
+external_scan_consistent_hash_spread_num 0 1 1
+
+-- !legacy --
+external_scan_consistent_hash_spread_num 1 1 0
+
+-- !after_errors --
+external_scan_consistent_hash_spread_num 1 1 0
+
+-- !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 00000000000..bdf76c6208c
--- /dev/null
+++
b/regression-test/suites/external_table_p0/tvf/test_external_scan_consistent_hash_spread.groovy
@@ -0,0 +1,88 @@
+// 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}",
+ "use_path_style" = "${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 = false"
+ 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")
+ sql "set enable_file_cache = false"
+ sql "set use_consistent_hash_for_external_scan = true"
+ for (int candidates : [0, 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}",
+ "use_path_style" = "${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 00000000000..6a608af1adf
--- /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'"
+
+ 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 0"
+ }
+ 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'"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
