feat: display precision and scale when describing DECIMAL columns

ksqldb-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java

@@ -1001,7 +1001,13 @@ public void shouldPrintTypesList() {
                 SqlBaseType.STRUCT,
                 ImmutableList.of(
                     new FieldInfo("f1", new SchemaInfo(SqlBaseType.STRING, null, null), Optional.empty())),
-                null)
+                null),
+            "typeC", new SchemaInfo(
+                    SqlBaseType.DECIMAL,
+                    null,
+                    null,
+                    ImmutableMap.of("precision", 10, "scale", 9)
+                )
         ))
     ));
 
@@ -1035,6 +1041,15 @@ public void shouldPrintTypesList() {
           + "        }" + NEWLINE
           + "      } ]," + NEWLINE
           + "      \"memberSchema\" : null" + NEWLINE
+          + "    }," + NEWLINE
+          + "    \"typeC\" : {" + NEWLINE
+          + "      \"type\" : \"DECIMAL\"," + NEWLINE
+          + "      \"fields\" : null," + NEWLINE
+          + "      \"memberSchema\" : null," + NEWLINE
+          + "      \"parameters\" : {" + NEWLINE
+          + "        \"precision\" : 10," + NEWLINE
+          + "        \"scale\" : 9" + NEWLINE
+          + "      }" + NEWLINE
           + "    }" + NEWLINE
           + "  }," + NEWLINE
           + "  \"warnings\" : [ ]" + NEWLINE
@@ -1045,6 +1060,7 @@ public void shouldPrintTypesList() {
           + "----------------------------------------" + NEWLINE
           + " typeA     | STRUCT<f1 VARCHAR(STRING)> " + NEWLINE
           + " typeB     | ARRAY<VARCHAR(STRING)>     " + NEWLINE
+          + " typeC     | DECIMAL(10, 9)             " + NEWLINE
           + "----------------------------------------" + NEWLINE));
     }
   }

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/util/EntityUtil.java

@@ -24,10 +24,12 @@
 import io.confluent.ksql.schema.ksql.SqlTypeWalker;
 import io.confluent.ksql.schema.ksql.types.SqlArray;
 import io.confluent.ksql.schema.ksql.types.SqlBaseType;
+import io.confluent.ksql.schema.ksql.types.SqlDecimal;
 import io.confluent.ksql.schema.ksql.types.SqlMap;
 import io.confluent.ksql.schema.ksql.types.SqlStruct;
 import io.confluent.ksql.schema.ksql.types.SqlStruct.Field;
 import io.confluent.ksql.schema.ksql.types.SqlType;
+
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
@@ -63,6 +65,7 @@ private static Optional<FieldType> fieldType(final Column column) {
         : Optional.empty();
   }
 
+
   private static final class Converter implements SqlTypeWalker.Visitor<SchemaInfo, FieldInfo> {
 
     public SchemaInfo visitType(final SqlType schema) {
@@ -84,5 +87,14 @@ public SchemaInfo visitStruct(final SqlStruct type, final List<? extends FieldIn
     public FieldInfo visitField(final Field field, final SchemaInfo type) {
       return new FieldInfo(field.name(), type, Optional.empty());
     }
+
+    public SchemaInfo visitDecimal(final SqlDecimal type) {
+      return new SchemaInfo(
+              SqlBaseType.DECIMAL,
+              null,
+              null,
+              type.toParametersMap()
+      );
+    }
   }
 }

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/util/EntityUtilTest.java

@@ -23,6 +23,7 @@
 import io.confluent.ksql.rest.entity.FieldInfo;
 import io.confluent.ksql.rest.entity.FieldInfo.FieldType;
 import io.confluent.ksql.schema.ksql.LogicalSchema;
+import io.confluent.ksql.schema.ksql.types.SqlDecimal;
 import io.confluent.ksql.schema.ksql.types.SqlType;
 import io.confluent.ksql.schema.ksql.types.SqlTypes;
 import java.util.List;
@@ -76,6 +77,27 @@ public void shouldBuildCorrectMapField() {
         equalTo("INTEGER"));
   }
 
