confluentinc#1439 TIMESTAMP_FORMAT does not respect escaping (conflue…

…ntinc#1781)

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

@@ -22,13 +22,15 @@ public class StringUtil {
 
   public static String cleanQuotes(final String stringWithQuotes) {
     // TODO: move check to grammar
-    if (stringWithQuotes.startsWith("'") && stringWithQuotes.endsWith("'")) {
-      return stringWithQuotes.substring(1, stringWithQuotes.length() - 1);
+    if (!stringWithQuotes.startsWith("'") || !stringWithQuotes.endsWith("'")) {
+      return stringWithQuotes;
     }
-    return stringWithQuotes;
+    return stringWithQuotes
+        .substring(1, stringWithQuotes.length() - 1)
+        .replaceAll("''", "'");
   }
 
-  public static String join(final String delimiter, final List<? extends Object> objs) {
+  public static String join(final String delimiter, final List<?> objs) {
     final StringBuilder sb = new StringBuilder();
     int cnt = 0;
     for (final Object obj : objs) {

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

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+package io.confluent.ksql.util;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import org.junit.Test;
+
+public class StringUtilTest {
+
+  @Test
+  public void testNotCleanIfMissingQuotePrefix() {
+    assertCleaned("COLUMN_NAME'", "COLUMN_NAME'");
+  }
+
+  @Test
+  public void testNotCleanIfMissingQuotePostfix() {
+    assertCleaned("'COLUMN_NAME", "'COLUMN_NAME");
+  }
+
+  @Test
+  public void testCleanQuotesIfQuoted() {
+    assertCleaned("'COLUMN_NAME'", "COLUMN_NAME");
+  }
+
+  @Test
+  public void testDoesNotReduceAnyQuotesIfNotQuotedString() {
+    assertCleaned("prefix ''Something in quotes'' postfix",
+        "prefix ''Something in quotes'' postfix");
+  }
+
+  @Test
+  public void testReducesDoubleSingleQuotesInQuotedStringToSingleSingleQuotes() {
+    assertCleaned("'prefix ''Something in quotes'' postfix'",
+        "prefix 'Something in quotes' postfix");
+
+    assertCleaned("'a '''' b ''' c '''' d '' e '' f '''",
+        "a '' b '' c '' d ' e ' f '");
+  }
+
+  @Test
+  public void testCleanDateFormat() {
+    assertCleaned("'yyyy-MM-dd''T''HH:mm:ssX'", "yyyy-MM-dd'T'HH:mm:ssX");
+    assertCleaned("'yyyy.MM.dd G ''at'' HH:mm:ss z'", "yyyy.MM.dd G 'at' HH:mm:ss z");
+    assertCleaned("'EEE, MMM d, ''''yy'", "EEE, MMM d, ''yy");
+    assertCleaned("'hh ''o''clock'' a, zzzz'", "hh 'o'clock' a, zzzz");
+    assertCleaned("'YYYY-''W''ww-u'", "YYYY-'W'ww-u");
+  }
+
+  private static void assertCleaned(final String input, final String expected){
+    final String result = StringUtil.cleanQuotes(input);
+    assertThat(result, is(expected));
+  }
+}

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

@@ -0,0 +1,31 @@
+{
+  "comments": [
+    "You can specify multiple statements per test case, i.e., to set up the various streams needed",
+    "for joins etc, but currently only the final topology will be verified. This should be enough",
+    "for most tests as we can simulate the outputs from previous stages into the final stage. If we",
+    "take a modular approach to testing we can still verify that it all works correctly, i.e, if we",
+    "verify the output of a select or aggregate is correct, we can use simulated output to feed into",
+    "a join or another aggregate."
+  ],
+  "tests": [
+    {
+      "name": "timestamp format",
+      "statements": [
+        "CREATE STREAM TEST (ID bigint, event_timestamp VARCHAR) WITH (kafka_topic='test_topic', value_format='JSON', timestamp='event_timestamp', timestamp_format='yyyy-MM-dd''T''HH:mm:ssX');",
+        "CREATE STREAM TS AS select id, stringtotimestamp(event_timestamp, 'yyyy-MM-dd''T''HH:mm:ssX') as ets from test;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "key": 0, "value": {"ID": 1, "event_timestamp": "2018-05-11T21:58:33Z"}, "timestamp": 1526075913000},
+        {"topic": "test_topic", "key": 0, "value": {"ID": 1, "event_timestamp": "2019-05-11T21:58:33Z"}, "timestamp": 1557611913000},
+        {"topic": "test_topic", "key": 0, "value": {"ID": 1, "event_timestamp": "2020-05-11T21:58:33Z"}, "timestamp": 1589234313000}
+      ],
+      "outputs": [
+        {"topic": "TS", "key": 0, "value": {"ID": 1, "ETS": 1526075913000}, "timestamp": 1526075913000},
+        {"topic": "TS", "key": 0, "value": {"ID": 1, "ETS": 1557611913000}, "timestamp": 1557611913000},
+        {"topic": "TS", "key": 0, "value": {"ID": 1, "ETS": 1589234313000}, "timestamp": 1589234313000}
+      ]
+    }
+  ]
+}
+
+