[ES-1527223] Fixes parameter counting for prepared statement queries according to JDBC standards (#907)

## Description
<!-- Provide a brief summary of the changes made and the issue they aim
to address.-->

- **Enhanced `DatabricksParameterMetaData`**:
  - Added constructor accepting SQL string for parameter analysis
- Implemented robust `countParameters(String sql)` method with
comprehensive SQL parsing
- Modified `getParameterCount()` to return statically-determined
parameter count
  
- **Updated `DatabricksPreparedStatement`**:
- Modified to pass SQL statement to `DatabricksParameterMetaData` during
initialization
  - Ensures consistent parameter counting across batch operations

### SQL Parser Capabilities

- Accurately identifies parameter markers (`?`) while respecting SQL
syntax rules
- Properly handles quoted literals (single and double quotes with escape
sequences)
- Correctly processes SQL comments (line comments `--` and block
comments `/* */`)
- Supports complex SQL constructs including multi-line statements and
nested structures


## Testing
<!-- Describe how the changes have been tested-->

Unit tests

## Additional Notes to the Reviewer
<!-- Share any additional context or insights that may help the reviewer
understand the changes better. This could include challenges faced,
limitations, or compromises made during the development process.
Also, mention any areas of the code that you would like the reviewer to
focus on specifically. -->

---------

Signed-off-by: Madhav Sainanee <[email protected]>

Author: madhav-db

NEXT_CHANGELOG.md

@@ -21,6 +21,7 @@
 - Fixed Bouncy Castle registration conflicts by using local provider instance instead of global security registration.
 - Fixed Azure U2M authentication issue.
 - Fixed unchecked exception thrown in delete session
+- Fixed ParameterMetaData.getParameterCount() to return total parameter count from SQL parsing instead of bound parameter count, aligning with JDBC standards
 
 ---
 *Note: When making changes, please add your change under the appropriate section with a brief description.* 

src/main/java/com/databricks/jdbc/api/impl/DatabricksParameterMetaData.java

@@ -18,9 +18,15 @@ public class DatabricksParameterMetaData implements ParameterMetaData {
   private static final JdbcLogger LOGGER =
       JdbcLoggerFactory.getLogger(DatabricksParameterMetaData.class);
   private final Map<Integer, ImmutableSqlParameter> parameterBindings;
+  private final int parameterCount;
 
   public DatabricksParameterMetaData() {
+    this(null);
+  }
+
+  public DatabricksParameterMetaData(String sql) {
     this.parameterBindings = new HashMap<>();
+    this.parameterCount = countParameters(sql);
   }
 
   public void put(int param, ImmutableSqlParameter value) {
@@ -37,7 +43,24 @@ public void clear() {
 
   @Override
   public int getParameterCount() throws SQLException {
-    return parameterBindings.size();
+    validateParameterBindings();
+    return parameterCount;
+  }
+
+  /**
+   * Validates that parameter bindings do not exceed the parameter count.
+   *
+   * @throws SQLException if there are more parameter bindings than expected parameters
+   */
+  private void validateParameterBindings() throws SQLException {
+    if (parameterBindings.size() > parameterCount) {
+      throw new SQLException(
+          "Number of parameter bindings ("
+              + parameterBindings.size()
+              + ") exceeds parameter count ("
+              + parameterCount
+              + ")");
+    }
   }
 
   @Override
@@ -111,4 +134,80 @@ private ImmutableSqlParameter getObject(int param) {
     }
     return parameterBindings.get(param);
   }
+
+  /**
+   * Counts the number of parameter markers (?) in the SQL statement. This is currently a hacky
+   * implementation that may need improvement for complex SQL.
+   *
+   * @param sql The SQL statement to analyze
+   * @return The number of parameter markers in the SQL statement
+   */
+  private int countParameters(String sql) {
+    if (sql == null || sql.isEmpty()) {
+      return 0;
+    }
+
+    int count = 0;
+    boolean inSingleQuote = false;
+    boolean inDoubleQuote = false;
+    boolean inLineComment = false;
+    boolean inBlockComment = false;
+
+    for (int i = 0; i < sql.length(); i++) {
+      char c = sql.charAt(i);
+      char next = (i < sql.length() - 1) ? sql.charAt(i + 1) : '\0';
+
+      // Handle comments
+      if (!inSingleQuote && !inDoubleQuote) {
+        if (!inBlockComment && !inLineComment && c == '-' && next == '-') {
+          inLineComment = true;
+          i++; // Skip next dash
+          continue;
+        } else if (!inBlockComment && !inLineComment && c == '/' && next == '*') {
+          inBlockComment = true;
+          i++; // Skip next asterisk
+          continue;
+        } else if (inLineComment && (c == '\n' || c == '\r')) {
+          inLineComment = false;
+        } else if (inBlockComment && c == '*' && next == '/') {
+          inBlockComment = false;
+          i++; // Skip next slash
+          continue;
+        }
+      }
+
+      // Skip if in comment
+      if (inLineComment || inBlockComment) {
+        continue;
+      }
+
+      // Handle quotes with escaped quotes
+      if (c == '\'') {
+        if (!inDoubleQuote) {
+          // Check for escaped single quote
+          if (inSingleQuote && i < sql.length() - 1 && sql.charAt(i + 1) == '\'') {
+            i++; // Skip the escaped quote
+          } else {
+            inSingleQuote = !inSingleQuote;
+          }
+        }
+      } else if (c == '"') {
+        if (!inSingleQuote) {
+          // Check for escaped double quote
+          if (inDoubleQuote && i < sql.length() - 1 && sql.charAt(i + 1) == '"') {
+            i++; // Skip the escaped quote
+          } else {
+            inDoubleQuote = !inDoubleQuote;
+          }
+        }
+      }
+
+      // Count parameter markers only when not inside quotes or comments
+      if (c == '?' && !inSingleQuote && !inDoubleQuote) {
+        count++;
+      }
+    }
+
+    return count;
+  }
 }

src/main/java/com/databricks/jdbc/api/impl/DatabricksPreparedStatement.java

@@ -41,7 +41,7 @@ public DatabricksPreparedStatement(DatabricksConnection connection, String sql)
     super(connection);
     this.sql = sql;
     this.interpolateParameters = connection.getConnectionContext().supportManyParameters();
-    this.databricksParameterMetaData = new DatabricksParameterMetaData();
+    this.databricksParameterMetaData = new DatabricksParameterMetaData(sql);
     this.databricksBatchParameterMetaData = new ArrayList<>();
   }
 
@@ -345,14 +345,14 @@ public boolean execute() throws SQLException {
   public void addBatch() {
     LOGGER.debug("public void addBatch()");
     this.databricksBatchParameterMetaData.add(databricksParameterMetaData);
-    this.databricksParameterMetaData = new DatabricksParameterMetaData();
+    this.databricksParameterMetaData = new DatabricksParameterMetaData(sql);
   }
 
   @Override
   public void clearBatch() throws DatabricksSQLException {
     LOGGER.debug("public void clearBatch()");
     checkIfClosed();
-    this.databricksParameterMetaData = new DatabricksParameterMetaData();
+    this.databricksParameterMetaData = new DatabricksParameterMetaData(sql);
     this.databricksBatchParameterMetaData = new ArrayList<>();
   }
 

src/test/java/com/databricks/jdbc/api/impl/DatabricksParameterMetaDataTest.java

@@ -67,4 +67,186 @@ public void testGetParameterType() throws SQLException {
     assertEquals(Types.VARCHAR, metaData.getParameterType(1));
     assertEquals(metaData.getParameterType(2), Types.INTEGER);
   }
+
+  @Test
+  public void testConstructorWithNullSql() throws SQLException {
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(null);
+    assertEquals(0, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testConstructorWithEmptySql() throws SQLException {
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData("");
+    assertEquals(0, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithSimpleQuery() throws SQLException {
+    String sql = "SELECT * FROM table WHERE id = ? AND name = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(2, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithNoParameters() throws SQLException {
+    String sql = "SELECT * FROM table WHERE id = 1 AND name = 'test'";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(0, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountIgnoresParametersInSingleQuotes() throws SQLException {
+    String sql = "SELECT * FROM table WHERE name = 'has ? in quotes' AND id = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(1, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountIgnoresParametersInDoubleQuotes() throws SQLException {
+    String sql = "SELECT * FROM table WHERE name = \"has ? in quotes\" AND id = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(1, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithEscapedSingleQuotes() throws SQLException {
+    String sql =
+        "SELECT * FROM table WHERE name = 'has '' escaped quote and ? parameter' AND id = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(1, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithEscapedDoubleQuotes() throws SQLException {
+    String sql =
+        "SELECT * FROM table WHERE name = \"has \"\" escaped quote and ? parameter\" AND id = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(1, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountIgnoresParametersInLineComments() throws SQLException {
+    String sql = "SELECT * FROM table WHERE id = ? -- comment with ? parameter\nAND name = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(2, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountIgnoresParametersInBlockComments() throws SQLException {
+    String sql =
+        "SELECT * FROM table WHERE id = ? /* block comment with ? parameter */ AND name = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(2, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithMultilineBlockComment() throws SQLException {
+    String sql =
+        "SELECT * FROM table WHERE id = ? /* multi-line\n comment with ? parameter\n */ AND name = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(2, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithComplexQuery() throws SQLException {
+    String sql =
+        "SELECT t1.id, t2.name "
+            + "FROM table1 t1 "
+            + "JOIN table2 t2 ON t1.id = t2.id "
+            + "WHERE t1.status = ? "
+            + "AND t2.created_date > ? "
+            + "AND t1.description NOT LIKE '%test?value%' -- comment with ?\n"
+            + "ORDER BY t1.id LIMIT ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(3, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithMixedQuotesAndComments() throws SQLException {
+    String sql =
+        "SELECT * FROM table WHERE col1 = 'value with ? mark' AND col2 = ? /* comment ? */ AND col3 = \"another ? value\" AND col4 = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(2, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountAfterBindingParameters() throws SQLException {
+    String sql = "SELECT * FROM table WHERE id = ? AND name = ? AND status = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+
+    // Initially should return count based on SQL parsing
+    assertEquals(3, metadata.getParameterCount());
+
+    // Add some bindings - count should still be based on SQL, not bindings
+    metadata.put(
+        1,
+        ImmutableSqlParameter.builder()
+            .type(ColumnInfoTypeName.INT)
+            .cardinal(1)
+            .value(123)
+            .build());
+
+    assertEquals(3, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithInsertStatement() throws SQLException {
+    String sql = "INSERT INTO table (col1, col2, col3) VALUES (?, ?, ?)";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(3, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithUpdateStatement() throws SQLException {
+    String sql = "UPDATE table SET col1 = ?, col2 = ? WHERE id = ? AND status = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(4, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithNestedQuotes() throws SQLException {
+    String sql =
+        "SELECT * FROM table WHERE col1 = 'outer ''inner ? not counted'' outer' AND col2 = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(1, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountWithCarriageReturnInComment() throws SQLException {
+    String sql = "SELECT * FROM table WHERE id = ? -- comment with ? \r\nAND name = ?";
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+    assertEquals(2, metadata.getParameterCount());
+  }
+
+  @Test
+  public void testParameterCountExceedsBindingsThrowsException() {
+    String sql = "SELECT * FROM table WHERE id = ? AND name = ?"; // 2 parameters
+    DatabricksParameterMetaData metadata = new DatabricksParameterMetaData(sql);
+
+    // Add 3 parameter bindings (more than the 2 parameters in SQL)
+    metadata.put(
+        1,
+        ImmutableSqlParameter.builder().type(ColumnInfoTypeName.INT).cardinal(1).value(1).build());
+    metadata.put(
+        2,
+        ImmutableSqlParameter.builder()
+            .type(ColumnInfoTypeName.STRING)
+            .cardinal(2)
+            .value("test")
+            .build());
+    metadata.put(
+        3,
+        ImmutableSqlParameter.builder()
+            .type(ColumnInfoTypeName.STRING)
+            .cardinal(3)
+            .value("extra")
+            .build());
+
+    // getParameterCount should throw SQLException due to too many bindings
+    SQLException exception = assertThrows(SQLException.class, metadata::getParameterCount);
+    assertTrue(
+        exception
+            .getMessage()
+            .contains("Number of parameter bindings (3) exceeds parameter count (2)"));
+  }
 }