+  @Test
+  public void shouldBuildCorrectDecimalField() {
+    // Given:
+    final SqlDecimal decimal  = SqlTypes.decimal(10, 9);
+    final LogicalSchema schema = LogicalSchema.builder()
+            .valueColumn(ColumnName.of("field"), decimal)
+            .build();
+
+    // When:
+    final List<FieldInfo> fields = EntityUtil.buildSourceSchemaEntity(schema);
+
+    // Then:
+    assertThat(fields, hasSize(1));
+    assertThat(fields.get(0).getName(), equalTo("field"));
+    assertThat(fields.get(0).getSchema().getTypeName(), equalTo("DECIMAL"));
+    assertThat(fields.get(0).getSchema().getFields(), equalTo(Optional.empty()));
+    assertThat(fields.get(0).getSchema().getParameters().get(SqlDecimal.SCALE), equalTo(decimal.getScale()));
+    assertThat(fields.get(0).getSchema().getParameters().get(SqlDecimal.PRECISION), equalTo(decimal.getPrecision()));
+  }
+
+
   @Test
   public void shouldBuildCorrectArrayField() {
     // Given:

ksqldb-rest-model/pom.xml

@@ -59,6 +59,10 @@
             <version>${io.confluent.ksql.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-guava</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

ksqldb-rest-model/src/main/java/io/confluent/ksql/rest/ApiJsonMapper.java

@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.datatype.guava.GuavaModule;
 import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import io.confluent.ksql.json.KsqlTypesSerializationModule;
 import io.confluent.ksql.json.StructSerializationModule;
@@ -38,6 +39,7 @@ public enum ApiJsonMapper {
       .registerModule(new Jdk8Module())
       .registerModule(new StructSerializationModule())
       .registerModule(new KsqlTypesSerializationModule())
+      .registerModule(new GuavaModule())
       .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
       .disable(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES)
       .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS)

ksqldb-rest-model/src/main/java/io/confluent/ksql/rest/entity/SchemaInfo.java

@@ -17,12 +17,16 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.errorprone.annotations.Immutable;
 import io.confluent.ksql.schema.ksql.types.SqlBaseType;
+import io.confluent.ksql.schema.ksql.types.SqlDecimal;
 import io.confluent.ksql.testing.EffectivelyImmutable;
+import io.confluent.ksql.util.KsqlPreconditions;
+
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -31,23 +35,37 @@
 
 @Immutable
 @JsonIgnoreProperties(ignoreUnknown = true)
+
 public class SchemaInfo {
 
   private final SqlBaseType type;
   private final ImmutableList<FieldInfo> fields;
   private final SchemaInfo memberSchema;
+  @JsonInclude(JsonInclude.Include.NON_EMPTY)
+  @EffectivelyImmutable
+  private final ImmutableMap<String, Object> parameters;
 
   @JsonCreator
   public SchemaInfo(
       @JsonProperty("type") final SqlBaseType type,
       @JsonProperty("fields") final List<? extends FieldInfo> fields,
-      @JsonProperty("memberSchema") final SchemaInfo memberSchema) {
+      @JsonProperty("memberSchema") final SchemaInfo memberSchema,
+      @JsonProperty("parameters") final ImmutableMap<String, Object> parameters) {
     Objects.requireNonNull(type);
     this.type = type;
     this.fields = fields == null
         ? null
         : ImmutableList.copyOf(fields);
     this.memberSchema = memberSchema;
+    this.parameters = parameters;
+
+  }
+
+  public SchemaInfo(
+      @JsonProperty("type") final SqlBaseType type,
+      @JsonProperty("fields") final List<? extends FieldInfo> fields,
+      @JsonProperty("memberSchema") final SchemaInfo memberSchema) {
+    this(type, fields, memberSchema, ImmutableMap.of());
   }
 
   public SqlBaseType getType() {
@@ -67,6 +85,10 @@ public Optional<SchemaInfo> getMemberSchema() {
     return Optional.ofNullable(memberSchema);
   }
 
+  public ImmutableMap<String, Object> getParameters() {
+    return parameters;
+  }
+
   @Override
   public boolean equals(final Object o) {
     if (this == o) {
@@ -105,6 +127,24 @@ public int hashCode() {
                   .stream()
                   .map(f -> f.getName() + " " + f.getSchema().toTypeString())
                   .collect(Collectors.joining(", ", SqlBaseType.STRUCT + "<", ">")))
+          .put(
+              SqlBaseType.DECIMAL,
+              si -> {
+                // Backwards compatibility case when the backend does not return parameters
+                if (si.getParameters().isEmpty()) {
+                  return String.valueOf(SqlBaseType.DECIMAL);
+                }
+
+                final Object precision = si.getParameters().get(SqlDecimal.PRECISION);
+                final Object scale = si.getParameters().get(SqlDecimal.SCALE);
+                final String parameterString = String.format("(%s, %s)", precision, scale);
+                KsqlPreconditions.checkArgument(
+                    (precision != null & scale != null),
+                    "Either one of precision and scale missing: "
+                        + parameterString
+                );
+                return SqlBaseType.DECIMAL + parameterString;
+              })
           .build();
 
   public String toTypeString() {

ksqldb-rest-model/src/test/java/io/confluent/ksql/rest/entity/SchemaInfoTest.java

@@ -0,0 +1,33 @@
+package io.confluent.ksql.rest.entity;
+
+import com.google.common.collect.ImmutableMap;
+import io.confluent.ksql.schema.ksql.types.SqlBaseType;
+import org.junit.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+public class SchemaInfoTest {
+
+  @Test
+  public void shouldCorrectlyFormatDecimalsWithPrecisionAndScale() {
+
+    final SchemaInfo schemaInfo= new SchemaInfo(
+            SqlBaseType.DECIMAL,
+            null,
+            null,
+            ImmutableMap.of("precision", 10, "scale", 9)
+    );
+    assertThat(schemaInfo.toTypeString(), equalTo("DECIMAL(10, 9)"));
+  }
+
+  @Test
+  public void shouldCorrectlyFormatDecimalsWithoutParameters() {
+    final SchemaInfo schemaInfo= new SchemaInfo(
+            SqlBaseType.DECIMAL,
+            null,
+            null
+    );
+    assertThat(schemaInfo.toTypeString(), equalTo("DECIMAL"));
+  }
+}

ksqldb-udf/src/main/java/io/confluent/ksql/schema/ksql/types/SqlDecimal.java

@@ -15,6 +15,7 @@
 
 package io.confluent.ksql.schema.ksql.types;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.errorprone.annotations.Immutable;
 import io.confluent.ksql.schema.utils.FormatOptions;
 import io.confluent.ksql.schema.utils.SchemaException;
@@ -25,6 +26,9 @@ public final class SqlDecimal extends SqlType {
 
   private final int precision;
   private final int scale;
+  public static final String PRECISION = "precision";
+  public static final String SCALE = "scale";
+
 
   public static SqlDecimal of(final int precision, final int scale) {
     return new SqlDecimal(precision, scale);
@@ -84,6 +88,10 @@ public String toString(final FormatOptions formatOptions) {
     return toString();
   }
 
+  public ImmutableMap<String, Object> toParametersMap() {
+    return ImmutableMap.of(SqlDecimal.PRECISION, precision, SqlDecimal.SCALE, scale);
+  }
+
   /**
    * Determine the decimal type should two decimals be added together.
    *