[core] Speed up row-id manifest sorting

docs/generated/core_configuration.html

@@ -1023,6 +1023,12 @@
             <td>String</td>
             <td>Partition field name to sort manifest entries by. Validated by schema validation, if not configured, defaults to the first partition field.</td>
         </tr>
+        <tr>
+            <td><h5>manifest-sort.run-merge-optimize.enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Whether to use streaming run merge for RowID-based manifest sorting. When disabled, the external sorter is used without changing the RowID sort semantics.</td>
+        </tr>
         <tr>
             <td><h5>manifest.compression</h5></td>
             <td style="word-wrap: break-word;">"zstd"</td>

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

@@ -595,6 +595,15 @@ public InlineElement getDescription() {
                                     + " skipped. Set to a larger value to allow more aggressive"
                                     + " sort rewriting. The cap only limits the sorted rewrite portion and full/minor cleanup may still happen beyond it.");
 
+    public static final ConfigOption<Boolean> MANIFEST_SORT_RUN_MERGE_OPTIMIZE_ENABLED =
+            key("manifest-sort.run-merge-optimize.enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether to use streaming run merge for RowID-based manifest sorting."
+                                    + " When disabled, the external sorter is used without changing"
+                                    + " the RowID sort semantics.");
+
     public static final ConfigOption<String> PARTITION_DEFAULT_NAME =
             key("partition.default-name")
                     .stringType()
@@ -3072,6 +3081,10 @@ public long manifestSortMaxRewriteSize() {
         return options.get(MANIFEST_SORT_MAX_REWRITE_SIZE).getBytes();
     }
 
+    public boolean manifestSortRunMergeOptimizeEnabled() {
+        return options.get(MANIFEST_SORT_RUN_MERGE_OPTIMIZE_ENABLED);
+    }
+
     public String partitionDefaultName() {
         return options.get(PARTITION_DEFAULT_NAME);
     }

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

@@ -47,8 +47,7 @@ byte[] bytes() {
     public boolean equals(Object obj) {
         return obj == this
                 || (obj instanceof ByteArrayKey && Arrays.equals(bytes, ((ByteArrayKey) obj).bytes))
-                || (obj instanceof ByteArrayLookupKey
-                        && Arrays.equals(bytes, ((ByteArrayLookupKey) obj).bytes()));
+                || (obj instanceof ByteArrayLookupKey && obj.equals(this));
     }
 
     @Override

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

@@ -20,8 +20,6 @@
 
 import javax.annotation.Nullable;
 
-import java.util.Arrays;
-
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /**
@@ -33,6 +31,8 @@
 public final class ByteArrayLookupKey {
 
     private @Nullable byte[] bytes;
+    private int offset;
+    private int length;
     private int hash;
 
     public ByteArrayLookupKey() {}
@@ -43,12 +43,26 @@ public ByteArrayLookupKey(byte[] bytes) {
 
     public void reset(byte[] bytes) {
         checkArgument(bytes != null, "Byte array cannot be null.");
+        reset(bytes, 0, bytes.length);
+    }
+
+    public void reset(byte[] bytes, int offset, int length) {
+        checkArgument(bytes != null, "Byte array cannot be null.");
+        checkArgument(offset >= 0 && length >= 0 && offset <= bytes.length - length);
         this.bytes = bytes;
-        this.hash = Arrays.hashCode(bytes);
+        this.offset = offset;
+        this.length = length;
+        int hash = 1;
+        for (int i = offset; i < offset + length; i++) {
+            hash = 31 * hash + bytes[i];
+        }
+        this.hash = hash;
     }
 
     public void clear() {
         bytes = null;
+        offset = 0;
+        length = 0;
         hash = 0;
     }
 
@@ -62,14 +76,38 @@ public boolean equals(Object obj) {
         return obj == this
                 || (bytes != null
                         && obj instanceof ByteArrayKey
-                        && Arrays.equals(bytes, ((ByteArrayKey) obj).bytes()))
+                        && equals(((ByteArrayKey) obj).bytes()))
                 || (bytes != null
                         && obj instanceof ByteArrayLookupKey
-                        && Arrays.equals(bytes, ((ByteArrayLookupKey) obj).bytes));
+                        && equals((ByteArrayLookupKey) obj));
     }
 
     @Override
     public int hashCode() {
         return hash;
     }
+
+    private boolean equals(byte[] other) {
+        if (length != other.length) {
+            return false;
+        }
+        for (int i = 0; i < length; i++) {
+            if (bytes[offset + i] != other[i]) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private boolean equals(ByteArrayLookupKey other) {
+        if (other.bytes == null || length != other.length) {
+            return false;
+        }
+        for (int i = 0; i < length; i++) {
+            if (bytes[offset + i] != other.bytes[other.offset + i]) {
+                return false;
+            }
+        }
+        return true;
+    }
 }

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

@@ -59,6 +59,23 @@ void testReusableMapLookup() {
         assertThat(lookup.hashCode()).isZero();
     }
 
+    @Test
+    void testReusableSliceLookup() {
+        Map<ByteArrayKey, String> values = new HashMap<>();
+        ByteArrayKey key = new ByteArrayKey(new byte[] {1, 2, 3});
+        values.put(key, "value");
+        ByteArrayLookupKey lookup = new ByteArrayLookupKey();
+
+        lookup.reset(new byte[] {9, 1, 2, 3, 8}, 1, 3);
+        assertThat(lookup).isEqualTo(key);
+        assertThat(key).isEqualTo(lookup);
+        assertThat(lookup.hashCode()).isEqualTo(key.hashCode());
+        assertThat(values.get(lookup)).isEqualTo("value");
+
+        lookup.clear();
+        assertThat(new ByteArrayLookupKey(new byte[] {1, 2, 3})).isNotEqualTo(lookup);
+    }
+
     @Test
     void testLookupEqualityLifecycle() {
         ByteArrayLookupKey first = new ByteArrayLookupKey(new byte[] {1});

paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java

@@ -113,7 +113,8 @@ private static Projection createDeleteEntryProjection() {
                                                         DataFileMeta.LEVEL,
                                                         DataFileMeta.EXTRA_FILES,
                                                         DataFileMeta.EMBEDDED_FILE_INDEX,
-                                                        DataFileMeta.EXTERNAL_PATH)))));
+                                                        DataFileMeta.EXTERNAL_PATH,
+                                                        DataFileMeta.FIRST_ROW_ID)))));
     }
 
     private static Projection createRowRangeProjection() {
@@ -310,6 +311,19 @@ public ReusableIdentifier replaceWithPartition(BinaryManifestEntry entry) {
             return appendEntryFields(entry);
         }
 
+        /** Replaces this encoding with an already serialized identifier. */
+        public ReusableIdentifier replace(byte[] value, int offset, int valueLength) {
+            checkArgument(value != null, "Serialized identifier cannot be null.");
+            checkArgument(
+                    offset >= 0 && valueLength >= 0 && offset <= value.length - valueLength,
+                    "Identifier byte range is invalid.");
+            length = 0;
+            ensureCapacity(valueLength);
+            System.arraycopy(value, offset, bytes, 0, valueLength);
+            length = valueLength;
+            return this;
+        }
+
         private ReusableIdentifier appendEntryFields(BinaryManifestEntry entry) {
             putInt(entry.bucket());
             BinaryDataFileMeta file = entry.file();