[2/7] Add native batching configuration and parameter model

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

@@ -0,0 +1,77 @@
+package com.databricks.jdbc.api.impl;
+
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * Immutable, position-ordered snapshot of one prepared-statement parameter set.
+ *
+ * <p>This model normalizes JDBC's one-based parameter indexes to zero-based wire ordinals. It does
+ * not validate parameter completeness, index continuity, or consistency with other parameter sets;
+ * those validations remain the backend's responsibility.
+ */
+public final class BatchParameterSet {
+
+  private final List<ImmutableSqlParameter> parameters;
+
+  private BatchParameterSet(List<ImmutableSqlParameter> parameters) {
+    this.parameters = List.copyOf(parameters);
+  }
+
+  public static BatchParameterSet from(Map<Integer, ImmutableSqlParameter> parameterBindings) {
+    Objects.requireNonNull(parameterBindings, "parameterBindings");
+    List<ImmutableSqlParameter> orderedParameters =
+        parameterBindings.entrySet().stream()
+            .sorted(Comparator.comparingInt(Map.Entry::getKey))
+            .map(BatchParameterSet::snapshotParameter)
+            .collect(Collectors.toList());
+    return new BatchParameterSet(orderedParameters);
+  }
+
+  public List<ImmutableSqlParameter> getParameters() {
+    return parameters;
+  }
+
+  public int size() {
+    return parameters.size();
+  }
+
+  public boolean isEmpty() {
+    return parameters.isEmpty();
+  }
+
+  private static ImmutableSqlParameter snapshotParameter(
+      Map.Entry<Integer, ImmutableSqlParameter> entry) {
+    ImmutableSqlParameter parameter = entry.getValue();
+    return ImmutableSqlParameter.builder()
+        .cardinal(entry.getKey() - 1)
+        .type(parameter.type())
+        .value(snapshotValue(parameter.value()))
+        .build();
+  }
+
+  private static Object snapshotValue(Object value) {
+    if (value instanceof Timestamp) {
+      Timestamp timestamp = (Timestamp) value;
+      Timestamp copy = new Timestamp(timestamp.getTime());
+      copy.setNanos(timestamp.getNanos());
+      return copy;
+    }
+    if (value instanceof Date) {
+      return new Date(((Date) value).getTime());
+    }
+    if (value instanceof Time) {
+      return new Time(((Time) value).getTime());
+    }
+    if (value instanceof byte[]) {
+      return ((byte[]) value).clone();
+    }
+    return value;
+  }
+}

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

@@ -1480,6 +1480,11 @@ public boolean isBatchedInsertsEnabled() {
     return getParameter(DatabricksJdbcUrlParams.ENABLE_BATCHED_INSERTS).equals("1");
   }
 
