[WIP] Support for server-side DML

pom.xml

@@ -11,7 +11,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>nl.topicus</groupId>
 	<artifactId>spanner-jdbc</artifactId>
-	<version>1.1.4-SNAPSHOT</version>
+	<version>1.2-SNAPSHOT</version>
 	<name>spanner-jdbc</name>
 	<description>JDBC Driver for Google Cloud Spanner</description>
 	<url>https://github.com/olavloite/spanner-jdbc</url>
@@ -41,12 +41,12 @@
 		<dependency>
 			<groupId>com.google.cloud</groupId>
 			<artifactId>google-cloud-spanner</artifactId>
-			<version>0.57.0-beta</version>
+			<version>0.66.0-beta</version>
 		</dependency>
 		<dependency>
 			<groupId>com.google.cloud</groupId>
 			<artifactId>google-cloud-storage</artifactId>
-			<version>1.39.0</version>
+			<version>1.48.0</version>
 		</dependency>
 		<dependency>
 			<groupId>org.json</groupId>

src/main/java/nl/topicus/jdbc/CloudSpannerConnection.java

@@ -137,6 +137,9 @@ public boolean equals(Object o) {
 
   private final Properties suppliedProperties;
 
+  private boolean originalUseServerDML;
+  private boolean useServerDML;
+
   private boolean originalAllowExtendedMode;
   private boolean allowExtendedMode;
 
@@ -657,6 +660,25 @@ public Properties getSuppliedProperties() {
     return suppliedProperties;
   }
 
+  @Override
+  public boolean isUseServerDML() {
+    return useServerDML;
+  }
+
+  @Override
+  public int setUseServerDML(boolean useServerDML) {
+    this.useServerDML = useServerDML;
+    return 1;
+  }
+
+  boolean isOriginalUseServerDML() {
+    return originalUseServerDML;
+  }
+
+  void setOriginalUseServerDML(boolean useServerDML) {
+    this.originalUseServerDML = useServerDML;
+  }
+
   @Override
   public boolean isAllowExtendedMode() {
     return allowExtendedMode;
@@ -761,6 +783,10 @@ public int resetDynamicConnectionProperty(String propertyName) throws SQLExcepti
   }
 
   private Supplier<Boolean> getOriginalValueGetter(String propertyName) {
+    if (propertyName.equalsIgnoreCase(
+        ConnectionProperties.getPropertyName(ConnectionProperties.USE_SERVER_DML))) {
+      return this::isOriginalUseServerDML;
+    }
     if (propertyName.equalsIgnoreCase(
         ConnectionProperties.getPropertyName(ConnectionProperties.ALLOW_EXTENDED_MODE))) {
       return this::isOriginalAllowExtendedMode;
@@ -791,6 +817,10 @@ static interface SqlFunction<T, R> {
   }
 
   private SqlFunction<Boolean, Integer> getPropertySetter(String propertyName) {
+    if (propertyName.equalsIgnoreCase(
+        ConnectionProperties.getPropertyName(ConnectionProperties.USE_SERVER_DML))) {
+      return this::setUseServerDML;
+    }
     if (propertyName.equalsIgnoreCase(
         ConnectionProperties.getPropertyName(ConnectionProperties.ALLOW_EXTENDED_MODE))) {
       return this::setAllowExtendedMode;
@@ -823,6 +853,11 @@ public ResultSet getDynamicConnectionProperties(CloudSpannerStatement statement)
   public ResultSet getDynamicConnectionProperty(CloudSpannerStatement statement,
       String propertyName) throws SQLException {
     Map<String, String> values = new HashMap<>();
+    if (propertyName == null || propertyName.equalsIgnoreCase(
+        ConnectionProperties.getPropertyName(ConnectionProperties.USE_SERVER_DML))) {
+      values.put(ConnectionProperties.getPropertyName(ConnectionProperties.USE_SERVER_DML),
+          String.valueOf(isUseServerDML()));
+    }
     if (propertyName == null || propertyName.equalsIgnoreCase(
         ConnectionProperties.getPropertyName(ConnectionProperties.ALLOW_EXTENDED_MODE))) {
       values.put(ConnectionProperties.getPropertyName(ConnectionProperties.ALLOW_EXTENDED_MODE),

src/main/java/nl/topicus/jdbc/CloudSpannerDriver.java

@@ -143,6 +143,8 @@ public CloudSpannerConnection connect(String url, Properties info) throws SQLExc
     connection.setSimulateProductName(properties.productName);
     connection.setSimulateMajorVersion(properties.majorVersion);
     connection.setSimulateMinorVersion(properties.minorVersion);
+    connection.setUseServerDML(properties.useServerDML);
+    connection.setOriginalUseServerDML(properties.useServerDML);
     connection.setAllowExtendedMode(properties.allowExtendedMode);
     connection.setOriginalAllowExtendedMode(properties.allowExtendedMode);
     connection.setAsyncDdlOperations(properties.asyncDdlOperations);

src/main/java/nl/topicus/jdbc/ConnectionProperties.java

@@ -7,7 +7,7 @@
 import nl.topicus.jdbc.exception.CloudSpannerSQLException;
 
 final class ConnectionProperties {
-  public static final int NUMBER_OF_PROPERTIES = 14;
+  public static final int NUMBER_OF_PROPERTIES = 15;
 
   static String getPropertyName(String propertyPart) {
     return propertyPart.substring(0, propertyPart.length() - 1);
@@ -27,6 +27,7 @@ static String getPropertyName(String propertyPart) {
   static final String SIMULATE_PRODUCT_MAJOR_VERSION = "SimulateProductMajorVersion=";
   static final String SIMULATE_PRODUCT_MINOR_VERSION = "SimulateProductMinorVersion=";
 
+  static final String USE_SERVER_DML = "UseServerDML=";
   static final String ALLOW_EXTENDED_MODE = "AllowExtendedMode=";
   static final String ASYNC_DDL_OPERATIONS = "AsyncDdlOperations=";
   static final String AUTO_BATCH_DDL_OPERATIONS = "AutoBatchDdlOperations=";
@@ -42,6 +43,7 @@ static String getPropertyName(String propertyPart) {
   String productName = null;
   Integer majorVersion = null;
   Integer minorVersion = null;
+  boolean useServerDML = false;
   boolean allowExtendedMode = false;
   boolean asyncDdlOperations = false;
   boolean autoBatchDdlOperations = false;
@@ -76,6 +78,8 @@ else if (conPartLower.startsWith(SIMULATE_PRODUCT_MAJOR_VERSION.toLowerCase()))
         else if (conPartLower.startsWith(SIMULATE_PRODUCT_MINOR_VERSION.toLowerCase()))
           res.minorVersion =
               parseInteger(conPart.substring(SIMULATE_PRODUCT_MINOR_VERSION.length()));
+        else if (conPartLower.startsWith(USE_SERVER_DML.toLowerCase()))
+          res.useServerDML = Boolean.valueOf(conPart.substring(USE_SERVER_DML.length()));
         else if (conPartLower.startsWith(ALLOW_EXTENDED_MODE.toLowerCase()))
           res.allowExtendedMode = Boolean.valueOf(conPart.substring(ALLOW_EXTENDED_MODE.length()));
         else if (conPartLower.startsWith(ASYNC_DDL_OPERATIONS.toLowerCase()))
@@ -142,6 +146,9 @@ void setAdditionalConnectionProperties(Properties info) {
               SIMULATE_PRODUCT_MINOR_VERSION
                   .substring(0, SIMULATE_PRODUCT_MINOR_VERSION.length() - 1).toLowerCase(),
               defaultString(minorVersion)));
+      useServerDML = Boolean.valueOf(lowerCaseInfo.getProperty(
+          USE_SERVER_DML.substring(0, USE_SERVER_DML.length() - 1).toLowerCase(),
+          String.valueOf(useServerDML)));
       allowExtendedMode = Boolean.valueOf(lowerCaseInfo.getProperty(
           ALLOW_EXTENDED_MODE.substring(0, ALLOW_EXTENDED_MODE.length() - 1).toLowerCase(),
           String.valueOf(allowExtendedMode)));
@@ -201,34 +208,38 @@ DriverPropertyInfo[] getPropertyInfo() {
         defaultString(minorVersion));
     res[7].description =
         "Use this property to make the driver return a different minor version number, for example if you are using a framework like Spring that use this property to determine how to a generate data model for Spring Batch";
-    res[8] =
+    res[8] = new DriverPropertyInfo(USE_SERVER_DML.substring(0, USE_SERVER_DML.length() - 1),
+        String.valueOf(useServerDML));
+    res[8].description =
+        "Should the user send DML statements to the server, or should it try to parse and process these locally.";
+    res[9] =
         new DriverPropertyInfo(ALLOW_EXTENDED_MODE.substring(0, ALLOW_EXTENDED_MODE.length() - 1),
             String.valueOf(allowExtendedMode));
-    res[8].description =
+    res[9].description =
         "Allow the driver to enter 'extended' mode for bulk operations. A value of false (default) indicates that the driver should never enter extended mode. If this property is set to true, the driver will execute all bulk DML-operations in a separate transaction when the number of records affected is greater than what will exceed the limitations of Cloud Spanner.";
-    res[9] =
+    res[10] =
         new DriverPropertyInfo(ASYNC_DDL_OPERATIONS.substring(0, ASYNC_DDL_OPERATIONS.length() - 1),
             String.valueOf(asyncDdlOperations));
-    res[9].description =
+    res[10].description =
         "Run DDL-operations (CREATE TABLE, ALTER TABLE, DROP TABLE, etc.) in asynchronous mode. When set to true, DDL-statements will be checked for correct syntax and other basic checks before the call returns. It can take up to several minutes before the statement has actually finished executing. The status of running DDL-operations can be queried by issuing a SHOW_DDL_OPERATIONS statement. DDL-operations that have finished can be cleared from this view by issuing a CLEAN_DDL_OPERATIONS statement.";
-    res[10] = new DriverPropertyInfo(
+    res[11] = new DriverPropertyInfo(
         AUTO_BATCH_DDL_OPERATIONS.substring(0, AUTO_BATCH_DDL_OPERATIONS.length() - 1),
         String.valueOf(autoBatchDdlOperations));
-    res[10].description =
+    res[11].description =
         "Automatically batch DDL-operations (CREATE TABLE, ALTER TABLE, DROP TABLE, etc.). When set to true, DDL-statements that are submitted through a Statement (not PreparedStatement) will automatically be batched together and only executed after an EXECUTE_DDL_BATCH statement. This property can be used in combination with the AsyncDdlOperations property to run a batch asynchronously or synchronously.";
-    res[11] = new DriverPropertyInfo(
+    res[12] = new DriverPropertyInfo(
         REPORT_DEFAULT_SCHEMA_AS_NULL.substring(0, REPORT_DEFAULT_SCHEMA_AS_NULL.length() - 1),
         String.valueOf(reportDefaultSchemaAsNull));
-    res[11].description =
+    res[12].description =
         "Report the default schema and catalog as null (true) or as an empty string (false).";
-    res[12] =
+    res[13] =
         new DriverPropertyInfo(BATCH_READ_ONLY_MODE.substring(0, BATCH_READ_ONLY_MODE.length() - 1),
             String.valueOf(batchReadOnlyMode));
-    res[12].description =
+    res[13].description =
         "Run queries in batch-read-only-mode. Use this mode when downloading large amounts of data from Cloud Spanner in combination with the methods Statement#execute(String) or PreparedStatement#execute()";
-    res[13] = new DriverPropertyInfo(USE_CUSTOM_HOST.substring(0, USE_CUSTOM_HOST.length() - 1),
+    res[14] = new DriverPropertyInfo(USE_CUSTOM_HOST.substring(0, USE_CUSTOM_HOST.length() - 1),
         String.valueOf(useCustomHost));
-    res[13].description =
+    res[14].description =
         "Connect to a custom host instead of https://spanner.googleapis.com. This enables the use of a local emulator instead of Google Cloud Spanner";
 
     return res;

src/main/java/nl/topicus/jdbc/ICloudSpannerConnection.java

@@ -24,6 +24,10 @@ public interface ICloudSpannerConnection extends Connection {
 
   public Properties getSuppliedProperties();
 
+  public boolean isUseServerDML();
+
+  public int setUseServerDML(boolean useServerDML);
+
   public boolean isAllowExtendedMode();
 
   public int setAllowExtendedMode(boolean allowExtendedMode);

src/main/java/nl/topicus/jdbc/statement/CloudSpannerPreparedStatement.java

@@ -15,6 +15,7 @@
 import com.google.cloud.spanner.Mutation.WriteBuilder;
 import com.google.cloud.spanner.Partition;
 import com.google.cloud.spanner.ReadContext;
+import com.google.cloud.spanner.ValueBinder;
 import com.google.rpc.Code;
 import net.sf.jsqlparser.JSQLParserException;
 import net.sf.jsqlparser.expression.Expression;
@@ -78,6 +79,7 @@ public class CloudSpannerPreparedStatement extends AbstractCloudSpannerPreparedS
   private boolean forceUpdate;
 
   private List<Mutations> batchMutations = new ArrayList<>();
+  private List<com.google.cloud.spanner.Statement> batchStatements = new ArrayList<>();
 
   public CloudSpannerPreparedStatement(String sql, CloudSpannerConnection connection,
       DatabaseClient dbClient) {
@@ -154,6 +156,45 @@ public ResultSet executeQuery() throws SQLException {
         Code.INVALID_ARGUMENT);
   }
 
+  private com.google.cloud.spanner.Statement.Builder createDMLBuilder(String sql) {
+    String namedSql = convertPositionalParametersToNamedParameters(sql);
+    com.google.cloud.spanner.Statement.Builder builder =
+        com.google.cloud.spanner.Statement.newBuilder(namedSql);
+    setDMLParameters(namedSql, builder);
+
+    return builder;
+  }
+
+  private void setDMLParameters(String sql, com.google.cloud.spanner.Statement.Builder builder) {
+    Character currentEndChar = null;
+    int i = 0;
+    int parIndex = 1;
+
+    while (i < sql.length()) {
+      char c = sql.charAt(i);
+      if (currentEndChar == null) {
+        if (c == '\'' || c == '"' || c == '{') {
+          currentEndChar = c == '{' ? '}' : c;
+        } else if (c == '@') {
+          ValueBinder<com.google.cloud.spanner.Statement.Builder> binder =
+              builder.bind("p" + parIndex);
+          setParamValue(binder, parIndex);
+          parIndex++;
+        }
+      } else if (c == currentEndChar) {
+        currentEndChar = null;
+      }
+      i++;
+    }
+  }
+
+  private <R> void setParamValue(ValueBinder<R> binder, int parIndex) {
+    ValueBinderExpressionVisitorAdapter<R> adapter =
+        new ValueBinderExpressionVisitorAdapter<>(getParameterStore(), binder, null);
+    adapter.setValue(getParameterStore().getParameter(parIndex),
+        getParameterStore().getType(parIndex));
+  }
+
   private com.google.cloud.spanner.Statement.Builder createSelectBuilder(Statement statement,
       String sql) {
     String namedSql = convertPositionalParametersToNamedParameters(sql);
@@ -330,26 +371,52 @@ public void addBatch() throws SQLException {
     if (isSelectStatement(sqlTokens)) {
       throw new SQLFeatureNotSupportedException("SELECT statements may not be batched");
     }
-    Mutations mutations = createMutations();
-    batchMutations.add(mutations);
+    if (getConnection().isUseServerDML()) {
+      if (!batchMutations.isEmpty()) {
+        throw new CloudSpannerSQLException(
+            "Mixing batched mutations and dml statements is not allowed", Code.FAILED_PRECONDITION);
+      }
+      batchStatements.add(createDMLBuilder(sql).build());
+    } else {
+      if (!batchStatements.isEmpty()) {
+        throw new CloudSpannerSQLException(
+            "Mixing batched mutations and dml statements is not allowed", Code.FAILED_PRECONDITION);
+      }
+      Mutations mutations = createMutations();
+      batchMutations.add(mutations);
+    }
     getParameterStore().clearParameters();
   }
 
   @Override
   public void clearBatch() throws SQLException {
     batchMutations.clear();
+    batchStatements.clear();
     getParameterStore().clearParameters();
   }
 
   @Override
   public int[] executeBatch() throws SQLException {
-    int[] res = new int[batchMutations.size()];
-    int index = 0;
-    for (Mutations mutation : batchMutations) {
-      res[index] = (int) writeMutations(mutation);
-      index++;
+    int[] res;
+    if (!batchMutations.isEmpty()) {
+      res = new int[batchMutations.size()];
+      int index = 0;
+      for (Mutations mutation : batchMutations) {
+        res[index] = (int) writeMutations(mutation);
+        index++;
+      }
+      batchMutations.clear();
+    } else if (!batchStatements.isEmpty()) {
+      res = new int[batchStatements.size()];
+      int index = 0;
+      for (com.google.cloud.spanner.Statement statement : batchStatements) {
+        res[index] = (int) getConnection().getTransaction().executeUpdate(statement);
+        index++;
+      }
+      batchStatements.clear();
+    } else {
+      res = new int[0];
     }
-    batchMutations.clear();
     getParameterStore().clearParameters();
     return res;
   }
@@ -364,8 +431,12 @@ public int executeUpdate() throws SQLException {
       String ddl = formatDDLStatement(sql);
       return executeDDL(ddl);
     }
-    Mutations mutations = createMutations();
-    return (int) writeMutations(mutations);
+    if (getConnection().isUseServerDML()) {
+      return (int) getConnection().getTransaction().executeUpdate(createDMLBuilder(sql).build());
+    } else {
+      Mutations mutations = createMutations();
+      return (int) writeMutations(mutations);
+    }
   }
 
   private Mutations createMutations() throws SQLException {

src/main/java/nl/topicus/jdbc/transaction/CloudSpannerTransaction.java

@@ -234,6 +234,14 @@ public void buffer(Iterable<Mutation> mutations) {
     transactionThread.buffer(mutations);
   }
 
+  @Override
+  public long executeUpdate(Statement statement) {
+    checkTransaction();
+    if (transactionThread == null)
+      throw new IllegalStateException("Updates are not allowed in read-only mode");
+    return transactionThread.executeUpdate(statement);
+  }
+
   @Override
   public ResultSet executeQuery(Statement statement, QueryOption... options) {
     checkTransaction();