Added check for +Inf/-Inf for every decimal type

Author: alex268

table/src/main/java/tech/ydb/table/values/DecimalType.java

@@ -16,14 +16,41 @@ public class DecimalType implements Type {
 
     public static final int MAX_PRECISION = 35;
 
-    private static final DecimalType YDB_DEFAULT = DecimalType.of(22, 9);
+    private static final InfValues[] INF_VALUES;
+
+    private static final DecimalType YDB_DEFAULT;
+
+    static {
+        INF_VALUES = new InfValues[DecimalType.MAX_PRECISION];
+
+        long mask32 = 0xFFFFFFFFL;
+        long infHigh = 0;
+        long infLow = 10;
+
+        for (int precision = 1; precision <= DecimalType.MAX_PRECISION; precision++) {
+            INF_VALUES[precision - 1] = new InfValues(infHigh, infLow);
+
+            // multiply by 10
+            long ll = 10 * (infLow & mask32);
+            long lh = 10 * (infLow >>> 32) + (ll >>> 32);
+            long hl = 10 * (infHigh & mask32) + (lh >>> 32);
+            long hh = 10 * (infHigh >>> 32) + (hl >>> 32);
+
+            infLow = (lh << 32) + (ll & mask32);
+            infHigh = (hh << 32) + (hl & mask32);
+        }
+
+        YDB_DEFAULT = DecimalType.of(22, 9);
+    }
 
     private final int precision;
     private final int scale;
+    private final InfValues inf;
 
     private DecimalType(int precision, int scale) {
         this.precision = precision;
         this.scale = scale;
+        this.inf = INF_VALUES[precision - 1];
     }
 
     public static DecimalType getDefault() {
@@ -115,4 +142,26 @@ public DecimalValue newValueUnscaled(BigInteger value) {
     public DecimalValue newValue(String value) {
         return DecimalValue.fromString(this, value);
     }
+
+    boolean isInf(long high, long low) {
+        return high > inf.posHigh || (high == inf.posHigh && Long.compareUnsigned(low, inf.posLow) >= 0);
+    }
+
+    boolean isNegInf(long high, long low) {
+        return high < inf.negHigh || (high == inf.negHigh && Long.compareUnsigned(low, inf.negLow) <= 0);
+    }
+
+    private static class InfValues {
+        private final long posHigh;
+        private final long posLow;
+        private final long negHigh;
+        private final long negLow;
+
+        InfValues(long infHigh, long infLow) {
+            this.posHigh = infHigh;
+            this.posLow = infLow;
+            this.negHigh = 0xFFFFFFFFFFFFFFFFL - infHigh;
+            this.negLow = 0xFFFFFFFFFFFFFFFFL - infLow + 1;
+        }
+    }
 }

table/src/main/java/tech/ydb/table/values/DecimalValue.java

@@ -272,18 +272,6 @@ private static long getLongBe(byte[] buf, int from, int to) {
         return r;
     }
 
-    private static boolean isNan(long high, long low) {
-        return NAN.high == high && NAN.low == low;
-    }
-
-    private static boolean isInf(long high, long low) {
-        return high > INF.high || (high == INF.high && Long.compareUnsigned(low, INF.low) >= 0);
-    }
-
-    private static boolean isNegInf(long high, long low) {
-        return high < NEG_INF.high || (high == NEG_INF.high && Long.compareUnsigned(low, NEG_INF.low) <= 0);
-    }
-
     private static DecimalValue newNan(DecimalType type) {
         return new DecimalValue(type, NAN.high, NAN.low);
     }
@@ -295,12 +283,27 @@ private static DecimalValue newInf(DecimalType type) {
     private static DecimalValue newNegInf(DecimalType type) {
         return new DecimalValue(type, NEG_INF.high, NEG_INF.low);
     }
-    static DecimalValue fromUnscaledLong(DecimalType type, long value) {
-        if (value == 0) {
+
+    private static boolean isNan(long high, long low) {
+        return NAN.high == high && NAN.low == low;
+    }
+
+    static DecimalValue fromUnscaledLong(DecimalType type, long low) {
+        if (low == 0) {
             return new DecimalValue(type, 0, 0);
         }
-        long high = value > 0 ? 0 : -1;
-        return new DecimalValue(type, high, value);
+
+        long high = low > 0 ? 0 : -1;
+
+        if (type.isInf(high, low)) {
+            return newInf(type);
+        }
+
+        if (type.isNegInf(high, low)) {
+            return newNegInf(type);
+        }
+
+        return new DecimalValue(type, high, low);
     }
 
     static DecimalValue fromBits(DecimalType type, long high, long low) {
@@ -312,11 +315,11 @@ static DecimalValue fromBits(DecimalType type, long high, long low) {
             return newNan(type);
         }
 
-        if (isInf(high, low)) {
+        if (type.isInf(high, low)) {
             return newInf(type);
         }
 
-        if (isNegInf(high, low)) {
+        if (type.isNegInf(high, low)) {
             return newNegInf(type);
         }
 
@@ -515,5 +518,4 @@ static DecimalValue fromBigDecimal(DecimalType type, BigDecimal value) {
 
         return DecimalValue.fromUnscaledBigInteger(type, rawValue);
     }
-
 }