fix: simplify pull query error message (#5672)

fixes: #3615

Simplifies the error message when an invalid pull query is issued. Importantly, the phase

```
Add EMIT CHANGES if you intended to issue a push query.
```

Is the second line in every case. Hopefully calling out what needs to be done if the user meant to issue a push query.

Co-authored-by: Andy Coates <[email protected]>

ksqldb-common/src/main/java/io/confluent/ksql/links/DocumentationLinks.java

@@ -35,6 +35,8 @@ public final class DocumentationLinks {
 
   public static final String SYNTHETIC_JOIN_KEY_DOC_URL = "https://cnfl.io/2LV7ouS";
 
+  public static final String PUSH_PULL_QUERY_DOC_LINK = "https://cnfl.io/queries";
+
   /*
   Schema Registry
    */

ksqldb-engine/src/main/java/io/confluent/ksql/analyzer/PullQueryValidator.java

@@ -15,6 +15,8 @@
 
 package io.confluent.ksql.analyzer;
 
+import static io.confluent.ksql.links.DocumentationLinks.PUSH_PULL_QUERY_DOC_LINK;
+
 import com.google.common.collect.ImmutableList;
 import io.confluent.ksql.parser.tree.ResultMaterialization;
 import io.confluent.ksql.util.KsqlException;
@@ -24,46 +26,10 @@
 
 public class PullQueryValidator implements QueryValidator {
 
-  private static final String PUSH_PULL_QUERY_DOC_LINK = "https://cnfl.io/queries";
-
-  public static final String NEW_QUERY_SYNTAX_SHORT_HELP = ""
-      + "Refer to " + PUSH_PULL_QUERY_DOC_LINK + " for info on query types. "
-      + "If you intended to issue a push query, resubmit with the EMIT CHANGES clause";
-
-  public static final String NEW_QUERY_SYNTAX_ADDITIONAL_HELP = ""
-      + "Query syntax in KSQL has changed. There are now two broad categories of queries:"
-      + System.lineSeparator()
-      + "- Pull queries: query the current state of the system, return a result, and terminate. "
-      + System.lineSeparator()
-      + "- Push queries: query the state of the system in motion and continue to output "
-      + "results until they meet a LIMIT condition or are terminated by the user."
-      + System.lineSeparator()
-      + System.lineSeparator()
-      + "'EMIT CHANGES' is used to to indicate a query is a push query. "
-      + "To convert a pull query into a push query, which was the default behavior in older "
-      + "versions of KSQL, add `EMIT CHANGES` to the end of the statement before any LIMIT clause."
-      + System.lineSeparator()
-      + System.lineSeparator()
-      + "For example, the following are pull queries:"
-      + System.lineSeparator()
-      + "\t'SELECT * FROM X WHERE ROWKEY=Y;' (non-windowed table)"
-      + System.lineSeparator()
-      + "\t'SELECT * FROM X WHERE ROWKEY=Y AND WINDOWSTART>=Z;' (windowed table)"
-      + System.lineSeparator()
-      + System.lineSeparator()
-      + "The following is a push query:"
-      + System.lineSeparator()
-      + "\t'SELECT * FROM X EMIT CHANGES;'"
-      + System.lineSeparator()
-      + System.lineSeparator()
-      + "Note: Persistent queries, e.g. `CREATE TABLE AS ...`, have an implicit "
-      + "`EMIT CHANGES`, but we recommend adding `EMIT CHANGES` to these statements for clarify.";
-
-  private static final String NEW_QUERY_SYNTAX_LONG_HELP = ""
-      + NEW_QUERY_SYNTAX_SHORT_HELP
-      + System.lineSeparator()
+  public static final String PULL_QUERY_SYNTAX_HELP = " "
+      + "See " + PUSH_PULL_QUERY_DOC_LINK + " for more info."
       + System.lineSeparator()
-      + NEW_QUERY_SYNTAX_ADDITIONAL_HELP;
+      + "Add EMIT CHANGES if you intended to issue a push query.";
 
   private static final List<Rule> RULES = ImmutableList.of(
       Rule.of(
@@ -105,7 +71,7 @@ public void validate(final Analysis analysis) {
     try {
       RULES.forEach(rule -> rule.check(analysis));
     } catch (final KsqlException e) {
-      throw new KsqlException(e.getMessage() + " " + NEW_QUERY_SYNTAX_LONG_HELP, e);
+      throw new KsqlException(e.getMessage() + PULL_QUERY_SYNTAX_HELP, e);
     }
   }
 

ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/pull-queries-against-materialized-aggregates.json

@@ -983,7 +983,7 @@
       ],
       "expectedError": {
         "type": "io.confluent.ksql.rest.entity.KsqlStatementErrorMessage",
-        "message": "'X' is not materialized.",
+        "message": "Can't pull from `X` as it's not a materialized table.",
         "status": 400
       }
     },

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/PullQueryExecutor.java

@@ -171,8 +171,8 @@ public TableRows execute(
 
     if (!statement.getConfig().getBoolean(KsqlConfig.KSQL_PULL_QUERIES_ENABLE_CONFIG)) {
       throw new KsqlException(
-          "Pull queries are disabled. "
-              + PullQueryValidator.NEW_QUERY_SYNTAX_SHORT_HELP
+          "Pull queries are disabled."
+              + PullQueryValidator.PULL_QUERY_SYNTAX_HELP
               + System.lineSeparator()
               + "Please set " + KsqlConfig.KSQL_PULL_QUERIES_ENABLE_CONFIG + "=true to enable "
               + "this feature.");
@@ -1000,15 +1000,9 @@ private static SourceName getSourceName(final ImmutableAnalysis analysis) {
   }
 
   private static KsqlException notMaterializedException(final SourceName sourceTable) {
-    return new KsqlException("'"
-        + sourceTable.toString(FormatOptions.noEscape()) + "' is not materialized. "
-        + PullQueryValidator.NEW_QUERY_SYNTAX_SHORT_HELP
-        + System.lineSeparator()
-        + " KSQL currently only supports pull queries on materialized aggregate tables."
-        + " i.e. those created by a 'CREATE TABLE AS SELECT <fields>, <aggregate_functions> "
-        + "FROM <sources> GROUP BY <key>' style statement."
-        + System.lineSeparator()
-        + PullQueryValidator.NEW_QUERY_SYNTAX_ADDITIONAL_HELP
+    return new KsqlException(
+        "Can't pull from " + sourceTable + " as it's not a materialized table."
+        + PullQueryValidator.PULL_QUERY_SYNTAX_HELP
     );
   }
 
@@ -1019,7 +1013,7 @@ private static KsqlException invalidWhereClauseException(
     final String additional = !windowed
         ? ""
         : System.lineSeparator()
-            + " - limits the time bounds of the windowed table. This can be: "
+            + " - (optionally) limits the time bounds of the windowed table. This can be: "
             + System.lineSeparator()
             + "    + a single window lower bound, e.g. `WHERE WINDOWSTART = z`, or"
             + System.lineSeparator()
@@ -1032,11 +1026,11 @@ private static KsqlException invalidWhereClauseException(
             + " with an optional numeric 4-digit timezone, e.g. '+0100'";
 
     return new KsqlException(msg + ". "
-        + PullQueryValidator.NEW_QUERY_SYNTAX_SHORT_HELP
+        + PullQueryValidator.PULL_QUERY_SYNTAX_HELP
         + System.lineSeparator()
         + "Pull queries require a WHERE clause that:"
         + System.lineSeparator()
-        + " - limits the query to a single key, e.g. `SELECT * FROM X WHERE <key-column>>=Y;`."
+        + " - limits the query to a single key, e.g. `SELECT * FROM X WHERE <key-column>=Y;`."
         + additional
     );
   }