[common] Widen hilbert index bytes beyond 8 dimensions

paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java

@@ -317,7 +317,14 @@ public static byte[] hilbertCurvePosBytes(Long[] points) {
         long[] data = Arrays.stream(points).mapToLong(Long::longValue).toArray();
         HilbertCurve hilbertCurve = HilbertCurve.bits(BITS_NUM).dimensions(points.length);
         BigInteger index = hilbertCurve.index(data);
-        return ConvertBinaryUtil.paddingToNByte(index.toByteArray(), BITS_NUM);
+        // an N-dimensional 63-bit index needs up to 63*N/8 + 1 bytes: the extra byte covers
+        // BigInteger's sign byte when the top bit is set. paddingToNByte drops trailing bytes
+        // when the array is longer than the width, so a narrower width silently discards
+        // low-order bits — at 8 dimensions that also inverts the order, because the sign byte
+        // makes an upper-half index sort below a smaller lower-half one. Hilbert keys are
+        // transient sort keys in every consumer, so widening them breaks nothing.
+        int paddingBytes = BITS_NUM * data.length / 8 + 1;
+        return ConvertBinaryUtil.paddingToNByte(index.toByteArray(), paddingBytes);
     }
 
     /** Process function interface. */

paimon-common/src/test/java/org/apache/paimon/sort/hilbert/HilbertIndexerTest.java

@@ -23,9 +23,13 @@
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
 
+import org.davidmoten.hilbert.HilbertCurve;
 import org.junit.jupiter.api.Test;
 
+import java.math.BigInteger;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -60,6 +64,110 @@ public void testBooleanValuesDistinctFromNull() {
         assertThat(falseIndex).isNotEqualTo(trueIndex);
     }
 
+    @Test
+    public void testHighDimensionIndexKeepsAllBits() {
+        // 9 dimensions: the 63*9-bit index needs 71 bytes; distinct points that differ
+        // only in the low-order bits must stay distinct instead of being truncated away
+        Long[][] points = {
+            {0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L},
+            {0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 1L},
+        };
+        byte[] first = HilbertIndexer.hilbertCurvePosBytes(points[0]);
+        byte[] second = HilbertIndexer.hilbertCurvePosBytes(points[1]);
+        assertThat(first).hasSize(71);
+        assertThat(second).hasSize(71);
+        assertThat(first).isNotEqualTo(second);
+
+        // 16 dimensions: the top bit being set adds BigInteger's sign byte, so the width
+        // must cover it or the low byte is truncated away
+        Long[] highBits =
+                new Long[] {
+                    Long.MAX_VALUE, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L
+                };
+        Long[] highBitsVariant = highBits.clone();
+        highBitsVariant[15] = 1L;
+        byte[] highFirst = HilbertIndexer.hilbertCurvePosBytes(highBits);
+        byte[] highSecond = HilbertIndexer.hilbertCurvePosBytes(highBitsVariant);
+        assertThat(highFirst).hasSize(127);
+        assertThat(highSecond).hasSize(127);
+        assertThat(highFirst).isNotEqualTo(highSecond);
+
+        // the width is 63*N/8 + 1 for every N, so a 2-dimension key is 16 bytes and an
+        // 8-dimension one is 64 — the extra byte over the 63-byte magnitude is what makes
+        // room for BigInteger's sign byte
+        assertThat(HilbertIndexer.hilbertCurvePosBytes(new Long[] {0L, 0L})).hasSize(16);
+        assertThat(HilbertIndexer.hilbertCurvePosBytes(new Long[] {0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L}))
+                .hasSize(64);
+    }
+
+    /**
+     * At 8 dimensions the index fills 63 bytes, so the top half of the space carries BigInteger's
+     * sign byte and spills to 64. Truncating that back to 63 does not merely lose resolution: the
+     * leading zero makes a large index sort below a smaller one.
+     */
+    @Test
+    public void testEightDimensionKeysOrderLikeTheirIndex() {
+        List<Long[]> points = new ArrayList<>();
+        for (long i = 0; i < 24; i++) {
+            // spread the points over the whole space so some land in the top half
+            long v = Long.MAX_VALUE / 23 * i;
+            points.add(new Long[] {v, v / 3, i, Long.MAX_VALUE - v, v / 7, i * 31, v / 11, i});
+        }
+
+        for (Long[] left : points) {
+            for (Long[] right : points) {
+                int indexOrder = index(left).compareTo(index(right));
+                int keyOrder =
+                        compareUnsigned(
+                                HilbertIndexer.hilbertCurvePosBytes(left),
+                                HilbertIndexer.hilbertCurvePosBytes(right));
+                assertThat(Integer.signum(keyOrder))
+                        .as(
+                                "key order must follow index order for %s vs %s",
+                                Arrays.toString(left), Arrays.toString(right))
+                        .isEqualTo(Integer.signum(indexOrder));
+            }
+        }
+    }
+
+    /**
+     * Sizes and inequalities are proxies for the property the width exists to guarantee: the key
+     * carries the whole index. State it directly, at the two dimension counts where the index needs
+     * its last byte the most.
+     */
+    @Test
+    public void testKeyCarriesTheWholeIndex() {
+        for (int dimensions : new int[] {8, 9}) {
+            Long[] topOfSpace = new Long[dimensions];
+            Arrays.fill(topOfSpace, Long.MAX_VALUE);
+            Long[] oneLowBitOff = topOfSpace.clone();
+            oneLowBitOff[dimensions - 1] = Long.MAX_VALUE - 1;
+
+            for (Long[] point : new Long[][] {topOfSpace, oneLowBitOff}) {
+                byte[] key = HilbertIndexer.hilbertCurvePosBytes(point);
+                assertThat(new BigInteger(1, key))
+                        .as("key must round-trip the index at %s dimensions", dimensions)
+                        .isEqualTo(index(point));
+            }
+        }
+    }
+
+    private static BigInteger index(Long[] points) {
+        long[] data = Arrays.stream(points).mapToLong(Long::longValue).toArray();
+        return HilbertCurve.bits(63).dimensions(points.length).index(data);
+    }
+
+    private static int compareUnsigned(byte[] left, byte[] right) {
+        assertThat(left).hasSameSizeAs(right);
+        for (int i = 0; i < left.length; i++) {
+            int cmp = Integer.compare(left[i] & 0xFF, right[i] & 0xFF);
+            if (cmp != 0) {
+                return cmp;
+            }
+        }
+        return 0;
+    }
+
     private static GenericRow booleanRow(Boolean value) {
         GenericRow row = new GenericRow(2);
         row.setField(0, value);