Fix NPE for SQL queries when a query parameter is missing in the mid (#11900)

* Fix NPE for SQL queries when a query parameter is missing in the mid

* checkstyle

* Throw SqlPlanningException instead of IAE

integration-tests/src/test/java/org/apache/druid/tests/query/ITJdbcQueryTest.java

@@ -21,6 +21,7 @@ package org.apache.druid.tests.query;
 
 import com.google.common.collect.ImmutableList;
 import com.google.inject.Inject;
+import org.apache.calcite.avatica.AvaticaSqlException;
 import org.apache.druid.https.SSLClientConfig;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.logger.Logger;
@@ -163,7 +164,7 @@ public class ITJdbcQueryTest
         );
       }
       catch (SQLException throwables) {
-        Assert.assertFalse(true, throwables.getMessage());
+        Assert.fail(throwables.getMessage());
       }
     }
   }
@@ -185,7 +186,7 @@ public class ITJdbcQueryTest
         }
       }
       catch (SQLException throwables) {
-        Assert.assertFalse(true, throwables.getMessage());
+        Assert.fail(throwables.getMessage());
       }
     }
   }
@@ -208,7 +209,18 @@ public class ITJdbcQueryTest
         }
       }
       catch (SQLException throwables) {
-        Assert.assertFalse(true, throwables.getMessage());
+        Assert.fail(throwables.getMessage());
+      }
+    }
+  }
+
+  @Test(expectedExceptions = AvaticaSqlException.class, expectedExceptionsMessageRegExp = ".* Parameter at position\\[0] is not bound")
+  public void testJdbcPrepareStatementQueryMissingParameters() throws SQLException
+  {
+    for (String url : connections) {
+      try (Connection connection = DriverManager.getConnection(url, connectionProperties);
+           PreparedStatement statement = connection.prepareStatement(QUERY_PARAMETERIZED);
+           ResultSet resultSet = statement.executeQuery()) {
       }
     }
   }

sql/src/main/java/org/apache/druid/sql/SqlPlanningException.java

@@ -65,7 +65,7 @@ public class SqlPlanningException extends BadQueryException
     this(PlanningError.VALIDATION_ERROR, e.getMessage());
   }
 
-  private SqlPlanningException(PlanningError planningError, String errorMessage)
+  public SqlPlanningException(PlanningError planningError, String errorMessage)
   {
     this(planningError.errorCode, errorMessage, planningError.errorClass);
   }

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

@@ -43,7 +43,9 @@ import org.apache.calcite.rex.RexDynamicParam;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexShuttle;
 import org.apache.calcite.sql.type.SqlTypeName;
