feat: allow DDL with autocommit=false (#3057)

Adds support for running DDL statements when a connection is in autocommit=false mode. By default, DDL statements are only allowed when no transaction is active. That is; no query or DML statement has been executed which activated a read/write transaction.

A new flag is added that can be used to revert the behavior back to the original behavior where DDL is always refused when autocommit=false. The same flag can also be used to make the API behave the same as MySQL and Oracle, where any active transaction is automatically committed whenever a DDL statement is encountered.

Concretely this means that the following is now allowed:

```
set autocommit=false;
create table Singers (SingerId INT64, Name STRING(MAX)) PRIMARY KEY (SingerId);
```

The following is by default NOT allowed, unless
ddlInTransactionMode=AUTO_COMMIT_TRANSACTION

```
set autocommit=false;
select * from singers; -- This starts a transaction
create table Albums (AlbumId INT64) PRIMARY KEY (AlbumId); -- This is not allowed
```

google-cloud-spanner/clirr-ignored-differences.xml

@@ -656,5 +656,17 @@
     <className>com/google/cloud/spanner/connection/Connection</className>
     <method>com.google.cloud.spanner.Spanner getSpanner()</method>
   </difference>
+
+  <!-- Add DdlInTransactionMode -->
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/spanner/connection/Connection</className>
+    <method>void setDdlInTransactionMode(com.google.cloud.spanner.connection.DdlInTransactionMode)</method>
+  </difference>
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/spanner/connection/Connection</className>
+    <method>com.google.cloud.spanner.connection.DdlInTransactionMode getDdlInTransactionMode()</method>
+  </difference>
 
 </differences>

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/Connection.java

@@ -773,6 +773,12 @@ default boolean isDelayTransactionStartUntilFirstWrite() {
   /** Sets how savepoints should be supported on this connection. */
   void setSavepointSupport(SavepointSupport savepointSupport);
 
+  /** Returns the current {@link DdlInTransactionMode} for this connection. */
+  DdlInTransactionMode getDdlInTransactionMode();
+
+  /** Sets how the connection should behave if a DDL statement is executed during a transaction. */
+  void setDdlInTransactionMode(DdlInTransactionMode ddlInTransactionMode);
+
   /**
    * Creates a savepoint with the given name.
    *

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java

@@ -243,6 +243,7 @@ static UnitOfWorkType of(TransactionMode transactionMode) {
   private QueryOptions queryOptions = QueryOptions.getDefaultInstance();
   private RpcPriority rpcPriority = null;
   private SavepointSupport savepointSupport = SavepointSupport.FAIL_AFTER_ROLLBACK;
+  private DdlInTransactionMode ddlInTransactionMode;
 
   private String transactionTag;
   private String statementTag;
@@ -271,6 +272,7 @@ static UnitOfWorkType of(TransactionMode transactionMode) {
     this.autocommit = options.isAutocommit();
     this.queryOptions = this.queryOptions.toBuilder().mergeFrom(options.getQueryOptions()).build();
     this.rpcPriority = options.getRPCPriority();
+    this.ddlInTransactionMode = options.getDdlInTransactionMode();
     this.returnCommitStats = options.isReturnCommitStats();
     this.delayTransactionStartUntilFirstWrite = options.isDelayTransactionStartUntilFirstWrite();
     this.dataBoostEnabled = options.isDataBoostEnabled();
@@ -296,6 +298,7 @@ static UnitOfWorkType of(TransactionMode transactionMode) {
         new StatementExecutor(options.isUseVirtualThreads(), Collections.emptyList());
     this.spannerPool = Preconditions.checkNotNull(spannerPool);
     this.options = Preconditions.checkNotNull(options);
+    this.ddlInTransactionMode = options.getDdlInTransactionMode();
     this.spanner = spannerPool.getSpanner(options, this);
     this.ddlClient = Preconditions.checkNotNull(ddlClient);
     this.dbClient = Preconditions.checkNotNull(dbClient);
@@ -571,6 +574,21 @@ public RpcPriority getRPCPriority() {
     return this.rpcPriority;
   }
 
+  @Override
+  public DdlInTransactionMode getDdlInTransactionMode() {
+    return this.ddlInTransactionMode;
+  }
+
+  @Override
+  public void setDdlInTransactionMode(DdlInTransactionMode ddlInTransactionMode) {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    ConnectionPreconditions.checkState(
+        !isBatchActive(), "Cannot set DdlInTransactionMode while in a batch");
+    ConnectionPreconditions.checkState(
+        !isTransactionStarted(), "Cannot set DdlInTransactionMode while a transaction is active");
+    this.ddlInTransactionMode = Preconditions.checkNotNull(ddlInTransactionMode);
+  }
+
   @Override
   public void setStatementTimeout(long timeout, TimeUnit unit) {
     Preconditions.checkArgument(timeout > 0L, "Zero or negative timeout values are not allowed");
@@ -1639,28 +1657,55 @@ private ApiFuture<long[]> internalExecuteBatchUpdateAsync(
   }
 
   private UnitOfWork getCurrentUnitOfWorkOrStartNewUnitOfWork() {
-    return getCurrentUnitOfWorkOrStartNewUnitOfWork(false);
+    return getCurrentUnitOfWorkOrStartNewUnitOfWork(StatementType.UNKNOWN, false);
+  }
+
+  @VisibleForTesting
+  UnitOfWork getCurrentUnitOfWorkOrStartNewUnitOfWork(boolean isInternalMetadataQuery) {
+    return getCurrentUnitOfWorkOrStartNewUnitOfWork(StatementType.UNKNOWN, isInternalMetadataQuery);
+  }
+
+  private UnitOfWork getOrStartDdlUnitOfWork() {
+    return getCurrentUnitOfWorkOrStartNewUnitOfWork(StatementType.DDL, false);
   }
 
   /**
    * Returns the current {@link UnitOfWork} of this connection, or creates a new one based on the
    * current transaction settings of the connection and returns that.
    */
   @VisibleForTesting
-  UnitOfWork getCurrentUnitOfWorkOrStartNewUnitOfWork(boolean isInternalMetadataQuery) {
+  UnitOfWork getCurrentUnitOfWorkOrStartNewUnitOfWork(
+      StatementType statementType, boolean isInternalMetadataQuery) {
     if (isInternalMetadataQuery) {
       // Just return a temporary single-use transaction.
-      return createNewUnitOfWork(true);
+      return createNewUnitOfWork(/* isInternalMetadataQuery = */ true, /* forceSingleUse = */ true);
     }
+    maybeAutoCommitCurrentTransaction(statementType);
     if (this.currentUnitOfWork == null || !this.currentUnitOfWork.isActive()) {
-      this.currentUnitOfWork = createNewUnitOfWork(false);
+      this.currentUnitOfWork =
+          createNewUnitOfWork(
+              /* isInternalMetadataQuery = */ false,
+              /* forceSingleUse = */ statementType == StatementType.DDL
+                  && this.ddlInTransactionMode != DdlInTransactionMode.FAIL
+                  && !this.transactionBeginMarked);
     }
     return this.currentUnitOfWork;
   }
 
+  void maybeAutoCommitCurrentTransaction(StatementType statementType) {
+    if (this.currentUnitOfWork instanceof ReadWriteTransaction
+        && this.currentUnitOfWork.isActive()
+        && statementType == StatementType.DDL
+        && this.ddlInTransactionMode == DdlInTransactionMode.AUTO_COMMIT_TRANSACTION) {
+      commit();
+    }
+  }
+
   @VisibleForTesting
-  UnitOfWork createNewUnitOfWork(boolean isInternalMetadataQuery) {
-    if (isInternalMetadataQuery || (isAutocommit() && !isInTransaction() && !isInBatch())) {
+  UnitOfWork createNewUnitOfWork(boolean isInternalMetadataQuery, boolean forceSingleUse) {
+    if (isInternalMetadataQuery
+        || (isAutocommit() && !isInTransaction() && !isInBatch())
+        || forceSingleUse) {
       return SingleUseTransaction.newBuilder()
           .setInternalMetadataQuery(isInternalMetadataQuery)
           .setDdlClient(ddlClient)
@@ -1741,7 +1786,7 @@ private void popUnitOfWorkFromTransactionStack() {
   }
 
   private ApiFuture<Void> executeDdlAsync(CallType callType, ParsedStatement ddl) {
-    return getCurrentUnitOfWorkOrStartNewUnitOfWork().executeDdlAsync(callType, ddl);
+    return getOrStartDdlUnitOfWork().executeDdlAsync(callType, ddl);
   }
 
   @Override
@@ -1788,15 +1833,23 @@ public void startBatchDdl() {
     ConnectionPreconditions.checkState(
         !isReadOnly(), "Cannot start a DDL batch when the connection is in read-only mode");
     ConnectionPreconditions.checkState(
-        !isTransactionStarted(), "Cannot start a DDL batch while a transaction is active");
+        !isTransactionStarted()
+            || getDdlInTransactionMode() == DdlInTransactionMode.AUTO_COMMIT_TRANSACTION,
+        "Cannot start a DDL batch while a transaction is active");
     ConnectionPreconditions.checkState(
         !(isAutocommit() && isInTransaction()),
         "Cannot start a DDL batch while in a temporary transaction");
     ConnectionPreconditions.checkState(
         !transactionBeginMarked, "Cannot start a DDL batch when a transaction has begun");
+    ConnectionPreconditions.checkState(
+        isAutocommit() || getDdlInTransactionMode() != DdlInTransactionMode.FAIL,
+        "Cannot start a DDL batch when autocommit=false and ddlInTransactionMode=FAIL");
+
+    maybeAutoCommitCurrentTransaction(StatementType.DDL);
     this.batchMode = BatchMode.DDL;
     this.unitOfWorkType = UnitOfWorkType.DDL_BATCH;
-    this.currentUnitOfWork = createNewUnitOfWork(false);
+    this.currentUnitOfWork =
+        createNewUnitOfWork(/* isInternalMetadataQuery = */ false, /* forceSingleUse = */ false);
   }
 
   @Override
@@ -1814,7 +1867,8 @@ public void startBatchDml() {
     // Then create the DML batch.
     this.batchMode = BatchMode.DML;
     this.unitOfWorkType = UnitOfWorkType.DML_BATCH;
-    this.currentUnitOfWork = createNewUnitOfWork(false);
+    this.currentUnitOfWork =
+        createNewUnitOfWork(/* isInternalMetadataQuery = */ false, /* forceSingleUse = */ false);
   }
 
   @Override

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionOptions.java

@@ -183,6 +183,8 @@ public String[] getValidValues() {
   private static final String DEFAULT_OPTIMIZER_VERSION = "";
   private static final String DEFAULT_OPTIMIZER_STATISTICS_PACKAGE = "";
   private static final RpcPriority DEFAULT_RPC_PRIORITY = null;
+  private static final DdlInTransactionMode DEFAULT_DDL_IN_TRANSACTION_MODE =
+      DdlInTransactionMode.ALLOW_IN_EMPTY_TRANSACTION;
   private static final boolean DEFAULT_RETURN_COMMIT_STATS = false;
   private static final boolean DEFAULT_LENIENT = false;
   private static final boolean DEFAULT_ROUTE_TO_LEADER = true;
@@ -253,6 +255,8 @@ public String[] getValidValues() {
   public static final String LENIENT_PROPERTY_NAME = "lenient";
   /** Name of the 'rpcPriority' connection property. */
   public static final String RPC_PRIORITY_NAME = "rpcPriority";
+
+  public static final String DDL_IN_TRANSACTION_MODE_PROPERTY_NAME = "ddlInTransactionMode";
   /** Dialect to use for a connection. */
   private static final String DIALECT_PROPERTY_NAME = "dialect";
   /** Name of the 'databaseRole' connection property. */
@@ -374,6 +378,11 @@ private static String generateGuardedConnectionPropertyError(
                   ConnectionProperty.createStringProperty(
                       RPC_PRIORITY_NAME,
                       "Sets the priority for all RPC invocations from this connection (HIGH/MEDIUM/LOW). The default is HIGH."),
+                  ConnectionProperty.createStringProperty(
+                      DDL_IN_TRANSACTION_MODE_PROPERTY_NAME,
+                      "Sets the behavior of a connection when a DDL statement is executed in a read/write transaction. The default is "
+                          + DEFAULT_DDL_IN_TRANSACTION_MODE
+                          + "."),
                   ConnectionProperty.createStringProperty(
                       DIALECT_PROPERTY_NAME,
                       "Sets the dialect to use for new databases that are created by this connection."),
@@ -697,6 +706,7 @@ public static Builder newBuilder() {
   private final boolean autoConfigEmulator;
   private final Dialect dialect;
   private final RpcPriority rpcPriority;
+  private final DdlInTransactionMode ddlInTransactionMode;
   private final boolean delayTransactionStartUntilFirstWrite;
   private final boolean trackSessionLeaks;
   private final boolean trackConnectionLeaks;
@@ -757,6 +767,7 @@ private ConnectionOptions(Builder builder) {
         determineHost(
             matcher, parseEndpoint(this.uri), autoConfigEmulator, usePlainText, System.getenv());
     this.rpcPriority = parseRPCPriority(this.uri);
+    this.ddlInTransactionMode = parseDdlInTransactionMode(this.uri);
     this.delayTransactionStartUntilFirstWrite = parseDelayTransactionStartUntilFirstWrite(this.uri);
     this.trackSessionLeaks = parseTrackSessionLeaks(this.uri);
     this.trackConnectionLeaks = parseTrackConnectionLeaks(this.uri);
@@ -1195,6 +1206,14 @@ static RpcPriority parseRPCPriority(String uri) {
     return value != null ? RpcPriority.valueOf(value) : DEFAULT_RPC_PRIORITY;
   }
 
+  @VisibleForTesting
+  static DdlInTransactionMode parseDdlInTransactionMode(String uri) {
+    String value = parseUriProperty(uri, DDL_IN_TRANSACTION_MODE_PROPERTY_NAME);
+    return value != null
+        ? DdlInTransactionMode.valueOf(value.toUpperCase())
+        : DEFAULT_DDL_IN_TRANSACTION_MODE;
+  }
+
   @VisibleForTesting
   static String parseUriProperty(String uri, String property) {
     Pattern pattern = Pattern.compile(String.format("(?is)(?:;|\\?)%s=(.*?)(?:;|$)", property));
@@ -1466,6 +1485,10 @@ RpcPriority getRPCPriority() {
     return rpcPriority;
   }
 
+  DdlInTransactionMode getDdlInTransactionMode() {
+    return this.ddlInTransactionMode;
+  }
+
   /**
    * Whether connections created by this {@link ConnectionOptions} should delay the actual start of
    * a read/write transaction until the first write operation.

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/DdlInTransactionMode.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.connection;
+
+/** Enum used for setting the behavior of DDL in read/write transactions. */
+public enum DdlInTransactionMode {
+  /** All DDL statements in a read/write transaction fail. */
+  FAIL,
+  /**
+   * DDL statements in an empty transaction are allowed. That is; if the connection is in
+   * AutoCommit=false mode and no other statement has been executed, then executing a DDL statement
+   * or a DDL batch is allowed.
+   */
+  ALLOW_IN_EMPTY_TRANSACTION,
+  /**
+   * DDL statements automatically cause the current transaction to be committed and the DDL
+   * statement is subsequently executed without a transaction. This is equal to how MySQL and Oracle
+   * behave.
+   */
+  AUTO_COMMIT_TRANSACTION;
+}

google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionImplTest.java

@@ -1751,7 +1751,7 @@ public void testTransactionTagNotAllowedAfterTransactionStarted() {
         new ConnectionImpl(
             connectionOptions, spannerPool, ddlClient, dbClient, mock(BatchClient.class)) {
           @Override
-          UnitOfWork createNewUnitOfWork(boolean isInternalMetadataQuery) {
+          UnitOfWork createNewUnitOfWork(boolean isInternalMetadataQuery, boolean forceSingleUse) {
             return unitOfWork;
           }
         }) {

google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionImplTransactionalReadWriteTest.java

@@ -151,7 +151,8 @@ boolean isGetCommitTimestampAllowed() {
     boolean isExecuteAllowed(StatementType type) {
       return type == StatementType.CLIENT_SIDE
           || type == StatementType.QUERY
-          || type == StatementType.UPDATE;
+          || type == StatementType.UPDATE
+          || type == StatementType.DDL;
     }
 
     @Override
@@ -765,7 +766,8 @@ boolean isGetCommitTimestampAllowed() {
     boolean isExecuteAllowed(StatementType type) {
       return type == StatementType.CLIENT_SIDE
           || type == StatementType.QUERY
-          || type == StatementType.UPDATE;
+          || type == StatementType.UPDATE
+          || type == StatementType.DDL;
     }
 
     @Override
@@ -920,7 +922,8 @@ boolean isGetCommitTimestampAllowed() {
     boolean isExecuteAllowed(StatementType type) {
       return type == StatementType.CLIENT_SIDE
           || type == StatementType.QUERY
-          || type == StatementType.UPDATE;
+          || type == StatementType.UPDATE
+          || type == StatementType.DDL;
     }
 
     @Override
@@ -1074,7 +1077,8 @@ boolean isGetCommitTimestampAllowed() {
     boolean isExecuteAllowed(StatementType type) {
       return type == StatementType.CLIENT_SIDE
           || type == StatementType.QUERY
-          || type == StatementType.UPDATE;
+          || type == StatementType.UPDATE
+          || type == StatementType.DDL;
     }
 
     @Override
@@ -1378,7 +1382,8 @@ boolean isGetCommitTimestampAllowed() {
     boolean isExecuteAllowed(StatementType type) {
       return type == StatementType.CLIENT_SIDE
           || type == StatementType.QUERY
-          || type == StatementType.UPDATE;
+          || type == StatementType.UPDATE
+          || type == StatementType.DDL;
     }
 
     @Override
@@ -1829,7 +1834,8 @@ boolean isGetCommitTimestampAllowed() {
     boolean isExecuteAllowed(StatementType type) {
       return type == StatementType.CLIENT_SIDE
           || type == StatementType.QUERY
-          || type == StatementType.UPDATE;
+          || type == StatementType.UPDATE
+          || type == StatementType.DDL;
     }
 
     @Override
@@ -1979,7 +1985,8 @@ boolean isGetCommitTimestampAllowed() {
     boolean isExecuteAllowed(StatementType type) {
       return type == StatementType.CLIENT_SIDE
           || type == StatementType.QUERY
-          || type == StatementType.UPDATE;
+          || type == StatementType.UPDATE
+          || type == StatementType.DDL;
     }
 
     @Override