[core] Optimize data evolution row-id assignment

docs/generated/core_configuration.html

@@ -170,6 +170,12 @@
             <td>Boolean</td>
             <td>Whether enabled chain table.</td>
         </tr>
+        <tr>
+            <td><h5>chain-table.split.key-range-enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>If true, a batch chain-table scan splits each bucket's snapshot and delta files into multiple splits by key range to improve read parallelism. Files with intersecting key ranges always stay in the same split so that all versions of a key across the snapshot and delta branches are merged together. Set to false to fall back to one split per bucket.</td>
+        </tr>
         <tr>
             <td><h5>chain-table.streaming.merge-snapshot</h5></td>
             <td style="word-wrap: break-word;">false</td>
@@ -506,6 +512,12 @@
             <td>Boolean</td>
             <td>Whether to persist source when process merge into action on data evolution table.</td>
         </tr>
+        <tr>
+            <td><h5>data-evolution.reassign.skip-contiguous-row-count</h5></td>
+            <td style="word-wrap: break-word;">1000000000</td>
+            <td>Long</td>
+            <td>Strictly contiguous same-partition logical row-id runs containing more than this number of rows are excluded from row-id reassignment. Set to 0 to disable this filtering.</td>
+        </tr>
         <tr>
             <td><h5>data-evolution.row-sidecar.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>
@@ -1416,12 +1428,6 @@
             <td>Long</td>
             <td>After configuring this time, only the data files created after this time will be read. It is independent of snapshots, but it is imprecise filtering (depending on whether or not compaction occurs).</td>
         </tr>
-        <tr>
-            <td><h5>chain-table.split.key-range-enabled</h5></td>
-            <td style="word-wrap: break-word;">true</td>
-            <td>Boolean</td>
-            <td>If true, a batch chain-table scan splits each bucket's snapshot and delta files into multiple splits by key range to improve read parallelism. Files with intersecting key ranges always stay in the same split so that all versions of a key across the snapshot and delta branches are merged together. Set to false to fall back to one split per bucket.</td>
-        </tr>
         <tr>
             <td><h5>scan.ignore-corrupt-files</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2497,6 +2497,15 @@ public String toString() {
                     .defaultValue(false)
                     .withDescription("Whether enable data evolution for row tracking table.");
 
+    public static final ConfigOption<Long> DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT =
+            key("data-evolution.reassign.skip-contiguous-row-count")
+                    .longType()
+                    .defaultValue(1_000_000_000L)
+                    .withDescription(
+                            "Strictly contiguous same-partition logical row-id runs containing "
+                                    + "more than this number of rows are excluded from row-id "
+                                    + "reassignment. Set to 0 to disable this filtering.");
+
     public static final ConfigOption<Boolean> DATA_EVOLUTION_ROW_SIDECAR_ENABLED =
             key("data-evolution.row-sidecar.enabled")
                     .booleanType()
@@ -4217,6 +4226,15 @@ public boolean dataEvolutionEnabled() {
         return options.get(DATA_EVOLUTION_ENABLED);
     }
 
+    public long dataEvolutionReassignSkipContiguousRowCount() {
+        long threshold = options.get(DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT);
+        checkArgument(
+                threshold >= 0,
+                "The option %s cannot be negative.",
+                DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT.key());
+        return threshold;
+    }
+
     public boolean dataEvolutionRowSidecarEnabled() {
         return options.get(DATA_EVOLUTION_ROW_SIDECAR_ENABLED);
     }

paimon-common/src/main/java/org/apache/paimon/utils/PrimitiveRowRanges.java

@@ -119,6 +119,26 @@ public void normalizeOverlapping() {
         normalized = true;
     }
 
+    /**
+     * Returns whether any stored range overlaps the inclusive range from {@code start} to {@code
+     * end}.
+     */
+    public boolean overlaps(long start, long end) {
+        checkArgument(start <= end, "Invalid row range [%s, %s].", start, end);
+        normalizeOverlapping();
+        int lower = 0;
+        int upper = size;
+        while (lower < upper) {
+            int middle = lower + ((upper - lower) >>> 1);
+            if (ends[middle] < start) {
+                lower = middle + 1;
+            } else {
+                upper = middle;
+            }
+        }
+        return lower < size && starts[lower] <= end;
+    }
+
     /**
      * Returns whether these ranges fully cover the inclusive range from {@code start} to {@code
      * end}.

paimon-common/src/test/java/org/apache/paimon/utils/PrimitiveRowRangesTest.java

@@ -78,6 +78,22 @@ void testDoesNotMergeAdjacentRanges() {
         assertThat(ranges.size()).isEqualTo(2);
     }
 
+    @Test
+    void testOverlaps() {
+        PrimitiveRowRanges ranges = new PrimitiveRowRanges(4);
+        ranges.add(20L, 25L);
+        ranges.add(5L, 8L);
+        ranges.add(8L, 12L);
+        ranges.add(30L, 35L);
+
+        assertThat(ranges.overlaps(0L, 4L)).isFalse();
+        assertThat(ranges.overlaps(4L, 5L)).isTrue();
+        assertThat(ranges.overlaps(12L, 19L)).isTrue();
+        assertThat(ranges.overlaps(13L, 19L)).isFalse();
+        assertThat(ranges.overlaps(24L, 31L)).isTrue();
+        assertThat(ranges.overlaps(36L, Long.MAX_VALUE)).isFalse();
+    }
+
     @Test
     void testCovers() {
         PrimitiveRowRanges ranges = new PrimitiveRowRanges(4);
@@ -111,6 +127,8 @@ void testRejectsInvalidInput() {
         assertThatThrownBy(() -> ranges.add(2L, 1L)).isInstanceOf(IllegalArgumentException.class);
         assertThatThrownBy(() -> ranges.start(0)).isInstanceOf(IllegalArgumentException.class);
         assertThatThrownBy(() -> ranges.append(null)).isInstanceOf(IllegalArgumentException.class);
+        assertThatThrownBy(() -> ranges.overlaps(2L, 1L))
+                .isInstanceOf(IllegalArgumentException.class);
         assertThatThrownBy(() -> ranges.covers(2L, 1L))
                 .isInstanceOf(IllegalArgumentException.class);
     }