From 0bbab5c5be4a18574d5f4be44d99648348068db1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sergio=20Pe=C3=B1a?= <sergio@confluent.io>
Date: Wed, 14 Aug 2019 17:31:12 -0500
Subject: [PATCH] fix: create persistent queries fail with reserved words

---
 .../expression/formatter/ExpressionFormatter.java      |  7 ++++++-
 .../io/confluent/ksql/parser/SqlFormatterTest.java     | 10 ++++++++++
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/ksql-execution/src/main/java/io/confluent/ksql/execution/expression/formatter/ExpressionFormatter.java b/ksql-execution/src/main/java/io/confluent/ksql/execution/expression/formatter/ExpressionFormatter.java
index 55c1db418020..bdc2135012c3 100644
--- a/ksql-execution/src/main/java/io/confluent/ksql/execution/expression/formatter/ExpressionFormatter.java
+++ b/ksql-execution/src/main/java/io/confluent/ksql/execution/expression/formatter/ExpressionFormatter.java
@@ -153,11 +153,16 @@ public String visitDereferenceExpression(
         final Context context) {
       final String baseString = process(node.getBase(), context);
       if (node.getBase() instanceof QualifiedNameReference) {
-        return baseString + KsqlConstants.DOT + formatIdentifier(node.getFieldName());
+        return baseString + KsqlConstants.DOT
+            + quoteReservedWord(formatIdentifier(node.getFieldName()), context);
       }
       return baseString + KsqlConstants.STRUCT_FIELD_REF + formatIdentifier(node.getFieldName());
     }
 
+    private String quoteReservedWord(final String s, final Context context) {
+      return context.isReserved.test(s) ? "`" + s + "`" : s;
+    }
+
     private static String formatQualifiedName(final QualifiedName name) {
       final List<String> parts = new ArrayList<>();
       for (final String part : name.getParts()) {
diff --git a/ksql-parser/src/test/java/io/confluent/ksql/parser/SqlFormatterTest.java b/ksql-parser/src/test/java/io/confluent/ksql/parser/SqlFormatterTest.java
index d686cddada1c..5e5ec079823b 100644
--- a/ksql-parser/src/test/java/io/confluent/ksql/parser/SqlFormatterTest.java
+++ b/ksql-parser/src/test/java/io/confluent/ksql/parser/SqlFormatterTest.java
@@ -113,6 +113,7 @@ public class SqlFormatterTest {
       .valueField("ARRAYCOL", SqlTypes.array(SqlTypes.DOUBLE))
       .valueField("MAPCOL", SqlTypes.map(SqlTypes.DOUBLE))
       .valueField("ADDRESS", addressSchema)
+      .valueField("SIZE", SqlTypes.INTEGER) // Reserved word
       .build();
 
   private static final CreateSourceProperties SOME_WITH_PROPS = CreateSourceProperties.from(
@@ -358,6 +359,15 @@ public void shouldFormatSelectStarCorrectly() {
             + "FROM ADDRESS ADDRESS"));
   }
 
+  @Test
+  public void shouldFormatCSASWithReservedWords() {
+    final String statementString = "CREATE STREAM S AS SELECT ITEMID, \"SIZE\" FROM address;";
+    final Statement statement = parseSingle(statementString);
+    assertThat(SqlFormatter.formatSql(statement),
+        equalTo("CREATE STREAM S AS SELECT\n" +
+            "  ADDRESS.ITEMID \"ITEMID\",\n  ADDRESS.`SIZE` \"SIZE\"\nFROM ADDRESS ADDRESS"));
+  }
+
   @Test
   public void shouldFormatSelectStarCorrectlyWithOtherFields() {
     final String statementString = "CREATE STREAM S AS SELECT *, address AS city FROM address;";