[core] Speed up row-id manifest sorting

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/DeletedRowIdSet.java

@@ -23,7 +23,7 @@
 import java.util.Arrays;
 
 /** Primitive set used by RowID compaction to avoid rebuilding file identifiers. */
-final class DeletedRowIdSet {
+public final class DeletedRowIdSet {
 
     private static final long EMPTY = Long.MIN_VALUE;
 
@@ -32,7 +32,7 @@ final class DeletedRowIdSet {
     private boolean containsMinValue;
     private @Nullable long[] sortedRowIds;
 
-    void add(long value) {
+    public void add(long value) {
         if (value == EMPTY) {
             if (!containsMinValue) {
                 containsMinValue = true;
@@ -56,7 +56,7 @@ void add(long value) {
         sortedRowIds = null;
     }
 
-    void addAll(DeletedRowIdSet other) {
+    public void addAll(DeletedRowIdSet other) {
         if (other.containsMinValue) {
             add(EMPTY);
         }
@@ -67,7 +67,7 @@ void addAll(DeletedRowIdSet other) {
         }
     }
 
-    boolean contains(long value) {
+    public boolean contains(long value) {
         if (value == EMPTY) {
             return containsMinValue;
         }
@@ -81,7 +81,7 @@ boolean contains(long value) {
         return false;
     }
 
-    boolean intersects(long minInclusive, long maxInclusive) {
+    public boolean intersects(long minInclusive, long maxInclusive) {
         if (minInclusive > maxInclusive) {
             return true;
         }
@@ -93,12 +93,12 @@ boolean intersects(long minInclusive, long maxInclusive) {
         return position < values.length && values[position] <= maxInclusive;
     }
 
-    void prepareRangeIndex() {
+    public void prepareRangeIndex() {
         // Publish the immutable sorted snapshot before concurrent manifest planning starts.
         sortedRowIds();
     }
 
-    void releaseRangeIndex() {
+    public void releaseRangeIndex() {
         sortedRowIds = null;
     }
 

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

@@ -213,6 +213,19 @@ public ReusableIdentifier replaceWithPartition(ProjectedManifestEntry 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(ProjectedManifestEntry entry) {
             putInt(entry.bucket());
             ProjectedDataFileMeta file = entry.file();

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

@@ -212,6 +212,11 @@ public ManifestAvroReader scanAvroBlocks(String fileName, @Nullable Long fileSiz
         }
     }
 
+    /** Opens a low-allocation reader for the encoded manifest fields needed by run merge. */
+    public ManifestAvroReader scanForRunMerge(String fileName, @Nullable Long fileSize) {
+        return scanAvroBlocks(fileName, fileSize);
+    }
+
     @VisibleForTesting
     public long suggestedFileSize() {
         return suggestedFileSize;