From c2dee53797933d38ecc4b3f2483514f5d6ad4881 Mon Sep 17 00:00:00 2001
From: Almog Gavra <almog@confluent.io>
Date: Mon, 15 Jul 2019 16:53:07 -0700
Subject: [PATCH] fix: `DESCRIBE` now works for sources with decimal types

---
 .../main/java/io/confluent/ksql/rest/util/EntityUtil.java  | 7 +++++++
 .../io/confluent/ksql/rest/server/TemporaryEngine.java     | 2 ++
 2 files changed, 9 insertions(+)

diff --git a/ksql-rest-app/src/main/java/io/confluent/ksql/rest/util/EntityUtil.java b/ksql-rest-app/src/main/java/io/confluent/ksql/rest/util/EntityUtil.java
index 4a5e8e2fe3e3..2fe6e30ba036 100644
--- a/ksql-rest-app/src/main/java/io/confluent/ksql/rest/util/EntityUtil.java
+++ b/ksql-rest-app/src/main/java/io/confluent/ksql/rest/util/EntityUtil.java
@@ -20,6 +20,7 @@
 import io.confluent.ksql.schema.connect.SchemaWalker;
 import io.confluent.ksql.schema.ksql.LogicalSchema;
 import io.confluent.ksql.schema.ksql.SqlBaseType;
+import io.confluent.ksql.util.DecimalUtil;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -86,6 +87,12 @@ public SchemaInfo visitString(final Schema schema) {
       return primitive(SqlBaseType.STRING);
     }
 
+    @Override
+    public SchemaInfo visitBytes(final Schema schema) {
+      DecimalUtil.requireDecimal(schema);
+      return new SchemaInfo(SqlBaseType.DECIMAL, null, null);
+    }
+
     public SchemaInfo visitArray(final Schema schema, final SchemaInfo element) {
       return new SchemaInfo(SqlBaseType.ARRAY, null, element);
     }
diff --git a/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/TemporaryEngine.java b/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/TemporaryEngine.java
index 6ed19a3ab151..af7f65a91bce 100644
--- a/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/TemporaryEngine.java
+++ b/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/TemporaryEngine.java
@@ -36,6 +36,7 @@
 import io.confluent.ksql.services.ServiceContext;
 import io.confluent.ksql.services.TestServiceContext;
 import io.confluent.ksql.statement.ConfiguredStatement;
+import io.confluent.ksql.util.DecimalUtil;
 import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.timestamp.MetadataTimestampExtractionPolicy;
 import io.confluent.rest.RestConfig;
@@ -50,6 +51,7 @@ public class TemporaryEngine extends ExternalResource {
 
   public static final LogicalSchema SCHEMA = LogicalSchema.of(SchemaBuilder.struct()
       .field("val", Schema.OPTIONAL_STRING_SCHEMA)
+      .field("val2", DecimalUtil.builder(2, 1).build())
       .build());
 
   private MutableMetaStore metaStore;