Gabriel39 · Oct 15, 2026 · Oct 8, 2026 · Oct 8, 2026 · Oct 8, 2026 · Oct 9, 2026
@@ -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>

@@ -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<String, String> getProperties() {
         return properties;

@@ -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

@@ -178,6 +178,13 @@ public Optional<String> 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();

@@ -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<String, String> getProperties() {
         return properties;

@@ -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

@@ -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 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

@@ -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<String> getPath() {
         return Optional.ofNullable(path);

@@ -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()

@@ -74,6 +74,12 @@ public Optional<String> 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;

@@ -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

@@ -125,6 +125,11 @@ public Optional<String> getPath() {
         return Optional.ofNullable(path);
     }
 
+    @Override
+    public String getSplitIdentity() {
+        return path != null ? path : properties.getOrDefault("paimon.split", "paimon");
+    }
+
     @Override
     public long getStart() {
         return start;

@@ -46,6 +46,14 @@ default Optional<String> 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;

@@ -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,

@@ -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

@@ -79,6 +79,11 @@ public String getTableFormatType() {
         return "trino_connector";
     }
 
+    @Override
+    public String getSplitIdentity() {
+        return properties.getOrDefault(KEY_SPLIT, "trino_connector");
+    }
+
     @Override
     public Map<String, String> getProperties() {
         return properties;

@@ -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>

@@ -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;
     }

@@ -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) {