diff --git a/fe/be-java-extensions/paimon-s3-shade/pom.xml b/fe/be-java-extensions/paimon-s3-shade/pom.xml index 02a80fc2bd386b..427e4f66b3abc6 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. + + + coverage + + + + org.jacoco + jacoco-maven-plugin + + + + **/META-INF/versions/** + + + + + + + + paimon-s3-shade ${project.basedir}/target/ 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 9dfc77fd01f1e0..731565ee2a4548 100644 --- a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcScanRange.java +++ b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcScanRange.java @@ -88,6 +88,12 @@ public String getTableFormatType() { return TABLE_FORMAT_TYPE; } + @Override + public String getSplitIdentity() { + return properties.getOrDefault(PARAM_PARTITION_DESCRIPTOR, + properties.getOrDefault(PARAM_QUERY_SQL, TABLE_FORMAT_TYPE)); + } + @Override public Map getProperties() { return properties; diff --git a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcScanRangeTest.java b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcScanRangeTest.java index b2db701e28f6e1..be8b51b34c5eb8 100644 --- a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcScanRangeTest.java +++ b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcScanRangeTest.java @@ -106,6 +106,22 @@ void carriesAPartitionDescriptorInsteadOfAStatement() { Assertions.assertFalse(params.containsKey("query_sql")); } + @Test + void usesPartitionDescriptorAsSchedulingIdentity() { + AdbcScanRange first = new AdbcScanRange.Builder() + .driverPath("/opt/doris/plugins/adbc_drivers/libadbc_driver_flightsql.so") + .uri("grpc://remote:9090") + .partitionDescriptor("partition-1") + .build(); + AdbcScanRange second = new AdbcScanRange.Builder() + .driverPath("/opt/doris/plugins/adbc_drivers/libadbc_driver_flightsql.so") + .uri("grpc://remote:9090") + .partitionDescriptor("partition-2") + .build(); + + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + } + @Test void 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 549d7fb5e48b2a..47699d4e2d40d6 100644 --- a/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java +++ b/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java @@ -178,6 +178,13 @@ public Optional getPath() { 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 0f720b91863607..009a7acb303de1 100644 --- a/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussScanRange.java +++ b/fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussScanRange.java @@ -177,6 +177,16 @@ public String getTableFormatType() { return TABLE_FORMAT; } + @Override + public String getSplitIdentity() { + return properties.getOrDefault(PROP_RANGE_TYPE, "") + "\u0001" + + properties.getOrDefault(PROP_PARTITION_ID, "") + "\u0001" + + properties.getOrDefault(PROP_BUCKET_ID, "") + "\u0001" + + properties.getOrDefault(PROP_LOG_START_OFFSET, "") + "\u0001" + + properties.getOrDefault(PROP_LOG_STOP_OFFSET, "") + "\u0001" + + properties.getOrDefault(PROP_KV_SNAPSHOT_ID, ""); + } + @Override public Map getProperties() { return properties; diff --git a/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussLakeRangeTest.java b/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussLakeRangeTest.java index 2494adc6f6546e..31e7a824a26f2b 100644 --- a/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussLakeRangeTest.java +++ b/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussLakeRangeTest.java @@ -174,6 +174,17 @@ public void theFormatNameIsTheScansOwn() { Assertions.assertEquals("fluss", FlussLakeRange.plain(new LakeSplit()).getTableFormatType()); } + @Test + public void changingTailDoesNotChangeLakeFileIdentity() { + ConnectorScanRange lake = new LakeSplit(); + FlussLakeRange first = FlussLakeRange.suppressed(lake, + new FlussLakeRange.Tail(DT_20260101, 2, 100L, 105L)); + FlussLakeRange second = FlussLakeRange.suppressed(lake, + new FlussLakeRange.Tail(DT_20260101, 2, 100L, 205L)); + + Assertions.assertEquals(first.getSplitIdentity(), second.getSplitIdentity()); + } + /** * The payload is both halves, in their own thrift fields: whatever the sibling wrote, untouched, plus * the fluss descriptor. Asserted whole rather than by spot check — this is an untyped string map that diff --git a/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussScanRangeTest.java b/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussScanRangeTest.java index 99b4a058a99a60..8089638b3da60e 100644 --- a/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussScanRangeTest.java +++ b/fe/fe-connector/fe-connector-fluss/src/test/java/org/apache/doris/connector/fluss/FlussScanRangeTest.java @@ -50,6 +50,7 @@ public class FlussScanRangeTest { @Test public void logRangeCarriesOffsetsAndNothingElse() { FlussScanRange range = FlussScanRange.log(FlussScanRange.Partition.NONE, 3, 10L, 42L); + FlussScanRange otherRange = FlussScanRange.log(FlussScanRange.Partition.NONE, 4, 10L, 42L); Map expected = new LinkedHashMap<>(); expected.put("fluss.range_type", "LOG"); @@ -58,6 +59,7 @@ public void logRangeCarriesOffsetsAndNothingElse() { expected.put("fluss.log_stop_offset", "42"); Assertions.assertEquals(expected, range.getProperties()); Assertions.assertEquals(FlussScanRange.RangeType.LOG, range.getRangeType()); + Assertions.assertNotEquals(range.getSplitIdentity(), otherRange.getSplitIdentity()); } @Test diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java index 9d8fd8312738ae..a22ac6341226c2 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java @@ -156,6 +156,12 @@ private IcebergScanRange(Builder builder) { this.positionDeleteContentSizeInBytes = builder.positionDeleteContentSizeInBytes; } + @Override + public String getSplitIdentity() { + // JNI metadata tasks share a dummy path; native ranges retain their actual file identity. + return serializedSplit != null ? serializedSplit : ConnectorScanRange.super.getSplitIdentity(); + } + @Override public Optional getPath() { return Optional.ofNullable(path); diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanRangeTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanRangeTest.java index 2902283d70d957..e657951100b4f6 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanRangeTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanRangeTest.java @@ -42,6 +42,26 @@ */ public class IcebergScanRangeTest { + @Test + public void metadataTasksUseTaskIdentityAndNativeRangesKeepFileIdentity() { + IcebergScanRange first = new IcebergScanRange.Builder().path("/dummyPath") + .serializedSplit("task-a").build(); + IcebergScanRange second = new IcebergScanRange.Builder().path("/dummyPath") + .serializedSplit("task-b").build(); + IcebergScanRange sameWork = new IcebergScanRange.Builder().path("/dummyPath") + .serializedSplit("task-a").build(); + Assertions.assertEquals(first.getPath(), second.getPath()); + Assertions.assertEquals(first.getStart(), second.getStart()); + Assertions.assertEquals(first.getLength(), second.getLength()); + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + Assertions.assertEquals(first.getSplitIdentity(), sameWork.getSplitIdentity()); + IcebergScanRange nativeRange = new IcebergScanRange.Builder().path("s3://bucket/file.parquet").build(); + Assertions.assertEquals("s3://bucket/file.parquet", nativeRange.getSplitIdentity()); + IcebergScanRange nativeDeletes = new IcebergScanRange.Builder().path("s3://bucket/delete.parquet") + .positionDeleteSysTableSplit(1, TFileFormatType.FORMAT_PARQUET, "s3://bucket/delete.parquet").build(); + Assertions.assertEquals("s3://bucket/delete.parquet", nativeDeletes.getSplitIdentity()); + } + @Test public void builderProducesFileScanRangeWithFileFields() { IcebergScanRange range = new IcebergScanRange.Builder() diff --git a/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputeScanRange.java b/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputeScanRange.java index 7e00f3b19875f6..f74d2e67980bfd 100644 --- a/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputeScanRange.java +++ b/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputeScanRange.java @@ -74,6 +74,12 @@ public Optional getPath() { return Optional.of("/row_offset"); } + @Override + public String getSplitIdentity() { + // Offsets and split indices are relative to the ODPS read session, not a physical file. + return getPath().get() + "/" + properties.get(PROP_SESSION_ID); + } + @Override public long getStart() { return start; diff --git a/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputeScanRangeTest.java b/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputeScanRangeTest.java index 3ce6e9b34bd9e3..55b9e1edfb6aef 100644 --- a/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputeScanRangeTest.java +++ b/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputeScanRangeTest.java @@ -63,6 +63,24 @@ */ public class MaxComputeScanRangeTest { + @Test + public void equalCoordinatesFromDifferentSessionsHaveDifferentIdentities() { + for (String type : new String[] {MaxComputeScanRange.SPLIT_TYPE_ROW_OFFSET, + MaxComputeScanRange.SPLIT_TYPE_BYTE_SIZE}) { + MaxComputeScanRange first = MaxComputeScanRange.builder().splitType(type) + .sessionId("session-a").start(0).length(1000).build(); + MaxComputeScanRange second = MaxComputeScanRange.builder().splitType(type) + .sessionId("session-b").start(0).length(1000).build(); + MaxComputeScanRange sameWork = MaxComputeScanRange.builder().splitType(type) + .sessionId("session-a").start(0).length(1000).build(); + Assertions.assertEquals(first.getPath(), second.getPath()); + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + Assertions.assertEquals(first.getSplitIdentity(), sameWork.getSplitIdentity()); + Assertions.assertEquals(0, first.getStart()); + Assertions.assertEquals(1000, first.getLength()); + } + } + private static final long SPLIT_BYTE_SIZE = 268435456L; // ODPS default byte-size split @Test diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanRange.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanRange.java index 77c0d073659bfa..ecb351d61c1fd7 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanRange.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanRange.java @@ -125,6 +125,11 @@ public Optional getPath() { return Optional.ofNullable(path); } + @Override + public String getSplitIdentity() { + return path != null ? path : properties.getOrDefault("paimon.split", "paimon"); + } + @Override public long getStart() { return start; diff --git a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java index 572a78ba1e0e78..a4ac22e52c0d73 100644 --- a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java +++ b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java @@ -46,6 +46,14 @@ default Optional getPath() { return Optional.empty(); } + /** + * Returns a stable identity used to assign connector ranges to external-scan backends. Connectors whose JNI + * ranges do not have a path should override this with their serialized split identity. + */ + default String getSplitIdentity() { + return getPath().orElse(""); + } + /** Returns the byte offset to start reading from. */ default long getStart() { return 0; diff --git a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java index fa459f99076f11..f76a5378caccee 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java +++ b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java @@ -87,8 +87,8 @@ public void connectorApiMajorTracksTheRecordedSurfaceChange() throws IOException Assertions.assertNotNull(in, "missing connector plugin API version resource"); version.load(in); } - // Major 13 carries textual write-input semantics on ConnectorColumn. - 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 void connectorApiMajorTracksTheRecordedSurfaceChange() throws IOException org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot.class, org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot.Builder.class, ConnectorScanPlanProvider.class, + org.apache.doris.connector.spi.scan.ConnectorScanRange.class, ConnectorWriteHandle.class, ConnectorChangelogMode.class, ConnectorRowLevelDmlRequest.class, diff --git a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt index f0ae0af27d158c..7776066bcf1c31 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt +++ b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt @@ -168,6 +168,24 @@ org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#supportsSystemTabl org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#supportsSystemTableTimeTravel():boolean org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#supportsTableSample():boolean org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#usesHiveParquetInt96TimeZone():boolean +org.apache.doris.connector.spi.scan.ConnectorScanRange#getFileFormat():java.lang.String +org.apache.doris.connector.spi.scan.ConnectorScanRange#getFileSize():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getHosts():java.util.List +org.apache.doris.connector.spi.scan.ConnectorScanRange#getLength():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getModificationTime():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getPartitionValues():java.util.Map +org.apache.doris.connector.spi.scan.ConnectorScanRange#getPath():java.util.Optional +org.apache.doris.connector.spi.scan.ConnectorScanRange#getProperties():java.util.Map +org.apache.doris.connector.spi.scan.ConnectorScanRange#getPushDownRowCount():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getSelfSplitWeight():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getSplitIdentity():java.lang.String +org.apache.doris.connector.spi.scan.ConnectorScanRange#getStart():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#getTableFormatType():java.lang.String +org.apache.doris.connector.spi.scan.ConnectorScanRange#getTargetSplitSize():long +org.apache.doris.connector.spi.scan.ConnectorScanRange#isNativeReadRange():boolean +org.apache.doris.connector.spi.scan.ConnectorScanRange#isPartitionBearing():boolean +org.apache.doris.connector.spi.scan.ConnectorScanRange#isSingleUse():boolean +org.apache.doris.connector.spi.scan.ConnectorScanRange#populateRangeParams(org.apache.doris.thrift.TTableFormatFileDesc,org.apache.doris.thrift.TFileRangeDesc):void org.apache.doris.connector.spi.scan.ScanNodePropertyKeys#field:FILE_FORMAT_TYPE:java.lang.String=file_format_type org.apache.doris.connector.spi.scan.ScanNodePropertyKeys#field:LOCATION_PREFIX:java.lang.String=location. org.apache.doris.connector.spi.scan.ScanNodePropertyKeys#field:PATH_PARTITION_KEYS:java.lang.String=path_partition_keys diff --git a/fe/fe-connector/fe-connector-trino/src/main/java/org/apache/doris/connector/trino/TrinoScanRange.java b/fe/fe-connector/fe-connector-trino/src/main/java/org/apache/doris/connector/trino/TrinoScanRange.java index 0f4d363b09052f..ebac65fa310690 100644 --- a/fe/fe-connector/fe-connector-trino/src/main/java/org/apache/doris/connector/trino/TrinoScanRange.java +++ b/fe/fe-connector/fe-connector-trino/src/main/java/org/apache/doris/connector/trino/TrinoScanRange.java @@ -79,6 +79,11 @@ public String getTableFormatType() { return "trino_connector"; } + @Override + public String getSplitIdentity() { + return properties.getOrDefault(KEY_SPLIT, "trino_connector"); + } + @Override public Map getProperties() { return properties; diff --git a/fe/fe-connector/pom.xml b/fe/fe-connector/pom.xml index 9bf53c1050bfa3..913f9ba889077e 100644 --- a/fe/fe-connector/pom.xml +++ b/fe/fe-connector/pom.xml @@ -55,7 +55,7 @@ under the License. of the latter two means bumping this property as well (and fe-extension-spi means bumping all five families). --> - 13.0 + 14.0 diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java index 3cccf8821f9257..58047dec8a9528 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java @@ -22,16 +22,23 @@ import org.apache.doris.datasource.split.FileSplit; import java.nio.ByteBuffer; +import java.util.Base64; public class RemoteDorisSplit extends FileSplit { private static final LocationPath DUMMY_PATH = LocationPath.of("/dummyPath"); private final String location; private final ByteBuffer ticket; + private final String splitIdentity; public RemoteDorisSplit(String location, ByteBuffer ticket) { super(DUMMY_PATH, 0, 0, 0, 0, null, null); this.location = location; this.ticket = ticket; + // Capture the work identity without consuming the buffer used to populate the execution range. + ByteBuffer identityTicket = ticket.asReadOnlyBuffer(); + byte[] ticketBytes = new byte[identityTicket.remaining()]; + identityTicket.get(ticketBytes); + this.splitIdentity = location.length() + ":" + location + Base64.getEncoder().encodeToString(ticketBytes); this.tableFormatType = TableFormatType.REMOTE_DORIS; } @@ -43,6 +50,11 @@ public String getLocation() { 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 c378c285dc55b2..20269cd67c1c8a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/ExternalScanNode.java @@ -44,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 4b4572b2ce76f3..b5080ef343385d 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.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 Map getAssignedWeightPerBackend() { } private Map assignedWeightPerBackend = Maps.newHashMap(); + private final NavigableMap assignedWeightCounts = new TreeMap<>(); protected ConsistentHash consistentHash; @@ -90,6 +95,8 @@ public Map getAssignedWeightPerBackend() { private NodeSelectionStrategy nodeSelectionStrategy; private boolean enableSplitsRedistribution = true; + private final int consistentHashSpreadNum; + private final IntUnaryOperator randomIndex; // Create a ConsistentHash ring may be a time-consuming operation, so we cache it. private static LoadingCache> consistentHashCache; @@ -142,7 +149,20 @@ public String toString() { } public FederationBackendPolicy(NodeSelectionStrategy nodeSelectionStrategy) { + this(nodeSelectionStrategy, 1); + } + + public FederationBackendPolicy(NodeSelectionStrategy nodeSelectionStrategy, int consistentHashSpreadNum) { + this(nodeSelectionStrategy, consistentHashSpreadNum, bound -> ThreadLocalRandom.current().nextInt(bound)); + } + + @VisibleForTesting + FederationBackendPolicy(NodeSelectionStrategy nodeSelectionStrategy, int consistentHashSpreadNum, + IntUnaryOperator randomIndex) { + Preconditions.checkArgument(consistentHashSpreadNum >= 0, "consistentHashSpreadNum must be at least 0"); this.nodeSelectionStrategy = nodeSelectionStrategy; + this.consistentHashSpreadNum = consistentHashSpreadNum; + this.randomIndex = randomIndex; } public FederationBackendPolicy() { @@ -191,10 +211,12 @@ public void init(BeSelectionPolicy policy) throws UserException { 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 Multimap computeScanRangeAssignment(List splits) t Optional chosenNode = candidateNodes.stream() .min(Comparator.comparingLong(ownerNode -> assignedWeightPerBackend.get(ownerNode))); - if (chosenNode.isPresent()) { + // A host hint is a local preference for remotely readable splits. In spread mode, do not + // let a repeated hint monopolize one backend while other eligible backends are idle. + boolean localNodeIsWithinGlobalMinimum = chosenNode.isPresent(); + if (localNodeIsWithinGlobalMinimum && isSpreadEnabled()) { + long minimumAssignedWeight = 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 Multimap computeScanRangeAssignment(List splits) t } case RANDOM: { randomCandidates.reset(); - candidateNodes = selectNodes(Config.split_assigner_min_random_candidate_num, randomCandidates); + candidateNodes = consistentHashSpreadNum == 0 ? backends + : selectNodes(Config.split_assigner_min_random_candidate_num, randomCandidates); break; } case CONSISTENT_HASHING: { - candidateNodes = consistentHash.getNode(split, - 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 Multimap computeScanRangeAssignment(List splits) t throw new UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG); } - Backend selectedBackend = chooseNodeForSplit(candidateNodes); - List 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 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 @@ private void equateDistribution(ListMultimap assignment) { // move split from max to min Split redistributedSplit = redistributeSplit(assignment, maxNode, minNode); - assignedWeightPerBackend.put(maxNode, - assignedWeightPerBackend.get(maxNode) - redistributedSplit.getSplitWeight().getRawValue()); - assignedWeightPerBackend.put(minNode, Math.addExact( - assignedWeightPerBackend.get(minNode), redistributedSplit.getSplitWeight().getRawValue())); + addAssignedWeight(maxNode, -redistributedSplit.getSplitWeight().getRawValue()); + addAssignedWeight(minNode, redistributedSplit.getSplitWeight().getRawValue()); // add max back into maxNodes only if it still has assignments if (assignment.containsKey(maxNode)) { @@ -489,6 +532,50 @@ private Backend chooseNodeForSplit(List candidateNodes) { 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 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 void funnel(Backend backend, PrimitiveSink primitiveSink) { private static class SplitHash implements Funnel { @Override public void funnel(Split split, PrimitiveSink primitiveSink) { - primitiveSink.putBytes(split.getConsistentHashString().getBytes(StandardCharsets.UTF_8)); + primitiveSink.putBytes(split.getSplitIdentity().getBytes(StandardCharsets.UTF_8)); primitiveSink.putLong(split.getStart()); primitiveSink.putLong(split.getLength()); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/PluginDrivenSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/PluginDrivenSplit.java index 2504b09adf0b9f..85a8c2b2c270a7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/PluginDrivenSplit.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/PluginDrivenSplit.java @@ -35,6 +35,7 @@ public class PluginDrivenSplit extends FileSplit { private final ConnectorScanRange connectorScanRange; + private final String splitIdentity; public PluginDrivenSplit(ConnectorScanRange scanRange) { super(buildPath(scanRange), @@ -45,6 +46,7 @@ public PluginDrivenSplit(ConnectorScanRange scanRange) { scanRange.getHosts().toArray(new String[0]), buildPartitionValues(scanRange)); this.connectorScanRange = scanRange; + this.splitIdentity = buildSplitIdentity(scanRange); // FIX-A1: thread the connector's proportional split weight into the FileSplit scheduling fields so // FederationBackendPolicy distributes by size (legacy parity) instead of uniform standard() weight. // Set ONLY when the connector provides BOTH a weight (>= 0; 0 is a real weight) and a positive @@ -73,6 +75,21 @@ public String getPathString() { return connectorScanRange.getPath().orElse("connector://virtual"); } + /** + * Returns a stable connector supplied identity for consistent-hash scheduling. JNI ranges often have no + * file path or byte range, so include their immutable connector metadata instead of collapsing all of them + * onto the same virtual path. + */ + @Override + public String getSplitIdentity() { + return splitIdentity; + } + + private static String buildSplitIdentity(ConnectorScanRange scanRange) { + String identity = scanRange.getSplitIdentity(); + return identity.isEmpty() ? scanRange.getTableFormatType() : identity; + } + private static LocationPath buildPath(ConnectorScanRange scanRange) { String pathStr = scanRange.getPath().orElse("connector://virtual"); return LocationPath.of(pathStr); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index 3f04a2b4ac29fc..405d76e0908a98 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 enum IgnoreSplitType { public static final String USE_CONSISTENT_HASHING_FOR_EXTERNAL_SCAN = "use_consistent_hash_for_external_scan"; @VarAttrDef.VarAttr(name = USE_CONSISTENT_HASHING_FOR_EXTERNAL_SCAN, + needForward = true, description = "Use consistent hashing to split the appearance for external scan") public boolean useConsistentHashForExternalScan = false; + 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 boolean getUseConsistentHashForExternalScan() { 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 48b6acd68e39c2..a5aa55186a237d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/spi/Split.java +++ b/fe/fe-core/src/main/java/org/apache/doris/spi/Split.java @@ -48,7 +48,7 @@ default boolean isRemotelyAccessible() { void setAlternativeHosts(List alternativeHosts); - default String getConsistentHashString() { + default String getSplitIdentity() { return getPathString(); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/source/RemoteDorisSplitTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/source/RemoteDorisSplitTest.java new file mode 100644 index 00000000000000..4d15b9e04c677a --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/source/RemoteDorisSplitTest.java @@ -0,0 +1,72 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource.doris.source; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; + +public class RemoteDorisSplitTest { + @Test + public void sameLocationWithDifferentTicketsHasDifferentIdentity() { + RemoteDorisSplit first = new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 2})); + RemoteDorisSplit second = new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 3})); + RemoteDorisSplit sameWork = new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 2})); + RemoteDorisSplit otherLocation = new RemoteDorisSplit("grpc://other:8815", ByteBuffer.wrap(new byte[] {1, 2})); + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + Assertions.assertEquals(first.getSplitIdentity(), sameWork.getSplitIdentity()); + Assertions.assertNotEquals(first.getSplitIdentity(), otherLocation.getSplitIdentity()); + } + + @Test + public void identityCapturePreservesExecutionBufferAndRemainsImmutable() { + ByteBuffer ticket = ByteBuffer.wrap(new byte[] {9, 1, 2, 8}); + ticket.position(1); + ticket.limit(3); + ticket.mark(); + RemoteDorisSplit split = new RemoteDorisSplit("grpc://proxy:8815", ticket); + String identity = split.getSplitIdentity(); + Assertions.assertSame(ticket, split.getTicket()); + Assertions.assertEquals(1, ticket.position()); + Assertions.assertEquals(3, ticket.limit()); + Assertions.assertEquals(identity, + new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 2})).getSplitIdentity()); + ticket.reset(); + ticket.put(1, (byte) 7); + ticket.position(2); + Assertions.assertEquals(identity, split.getSplitIdentity()); + Assertions.assertEquals(2, ticket.position()); + Assertions.assertEquals(3, ticket.limit()); + } + + @Test + public void directReadOnlyTicketUsesTheRemainingBytes() { + ByteBuffer direct = ByteBuffer.allocateDirect(4); + direct.put(new byte[] {9, 1, 2, 8}); + direct.flip(); + direct.position(1); + direct.limit(3); + ByteBuffer ticket = direct.asReadOnlyBuffer(); + RemoteDorisSplit split = new RemoteDorisSplit("grpc://proxy:8815", ticket); + Assertions.assertEquals(1, ticket.position()); + Assertions.assertEquals(3, ticket.limit()); + Assertions.assertEquals(split.getSplitIdentity(), + new RemoteDorisSplit("grpc://proxy:8815", ByteBuffer.wrap(new byte[] {1, 2})).getSplitIdentity()); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FederationBackendPolicySpreadTest.java new file mode 100644 index 00000000000000..ea1b2c683d64c0 --- /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 mockedEnv; + private ConnectContext previousContext; + private ConnectContext context; + private ComputeGroup computeGroup; + private List backends; + private int originalCandidateNum; + private boolean originalLocalScheduling; + + @BeforeEach + public void setUp() { + previousContext = ConnectContext.get(); + originalCandidateNum = Config.split_assigner_min_consistent_hash_candidate_num; + originalLocalScheduling = Config.split_assigner_optimized_local_scheduling; + Config.split_assigner_min_consistent_hash_candidate_num = 2; + Config.split_assigner_optimized_local_scheduling = true; + mockedEnv = Mockito.mockStatic(Env.class); + mockedEnv.when(Env::getCurrentEnv).thenReturn(Mockito.mock(Env.class)); + backends = new ArrayList<>(); + long firstId = NEXT_BACKEND_ID.getAndAdd(3); + for (int i = 0; i < 3; i++) { + Backend backend = new Backend(firstId + i, "192.0.2." + (i + 1), 9050); + backend.setAlive(true); + backends.add(backend); + } + computeGroup = Mockito.mock(ComputeGroup.class); + Mockito.when(computeGroup.getBackendList()).thenAnswer(invocation -> backends); + context = new ConnectContext(); + context.setComputeGroup(computeGroup); + context.setThreadLocalInfo(); + } + + @AfterEach + public void tearDown() { + Config.split_assigner_min_consistent_hash_candidate_num = originalCandidateNum; + Config.split_assigner_optimized_local_scheduling = originalLocalScheduling; + mockedEnv.close(); + ConnectContext.remove(); + if (previousContext != null) { + previousContext.setThreadLocalInfo(); + } + } + + private FileSplit split() { + return new FileSplit(LocationPath.of("s3://bucket/hot.csv"), 0, 1000, 1000, + 0, new String[0], Collections.emptyList()); + } + + private FederationBackendPolicy policy(int candidates, IntUnaryOperator random) throws UserException { + FederationBackendPolicy policy = new FederationBackendPolicy( + NodeSelectionStrategy.CONSISTENT_HASHING, candidates, random); + policy.init(); + return policy; + } + + private Backend assign(FederationBackendPolicy policy, Split split) throws UserException { + Multimap assignment = policy.computeScanRangeAssignment( + new ArrayList<>(Collections.singletonList(split))); + Assertions.assertEquals(1, assignment.size()); + Assertions.assertEquals(1, assignment.keySet().size()); + Assertions.assertSame(split, assignment.values().iterator().next()); + return assignment.keySet().iterator().next(); + } + + private static IntUnaryOperator selectCandidate(int index) { + // 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 candidates = new HashSet<>(policy.consistentHash.getNode(split(), 2)); + for (int batch = 0; batch < 2; batch++) { + List splits = new ArrayList<>(); + for (int i = 0; i < 20; i++) { + splits.add(split()); + } + Multimap assignment = policy.computeScanRangeAssignment(splits); + Assertions.assertEquals(20, assignment.size()); + Assertions.assertTrue(candidates.containsAll(assignment.keySet())); + for (Backend backend : candidates) { + Assertions.assertEquals(10, assignment.get(backend).size()); + Assertions.assertEquals((batch + 1) * 1000L, policy.getAssignedWeightPerBackend().get(backend)); + } + } + for (Backend backend : backends) { + if (!candidates.contains(backend)) { + Assertions.assertEquals(0L, policy.getAssignedWeightPerBackend().get(backend)); + } + } + } + + @Test + public void testSyntheticPathsDoNotConfineDistinctWorkToOneCandidateSet() throws Exception { + backends = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + Backend backend = new Backend(NEXT_BACKEND_ID.getAndIncrement(), "192.0.2." + (i + 1), 9050); + backend.setAlive(true); + backends.add(backend); + } + for (String path : Arrays.asList("/row_offset", "/dummyPath")) { + FederationBackendPolicy policy = policy(2, bound -> 0); + Set targets = new HashSet<>(); + for (int i = 0; i < 100; i++) { + String identity = path + "/work-" + i; + PluginDrivenSplit split = new PluginDrivenSplit(new ConnectorScanRange() { + @Override + public Map getProperties() { + return Collections.emptyMap(); + } + + @Override + public Optional getPath() { + return Optional.of(path); + } + + @Override + public String getSplitIdentity() { + return identity; + } + }); + List candidates = policy.consistentHash.getNode(split, 2); + Backend selected = assign(policy, split); + Assertions.assertTrue(candidates.contains(selected)); + targets.add(selected); + } + Assertions.assertTrue(targets.size() > 2, + "Distinct work with equal synthetic paths must escape a single two-BE candidate set"); + } + } + + @Test + public void testRemoteDorisTicketsAtSameProxyUseDistinctCandidateSets() throws Exception { + backends = new ArrayList<>(); + for (int i = 0; i < 8; i++) { + Backend backend = new Backend(NEXT_BACKEND_ID.getAndIncrement(), "192.0.2." + (i + 1), 9050); + backend.setAlive(true); + backends.add(backend); + } + FederationBackendPolicy policy = policy(2, bound -> 0); + Set targets = new HashSet<>(); + for (int i = 0; i < 100; i++) { + ByteBuffer ticket = ByteBuffer.wrap(new byte[] {(byte) i}); + RemoteDorisSplit split = new RemoteDorisSplit("grpc://proxy:8815", ticket); + List candidates = policy.consistentHash.getNode(split, 2); + Backend selected = assign(policy, split); + Assertions.assertTrue(candidates.contains(selected)); + Assertions.assertEquals(0, ticket.position()); + targets.add(selected); + } + Assertions.assertTrue(targets.size() > 2, + "Different Flight tickets at the same proxy must escape a single candidate pair"); + } + + @Test + public void testWeightHistogramRemainsIncrementalAcrossSingletonBatches() throws Exception { + for (int spread : Arrays.asList(0, 1, 2)) { + FederationBackendPolicy policy = policy(spread, bound -> 0); + for (int batch = 0; batch < 30; batch++) { + assign(policy, split()); + Map expected = new TreeMap<>(); + for (Long weight : policy.getAssignedWeightPerBackend().values()) { + expected.merge(weight, 1, Integer::sum); + } + Assertions.assertEquals(expected, Deencapsulation.getField(policy, "assignedWeightCounts")); + } + } + } + + @Test + public void testMinimumWeightWinsBeforeRandomTieBreaking() throws Exception { + FederationBackendPolicy policy = policy(3, selectCandidate(2)); + List candidates = policy.consistentHash.getNode(split(), 3); + 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 candidates = disabled.consistentHash.getNode(split(), 2); + Assertions.assertEquals(candidates.get(1), assign(disabled, split())); + Assertions.assertEquals(candidates.get(1), assign(original, split())); + } + + @Test + public void testDefaultPreservesGlobalRedistribution() throws Exception { + FederationBackendPolicy disabled = policy(1, bound -> { + throw new AssertionError("Disabled spread must not use randomness"); + }); + List splits = new ArrayList<>(); + for (int i = 0; i < 30; i++) { + splits.add(split()); + } + Assertions.assertEquals(3, disabled.computeScanRangeAssignment(splits).keySet().size()); + } + + @Test + public void testCandidateCountIsCappedByEligibleBackends() throws Exception { + Set targets = new HashSet<>(); + for (int index = 0; index < 3; index++) { + targets.add(assign(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 candidates = policy.consistentHash.getNode(split, 2); + Backend local = backends.stream().filter(backend -> !candidates.contains(backend)).findFirst().get(); + split.setHosts(new String[] {local.getHost()}); + Assertions.assertEquals(local, assign(policy, split)); + } + + @Test + public void testNonRemoteSplitsRetainHostConstraintWithoutRedistribution() throws Exception { + for (boolean preferLocal : Arrays.asList(true, false)) { + Config.split_assigner_optimized_local_scheduling = preferLocal; + FederationBackendPolicy policy = policy(3, bound -> { + throw new AssertionError("Non-remote selection must not use spread randomness"); + }); + FileSplit split = new FileSplit(LocationPath.of("file:///hot.csv"), 0, 1000, 1000, + 0, new String[] {backends.get(0).getHost()}, Collections.emptyList()) { + @Override + public boolean isRemotelyAccessible() { + return false; + } + }; + List splits = new ArrayList<>(Collections.nCopies(12, split)); + Multimap assignment = policy.computeScanRangeAssignment(splits); + Assertions.assertEquals(12, assignment.size()); + Assertions.assertEquals(Collections.singleton(backends.get(0)), assignment.keySet()); + } + } + + @Test + public void testRemoteLocalityIsIgnoredWhenLocalPreferenceIsDisabled() throws Exception { + Config.split_assigner_optimized_local_scheduling = false; + FederationBackendPolicy policy = policy(2, selectCandidate(1)); + FileSplit split = split(); + List candidates = policy.consistentHash.getNode(split, 2); + Backend local = backends.stream().filter(backend -> !candidates.contains(backend)).findFirst().get(); + split.setHosts(new String[] {local.getHost()}); + Assertions.assertEquals(candidates.get(1), assign(policy, split)); + } + + @Test + public void testQueryAndLoadDisabledBackendsAreExcluded() throws Exception { + backends.get(1).setQueryDisabled(true); + backends.get(2).setLoadDisabled(true); + FederationBackendPolicy policy = policy(3, bound -> bound - 1); + Assertions.assertEquals(1, policy.numBackends()); + Assertions.assertEquals(backends.get(0), assign(policy, split())); + } + + @Test + public void testMissingRequiredHostFails() throws Exception { + FederationBackendPolicy policy = policy(3, bound -> bound - 1); + FileSplit split = new FileSplit(LocationPath.of("file:///hot.csv"), 0, 1000, 1000, + 0, new String[] {"192.0.2.200"}, Collections.emptyList()) { + @Override + public boolean isRemotelyAccessible() { + return false; + } + }; + Assertions.assertThrows(UserException.class, () -> assign(policy, split)); + } + + @Test + public void testNoEligibleBackendFails() { + backends.clear(); + Assertions.assertThrows(UserException.class, () -> policy(3, bound -> bound - 1)); + } + + @Test + public void testInvalidSpreadCountFails() { + Assertions.assertThrows(IllegalArgumentException.class, + () -> new FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING, -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 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 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 splits = new ArrayList<>(); + String localHost = backends.get(0).getHost(); + for (int i = 0; i < 30; i++) { + splits.add(new FileSplit(LocationPath.of("s3://bucket/file-" + i), 0, 1000, 1000, + 0, new String[] {localHost}, Collections.emptyList())); + } + + Multimap assignment = policy.computeScanRangeAssignment(splits); + + Assertions.assertTrue(assignment.keySet().size() > 1); + Assertions.assertTrue(assignment.containsKey(backends.get(0)), + "the first local preference should remain eligible in spread mode"); + } + + @Test + public void testLegacyDefaultAndAutomaticExternalScanStrategy() { + SessionVariable session = context.getSessionVariable(); + 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 6345c869de1cda..575d23915bc44d 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java @@ -64,6 +64,7 @@ public void mixedNativeRangesKeepParquetScannerContract() { TFileScanRangeParams params = new TFileScanRangeParams(); Deencapsulation.setField(node, "params", params); ConnectorScanRange scanRange = Mockito.mock(ConnectorScanRange.class); + Mockito.when(scanRange.getSplitIdentity()).thenReturn(""); PluginDrivenSplit split = new PluginDrivenSplit(scanRange); TFileRangeDesc range = new TFileRangeDesc(); range.setFormatType(TFileFormatType.FORMAT_ORC); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/PluginDrivenSplitPartitionValuesTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/PluginDrivenSplitPartitionValuesTest.java index 2deb928f853a7e..b17ecfc0ad99cc 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/PluginDrivenSplitPartitionValuesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/PluginDrivenSplitPartitionValuesTest.java @@ -93,4 +93,26 @@ public void nonEmptyMapYieldsValuesInOrder() { PluginDrivenSplit split = new PluginDrivenSplit(range(parts, true)); Assertions.assertEquals(java.util.Arrays.asList("1", "2"), split.getPartitionValues()); } + + @Test + public void connectorMetadataParticipatesInConsistentHashIdentity() { + PluginDrivenSplit first = new PluginDrivenSplit(rangeWithProperty("paimon.split", "split-1")); + PluginDrivenSplit second = new PluginDrivenSplit(rangeWithProperty("paimon.split", "split-2")); + + Assertions.assertNotEquals(first.getSplitIdentity(), second.getSplitIdentity()); + } + + private static ConnectorScanRange rangeWithProperty(String key, String value) { + return new ConnectorScanRange() { + @Override + public Map getProperties() { + return Collections.singletonMap(key, value); + } + + @Override + public String getSplitIdentity() { + return value; + } + }; + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java b/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java index 17d0e7c83a7051..4c06874d8d02be 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.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 @@ private static void assignBackends() { @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 83e12e73d97212..386b30da2ec5fc 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 void testAddBeDropBe() throws Exception { .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 void testAddBeDropBe() throws Exception { .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 void testAddBeDropBe() throws Exception { .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 void testAddBeDropBe() throws Exception { .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 void testAddBeDropBe() throws Exception { 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 void testPost1() throws Exception { .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 void testPost2() throws Exception { .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 void testGet1() throws Exception { .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 d5748699acc869..d937a89d649932 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 @@ void testInnerJoin() { @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 f584d59572a5ff..5f3477e7b98ddf 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 void testAdaptiveBatchSizeRejectsZeroByteValues() { 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 00000000000000..672d7229fffc2e --- /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 00000000000000..663ad55a92027d --- /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 00000000000000..bdf76c6208ca9f --- /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 00000000000000..6a608af1adf994 --- /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'" +}