Struct Data Type(Part-1): New struct type, DDL statements and Describe

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

@@ -157,7 +157,7 @@ private List<FieldSchemaInfo> buildTestSchema(int size) {
     List<FieldSchemaInfo> res = new ArrayList<>();
     List<Field> fields = dataSourceBuilder.build().fields();
     for (Field field : fields) {
-      res.add(new FieldSchemaInfo(field.name(), SchemaUtil.getSchemaFieldName(field)));
+      res.add(new FieldSchemaInfo(field.name(), SchemaUtil.getSchemaFieldType(field)));
     }
 
     return res;

ksql-common/src/main/java/io/confluent/ksql/util/SchemaUtil.java

@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.apache.avro.Schema.create;
 import static org.apache.avro.Schema.createArray;
@@ -171,17 +172,52 @@ public static Schema buildSchemaWithAlias(final Schema schema, final String alia
           .put("MAP", "MAP")
           .build();
 
-  public static String getSchemaFieldName(Field field) {
+  public static String getSchemaFieldType(Field field) {
     if (field.schema().type() == Schema.Type.ARRAY) {
-      return "ARRAY[" + TYPE_MAP.get(field.schema().valueSchema().type().name()) + "]";
+      return "ARRAY[" + getSchemaFieldType(field.schema().valueSchema().fields().get(0)) + "]";
     } else if (field.schema().type() == Schema.Type.MAP) {
-      return "MAP[" + TYPE_MAP.get(field.schema().keySchema().type().name()) + ","
-          + TYPE_MAP.get(field.schema().valueSchema().type().name()) + "]";
+      return "MAP[" + getSchemaFieldType(field.schema().keySchema().fields().get(0)) + ","
+          + getSchemaFieldType(field.schema().valueSchema().fields().get(0)) + "]";
+    } else if (field.schema().type() == Schema.Type.STRUCT) {
+      StringBuilder stringBuilder = new StringBuilder("STRUCT <");
+      stringBuilder.append(
+          field.schema().fields().stream()
+              .map(schemaField -> getSchemaFieldType(schemaField))
+              .collect(Collectors.joining(", ")));
+      stringBuilder.append(">");
+      return stringBuilder.toString();
     } else {
       return TYPE_MAP.get(field.schema().type().name());
     }
   }
 
+
+  //TODO: Improve the format with proper indentation.
+  public static String describeSchema(Schema schema) {
+    if (schema.type() == Schema.Type.ARRAY) {
+      return "ARRAY[" + describeSchema(schema.valueSchema()) + "]";
+    } else if (schema.type() == Schema.Type.MAP) {
+      return "MAP[" + describeSchema(schema.keySchema()) + ","
+             + describeSchema(schema.valueSchema()) + "]";
+    } else if (schema.type() == Schema.Type.STRUCT) {
+      StringBuilder stringBuilder = new StringBuilder("STRUCT < ");
+      boolean addComma = false;
+      for (Field structField: schema.fields()) {
+        if (addComma) {
+          stringBuilder.append(", ");
+        } else {
+          addComma = true;
+        }
+        stringBuilder
+            .append("\n\t " + structField.name() + " " + describeSchema(structField.schema()));
+      }
+      stringBuilder.append("\n >");
+      return stringBuilder.toString();
+    } else {
+      return TYPE_MAP.get(schema.type().name());
+    }
+  }
+
   public static String getJavaCastString(Schema schema) {
     switch (schema.type()) {
       case INT32:

ksql-common/src/test/java/io/confluent/ksql/util/SchemaUtilTest.java

@@ -295,4 +295,44 @@ public void shouldReturnFieldNameWithoutAliasAsIs() {
     assertThat("Invalid field name", SchemaUtil.getFieldNameWithNoAlias(schema.fields().get(0)),
         equalTo(schema.fields().get(0).name()));
   }
+
+  @Test
+  public void shouldCreateCorrectSchemaDescription() {
+    Schema addressSchema = SchemaBuilder.struct()
+        .field("NUMBER", Schema.INT64_SCHEMA)
+        .field("STREET", Schema.STRING_SCHEMA)
+        .field("CITY", Schema.STRING_SCHEMA)
+        .field("STATE", Schema.STRING_SCHEMA)
+        .field("ZIPCODE", Schema.INT64_SCHEMA)
+        .build();
+
+    SchemaBuilder schemaBuilder = SchemaBuilder.struct();
+    Schema structSchema = schemaBuilder
+        .field("ordertime", Schema.INT64_SCHEMA)
+        .field("orderid", Schema.INT64_SCHEMA)
+        .field("itemid", Schema.STRING_SCHEMA)
+        .field("orderunits", Schema.FLOAT64_SCHEMA)
+        .field("arraycol",schemaBuilder.array(Schema.FLOAT64_SCHEMA))
+        .field("mapcol", schemaBuilder.map(Schema.STRING_SCHEMA, Schema.FLOAT64_SCHEMA))
+        .field("address", addressSchema).build();
+
+    String schemaDescription = SchemaUtil.describeSchema(structSchema);
+
+    assertThat(schemaDescription, equalTo("STRUCT < \n"
+                                          + "\t ordertime BIGINT, \n"
+                                          + "\t orderid BIGINT, \n"
+                                          + "\t itemid VARCHAR(STRING), \n"
+                                          + "\t orderunits DOUBLE, \n"
+                                          + "\t arraycol ARRAY[DOUBLE], \n"
+                                          + "\t mapcol MAP[VARCHAR(STRING),DOUBLE], \n"
+                                          + "\t address STRUCT < \n"
+                                          + "\t NUMBER BIGINT, \n"
+                                          + "\t STREET VARCHAR(STRING), \n"
+                                          + "\t CITY VARCHAR(STRING), \n"
+                                          + "\t STATE VARCHAR(STRING), \n"
+                                          + "\t ZIPCODE BIGINT\n"
+                                          + " >\n"
+                                          + " >"));
+  }
+
 }

ksql-engine/src/main/java/io/confluent/ksql/ddl/commands/AbstractCreateStreamCommand.java

@@ -35,10 +35,12 @@
 import io.confluent.ksql.util.KsqlPreconditions;
 import io.confluent.ksql.util.SchemaUtil;
 import io.confluent.ksql.util.StringUtil;
+import io.confluent.ksql.util.TypeUtil;
 import io.confluent.ksql.util.timestamp.TimestampExtractionPolicy;
 import io.confluent.ksql.util.timestamp.TimestampExtractionPolicyFactory;
 
 
+
 /**
  * Base class of create table/stream command
  */
@@ -140,7 +142,7 @@ private SchemaBuilder getStreamTableSchema(List<TableElement> tableElementList)
       }
       tableSchema = tableSchema.field(
           tableElement.getName(),
-          SchemaUtil.getTypeSchema(tableElement.getType())
+          TypeUtil.getTypeSchema(tableElement.getType())
       );
     }
 

ksql-engine/src/main/java/io/confluent/ksql/util/AvroUtil.java

@@ -143,8 +143,8 @@ private AbstractStreamCreateStatement addAvroFields(
   ) {
     List<TableElement> elements = new ArrayList<>();
     for (Field field : schema.fields()) {
-      TableElement tableElement = new TableElement(field.name().toUpperCase(), SchemaUtil
-          .getSqlTypeName(field.schema()));
+      TableElement tableElement = new TableElement(field.name().toUpperCase(),
+                                                   TypeUtil.getKsqlType(field.schema()));
       elements.add(tableElement);
     }
     StringLiteral schemaIdLiteral = new StringLiteral(String.format("%d", schemaId));

ksql-engine/src/test/java/io/confluent/ksql/ddl/commands/CommandFactoriesTest.java

@@ -35,11 +35,13 @@
 import io.confluent.ksql.parser.tree.DropTable;
 import io.confluent.ksql.parser.tree.DropTopic;
 import io.confluent.ksql.parser.tree.Expression;
+import io.confluent.ksql.parser.tree.PrimitiveType;
 import io.confluent.ksql.parser.tree.QualifiedName;
 import io.confluent.ksql.parser.tree.RegisterTopic;
 import io.confluent.ksql.parser.tree.SetProperty;
 import io.confluent.ksql.parser.tree.StringLiteral;
 import io.confluent.ksql.parser.tree.TableElement;
+import io.confluent.ksql.parser.tree.Type;
 import io.confluent.ksql.util.KafkaTopicClient;
 import io.confluent.ksql.util.KsqlException;
 
@@ -92,8 +94,8 @@ public void shouldCreateCommandForCreateTable() {
     tableProperties.put(DdlConfig.KEY_NAME_PROPERTY, new StringLiteral("COL1"));
     final DdlCommand result = commandFactories.create(sqlExpression,
         new CreateTable(QualifiedName.of("foo"),
-                        Arrays.asList(new TableElement("COL1", "BIGINT"), new TableElement
-                            ("COL2", "VARCHAR")), true,
+                        Arrays.asList(new TableElement("COL1", new PrimitiveType(Type.KsqlType.BIGINT)), new TableElement
+                            ("COL2", new PrimitiveType(Type.KsqlType.STRING))), true,
                         tableProperties),
         Collections.emptyMap());
 
@@ -108,8 +110,8 @@ public void shouldFailCreateTableIfKeyNameIsIncorrect() {
     try {
       final DdlCommand result = commandFactories.create(sqlExpression,
           new CreateTable(QualifiedName.of("foo"),
-                          Arrays.asList(new TableElement("COL1", "BIGINT"), new TableElement
-                              ("COL2", "VARCHAR")), true,
+                          Arrays.asList(new TableElement("COL1", new PrimitiveType(Type.KsqlType.BIGINT)), new TableElement
+                              ("COL2", new PrimitiveType(Type.KsqlType.STRING))), true,
                           tableProperties),
           Collections.emptyMap());
 
@@ -128,8 +130,8 @@ public void shouldFailCreateTableIfTimestampColumnNameIsIncorrect() {
     try {
       commandFactories.create(sqlExpression,
           new CreateTable(QualifiedName.of("foo"),
-                          Arrays.asList(new TableElement("COL1", "BIGINT"), new TableElement
-                              ("COL2", "VARCHAR")), true,
+                          Arrays.asList(new TableElement("COL1", new PrimitiveType(Type.KsqlType.BIGINT)), new TableElement
+                              ("COL2", new PrimitiveType(Type.KsqlType.STRING))), true,
                           tableProperties),
           Collections.emptyMap());
 
@@ -144,8 +146,8 @@ public void shouldFailCreateTableIfKeyIsNotProvided() {
     try {
       commandFactories.create(sqlExpression,
           new CreateTable(QualifiedName.of("foo"),
-                          Arrays.asList(new TableElement("COL1", "BIGINT"), new TableElement
-                              ("COL2", "VARCHAR")), true, properties),
+                          Arrays.asList(new TableElement("COL1", new PrimitiveType(Type.KsqlType.BIGINT)), new TableElement
+                              ("COL2", new PrimitiveType(Type.KsqlType.STRING))), true, properties),
           Collections.emptyMap());
 
     } catch (KsqlException e) {

ksql-engine/src/test/java/io/confluent/ksql/util/AvroUtilTest.java

@@ -33,8 +33,11 @@
 import io.confluent.ksql.metastore.MetaStoreImpl;
 import io.confluent.ksql.parser.KsqlParser;
 import io.confluent.ksql.parser.tree.AbstractStreamCreateStatement;
+import io.confluent.ksql.parser.tree.Array;
+import io.confluent.ksql.parser.tree.Map;
 import io.confluent.ksql.parser.tree.Statement;
 import io.confluent.ksql.parser.tree.TableElement;
+import io.confluent.ksql.parser.tree.Type;
 import io.confluent.ksql.serde.avro.KsqlAvroTopicSerDe;
 
 import static org.easymock.EasyMock.anyObject;
@@ -70,19 +73,33 @@ public void shouldPassAvroCheck() throws Exception {
     expect(schemaRegistryClient.getLatestSchemaMetadata(anyString())).andReturn(schemaMetadata);
     replay(schemaRegistryClient);
     AbstractStreamCreateStatement abstractStreamCreateStatement = getAbstractStreamCreateStatement
-        ("CREATE STREAM S1 WITH "
-                                                                   + "(kafka_topic='s1_topic', "
+        ("CREATE STREAM S1 WITH (kafka_topic='s1_topic', "
                                      + "value_format='avro' );");
     Pair<AbstractStreamCreateStatement, String> checkResult = avroUtil.checkAndSetAvroSchema(abstractStreamCreateStatement, new HashMap<>(), schemaRegistryClient);
     AbstractStreamCreateStatement newAbstractStreamCreateStatement = checkResult.getLeft();
-    assertThat(newAbstractStreamCreateStatement.getElements(), equalTo(Arrays.asList(
-        new TableElement("ORDERTIME", "BIGINT"),
-        new TableElement("ORDERID", "BIGINT"),
-        new TableElement("ITEMID", "VARCHAR"),
-        new TableElement("ORDERUNITS", "DOUBLE"),
-        new TableElement("ARRAYCOL", "ARRAY<DOUBLE>"),
-        new TableElement("MAPCOL", "MAP<VARCHAR,DOUBLE>")
-        )));
+    List<TableElement> tableElements = newAbstractStreamCreateStatement.getElements();
+    assertThat(tableElements.size(), equalTo(6));
+    assertThat(tableElements.get(0).getName(), equalTo("ORDERTIME"));
+    assertThat(tableElements.get(0).getType().getKsqlType(), equalTo(Type.KsqlType.BIGINT));
+
+    assertThat(tableElements.get(1).getName(), equalTo("ORDERID"));
+    assertThat(tableElements.get(1).getType().getKsqlType(), equalTo(Type.KsqlType.BIGINT));
+
+    assertThat(tableElements.get(2).getName(), equalTo("ITEMID"));
+    assertThat(tableElements.get(2).getType().getKsqlType(), equalTo(Type.KsqlType.STRING));
+
+    assertThat(tableElements.get(3).getName(), equalTo("ORDERUNITS"));
+    assertThat(tableElements.get(3).getType().getKsqlType(), equalTo(Type.KsqlType.DOUBLE));
+
+    assertThat(tableElements.get(4).getName(), equalTo("ARRAYCOL"));
+    assertThat(tableElements.get(4).getType().getKsqlType(), equalTo(Type.KsqlType.ARRAY));
+    assertThat(((Array) tableElements.get(4).getType()).getItemType().getKsqlType(),
+               equalTo(Type.KsqlType.DOUBLE));
+
+    assertThat(tableElements.get(5).getName(), equalTo("MAPCOL"));
+    assertThat(tableElements.get(5).getType().getKsqlType(), equalTo(Type.KsqlType.MAP));
+    assertThat(((Map) tableElements.get(5).getType()).getValueType().getKsqlType(),
+               equalTo(Type.KsqlType.DOUBLE));
   }
 
   @Test

ksql-engine/src/test/resources/query-validation-tests/max-group-by.json

@@ -11,7 +11,7 @@
     {
       "name": "max integer group by",
       "statements": [
-      "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE int) WITH (kafka_topic='test_topic', value_format='DELIMITED', key='ID');",
+      "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE integer) WITH (kafka_topic='test_topic',value_format='DELIMITED', key='ID');",
       "CREATE TABLE S2 as SELECT id, max(value) FROM test group by id;"
       ],
       "inputs": [

ksql-engine/src/test/resources/query-validation-tests/min-group-by.json

@@ -11,7 +11,7 @@
     {
       "name": "min integer group by",
       "statements": [
-      "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE int) WITH (kafka_topic='test_topic', value_format='DELIMITED', key='ID');",
+      "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE integer) WITH (kafka_topic='test_topic',value_format='DELIMITED', key='ID');",
       "CREATE TABLE S2 as SELECT id, min(value) FROM test group by id;"
       ],
       "inputs": [

ksql-engine/src/test/resources/query-validation-tests/sum.json

@@ -11,7 +11,7 @@
     {
       "name": "sum int",
       "statements": [
-        "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE int) WITH (kafka_topic='test_topic', value_format='DELIMITED', key='ID');",
+        "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE integer) WITH (kafka_topic='test_topic',value_format='DELIMITED', key='ID');",
         "CREATE TABLE S2 as SELECT id, sum(value) FROM test group by id;"
       ],
       "inputs": [

ksql-engine/src/test/resources/query-validation-tests/topk-distinct.json

@@ -11,7 +11,7 @@
     {
       "name": "topk distinct integer",
       "statements": [
-        "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE int) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
+        "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE integer) WITH (kafka_topic='test_topic',value_format='JSON', key='ID');",
         "CREATE TABLE S2 as SELECT id, topkdistinct(value, 3) as topk FROM test group by id;"
       ],
       "inputs": [

ksql-engine/src/test/resources/query-validation-tests/topk-group-by.json

@@ -11,7 +11,7 @@
     {
       "name": "topk integer",
       "statements": [
-        "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE int) WITH (kafka_topic='test_topic', value_format='JSON', key='ID');",
+        "CREATE STREAM TEST (ID bigint, NAME varchar, VALUE integer) WITH (kafka_topic='test_topic',value_format='JSON', key='ID');",
         "CREATE TABLE S2 as SELECT id, topk(value, 3) as topk FROM test group by id;"
       ],
       "inputs": [

ksql-parser/src/main/antlr4/io/confluent/ksql/parser/SqlBase.g4

@@ -262,10 +262,8 @@ primaryExpression
     | STRING                                                                         #stringLiteral
     | BINARY_LITERAL                                                                 #binaryLiteral
     | POSITION '(' valueExpression IN valueExpression ')'                            #position
-    | '(' expression (',' expression)+ ')'                                           #rowConstructor
-    | ROW '(' expression (',' expression)* ')'                                       #rowConstructor
     | qualifiedName '(' ASTERISK ')' over?                                           #functionCall
-    | qualifiedName '(' (expression (',' expression)*)? ')' over?     #functionCall
+    | qualifiedName '(' (expression (',' expression)*)? ')' over?                    #functionCall
     | identifier '->' expression                                                     #lambda
     | '(' identifier (',' identifier)* ')' '->' expression                           #lambda
     | '(' query ')'                                                                  #subqueryExpression
@@ -310,7 +308,7 @@ type
     : type ARRAY
     | ARRAY '<' type '>'
     | MAP '<' type ',' type '>'
-    | ROW '(' identifier type (',' identifier type)* ')'
+    | STRUCT '<' identifier type (',' identifier type)* '>'
     | baseType ('(' typeParameter (',' typeParameter)* ')')?
     ;
 
@@ -386,7 +384,7 @@ number
 nonReserved
     : SHOW | TABLES | COLUMNS | COLUMN | PARTITIONS | FUNCTIONS | SCHEMAS | CATALOGS | SESSION
     | ADD
-    | OVER | PARTITION | RANGE | ROWS | PRECEDING | FOLLOWING | CURRENT | ROW | MAP | ARRAY
+    | OVER | PARTITION | RANGE | ROWS | PRECEDING | FOLLOWING | CURRENT | ROW | STRUCT | MAP | ARRAY
     | TINYINT | SMALLINT | INTEGER | DATE | TIME | TIMESTAMP | INTERVAL | ZONE
     | YEAR | MONTH | DAY | HOUR | MINUTE | SECOND
     | EXPLAIN | ANALYZE | FORMAT | TYPE | TEXT | GRAPHVIZ | LOGICAL | DISTRIBUTED
@@ -509,6 +507,7 @@ PRECEDING: 'PRECEDING';
 FOLLOWING: 'FOLLOWING';
 CURRENT: 'CURRENT';
 ROW: 'ROW';
+STRUCT: 'STRUCT';
 WITH: 'WITH';
 RECURSIVE: 'RECURSIVE';
 VALUES: 'VALUES';