From 8c2003c7b8470292f493f47714702ce662a12a05 Mon Sep 17 00:00:00 2001
From: Mahesh Balakrishnan <maheshb@gmail.com>
Date: Thu, 3 Feb 2022 01:02:10 -0800
Subject: [PATCH] fix: bug preventing decimals in (-1, 1) from being inserted /
 queried

---
 .../io/confluent/ksql/util/DecimalUtil.java   | 20 +++++++++++++++----
 .../confluent/ksql/util/DecimalUtilTest.java  | 15 +++++++++++++-
 .../insert-values.json                        | 12 +++++++++++
 3 files changed, 42 insertions(+), 5 deletions(-)

diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/util/DecimalUtil.java b/ksqldb-common/src/main/java/io/confluent/ksql/util/DecimalUtil.java
index c67b12b3629c..c9b62b62cc7c 100644
--- a/ksqldb-common/src/main/java/io/confluent/ksql/util/DecimalUtil.java
+++ b/ksqldb-common/src/main/java/io/confluent/ksql/util/DecimalUtil.java
@@ -274,10 +274,22 @@ public static SqlType fromValue(final BigDecimal value) {
         ? value.setScale(0, BigDecimal.ROUND_UNNECESSARY)
         : value;
 
-    /* When a BigDecimal has value 0, the built-in method precision() always returns 1. To account
-    for this edge case, we just take the scale and add one and use that for the precision instead.
-    */
-    if (decimal.compareTo(BigDecimal.ZERO) == 0) {
+    /* We can't use BigDecimal.precision() directly for all cases, since it defines
+     * precision differently from SQL Decimal.
+     * In particular, if the decimal is between -1 and 1, BigDecimal precision can be
+     * lower than scale, which is disallowed in SQL Decimal. For example, 0.005 in
+     * BigDecimal has a (precision, scale) of (1, 3); whereas we expect (4, 3).
+     * To account for this edge case, we just take the scale and add one and use that
+     * for the precision instead. This works since BigDecimal defines scale as the
+     * number of digits to the right of the period; which is one lower than the precision for
+     * anything in the range (-1, 1).
+     * This covers the case where BigDecimal has a value of 0.
+     * Note: This solution differs from the SQL definition in that it returns (4, 3) for
+     * both "0.005" and ".005", whereas SQL expects (3, 3) for the latter. This is unavoidable
+     * if we use BigDecimal as an intermediate representation, since the two strings are parsed
+     * identically by it to have precision 1.
+     */
+    if (decimal.compareTo(BigDecimal.ONE) < 0 && decimal.compareTo(BigDecimal.ONE.negate()) > 0) {
       return SqlTypes.decimal(decimal.scale() + 1, decimal.scale());
     }
     return SqlTypes.decimal(decimal.precision(), Math.max(decimal.scale(), 0));
diff --git a/ksqldb-common/src/test/java/io/confluent/ksql/util/DecimalUtilTest.java b/ksqldb-common/src/test/java/io/confluent/ksql/util/DecimalUtilTest.java
index 0b14280e7a32..808ba091469b 100644
--- a/ksqldb-common/src/test/java/io/confluent/ksql/util/DecimalUtilTest.java
+++ b/ksqldb-common/src/test/java/io/confluent/ksql/util/DecimalUtilTest.java
@@ -355,8 +355,12 @@ public void shouldGetSchemaFromDecimal2_2() {
     // When:
     final SqlType schema = DecimalUtil.fromValue(new BigDecimal(".12"));
 
+    // Note: this behavior is different from the SQL specification, where
+    // we expect precision = 2, scale = 2. This difference is because we use
+    // BigDecimal in our implementation, which treats precision differently.
+
     // Then:
-    assertThat(schema, is(SqlTypes.decimal(2, 2)));
+    assertThat(schema, is(SqlTypes.decimal(3, 2)));
   }
 
   @Test
@@ -386,6 +390,15 @@ public void shouldGetSchemaFromDecimal10_5() {
     assertThat(schema, is(SqlTypes.decimal(10, 5)));
   }
 
+  @Test
+  public void shouldGetSchemaFromDecimal4_3() {
+    // When:
+    final SqlType schema = DecimalUtil.fromValue(new BigDecimal("0.005"));
+
+    // Then:
+    assertThat(schema, is(SqlTypes.decimal(4, 3)));
+  }
+
   @Test
   public void shouldFailIfBuilderWithZeroPrecision() {
     // When:
diff --git a/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/insert-values.json b/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/insert-values.json
index 3d51547eee93..c29707d483dc 100644
--- a/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/insert-values.json
+++ b/ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/insert-values.json
@@ -357,6 +357,18 @@
         {"topic": "test_topic", "key": 1, "value": {"V0": 0.00}}
       ]
     },
+    {
+      "name": "should insert decimals between zero and one",
+      "statements": [
+        "CREATE STREAM S (ID INT KEY, V0 DECIMAL(12, 2))  WITH (kafka_topic='test_topic',partitions = 1, value_format='JSON');",
+        "INSERT INTO S (ID,V0) VALUES (1,0.05);"
+      ],
+      "inputs": [
+      ],
+      "outputs": [
+        {"topic": "test_topic", "key": 1, "value": {"V0": 0.05}}
+      ]
+    },
     {
       "name": "should fail on column reference",
       "statements": [