Improve error messages from SQL REPLACE syntax (#12523)

- Add user friendly error messages for missing or incorrect OVERWRITE clause for REPLACE SQL query
- Move validation of missing OVERWRITE clause at code level instead of parser for custom error message

sql/src/main/codegen/includes/replace.ftl

@@ -28,7 +28,7 @@ SqlNode DruidSqlReplaceEof() :
     // Using fully qualified name for Pair class, since Calcite also has a same class name being used in the Parser.jj
     org.apache.druid.java.util.common.Pair<Granularity, String> partitionedBy = new org.apache.druid.java.util.common.Pair(null, null);
     final Pair<SqlNodeList, SqlNodeList> p;
-    final SqlNode replaceTimeQuery;
+    SqlNode replaceTimeQuery = null;
 }
 {
     <REPLACE> { s = span(); }
@@ -41,8 +41,9 @@ SqlNode DruidSqlReplaceEof() :
             }
         }
     ]
-    <OVERWRITE>
-    replaceTimeQuery = ReplaceTimeQuery()
+    [
+        <OVERWRITE> replaceTimeQuery = ReplaceTimeQuery()
+    ]
     source = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)
     // PARTITIONED BY is necessary, but is kept optional in the grammar. It is asserted that it is not missing in the
     // DruidSqlInsert constructor so that we can return a custom error message.

sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlReplace.java

@@ -61,7 +61,7 @@ public class DruidSqlReplace extends SqlInsert
       @Nonnull SqlInsert insertNode,
       @Nullable Granularity partitionedBy,
       @Nullable String partitionedByStringForUnparse,
-      @Nonnull SqlNode replaceTimeQuery
+      @Nullable SqlNode replaceTimeQuery
   ) throws ParseException
   {
     super(
@@ -71,8 +71,11 @@ public class DruidSqlReplace extends SqlInsert
         insertNode.getSource(),
         insertNode.getTargetColumnList()
     );
+    if (replaceTimeQuery == null) {
+      throw new ParseException("Missing time chunk information in OVERWRITE clause for REPLACE, set it to OVERWRITE WHERE <__time based condition> or set it to overwrite the entire table with OVERWRITE ALL.");
+    }
     if (partitionedBy == null) {
-      throw new ParseException("REPLACE statements must specify PARTITIONED BY clause explictly");
+      throw new ParseException("REPLACE statements must specify PARTITIONED BY clause explicitly");
     }
     this.partitionedBy = partitionedBy;
 

sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java

@@ -899,7 +899,7 @@ public class DruidPlanner implements Closeable
 
       SqlNode replaceTimeQuery = druidSqlReplace.getReplaceTimeQuery();
       if (replaceTimeQuery == null) {
-        throw new ValidationException("Missing time chunk information in DELETE WHERE clause for replace.");
+        throw new ValidationException("Missing time chunk information in OVERWRITE clause for REPLACE, set it to OVERWRITE WHERE <__time based condition> or set it to overwrite the entire table with OVERWRITE ALL.");
       }
 
       Granularity ingestionGranularity = druidSqlReplace.getPartitionedBy();

sql/src/test/java/org/apache/druid/sql/calcite/CalciteReplaceDmlTest.java

@@ -368,6 +368,24 @@ public class CalciteReplaceDmlTest extends CalciteIngestionDmlTest
         .verify();
   }
 
+  @Test
+  public void testReplaceWithoutOverwriteClause()
+  {
+    testIngestionQuery()
+        .sql("REPLACE INTO dst SELECT * FROM foo PARTITIONED BY ALL TIME")
+        .expectValidationError(SqlPlanningException.class, "Missing time chunk information in OVERWRITE clause for REPLACE, set it to OVERWRITE WHERE <__time based condition> or set it to overwrite the entire table with OVERWRITE ALL.")
+        .verify();
+  }
+
+  @Test
+  public void testReplaceWithoutCompleteOverwriteClause()
+  {
+    testIngestionQuery()
+        .sql("REPLACE INTO dst OVERWRITE SELECT * FROM foo PARTITIONED BY ALL TIME")
+        .expectValidationError(SqlPlanningException.class, "Missing time chunk information in OVERWRITE clause for REPLACE, set it to OVERWRITE WHERE <__time based condition> or set it to overwrite the entire table with OVERWRITE ALL.")
+        .verify();
+  }
+
   @Test
   public void testReplaceIntoSystemTable()
   {