-import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.sql.SqlPlanningException;
+import org.apache.druid.sql.SqlPlanningException.PlanningError;
 
 /**
  * Traverse {@link RelNode} tree and replaces all {@link RexDynamicParam} with {@link org.apache.calcite.rex.RexLiteral}
@@ -198,6 +200,12 @@ public class RelParameterizerShuttle implements RelShuttle
       // if we have a value for dynamic parameter, replace with a literal, else add to list of unbound parameters
       if (plannerContext.getParameters().size() > dynamicParam.getIndex()) {
         TypedValue param = plannerContext.getParameters().get(dynamicParam.getIndex());
+        if (param == null) {
+          throw new SqlPlanningException(
+              PlanningError.VALIDATION_ERROR,
+              StringUtils.format("Parameter at position[%s] is not bound", dynamicParam.getIndex())
+          );
+        }
         if (param.value == null) {
           return builder.makeNullLiteral(typeFactory.createSqlType(SqlTypeName.NULL));
         }
@@ -208,7 +216,10 @@ public class RelParameterizerShuttle implements RelShuttle
             true
         );
       } else {
-        throw new ISE("Parameter: [%s] is not bound", dynamicParam.getName());
+        throw new SqlPlanningException(
+            PlanningError.VALIDATION_ERROR,
+            StringUtils.format("Parameter at position[%s] is not bound", dynamicParam.getIndex())
+        );
       }
     }
     return node;

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

@@ -26,6 +26,7 @@ import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.util.SqlShuttle;
 import org.apache.calcite.util.TimestampString;
+import org.apache.druid.java.util.common.IAE;
 
 /**
  * Replaces all {@link SqlDynamicParam} encountered in an {@link SqlNode} tree with a {@link SqlLiteral} if a value
@@ -51,6 +52,9 @@ public class SqlParameterizerShuttle extends SqlShuttle
     try {
       if (plannerContext.getParameters().size() > param.getIndex()) {
         TypedValue paramBinding = plannerContext.getParameters().get(param.getIndex());
+        if (paramBinding == null) {
+          throw new IAE("Parameter at position[%s] is not bound", param.getIndex());
+        }
         if (paramBinding.value == null) {
           return SqlLiteral.createNull(param.getParserPosition());
         }
@@ -67,6 +71,8 @@ public class SqlParameterizerShuttle extends SqlShuttle
         }
 
         return typeName.createLiteral(paramBinding.value, param.getParserPosition());
+      } else {
+        throw new IAE("Parameter at position[%s] is not bound", param.getIndex());
       }
     }
     catch (ClassCastException ignored) {

sql/src/main/java/org/apache/druid/sql/http/SqlQuery.java

@@ -36,7 +36,10 @@ public class SqlQuery
   public static List<TypedValue> getParameterList(List<SqlParameter> parameters)
   {
     return parameters.stream()
-                     .map(SqlParameter::getTypedValue)
+                     // null params are not good!
+                     // we pass them to the planner, so that it can generate a proper error message.
+                     // see SqlParameterizerShuttle and RelParameterizerShuttle.
+                     .map(p -> p == null ? null : p.getTypedValue())
                      .collect(Collectors.toList());
   }
 

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

@@ -38,12 +38,16 @@ import org.apache.druid.query.scan.ScanQuery;
 import org.apache.druid.query.scan.ScanQuery.ResultFormat;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.SqlPlanningException;
 import org.apache.druid.sql.calcite.filtration.Filtration;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.http.SqlParameter;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * This class has copied a subset of the tests in {@link CalciteQueryTest} and replaced various parts of queries with
  * dynamic parameters. It is NOT important that this file remains in sync with {@link CalciteQueryTest}, the tests
@@ -575,8 +579,8 @@ public class CalciteParameterQueryTest extends BaseCalciteQueryTest
   @Test
   public void testMissingParameter() throws Exception
   {
-    expectedException.expect(IllegalStateException.class);
+    expectedException.expect(SqlPlanningException.class);
-    expectedException.expectMessage("Parameter: [?0] is not bound");
+    expectedException.expectMessage("Parameter at position[0] is not bound");
     testQuery(
         "SELECT COUNT(*)\n"
         + "FROM druid.numfoo\n"
@@ -590,8 +594,8 @@ public class CalciteParameterQueryTest extends BaseCalciteQueryTest
   @Test
   public void testPartiallyMissingParameter() throws Exception
   {
-    expectedException.expect(IllegalStateException.class);
+    expectedException.expect(SqlPlanningException.class);
-    expectedException.expectMessage("Parameter: [?1] is not bound");
+    expectedException.expectMessage("Parameter at position[1] is not bound");
     testQuery(
         "SELECT COUNT(*)\n"
         + "FROM druid.numfoo\n"
@@ -602,6 +606,22 @@ public class CalciteParameterQueryTest extends BaseCalciteQueryTest
     );
   }
 
+  @Test
+  public void testPartiallyMissingParameterInTheMiddle() throws Exception
+  {
+    List<SqlParameter> params = new ArrayList<>();
+    params.add(null);
+    params.add(new SqlParameter(SqlType.INTEGER, 1));
+    expectedException.expect(SqlPlanningException.class);
+    expectedException.expectMessage("Parameter at position[0] is not bound");
+    testQuery(
+        "SELECT 1 + ?, dim1 FROM foo LIMIT ?",
+        ImmutableList.of(),
+        ImmutableList.of(),
+        params
+    );
+  }
+
   @Test
   public void testWrongTypeParameter() throws Exception
   {