fix: Expose Decimal Literal over Java (#3521)

Author: a10y

java/vortex-jni/src/main/java/dev/vortex/api/expressions/Literal.java

@@ -15,8 +15,10 @@
  */
 package dev.vortex.api.expressions;
 
-import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
 import dev.vortex.api.Expression;
+import java.math.BigDecimal;
+import java.util.Objects;
 import java.util.Optional;
 
 public abstract class Literal<T> implements Expression {
@@ -44,7 +46,7 @@ public int hashCode() {
     public boolean equals(Object o) {
         if (!(o instanceof Literal)) return false;
         Literal<?> literal = (Literal<?>) o;
-        return java.util.Objects.equals(value, literal.value);
+        return Objects.equals(value, literal.value);
     }
 
     public static Literal<Void> nullLit() {
@@ -79,6 +81,10 @@ public static Literal<Double> float64(Double value) {
         return new Float64Literal(value);
     }
 
+    public static Literal<BigDecimal> decimal(BigDecimal value, int precision, int scale) {
+        return new DecimalLiteral(value, precision, scale);
+    }
+
     public static Literal<String> string(String value) {
         return new StringLiteral(value);
     }
@@ -165,6 +171,8 @@ public interface LiteralVisitor<U> {
 
         U visitFloat64(Double literal);
 
+        U visitDecimal(BigDecimal decimal, int precision, int scale);
+
         U visitString(String literal);
 
         U visitBytes(byte[] literal);
@@ -260,6 +268,25 @@ public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
         }
     }
 
+    static final class DecimalLiteral extends Literal<BigDecimal> {
+        private final int precision;
+        private final int scale;
+
+        DecimalLiteral(BigDecimal value, int precision, int scale) {
+            super(value);
+            if (!Objects.isNull(value)) {
+                Preconditions.checkArgument(scale == value.scale(), "scale %s ≠ value scale %s", scale, value.scale());
+            }
+            this.precision = precision;
+            this.scale = scale;
+        }
+
+        @Override
+        public <U> U acceptLiteralVisitor(LiteralVisitor<U> visitor) {
+            return visitor.visitDecimal(getValue(), precision, scale);
+        }
+    }
+
     static final class StringLiteral extends Literal<String> {
         StringLiteral(String value) {
             super(value);

java/vortex-jni/src/main/java/dev/vortex/api/expressions/proto/DTypes.java

@@ -91,6 +91,16 @@ static DTypeProtos.DType float64(boolean nullable) {
                 .build();
     }
 
+    static DTypeProtos.DType decimal(boolean nullable, int precision, int scale) {
+        return DTypeProtos.DType.newBuilder()
+                .setDecimal(DTypeProtos.Decimal.newBuilder()
+                        .setNullable(nullable)
+                        .setPrecision(precision)
+                        .setScale(scale)
+                        .build())
+                .build();
+    }
+
     static DTypeProtos.DType string(boolean nullable) {
         return DTypeProtos.DType.newBuilder()
                 .setUtf8(DTypeProtos.Utf8.newBuilder().setNullable(nullable).build())

java/vortex-jni/src/main/java/dev/vortex/api/expressions/proto/EndianUtils.java

@@ -0,0 +1,70 @@
+/**
+ * (c) Copyright 2025 SpiralDB Inc. All rights reserved.
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.vortex.api.expressions.proto;
+
+import com.google.protobuf.ByteString;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+
+final class EndianUtils {
+    static byte[] reverse(ByteString src) {
+        byte[] dst = new byte[src.size()];
+        for (int i = 0; i < dst.length; i++) {
+            dst[i] = src.byteAt(dst.length - 1 - i);
+        }
+        return dst;
+    }
+
+    static byte[] littleEndianDecimal(BigDecimal decimal) {
+        BigInteger unscaled = decimal.unscaledValue();
+        byte[] bigEndianBytes = unscaled.toByteArray();
+
+        // Determine target size (1, 2, 4, 8, 16, or 32 bytes)
+        int targetSize;
+        if (bigEndianBytes.length <= 1) {
+            targetSize = 1;
+        } else if (bigEndianBytes.length <= 2) {
+            targetSize = 2;
+        } else if (bigEndianBytes.length <= 4) {
+            targetSize = 4;
+        } else if (bigEndianBytes.length <= 8) {
+            targetSize = 8;
+        } else if (bigEndianBytes.length <= 16) {
+            targetSize = 16;
+        } else if (bigEndianBytes.length <= 32) {
+            targetSize = 32;
+        } else {
+            throw new IllegalArgumentException(
+                    "BigDecimal with " + bigEndianBytes.length + " bytes overflows maximum Vortex decimal size");
+        }
+
+        byte[] result = new byte[targetSize];
+
+        // Copy bytes in reverse order (big endian to little endian)
+        for (int i = 0; i < bigEndianBytes.length; i++) {
+            result[i] = bigEndianBytes[bigEndianBytes.length - 1 - i];
+        }
+
+        // Sign extend if negative
+        if (unscaled.signum() < 0) {
+            for (int i = bigEndianBytes.length; i < targetSize; i++) {
+                result[i] = (byte) 0xFF;
+            }
+        }
+
+        return result;
+    }
+}

java/vortex-jni/src/main/java/dev/vortex/api/expressions/proto/ExpressionProtoDeserializer.java

@@ -17,11 +17,14 @@
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Iterables;
+import com.google.protobuf.ByteString;
 import dev.vortex.api.Expression;
 import dev.vortex.api.expressions.*;
 import dev.vortex.proto.DTypeProtos;
 import dev.vortex.proto.ExprProtos;
 import dev.vortex.proto.ScalarProtos;
+import java.math.BigDecimal;
+import java.math.BigInteger;
 import java.util.List;
 import java.util.Optional;
 
@@ -126,7 +129,18 @@ private static Expression deserializeLiteral(ExprProtos.Kind.Literal literal, Li
             case STRING_VALUE:
                 return Literal.string(scalarValue.getStringValue());
             case BYTES_VALUE:
-                return Literal.bytes(scalarValue.getBytesValue().toByteArray());
+                if (dtype.hasDecimal()) {
+                    ByteString littleEndian = scalarValue.getBytesValue();
+                    byte[] bigEndian = EndianUtils.reverse(littleEndian);
+                    BigDecimal value = new BigDecimal(
+                            new BigInteger(bigEndian), dtype.getDecimal().getScale());
+                    return Literal.decimal(
+                            value,
+                            dtype.getDecimal().getPrecision(),
+                            dtype.getDecimal().getScale());
+                } else {
+                    return Literal.bytes(scalarValue.getBytesValue().toByteArray());
+                }
             default:
                 throw new UnsupportedOperationException("Unsupported ScalarValue type encountered: " + scalarValue);
         }
@@ -229,6 +243,11 @@ private static Literal<?> nullLiteral(DTypeProtos.DType type) {
                     default:
                         throw new UnsupportedOperationException("Unsupported ScalarValue type encountered: " + type);
                 }
+            case DECIMAL:
+                return Literal.decimal(
+                        null,
+                        type.getDecimal().getPrecision(),
+                        type.getDecimal().getScale());
             case UTF8:
                 return Literal.string(null);
             case BINARY:

java/vortex-jni/src/main/java/dev/vortex/api/expressions/proto/LiteralToScalar.java

@@ -17,6 +17,7 @@
 
 import dev.vortex.api.expressions.Literal;
 import dev.vortex.proto.ScalarProtos;
+import java.math.BigDecimal;
 import java.util.Objects;
 import java.util.Optional;
 
@@ -174,6 +175,15 @@ public ScalarProtos.Scalar visitFloat64(Double literal) {
         }
     }
 
+    @Override
+    public ScalarProtos.Scalar visitDecimal(BigDecimal decimal, int precision, int scale) {
+        if (Objects.isNull(decimal)) {
+            return Scalars.nullDecimal(precision, scale);
+        } else {
+            return Scalars.decimal(decimal, precision, scale);
+        }
+    }
+
     @Override
     public ScalarProtos.Scalar visitString(String literal) {
         if (Objects.isNull(literal)) {

java/vortex-jni/src/main/java/dev/vortex/api/expressions/proto/Scalars.java

@@ -18,6 +18,7 @@
 import com.google.protobuf.ByteString;
 import com.google.protobuf.NullValue;
 import dev.vortex.proto.ScalarProtos;
+import java.math.BigDecimal;
 import java.util.Optional;
 
 final class Scalars {
@@ -174,6 +175,23 @@ static ScalarProtos.Scalar nullString() {
                 .build();
     }
 
+    static ScalarProtos.Scalar decimal(BigDecimal decimal, int precision, int scale) {
+        var littleEndian = EndianUtils.littleEndianDecimal(decimal);
+        return ScalarProtos.Scalar.newBuilder()
+                .setValue(ScalarProtos.ScalarValue.newBuilder()
+                        .setBytesValue(ByteString.copyFrom(littleEndian))
+                        .build())
+                .setDtype(DTypes.decimal(false, precision, scale))
+                .build();
+    }
+
+    static ScalarProtos.Scalar nullDecimal(int precision, int scale) {
+        return ScalarProtos.Scalar.newBuilder()
+                .setValue(ScalarProtos.ScalarValue.newBuilder().setNullValue(NullValue.NULL_VALUE))
+                .setDtype(DTypes.decimal(true, precision, scale))
+                .build();
+    }
+
     static ScalarProtos.Scalar bytes(byte[] value) {
         return ScalarProtos.Scalar.newBuilder()
                 .setValue(ScalarProtos.ScalarValue.newBuilder()

java/vortex-jni/src/test/java/dev/vortex/api/expressions/LiteralTest.java

@@ -0,0 +1,36 @@
+/**
+ * (c) Copyright 2025 SpiralDB Inc. All rights reserved.
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.vortex.api.expressions;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import dev.vortex.api.Expression;
+import dev.vortex.api.expressions.proto.ExpressionProtoDeserializer;
+import dev.vortex.api.expressions.proto.ExpressionProtoSerializer;
+import dev.vortex.proto.ExprProtos;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import org.junit.jupiter.api.Test;
+
+public final class LiteralTest {
+    @Test
+    public void testLiteral_decimals() {
+        Literal<BigDecimal> lit = Literal.decimal(new BigDecimal(BigInteger.valueOf(-1234L), 3), 38, 3);
+        ExprProtos.Expr serialized = ExpressionProtoSerializer.serialize(lit);
+        Expression out = ExpressionProtoDeserializer.deserialize(serialized);
+        assertEquals(lit, out);
+    }
+}

vortex-layout/src/segments/events.rs

@@ -151,7 +151,7 @@ impl SegmentSource for EventsSegmentSource {
 }
 
 /// A pending segment returned by the [`SegmentSource`].
-pub struct PendingSegment {
+struct PendingSegment {
     id: SegmentId,
     /// A weak shared future that we hand out to all requesters. Once all requesters have been
     /// dropped, typically because their row split has completed (or been pruned), then the weak