[common] Order decimal z-values through the signed long transform

paimon-common/src/main/java/org/apache/paimon/sort/zorder/ZIndexer.java

@@ -51,6 +51,7 @@
 import org.apache.paimon.types.VectorType;
 
 import java.io.Serializable;
+import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.LinkedHashSet;
@@ -76,7 +77,7 @@ public ZIndexer(RowType rowType, List<String> orderColumns) {
     public ZIndexer(RowType rowType, List<String> orderColumns, int varTypeSize) {
         List<String> fields = rowType.getFieldNames();
         fieldsIndex = new int[orderColumns.size()];
-        int varTypeCount = 0;
+        int total = 0;
         for (int i = 0; i < fieldsIndex.length; i++) {
             int index = fields.indexOf(orderColumns.get(i));
             if (index == -1) {
@@ -87,15 +88,10 @@ public ZIndexer(RowType rowType, List<String> orderColumns, int varTypeSize) {
                                 + fields);
             }
             fieldsIndex[i] = index;
-
-            if (isVarType(rowType.getFieldTypes().get(index))) {
-                varTypeCount++;
-            }
+            total += zBytes(rowType.getFieldTypes().get(index), varTypeSize);
         }
         this.functionSet = constructFunctionMap(rowType.getFields(), varTypeSize);
-        this.totalBytes =
-                PRIMITIVE_BUFFER_SIZE * (this.fieldsIndex.length - varTypeCount)
-                        + varTypeSize * varTypeCount;
+        this.totalBytes = total;
     }
 
     private static boolean isVarType(DataType dataType) {
@@ -105,6 +101,26 @@ private static boolean isVarType(DataType dataType) {
                 || dataType instanceof VarBinaryType;
     }
 
+    /**
+     * Bytes the z-value reserves for a column. A DECIMAL gets a width that holds any unscaled value
+     * of its precision, so it keeps full ordering resolution; 8 bytes cannot, and projecting into
+     * them collapses ordinary wide-decimal values to one key.
+     */
+    private static int zBytes(DataType type, int varTypeSize) {
+        if (isVarType(type)) {
+            return varTypeSize;
+        }
+        if (type instanceof DecimalType) {
+            return decimalBytes(((DecimalType) type).getPrecision());
+        }
+        return PRIMITIVE_BUFFER_SIZE;
+    }
+
+    /** Two's-complement byte width that holds any unscaled value with {@code precision} digits. */
+    private static int decimalBytes(int precision) {
+        return BigInteger.TEN.pow(precision).bitLength() / 8 + 1;
+    }
+
     public void open() {
         this.reuse = ByteBuffer.allocate(totalBytes);
         functionSet.forEach(RowProcessor::open);
@@ -138,8 +154,7 @@ public Set<RowProcessor> constructFunctionMap(List<DataField> fields, int varTyp
     public static RowProcessor zmapColumnToCalculator(DataField field, int index, int varTypeSize) {
         DataType type = field.type();
         return new RowProcessor(
-                type.accept(new TypeVisitor(index, varTypeSize)),
-                isVarType(type) ? varTypeSize : PRIMITIVE_BUFFER_SIZE);
+                type.accept(new TypeVisitor(index, varTypeSize)), zBytes(type, varTypeSize));
     }
 
     /** Type Visitor to generate function map from row column to z-index. */
@@ -239,18 +254,39 @@ public ZProcessFunction visit(VarBinaryType varBinaryType) {
         public ZProcessFunction visit(DecimalType decimalType) {
             final InternalRow.FieldGetter fieldGetter =
                     InternalRow.createFieldGetter(decimalType, fieldIndex);
+            // Encode the unscaled value as a fixed-width, sign-flipped big-endian two's complement.
+            // The width holds any value of this precision, so the encoding is order-preserving
+            // under the unsigned comparison a z-value gets and keeps full resolution: small
+            // separated values, their negatives, and values past the long range all get distinct
+            // keys. An 8-byte projection cannot, since one wide-decimal column exceeds 64 bits and
+            // every value below the shift collapses to a single key.
+            final int width = decimalBytes(decimalType.getPrecision());
+            final byte[] nullBytes = new byte[width];
             return (row, reuse) -> {
                 Object o = fieldGetter.getFieldOrNull(row);
-                return o == null
-                        ? NULL_BYTES
-                        : ZOrderByteUtils.byteTruncateOrFill(
-                                        ((Decimal) o).toUnscaledBytes(),
-                                        PRIMITIVE_BUFFER_SIZE,
-                                        reuse)
-                                .array();
+                if (o == null) {
+                    return nullBytes;
+                }
+                Decimal decimal = (Decimal) o;
+                BigInteger unscaled =
+                        decimal.isCompact()
+                                ? BigInteger.valueOf(decimal.toUnscaledLong())
+                                : decimal.toBigDecimal().unscaledValue();
+                return orderedDecimalBytes(unscaled, width, reuse);
             };
         }
 
+        private static byte[] orderedDecimalBytes(
+                BigInteger unscaled, int width, ByteBuffer reuse) {
+            byte[] buffer = ZOrderByteUtils.reuse(reuse, width).array();
+            Arrays.fill(buffer, 0, width, unscaled.signum() < 0 ? (byte) 0xFF : (byte) 0x00);
+            byte[] magnitude = unscaled.toByteArray();
+            System.arraycopy(magnitude, 0, buffer, width - magnitude.length, magnitude.length);
+            // Flip the sign bit so signed order becomes unsigned lexicographic order.
+            buffer[0] ^= (byte) 0x80;
+            return buffer;
+        }
+
         @Override
         public ZProcessFunction visit(TinyIntType tinyIntType) {
             return (row, reuse) ->

paimon-common/src/test/java/org/apache/paimon/sort/zorder/TestZOrderByteUtil.java

@@ -18,17 +18,22 @@
 
 package org.apache.paimon.sort.zorder;
 
+import org.apache.paimon.data.Decimal;
 import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DecimalType;
 import org.apache.paimon.types.RowType;
 
 import org.junit.Test;
 import org.testcontainers.shaded.com.google.common.primitives.UnsignedBytes;
 
+import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.List;
 import java.util.Random;
 
 import static org.apache.paimon.utils.RandomUtil.randomBytes;
@@ -94,6 +99,125 @@ private String interleaveStrings(String[] strings) {
         return result.toString();
     }
 
+    /** Decimal z-values must order by value and stay distinct from the null sentinel. */
+    @Test
+    public void testZIndexerDecimalOrdering() {
+        RowType rowType =
+                new RowType(
+                        Arrays.asList(
+                                new DataField(0, "a", new DecimalType(20, 2)),
+                                new DataField(1, "b", new DecimalType(20, 2))));
+        ZIndexer indexer = new ZIndexer(rowType, Arrays.asList("a", "b"));
+        indexer.open();
+
+        // Two rows whose decimal z-values must order by value: (-1, 0) then (0, 1).
+        // Old code fed minimal two's-complement arrays into unsigned comparison,
+        // making -1 > 1 and 0 collide with the null sentinel.
+        GenericRow row1 = new GenericRow(2);
+        row1.setField(0, Decimal.fromBigDecimal(new BigDecimal("-1.00"), 20, 2));
+        row1.setField(1, Decimal.fromBigDecimal(new BigDecimal("0.00"), 20, 2));
+        GenericRow row2 = new GenericRow(2);
+        row2.setField(0, Decimal.fromBigDecimal(new BigDecimal("0.00"), 20, 2));
+        row2.setField(1, Decimal.fromBigDecimal(new BigDecimal("1.00"), 20, 2));
+
+        byte[] z1 = Arrays.copyOf(indexer.index(row1), indexer.size());
+        byte[] z2 = Arrays.copyOf(indexer.index(row2), indexer.size());
+        // Interleaved bits: column a dominates the high bits of the z-value.
+        assertThat(compareUnsigned(z1, z2)).isLessThan(0);
+
+        GenericRow rowNull = new GenericRow(2);
+        rowNull.setField(0, null);
+        rowNull.setField(1, null);
+        byte[] zNull = Arrays.copyOf(indexer.index(rowNull), indexer.size());
+        // The null sentinel (all-zero bytes) sorts below every real value.
+        assertThat(compareUnsigned(zNull, z1)).isLessThan(0);
+        assertThat(compareUnsigned(zNull, z2)).isLessThan(0);
+
+        // A large magnitude still orders correctly against the small positives: the fixed-width
+        // encoding holds the whole unscaled value rather than projecting it into 8 bytes.
+        GenericRow big = new GenericRow(2);
+        big.setField(0, Decimal.fromBigDecimal(new BigDecimal("92233720368547758.08"), 20, 2));
+        big.setField(1, Decimal.fromBigDecimal(new BigDecimal("0.00"), 20, 2));
+        byte[] zBig = Arrays.copyOf(indexer.index(big), indexer.size());
+        assertThat(compareUnsigned(zBig, z2)).isGreaterThan(0);
+
+        // Its negative counterpart orders below the small negatives, but still above the null
+        // sentinel.
+        Decimal minUnscaled =
+                Decimal.fromBigDecimal(new BigDecimal("-92233720368547758.08"), 20, 2);
+        GenericRow negativeBig = new GenericRow(2);
+        negativeBig.setField(0, minUnscaled);
+        negativeBig.setField(1, minUnscaled);
+        byte[] zNegativeBig = Arrays.copyOf(indexer.index(negativeBig), indexer.size());
+        assertThat(compareUnsigned(zNull, zNegativeBig)).isLessThan(0);
+        assertThat(compareUnsigned(zNegativeBig, z1)).isLessThan(0);
+    }
+
+    /**
+     * High-precision decimals must keep full clustering resolution: ordinary small values, their
+     * negatives, and values past the long range all get distinct, correctly ordered z-keys.
+     */
+    @Test
+    public void testZIndexerHighPrecisionDecimalClustering() {
+        RowType rowType =
+                new RowType(
+                        Arrays.asList(
+                                new DataField(0, "a", new DecimalType(38, 18)),
+                                new DataField(1, "b", new DecimalType(38, 18))));
+        ZIndexer indexer = new ZIndexer(rowType, Arrays.asList("a", "b"));
+        indexer.open();
+
+        // Strictly ascending values, including small ones well below an 8-byte projection's
+        // overflow point (which collapsed 0..18 to a single key), sub-unit fractions, negatives,
+        // and a value past Long.MAX_VALUE. Each must get a strictly greater z-key than the last.
+        List<String> ascending =
+                Arrays.asList(
+                        "-90",
+                        "-10",
+                        "-2",
+                        "-1",
+                        "0",
+                        "0.001",
+                        "0.002",
+                        "1",
+                        "2",
+                        "3",
+                        "10",
+                        "20",
+                        "90",
+                        "12345678901234567890.123456789012345678");
+        byte[] previous = null;
+        for (String value : ascending) {
+            byte[] key = highPrecisionKey(indexer, value);
+            if (previous != null) {
+                assertThat(compareUnsigned(previous, key)).isLessThan(0);
+            }
+            previous = key;
+        }
+
+        // Null sorts below every real value.
+        byte[] zNull = Arrays.copyOf(indexer.index(new GenericRow(2)), indexer.size());
+        assertThat(compareUnsigned(zNull, highPrecisionKey(indexer, "-90"))).isLessThan(0);
+    }
+
+    private static byte[] highPrecisionKey(ZIndexer indexer, String value) {
+        GenericRow row = new GenericRow(2);
+        row.setField(0, Decimal.fromBigDecimal(new BigDecimal(value), 38, 18));
+        row.setField(1, Decimal.fromBigDecimal(new BigDecimal("0"), 38, 18));
+        return Arrays.copyOf(indexer.index(row), indexer.size());
+    }
+
+    private static int compareUnsigned(byte[] left, byte[] right) {
+        for (int i = 0; i < left.length && i < right.length; i++) {
+            int a = left[i] & 0xFF;
+            int b = right[i] & 0xFF;
+            if (a != b) {
+                return Integer.compare(a, b);
+            }
+        }
+        return Integer.compare(left.length, right.length);
+    }
+
     /**
      * Compares the result of a string based interleaving algorithm implemented above versus the
      * binary bit-shifting algorithm used in ZOrderByteUtils. Either both algorithms are identically