+  @Override
+  public boolean isNativeBatchingEnabled() {
+    return getParameter(DatabricksJdbcUrlParams.ENABLE_NATIVE_BATCHING).equals("1");
+  }
+
   @Override
   public List<String> getNonRowcountQueryPrefixes() {
     String prefixesStr = getParameter(DatabricksJdbcUrlParams.NON_ROWCOUNT_QUERY_PREFIXES);

src/main/java/com/databricks/jdbc/api/internal/IDatabricksConnectionContext.java

@@ -426,6 +426,9 @@ default int getHeartbeatIntervalSeconds() {
   /** Returns whether batched INSERT optimization is enabled */
   boolean isBatchedInsertsEnabled();
 
+  /** Returns whether native parameter batch execution is enabled */
+  boolean isNativeBatchingEnabled();
+
   /** Returns whether transaction-related method calls should be ignored */
   boolean getIgnoreTransactions();
 

src/main/java/com/databricks/jdbc/common/DatabricksJdbcUrlParams.java

@@ -194,6 +194,7 @@ public enum DatabricksJdbcUrlParams {
       "Timeout in seconds for metadata polling operations (e.g. GetTables, GetColumns). 0 means no timeout",
       "300"),
   ENABLE_BATCHED_INSERTS("EnableBatchedInserts", "Enable batched INSERT optimization", "0"),
+  ENABLE_NATIVE_BATCHING("EnableNativeBatching", "Enable native parameter batch execution", "0"),
   ENABLE_SQL_VALIDATION_FOR_IS_VALID(
       "EnableSQLValidationForIsValid",
       "Enable SQL query execution for connection validation in isValid() method",

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

@@ -0,0 +1,106 @@
+package com.databricks.jdbc.api.impl;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.databricks.jdbc.model.core.ColumnInfoTypeName;
+import java.sql.Timestamp;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+class BatchParameterSetTest {
+
+  @Test
+  void ordersParametersByJdbcIndexAndUsesZeroBasedOrdinals() {
+    Map<Integer, ImmutableSqlParameter> bindings = new HashMap<>();
+    bindings.put(3, parameter(99, "third", ColumnInfoTypeName.STRING));
+    bindings.put(1, parameter(99, "first", ColumnInfoTypeName.STRING));
+    bindings.put(2, parameter(99, "second", ColumnInfoTypeName.STRING));
+
+    BatchParameterSet parameterSet = BatchParameterSet.from(bindings);
+
+    assertEquals(List.of("first", "second", "third"), values(parameterSet));
+    assertEquals(List.of(0, 1, 2), ordinals(parameterSet));
+  }
+
+  @Test
+  void preservesSparseIndexesWithoutValidation() {
+    Map<Integer, ImmutableSqlParameter> bindings = new HashMap<>();
+    bindings.put(3, parameter(3, "third", ColumnInfoTypeName.STRING));
+    bindings.put(1, parameter(1, "first", ColumnInfoTypeName.STRING));
+
+    BatchParameterSet parameterSet = BatchParameterSet.from(bindings);
+
+    assertEquals(List.of("first", "third"), values(parameterSet));
+    assertEquals(List.of(0, 2), ordinals(parameterSet));
+  }
+
+  @Test
+  void allowsEmptyParameterSet() {
+    BatchParameterSet parameterSet = BatchParameterSet.from(Map.of());
+
+    assertTrue(parameterSet.isEmpty());
+    assertEquals(0, parameterSet.size());
+  }
+
+  @Test
+  void snapshotsBindingsAndMutableValues() {
+    Timestamp timestamp = Timestamp.valueOf("2026-08-10 12:34:56.123456789");
+    byte[] bytes = new byte[] {1, 2, 3};
+    Map<Integer, ImmutableSqlParameter> bindings = new HashMap<>();
+    bindings.put(1, parameter(1, timestamp, ColumnInfoTypeName.TIMESTAMP));
+    bindings.put(2, parameter(2, bytes, ColumnInfoTypeName.BINARY));
+
+    BatchParameterSet parameterSet = BatchParameterSet.from(bindings);
+    bindings.clear();
+    timestamp.setTime(0);
+    bytes[0] = 9;
+
+    assertFalse(parameterSet.isEmpty());
+    assertEquals(
+        Timestamp.valueOf("2026-08-10 12:34:56.123456789"),
+        parameterSet.getParameters().get(0).value());
+    assertArrayEquals(new byte[] {1, 2, 3}, (byte[]) parameterSet.getParameters().get(1).value());
+    assertThrows(
+        UnsupportedOperationException.class,
+        () -> parameterSet.getParameters().add(parameter(3, "extra", ColumnInfoTypeName.STRING)));
+  }
+
+  @Test
+  void preservesNullValueAndType() {
+    BatchParameterSet parameterSet =
+        BatchParameterSet.from(Map.of(1, parameter(1, null, ColumnInfoTypeName.DECIMAL)));
+
+    ImmutableSqlParameter parameter = parameterSet.getParameters().get(0);
+    assertNull(parameter.value());
+    assertEquals(ColumnInfoTypeName.DECIMAL, parameter.type());
+    assertEquals(0, parameter.cardinal());
+  }
+
+  private ImmutableSqlParameter parameter(
+      int cardinal, Object value, ColumnInfoTypeName columnInfoTypeName) {
+    return ImmutableSqlParameter.builder()
+        .cardinal(cardinal)
+        .value(value)
+        .type(columnInfoTypeName)
+        .build();
+  }
+
+  private List<Object> values(BatchParameterSet parameterSet) {
+    return parameterSet.getParameters().stream()
+        .map(ImmutableSqlParameter::value)
+        .collect(java.util.stream.Collectors.toList());
+  }
+
+  private List<Integer> ordinals(BatchParameterSet parameterSet) {
+    return parameterSet.getParameters().stream()
+        .map(ImmutableSqlParameter::cardinal)
+        .collect(java.util.stream.Collectors.toList());
+  }
+}

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

@@ -2081,4 +2081,23 @@ public void testRefreshTokenFlowDoesNotReadCredsFromUserPassword() throws Databr
     assertNull(ctx.getClientSecret());
     assertNull(ctx.getNullableClientId());
   }
+
+  @Test
+  public void testNativeBatchingDisabledByDefault() throws DatabricksSQLException {
+    IDatabricksConnectionContext context =
+        DatabricksConnectionContext.parse(TestConstants.VALID_URL_1, new Properties());
+
+    assertFalse(context.isNativeBatchingEnabled());
+  }
+
+  @ParameterizedTest
+  @CsvSource({"0, false", "1, true", "true, false"})
+  public void testNativeBatchingConnectionProperty(String value, boolean expected)
+      throws DatabricksSQLException {
+    String url = TestConstants.VALID_URL_1 + ";EnableNativeBatching=" + value;
+
+    IDatabricksConnectionContext context = DatabricksConnectionContext.parse(url, new Properties());
+
+    assertEquals(expected, context.isNativeBatchingEnabled());
+  }
 }