Repository: GoogleCloudPlatform/pgadapter
Pull request #147 (2022-05-05 13:00:57)

fix: clear statement tag before auto rollback
Clear any statement tag before executing an automatic rollback when a
transaction is aborted, as statement tags are not allowed for rollback
requests.

Fixes #146


Patch:
diff --git a/.ci/run-with-credentials.sh b/.ci/run-with-credentials.sh
index c54df692..71cc029e 100755
--- a/.ci/run-with-credentials.sh
+++ b/.ci/run-with-credentials.sh
@@ -76,7 +76,7 @@ e2e-psql)
   echo "Starting PGAdapter"
   ls -lh target
   UBER_JAR="pgadapter.jar"
-  (java -jar target/"${UBER_JAR}" -p "${GOOGLE_CLOUD_PROJECT}" -i "${GOOGLE_CLOUD_INSTANCE}" -d "${GOOGLE_CLOUD_DATABASE_WITH_VERSION}" -e "${GOOGLE_CLOUD_ENDPOINT}" -s 4242 -q > /dev/null 2>&1) &
+  (java -jar target/"${UBER_JAR}" -p "${GOOGLE_CLOUD_PROJECT}" -i "${GOOGLE_CLOUD_INSTANCE}" -d "${GOOGLE_CLOUD_DATABASE_WITH_VERSION}" -e "${GOOGLE_CLOUD_ENDPOINT}" -s 4242 -q -ddl AutocommitImplicitTransaction > /dev/null 2>&1) &
   BACK_PID=$!
   sleep 1
 #  execute psql and evaluate result
diff --git a/README.md b/README.md
index 60e191d4..c7932052 100644
--- a/README.md
+++ b/README.md
@@ -37,10 +37,10 @@ See [Options](#Options) for an explanation of all further options.
 
 ### Standalone with pre-built jar
 
-A pre-built jar containing all dependencies can be downloaded from https://storage.googleapis.com/pgadapter-jar-releases/pgadapter.jar
+A pre-built jar and all dependencies can be downloaded from https://storage.googleapis.com/pgadapter-jar-releases/pgadapter.tar.gz
 
 ```shell
-wget https://storage.googleapis.com/pgadapter-jar-releases/pgadapter.jar
+wget https://storage.googleapis.com/pgadapter-jar-releases/pgadapter.tar.gz && tar -xzvf pgadapter.tar.gz
 java -jar pgadapter.jar -p my-project -i my-instance -d my-database
 ```
 
@@ -51,22 +51,23 @@ You can also download a specific version of the jar. Example (replace `v0.4.0` w
 
 ```shell
 VERSION=v0.4.0
-wget https://storage.googleapis.com/pgadapter-jar-releases/pgadapter-${VERSION}.jar
-java -jar pgadapter-${VERSION}.jar -p my-project -i my-instance -d my-database
+wget https://storage.googleapis.com/pgadapter-jar-releases/pgadapter-${VERSION}.tar.gz && tar -xzvf pgadapter-${VERSION}.tar.gz
+java -jar pgadapter.jar -p my-project -i my-instance -d my-database
 ```
 
 See [Options](#Options) for an explanation of all further options.
 
 ### Standalone with locally built jar
-1. Build a jar file containing all dependencies by running
+1. Build a jar file and assemble all dependencies by running
 
 ```shell
-mvn package -P shade
+mvn package -P assembly
 ```
 
-2. Execute
+2. Execute (the binaries are in the target/pgadapter folder)
 ```shell
-java -jar <jar-file> <options>
+cd target/pgadapter
+java -jar pgadapter.jar -p my-project -i my-instance -d my-database
 ```
 
 See [Options](#Options) for an explanation of all further options.
@@ -243,8 +244,9 @@ The following options are optional:
 An example of a simple run string:
 
 ```shell
-java -jar <jar-file> -p <project name> -i <instance id> -d <database name> -c
-<path to credentials file> -s 5432 
+java -jar pgadapter.jar \
+     -p <project-id> -i <instance-id> -d <database-id> \
+     -c <path to credentials file> -s 5432
 ```
 
 ## Details
diff --git a/build/startup.sh b/build/startup.sh
index 3f9b2a31..73c7ce3e 100755
--- a/build/startup.sh
+++ b/build/startup.sh
@@ -13,6 +13,6 @@ do
 done
 
 cd /home/pgadapter
-COMMAND="java ${JAVA_ARGUMENTS} -cp \"pgadapter.jar:lib/*\" com.google.cloud.spanner.pgadapter.Server ${ARGUMENTS}"
+COMMAND="java ${JAVA_ARGUMENTS} -jar pgadapter.jar ${ARGUMENTS}"
 echo $COMMAND
 eval " $COMMAND"
diff --git a/docs/jdbc.md b/docs/jdbc.md
index b0619fc9..79873e37 100644
--- a/docs/jdbc.md
+++ b/docs/jdbc.md
@@ -8,7 +8,7 @@ version 42.0.0 and higher.
 First start PGAdapter in JDBC mode by adding `-jdbc` to the command line parameters:
 
 ```shell
-wget https://storage.googleapis.com/pgadapter-jar-releases/pgadapter.jar
+wget https://storage.googleapis.com/pgadapter-jar-releases/pgadapter.tar.gz && tar -xzvf pgadapter.tar.gz
 java -jar pgadapter.jar -p my-project -i my-instance -d my-database -jdbc
 ```
 
@@ -31,5 +31,5 @@ try (Connection connection = DriverManager.getConnection("jdbc:postgresql://loca
 
 ## Running PGAdapter
 
-This example uses the pre-built jar with dependencies to run PGAdapter as a standalone process.
+This example uses the pre-built jar to run PGAdapter as a standalone process.
 See [README](../README.md) for more possibilities on how to run PGAdapter.
diff --git a/pom.xml b/pom.xml
index 609819ef..88525f09 100644
--- a/pom.xml
+++ b/pom.xml
@@ -61,11 +61,6 @@
   </licenses>
 
   <dependencies>
-    <dependency>
-      <groupId>com.googlecode.json-simple</groupId>
-      <artifactId>json-simple</artifactId>
-      <version>1.1.1</version>
-    </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-spanner</artifactId>
@@ -76,6 +71,11 @@
       <artifactId>postgresql</artifactId>
       <version>42.3.4</version>
     </dependency>
+    <dependency>
+      <groupId>com.googlecode.json-simple</groupId>
+      <artifactId>json-simple</artifactId>
+      <version>1.1.1</version>
+    </dependency>
     <dependency>
       <groupId>commons-cli</groupId>
       <artifactId>commons-cli</artifactId>
@@ -86,11 +86,6 @@
       <artifactId>commons-csv</artifactId>
       <version>1.9.0</version>
     </dependency>
-    <dependency>
-      <groupId>org.json</groupId>
-      <artifactId>json</artifactId>
-      <version>20210307</version>
-    </dependency>
     <dependency>
       <groupId>org.apache.maven</groupId>
       <artifactId>maven-model</artifactId>
@@ -116,6 +111,7 @@
       <groupId>net.java.dev.jna</groupId>
       <artifactId>jna</artifactId>
       <version>5.10.0</version>
+      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>junit</groupId>
@@ -213,7 +209,7 @@
                   <goal>copy-resources</goal>
                 </goals>
                 <configuration>
-                  <outputDirectory>${basedir}/target/PGAdapter</outputDirectory>
+                  <outputDirectory>${project.build.directory}/pgadapter</outputDirectory>
                   <resources>
                     <resource>
                       <directory>resources</directory>
@@ -248,6 +244,14 @@
             <artifactId>maven-jar-plugin</artifactId>
             <configuration>
               <finalName>pgadapter/pgadapter</finalName>
+              <archive>
+                <index>false</index>
+                <manifest>
+                  <mainClass>com.google.cloud.spanner.pgadapter.Server</mainClass>
+                  <addClasspath>true</addClasspath>
+                  <classpathPrefix>lib/</classpathPrefix>
+                </manifest>
+              </archive>
             </configuration>
           </plugin>
         </plugins>
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java b/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
index 67e319d8..962f97a4 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
@@ -34,6 +34,27 @@ import org.json.simple.JSONObject;
 
 /** Metadata extractor for CLI. */
 public class OptionsMetadata {
+  public enum DdlTransactionMode {
+
+    // Disables all DDL batching and DDL statements in transactions. Only single DDL statements
+    // outside batches and transactions are allowed.
+    Single,
+
+    // Allows batches that contain only DDL statements outside explicit transactions.
+    // Does not allow mixing of DML, DQL and DDL statements in one batch.
+    // This is the default mode.
+    Batch,
+
+    // DDL statements automatically commit the active implicit transaction.
+    // Allows batches that contain mixes of DML, DQL and DDL statements.
+    // Fails if a DDL statement is executed in an explicit transaction.
+    AutocommitImplicitTransaction,
+
+    // DDL statements automatically commit the active explicit or implicit transaction.
+    // Allows batches that contain mixes of DML, DQL and DDL statements.
+    // Explicit transaction blocks that contain DDL statements will not be atomic.
+    AutocommitExplicitTransaction,
+  }
 
   private static final Logger logger = Logger.getLogger(OptionsMetadata.class.getName());
   private static final String DEFAULT_SERVER_VERSION = "1.0.0";
@@ -47,6 +68,7 @@ public class OptionsMetadata {
   private static final String OPTION_BINARY_FORMAT = "b";
   private static final String OPTION_AUTHENTICATE = "a";
   private static final String OPTION_PSQL_MODE = "q";
+  private static final String OPTION_DDL_TRANSACTION_MODE = "ddl";
   private static final String OPTION_JDBC_MODE = "jdbc";
   private static final String OPTION_COMMAND_METADATA_FILE = "j";
   private static final String OPTION_DISABLE_LOCALHOST_CHECK = "x";
@@ -69,6 +91,7 @@ public class OptionsMetadata {
   private final boolean binaryFormat;
   private final boolean authenticate;
   private final boolean requiresMatcher;
+  private final DdlTransactionMode ddlTransactionMode;
   private final boolean replaceJdbcMetadataQueries;
   private final boolean disableLocalhostCheck;
   private final JSONObject commandMetadataJSON;
@@ -92,6 +115,8 @@ public class OptionsMetadata {
     this.requiresMatcher =
         commandLine.hasOption(OPTION_PSQL_MODE)
             || commandLine.hasOption(OPTION_COMMAND_METADATA_FILE);
+    this.ddlTransactionMode =
+        parseDdlTransactionMode(commandLine.getOptionValue(OPTION_DDL_TRANSACTION_MODE));
     this.replaceJdbcMetadataQueries = commandLine.hasOption(OPTION_JDBC_MODE);
     this.commandMetadataJSON = buildCommandMetadataJSON(commandLine);
     this.propertyMap = parseProperties(commandLine.getOptionValue(OPTION_JDBC_PROPERTIES, ""));
@@ -117,6 +142,7 @@ public class OptionsMetadata {
     this.binaryFormat = forceBinary;
     this.authenticate = authenticate;
     this.requiresMatcher = requiresMatcher;
+    this.ddlTransactionMode = DdlTransactionMode.AutocommitImplicitTransaction;
     this.replaceJdbcMetadataQueries = replaceJdbcMetadataQueries;
     this.commandMetadataJSON = commandMetadata;
     this.propertyMap = new HashMap<>();
@@ -141,6 +167,19 @@ public class OptionsMetadata {
     return properties;
   }
 
+  private DdlTransactionMode parseDdlTransactionMode(String value) {
+    if (value == null) {
+      return DdlTransactionMode.Batch;
+    }
+    try {
+      return DdlTransactionMode.valueOf(value);
+    } catch (IllegalArgumentException e) {
+      // Catch and rethrow to give a better error message.
+      throw new IllegalArgumentException(
+          String.format("Invalid ddl-batching mode value specified: %s", value));
+    }
+  }
+
   /**
    * Takes the proxy port option result and parses it accordingly to fit port specs.
    *
@@ -307,6 +346,26 @@ public class OptionsMetadata {
             + " added performance cost. PSQL mode is implemented using predefined dynamic matchers"
             + " and as such cannot be used with the option -j. This mode should not be used for"
             + " production, and we do not guarantee its functionality beyond the basics.");
+    options.addOption(
+        OPTION_DDL_TRANSACTION_MODE,
+        "ddl-transaction-mode",
+        true,
+        String.format(
+            "Sets the way PGAdapter should handle DDL statements in implicit and explicit "
+                + "transactions. Cloud Spanner does not support DDL transactions. The possible modes "
+                + "are:\n"
+                + "%s: Only allows single DDL statements outside implicit and explicit transactions.\n"
+                + "%s: Allows batches that contain only DDL statements. "
+                + "Does not allow mixed batches of DDL and other statements, or DDL statements in transactions.\n"
+                + "%s: Allows mixed batches of DDL and other statements. "
+                + "Automatically commits the implicit transaction when a DDL statement is encountered in a batch. "
+                + "DDL statements in explicit transactions are not allowed.\n"
+                + "%s: Allows mixed batches of DDL and other statements. "
+                + "Automatically commits the current transaction when a DDL statement is encountered.",
+            DdlTransactionMode.Single,
+            DdlTransactionMode.Batch,
+            DdlTransactionMode.AutocommitImplicitTransaction,
+            DdlTransactionMode.AutocommitExplicitTransaction));
     options.addOption(
         OPTION_JDBC_MODE,
         "jdbc-mode",
@@ -372,6 +431,7 @@ public class OptionsMetadata {
 
     CommandLineParser parser = new DefaultParser();
     HelpFormatter help = new HelpFormatter();
+    help.setWidth(120);
     try {
       CommandLine commandLine = parser.parse(options, args);
       if (commandLine.hasOption(OPTION_HELP)) {
@@ -385,6 +445,10 @@ public class OptionsMetadata {
     }
   }
 
+  public DdlTransactionMode getDdlTransactionMode() {
+    return this.ddlTransactionMode;
+  }
+
   public boolean isBinaryFormat() {
     return this.binaryFormat;
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyStatement.java
index 59b87ce8..30836937 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyStatement.java
@@ -301,7 +301,7 @@ public class CopyStatement extends IntermediateStatement {
 
   @Override
   public void execute() {
-    this.executed = true;
+    this.executedCount++;
     try {
       parseCopyStatement();
       queryInformationSchema();
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePreparedStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePreparedStatement.java
index 3375fdcf..5e10f5fb 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePreparedStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePreparedStatement.java
@@ -77,10 +77,10 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
 
   @Override
   public void execute() {
-    this.executed = true;
     // If the portal has already been described, the statement has already been executed, and we
     // don't need to do that once more.
     if (getStatementResult(0) == null) {
+      this.executedCount++;
       try {
         if (!connection.isInTransaction()
             // TODO(230579451): Refactor to use ClientSideStatement information.
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
index 7b5fa312..f83931e4 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
@@ -103,7 +103,7 @@ public class QueryMessage extends ControlMessage {
    */
   public void handleQuery() throws Exception {
     // Skip unexecuted statements, as no response needs be returned
-    for (int index = 0; index < statement.getStatements().size(); index++) {
+    for (int index = 0; index < statement.getExecutedCount(); index++) {
       if (this.statement.hasException(index)) {
         new ErrorResponse(
                 this.outputStream, this.statement.getException(index), State.InternalError)
@@ -140,6 +140,8 @@ public class QueryMessage extends ControlMessage {
       if (connection.getStatus() == ConnectionStatus.TRANSACTION_ABORTED) {
         transactionStatus = Status.FAILED;
         // Actively rollback the aborted transaction but still block clients
+        // Clear any statement tags, as these are not allowed for rollbacks.
+        connection.getSpannerConnection().setStatementTag(null);
         connection.getSpannerConnection().rollback();
       } else {
         transactionStatus = Status.TRANSACTION;


Test patch:
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
index ded14f9f..fd1b1018 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
@@ -28,6 +28,7 @@ import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType
 import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.connection.PostgreSQLStatementParser;
 import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.connection.TransactionMode;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.metadata.DescribeMetadata;
@@ -46,6 +47,8 @@ import java.util.Objects;
  */
 @InternalApi
 public class IntermediateStatement {
+  public static final String TRANSACTION_ABORTED_ERROR =
+      "current transaction is aborted, commands ignored until end of transaction block";
   protected static final PostgreSQLStatementParser PARSER =
       (PostgreSQLStatementParser) AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
 
@@ -57,7 +60,7 @@ public class IntermediateStatement {
   protected final ParsedStatement parsedStatement;
   protected final ImmutableList<String> commands;
   protected final List<String> commandTags;
-  protected boolean executed = false;
+  protected int executedCount = 0;
   protected final Connection connection;
   protected long[] updateCounts;
   protected final ImmutableList<ParsedStatement> statements;
@@ -218,7 +221,11 @@ public class IntermediateStatement {
 
   /** @return True if this statement was executed, False otherwise. */
   public boolean isExecuted() {
-    return executed;
+    return executedCount > 0;
+  }
+
+  public int getExecutedCount() {
+    return this.executedCount;
   }
 
   /**
@@ -346,61 +353,180 @@ public class IntermediateStatement {
 
   /** Execute the SQL statement, storing metadata. */
   public void execute() {
-    executed = true;
     if (connection.isInTransaction()) {
       executionStatus = ExecutionStatus.EXPLICIT_TRANSACTION;
     } else {
-      executionStatus = ExecutionStatus.AUTOCOMMIT;
+      executionStatus = ExecutionStatus.IMPLICIT_TRANSACTION;
     }
-    int index = 0;
-    while (index < getStatementCount()) {
+    while (executedCount < getStatementCount()) {
+      maybeStartImplicitTransaction(executedCount);
+
       if (connectionHandler.getStatus() == ConnectionStatus.TRANSACTION_ABORTED
-          && !"ROLLBACK".equals(getCommand(index))
-          && !"COMMIT".equals(getCommand(index))) {
+          && !"ROLLBACK".equals(getCommand(executedCount))
+          && !"COMMIT".equals(getCommand(executedCount))) {
         handleExecutionException(
-            index,
+            executedCount,
             SpannerExceptionFactory.newSpannerException(
-                ErrorCode.INVALID_ARGUMENT,
-                "ERROR: current transaction is aborted, commands ignored until end of transaction block"));
-        index++;
+                ErrorCode.INVALID_ARGUMENT, TRANSACTION_ABORTED_ERROR));
+        executedCount++;
         continue;
       }
-      if (getStatementType(index) == StatementType.DDL
-          && executionStatus == ExecutionStatus.EXPLICIT_TRANSACTION) {
-        handleExecutionExceptionAndTransactionStatus(
-            index,
-            SpannerExceptionFactory.newSpannerException(
-                ErrorCode.INVALID_ARGUMENT,
-                "ERROR: DDL statements are only allowed outside explicit transactions."));
-        index++;
-        continue;
+      if (getStatementType(executedCount) == StatementType.DDL) {
+        try {
+          prepareExecuteDdl(executedCount);
+        } catch (SpannerException e) {
+          handleExecutionExceptionAndTransactionStatus(executedCount, e);
+          executedCount++;
+          break;
+        }
       }
       boolean canUseBatch = false;
-      if (index < (getStatementCount() - 1)) {
-        StatementType statementType = getStatementType(index);
-        StatementType nextStatementType = getStatementType(index + 1);
+      if (executedCount < (getStatementCount() - 1)) {
+        StatementType statementType = getStatementType(executedCount);
+        StatementType nextStatementType = getStatementType(executedCount + 1);
         canUseBatch = canBeBatchedTogether(statementType, nextStatementType);
       }
       if (canUseBatch) {
-        index += executeStatementsInBatch(index);
+        executedCount += executeStatementsInBatch(executedCount);
       } else {
-        executeSingleStatement(index);
-        index++;
+        executeSingleStatement(executedCount);
+        executedCount++;
+      }
+      // Return directly after any exception.
+      if (hasException(executedCount - 1)) {
+        break;
+      }
+    }
+    if (executionStatus == ExecutionStatus.IMPLICIT_TRANSACTION && connection.isInTransaction()) {
+      if (hasException(executedCount - 1)) {
+        connection.rollback();
+      } else {
+        connection.commit();
       }
     }
   }
 
+  /**
+   * Prepares the connection for executing a DDL statement. This can include committing the current
+   * transaction.
+   *
+   * <p>Executing the DDL statement may or may not be allowed depending on the state of the
+   * transaction and the selected {@link
+   * com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode}. The method
+   * will throw a {@link SpannerException} if executing a DDL statement at this point is not
+   * allowed.
+   */
+  protected void prepareExecuteDdl(int index) {
+    // Single statements are simpler to check, so we do that in a separate check.
+    if (statements.size() == 1) {
+      switch (options.getDdlTransactionMode()) {
+        case Single:
+        case Batch:
+        case AutocommitImplicitTransaction:
+          // Single DDL statements outside explicit transactions are always allowed. For a single
+          // statement, there can also not be an implicit transaction that needs to be committed.
+          if (connection.isInTransaction()) {
+            throw SpannerExceptionFactory.newSpannerException(
+                ErrorCode.FAILED_PRECONDITION,
+                "DDL statements are only allowed outside explicit transactions.");
+          }
+          break;
+        case AutocommitExplicitTransaction:
+          // DDL statements are allowed even in explicit transactions. Commit any transaction that
+          // might be active.
+          if (connection.isInTransaction()) {
+            connection.commit();
+          }
+      }
+      return;
+    }
+
+    // We are in a batch of statements.
+    switch (options.getDdlTransactionMode()) {
+      case Single:
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.FAILED_PRECONDITION,
+            "DDL statements are only allowed outside batches and transactions.");
+      case Batch:
+        if (connection.isInTransaction()
+            || statements.stream().anyMatch(statement -> !statement.isDdl())) {
+          throw SpannerExceptionFactory.newSpannerException(
+              ErrorCode.FAILED_PRECONDITION,
+              "DDL statements are not allowed in mixed batches or transactions.");
+        }
+        break;
+      case AutocommitImplicitTransaction:
+        if (connection.isInTransaction()
+            && executionStatus != ExecutionStatus.IMPLICIT_TRANSACTION) {
+          throw SpannerExceptionFactory.newSpannerException(
+              ErrorCode.FAILED_PRECONDITION,
+              "DDL statements are only allowed outside explicit transactions.");
+        }
+        if (connection.isInTransaction()) {
+          // Commit the implicit transaction before we continue.
+          connection.commit();
+        }
+        break;
+      case AutocommitExplicitTransaction:
+        // Commit any transaction that might be active and allow executing the statement.
+        // Switch the execution state to implicit transaction.
+        if (connection.isInTransaction()) {
+          connection.commit();
+          executionStatus = ExecutionStatus.IMPLICIT_TRANSACTION;
+        }
+    }
+  }
+
+  private void maybeStartImplicitTransaction(int index) {
+    if (connection.isInTransaction()) {
+      return;
+    }
+    if (executionStatus == ExecutionStatus.EXPLICIT_TRANSACTION) {
+      return;
+    }
+    // No need to start a transaction for the last statement.
+    if (index == statements.size() - 1) {
+      return;
+    }
+    // No need to start a transaction for DDL or client side statements.
+    if (statementTypes.get(index) == StatementType.DDL
+        || statementTypes.get(index) == StatementType.CLIENT_SIDE) {
+      return;
+    }
+    // If there are only DML statements left, those can be executed as an auto-commit dml batch.
+    if (hasOnlyDmlStatementsAfter(index)) {
+      return;
+    }
+    // We need to start an implicit transaction.
+    // Check if a read-only transaction suffices.
+    connection.beginTransaction();
+    if (!hasDmlStatementsAfter(index)) {
+      connection.setTransactionMode(TransactionMode.READ_ONLY_TRANSACTION);
+    }
+  }
+
+  private boolean hasDmlStatementsAfter(int index) {
+    return statementTypes.subList(index, statementTypes.size()).stream()
+        .anyMatch(type -> type == StatementType.UPDATE);
+  }
+
+  private boolean hasOnlyDmlStatementsAfter(int index) {
+    return statementTypes.subList(index, statementTypes.size()).stream()
+        .allMatch(type -> type == StatementType.UPDATE);
+  }
+
   private void executeSingleStatement(int index) {
     // Before executing the statement, handle specific statements that change the transaction status
     String command = getCommand(index);
     String statement = getStatement(index);
     if ("BEGIN".equals(command)) {
-      if (executionStatus == ExecutionStatus.EXPLICIT_TRANSACTION) {
-        // Executing a BEGIN statement when an explicit transaction is already active is a no-op
-        return;
-      } else {
-        executionStatus = ExecutionStatus.EXPLICIT_TRANSACTION;
+      // Executing a BEGIN statement when a transaction is already active will set the execution
+      // mode to EXPLICIT_TRANSACTION. The current transaction is not committed.
+      executionStatus = ExecutionStatus.EXPLICIT_TRANSACTION;
+      if (!connection.isInTransaction()) {
+        connection.execute(Statement.of(statement));
       }
+      return;
     }
     if ("COMMIT".equals(command) || "ROLLBACK".equals(command)) {
       if (connectionHandler.getStatus() == ConnectionStatus.TRANSACTION_ABORTED) {
@@ -409,11 +535,11 @@ public class IntermediateStatement {
         statement = "ROLLBACK";
         commandTags.set(index, "ROLLBACK");
       }
-      if (executionStatus == ExecutionStatus.AUTOCOMMIT) {
-        // Executing a COMMIT/ROLLBACK statement in autocommit mode is a no-op
+      // Executing ROLLBACK or COMMIT when there is no active transaction is a no-op, but will
+      // change the transaction mode to implicit.
+      executionStatus = ExecutionStatus.IMPLICIT_TRANSACTION;
+      if (!connection.isInTransaction()) {
         return;
-      } else {
-        executionStatus = ExecutionStatus.AUTOCOMMIT;
       }
     }
     // Ignore empty statements.
@@ -473,16 +599,7 @@ public class IntermediateStatement {
       long[] counts = e.getUpdateCounts();
       updateBatchResultCount(fromIndex, counts);
       handleExecutionExceptionAndTransactionStatus(fromIndex + counts.length, e);
-      // Skip remaining statements in DML/DDL batch
-      // TODO: May want to start off from the failure in order to be compatible with the
-      // behavior of PG.
-      for (int i = fromIndex + counts.length + 1; i < index; i++) {
-        handleExecutionException(
-            i,
-            SpannerExceptionFactory.newSpannerException(
-                ErrorCode.INTERNAL,
-                "ERROR: Statement is not executed due to the preceding error in DML/DDL batch."));
-      }
+      return counts.length + 1;
     }
     return index - fromIndex;
   }
@@ -529,6 +646,6 @@ public class IntermediateStatement {
 
   enum ExecutionStatus {
     EXPLICIT_TRANSACTION,
-    AUTOCOMMIT
+    IMPLICIT_TRANSACTION
   }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitExplicitTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitExplicitTest.java
new file mode 100644
index 00000000..bd030db5
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitExplicitTest.java
@@ -0,0 +1,165 @@
+// Copyright 2022 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.pgadapter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.spanner.v1.CommitRequest;
+import com.google.spanner.v1.ExecuteBatchDmlRequest;
+import com.google.spanner.v1.ExecuteSqlRequest;
+import com.google.spanner.v1.RollbackRequest;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.List;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class DdlTransactionModeAutocommitExplicitTest
+    extends DdlTransactionModeAutocommitImplicitTest {
+  @BeforeClass
+  public static void startMockSpannerAndPgAdapterServers() throws Exception {
+    doStartMockSpannerAndPgAdapterServers(
+        "d", Collections.singleton("-ddl=AutocommitExplicitTransaction"));
+  }
+
+  @Test
+  public void testMixedBatchWithExplicitTransaction() throws SQLException {
+    String sql =
+        "BEGIN; INSERT INTO FOO VALUES (1); CREATE TABLE foo (id bigint primary key); COMMIT;";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(1, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+
+        assertTrue(statement.execute("show transaction isolation level"));
+        assertNotNull(statement.getResultSet());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testDdlInExplicitTransaction() throws SQLException {
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Start an explicit transaction before executing batch
+        assertFalse(statement.execute("BEGIN; SELECT 1"));
+        assertEquals(0, statement.getUpdateCount());
+
+        assertTrue(statement.getMoreResults());
+        try (ResultSet resultSet = statement.getResultSet()) {
+          assertTrue(resultSet.next());
+          assertEquals(1L, resultSet.getLong(1));
+          assertFalse(resultSet.next());
+        }
+
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+
+        // Execute batch
+        assertTrue(statement.execute("SELECT 2; CREATE TABLE BAR (id bigint primary key);"));
+        assertNotNull(statement.getResultSet());
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+
+        // Commit the explicit transaction.
+        assertFalse(statement.execute("COMMIT"));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(2, requests.size());
+    assertEquals(SELECT1.getSql(), requests.get(0).getSql());
+    assertTrue(requests.get(0).getTransaction().hasBegin());
+    assertTrue(requests.get(0).getTransaction().getBegin().hasReadWrite());
+    // Verify that the explicit transaction is active while batch starts
+    assertEquals(SELECT2.getSql(), requests.get(1).getSql());
+    assertFalse(requests.get(1).getTransaction().hasBegin());
+    assertTrue(requests.get(1).getTransaction().hasId());
+    assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
+  }
+
+  @Test
+  public void testDdlInExplicitTransactionSwitchedToImplicit() throws SQLException {
+    String sql =
+        String.format(
+            "BEGIN; %s; CREATE TABLE foo (id bigint primary key); %s; %s; COMMIT;",
+            INSERT_STATEMENT, UPDATE_STATEMENT, INVALID_DML);
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // The invalid DML statement will cause the batch to fail. The statement is in an explicit
+        // transaction block, but that block also contains a DDL statement. That statement auto
+        // commits the explicit transaction, which means that we continue with an implicit
+        // transaction after that. That transaction is therefore rolled back when the batch fails.
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(exception.getMessage(), exception.getMessage().contains(EXCEPTION.getMessage()));
+
+        // The connection should not be in the aborted transaction state.
+        assertTrue(statement.execute("show transaction isolation level"));
+      }
+    }
+
+    // The first statement is executed as a single DML statement.
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(1, requests.size());
+    assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
+    assertTrue(requests.get(0).getTransaction().hasBegin());
+    assertTrue(requests.get(0).getTransaction().getBegin().hasReadWrite());
+    // The transaction is automatically committed by the CREATE TABLE statement.
+    assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
+
+    List<ExecuteBatchDmlRequest> batchDmlRequests =
+        mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class);
+    assertEquals(1, batchDmlRequests.size());
+    assertEquals(2, batchDmlRequests.get(0).getStatementsCount());
+    assertTrue(batchDmlRequests.get(0).getTransaction().hasBegin());
+    assertTrue(batchDmlRequests.get(0).getTransaction().getBegin().hasReadWrite());
+    // The transaction fails and is automatically rolled back.
+    assertEquals(1, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitImplicitTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitImplicitTest.java
new file mode 100644
index 00000000..ef2f127d
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitImplicitTest.java
@@ -0,0 +1,407 @@
+// Copyright 2022 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.pgadapter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
+import com.google.spanner.admin.database.v1.UpdateDatabaseDdlRequest;
+import com.google.spanner.v1.BeginTransactionRequest;
+import com.google.spanner.v1.CommitRequest;
+import com.google.spanner.v1.ExecuteBatchDmlRequest;
+import com.google.spanner.v1.ExecuteSqlRequest;
+import com.google.spanner.v1.RollbackRequest;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class DdlTransactionModeAutocommitImplicitTest extends DdlTransactionModeBatchTest {
+  @BeforeClass
+  public static void startMockSpannerAndPgAdapterServers() throws Exception {
+    doStartMockSpannerAndPgAdapterServers(
+        "d", Collections.singleton("-ddl=AutocommitImplicitTransaction"));
+  }
+
+  @Test
+  public void testMixedBatchDdlFirst() throws SQLException {
+    String sql =
+        "CREATE TABLE foo (id bigint primary key); INSERT INTO FOO VALUES (1); UPDATE FOO SET BAR=1 WHERE BAZ=2;";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(1, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(2, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteBatchDmlRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testMixedBatchDmlFirst() throws SQLException {
+    String sql =
+        "INSERT INTO FOO VALUES (1); UPDATE FOO SET BAR=1 WHERE BAZ=2; CREATE TABLE foo (id bigint primary key);";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(1, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(2, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteBatchDmlRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testMixedBatchDqlFirst() throws SQLException {
+    String sql = "SELECT 1; CREATE TABLE foo (id bigint primary key);";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertTrue(statement.execute(sql));
+        assertNotNull(statement.getResultSet());
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    assertEquals(1, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
+    assertTrue(
+        mockSpanner
+            .getRequestsOfType(BeginTransactionRequest.class)
+            .get(0)
+            .getOptions()
+            .hasReadOnly());
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testMixedBatchDqlFirstDmlAfter() throws SQLException {
+    String sql = "SELECT 1; CREATE TABLE foo (id bigint primary key); INSERT INTO FOO VALUES (1);";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertTrue(statement.execute(sql));
+        assertNotNull(statement.getResultSet());
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(1, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    // We get 2 commit requests, because the first implicit transaction is automatically committed
+    // by the DDL statement. The DML statement is also automatically committed at the end of the
+    // batch.
+    assertEquals(2, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testMixedBatchWithExplicitTransaction() throws SQLException {
+    String sql =
+        "BEGIN; INSERT INTO FOO VALUES (1); CREATE TABLE foo (id bigint primary key); COMMIT;";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are only allowed outside explicit transactions."));
+
+        exception =
+            assertThrows(
+                SQLException.class, () -> statement.execute("show transaction isolation level"));
+        assertTrue(
+            exception.getMessage(),
+            exception.getMessage().contains(IntermediateStatement.TRANSACTION_ABORTED_ERROR));
+      }
+    }
+
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testMixedBatch() throws SQLException {
+    String sql =
+        "CREATE TABLE foo (id bigint primary key); INSERT INTO FOO VALUES (1); UPDATE FOO SET BAR=1 WHERE BAZ=2;";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+
+        // getMoreResults() returns false as the next result is an update count.
+        assertFalse(statement.getMoreResults());
+        assertEquals(1, statement.getUpdateCount());
+
+        // getMoreResults() returns false as the next result is an update count.
+        assertFalse(statement.getMoreResults());
+        assertEquals(2, statement.getUpdateCount());
+
+        // getMoreResults() should now return false. We should also check getUpdateCount() as that
+        // method should return -1 to indicate that there is also no update count available.
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "CREATE TABLE foo (id bigint primary key)",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+
+    // Verify that the DML statements were batched together by PgAdapter.
+    List<ExecuteBatchDmlRequest> requests =
+        mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class);
+    assertEquals(1, requests.size());
+    ExecuteBatchDmlRequest request = requests.get(0);
+    assertEquals(2, request.getStatementsCount());
+    assertEquals(INSERT_STATEMENT.getSql(), request.getStatements(0).getSql());
+    assertEquals(UPDATE_STATEMENT.getSql(), request.getStatements(1).getSql());
+
+    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
+    assertEquals(1, commitRequests.size());
+  }
+
+  @Test
+  public void testErrorHandlingOfDdl() throws SQLException {
+    addDdlExceptionToSpannerAdmin();
+    String sql = String.format("%s; %s; %s", INSERT_STATEMENT, INVALID_DDL, UPDATE_STATEMENT);
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception.getMessage().contains("INVALID_ARGUMENT: Statement is invalid."));
+      }
+    }
+
+    // Verify that the execution is stopped after the first error.
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(1, requests.size());
+    assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
+
+    // The first insert statement is committed when the implicit transaction encounters a DDL
+    // statement. This deviates from PostgreSQL, as PostgreSQL would have included the DDL statement
+    // in the implicit transaction, and then rolled back the transaction when the DDL statement
+    // failed.
+    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
+    assertEquals(1, commitRequests.size());
+  }
+
+  @Test
+  public void testSelectAndDdlInBatch() throws SQLException {
+    String sql = "SELECT 1; SELECT 2; CREATE TABLE foo (id bigint primary key);";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertTrue(statement.execute(sql));
+
+        try (ResultSet resultSet = statement.getResultSet()) {
+          assertTrue(resultSet.next());
+          assertEquals(1L, resultSet.getLong(1));
+          assertFalse(resultSet.next());
+        }
+
+        assertTrue(statement.getMoreResults());
+        try (ResultSet resultSet = statement.getResultSet()) {
+          assertTrue(resultSet.next());
+          assertEquals(2L, resultSet.getLong(1));
+          assertFalse(resultSet.next());
+        }
+
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
+
+        // getMoreResults() should now return false. We should also check getUpdateCount() as that
+        // method should return -1 to indicate that there is also no update count available.
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "CREATE TABLE foo (id bigint primary key)",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+
+    // The first SELECT statement should start a read-only implicit transaction.
+    assertEquals(1, mockSpanner.getRequestsOfType(BeginTransactionRequest.class).size());
+    BeginTransactionRequest beginRequest =
+        mockSpanner.getRequestsOfType(BeginTransactionRequest.class).get(0);
+    assertTrue(beginRequest.getOptions().hasReadOnly());
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(2, requests.size());
+    assertEquals(SELECT1.getSql(), requests.get(0).getSql());
+    assertTrue(requests.get(0).getTransaction().hasId());
+    assertEquals(SELECT2.getSql(), requests.get(1).getSql());
+    assertTrue(requests.get(1).getTransaction().hasId());
+  }
+
+  @Test
+  public void testDdlAndSelectInBatch() throws SQLException {
+    String sql = "CREATE TABLE foo (id bigint primary key); SELECT 1; SELECT 2;";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+
+        assertTrue(statement.getMoreResults());
+        try (ResultSet resultSet = statement.getResultSet()) {
+          assertTrue(resultSet.next());
+          assertEquals(1L, resultSet.getLong(1));
+          assertFalse(resultSet.next());
+        }
+
+        assertTrue(statement.getMoreResults());
+        try (ResultSet resultSet = statement.getResultSet()) {
+          assertTrue(resultSet.next());
+          assertEquals(2L, resultSet.getLong(1));
+          assertFalse(resultSet.next());
+        }
+
+        // getMoreResults() should now return false. We should also check getUpdateCount() as that
+        // method should return -1 to indicate that there is also no update count available.
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "CREATE TABLE foo (id bigint primary key)",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(2, requests.size());
+    assertEquals(SELECT1.getSql(), requests.get(0).getSql());
+    // The first statement after the DDL statement should start an implicit transaction.
+    // That transaction will be a read-only transaction, and the begin of those are not inlined
+    // with the first statement.
+    assertTrue(requests.get(0).getTransaction().hasId());
+    assertEquals(SELECT2.getSql(), requests.get(1).getSql());
+    assertTrue(requests.get(1).getTransaction().hasId());
+    assertEquals(1, mockSpanner.getRequestsOfType(BeginTransactionRequest.class).size());
+    BeginTransactionRequest beginRequest =
+        mockSpanner.getRequestsOfType(BeginTransactionRequest.class).get(0);
+    assertTrue(beginRequest.getOptions().hasReadOnly());
+  }
+
+  @Test
+  public void testDdlInExplicitTransaction() throws SQLException {
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Start an explicit transaction before executing batch
+        assertFalse(statement.execute("BEGIN; SELECT 1"));
+        assertEquals(0, statement.getUpdateCount());
+
+        assertTrue(statement.getMoreResults());
+        try (ResultSet resultSet = statement.getResultSet()) {
+          assertTrue(resultSet.next());
+          assertEquals(1L, resultSet.getLong(1));
+          assertFalse(resultSet.next());
+        }
+
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+
+        // Execute batch
+        SQLException exception =
+            assertThrows(
+                SQLException.class,
+                () -> statement.execute("SELECT 2; CREATE TABLE BAR (id bigint primary key);"));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are only allowed outside explicit transactions."));
+      }
+    }
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(2, requests.size());
+    assertEquals(SELECT1.getSql(), requests.get(0).getSql());
+    assertTrue(requests.get(0).getTransaction().hasBegin());
+    // Verify that the explicit transaction is active while batch starts
+    assertEquals(SELECT2.getSql(), requests.get(1).getSql());
+    assertFalse(requests.get(1).getTransaction().hasBegin());
+    assertTrue(requests.get(1).getTransaction().hasId());
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeBatchTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeBatchTest.java
new file mode 100644
index 00000000..2cc92e23
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeBatchTest.java
@@ -0,0 +1,170 @@
+// Copyright 2022 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.pgadapter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
+import com.google.spanner.v1.BeginTransactionRequest;
+import com.google.spanner.v1.CommitRequest;
+import com.google.spanner.v1.RollbackRequest;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collections;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class DdlTransactionModeBatchTest extends DdlTransactionModeNoneTest {
+
+  @BeforeClass
+  public static void startMockSpannerAndPgAdapterServers() throws Exception {
+    doStartMockSpannerAndPgAdapterServers("d", Collections.singleton("-ddl=Batch"));
+  }
+
+  @Override
+  @Test
+  public void testDdlBatch() throws SQLException {
+    addDdlResponseToSpannerAdmin();
+    String sql =
+        "CREATE TABLE foo (id bigint primary key); CREATE TABLE bar (id bigint primary key);";
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+  }
+
+  @Test
+  public void testMixedBatchDdlFirst() throws SQLException {
+    String sql =
+        "CREATE TABLE foo (id bigint primary key); INSERT INTO FOO VALUES (1); UPDATE FOO SET BAR=1 WHERE BAZ=2;";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are not allowed in mixed batches or transactions."));
+      }
+    }
+  }
+
+  @Test
+  public void testMixedBatchDmlFirst() throws SQLException {
+    String sql =
+        "INSERT INTO FOO VALUES (1); UPDATE FOO SET BAR=1 WHERE BAZ=2; CREATE TABLE foo (id bigint primary key);";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are not allowed in mixed batches or transactions."));
+      }
+    }
+
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testMixedBatchDqlFirst() throws SQLException {
+    String sql = "SELECT 1; CREATE TABLE foo (id bigint primary key);";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are not allowed in mixed batches or transactions."));
+      }
+    }
+
+    assertEquals(1, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
+    assertTrue(
+        mockSpanner
+            .getRequestsOfType(BeginTransactionRequest.class)
+            .get(0)
+            .getOptions()
+            .hasReadOnly());
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testMixedBatchDqlFirstDmlAfter() throws SQLException {
+    String sql = "SELECT 1; CREATE TABLE foo (id bigint primary key); INSERT INTO FOO VALUES (1);";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are not allowed in mixed batches or transactions."));
+      }
+    }
+
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+
+  @Test
+  public void testMixedBatchWithExplicitTransaction() throws SQLException {
+    String sql =
+        "BEGIN; INSERT INTO FOO VALUES (1); CREATE TABLE foo (id bigint primary key); COMMIT;";
+    addDdlResponseToSpannerAdmin();
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are not allowed in mixed batches or transactions."));
+
+        exception =
+            assertThrows(
+                SQLException.class, () -> statement.execute("show transaction isolation level"));
+        assertTrue(
+            exception.getMessage(),
+            exception.getMessage().contains(IntermediateStatement.TRANSACTION_ABORTED_ERROR));
+      }
+    }
+
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(RollbackRequest.class));
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeNoneTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeNoneTest.java
new file mode 100644
index 00000000..ceb9a592
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeNoneTest.java
@@ -0,0 +1,99 @@
+// Copyright 2022 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.pgadapter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.spanner.admin.database.v1.UpdateDatabaseDdlRequest;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class DdlTransactionModeNoneTest extends AbstractMockServerTest {
+  @BeforeClass
+  public static void loadPgJdbcDriver() throws Exception {
+    // Make sure the PG JDBC driver is loaded.
+    Class.forName("org.postgresql.Driver");
+  }
+
+  @BeforeClass
+  public static void startMockSpannerAndPgAdapterServers() throws Exception {
+    doStartMockSpannerAndPgAdapterServers("d", Collections.singleton("-ddl=Single"));
+  }
+
+  /**
+   * Creates a JDBC connection string that instructs the PG JDBC driver to use the default simple
+   * mode for queries and DML statements. This makes the JDBC driver behave in (much) the same way
+   * as psql.
+   */
+  protected String createUrl() {
+    return String.format(
+        "jdbc:postgresql://localhost:%d/?preferQueryMode=simple", pgServer.getLocalPort());
+  }
+
+  @Test
+  public void testSingleDdlStatement() throws SQLException {
+    String sql = "CREATE TABLE foo (id bigint primary key)";
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "CREATE TABLE foo (id bigint primary key)",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testDdlBatch() throws SQLException {
+    String sql =
+        "CREATE TABLE foo (id bigint primary key); CREATE TABLE bar (id bigint primary key);";
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are only allowed outside batches and transactions."));
+      }
+    }
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java
index 48e58235..13e26b70 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java
@@ -14,6 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import static com.google.cloud.spanner.pgadapter.statements.IntermediateStatement.TRANSACTION_ABORTED_ERROR;
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
@@ -24,6 +25,7 @@ import static org.junit.Assert.assertTrue;
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
 import com.google.spanner.admin.database.v1.UpdateDatabaseDdlRequest;
+import com.google.spanner.v1.BeginTransactionRequest;
 import com.google.spanner.v1.CommitRequest;
 import com.google.spanner.v1.ExecuteBatchDmlRequest;
 import com.google.spanner.v1.ExecuteSqlRequest;
@@ -308,7 +310,7 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
             assertThrows(
                 SQLException.class,
                 () -> statement.execute(String.format("%s; %s;", INSERT_STATEMENT, INVALID_DML)));
-        assertEquals("ERROR: INVALID_ARGUMENT: Statement is invalid.", exception.getMessage());
+        assertTrue(exception.getMessage().contains("INVALID_ARGUMENT: Statement is invalid."));
       }
     }
 
@@ -337,14 +339,27 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
         SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
         assertThat(
             exception.getMessage(), containsString("INVALID_ARGUMENT: Statement is invalid."));
+
+        // Execute a client side statement to verify that the transaction is in the aborted state.
+        exception =
+            assertThrows(
+                SQLException.class, () -> statement.execute("show transaction isolation level"));
+        assertTrue(
+            exception.getMessage(), exception.getMessage().contains(TRANSACTION_ABORTED_ERROR));
+        // Rollback the transaction and verify that we can get out of the aborted state.
+        statement.execute("rollback work");
+        assertTrue(statement.execute("show transaction isolation level"));
       }
     }
 
-    // Verify that the DML statements were batched together by PgAdapter.
+    // The first DML statement is executed separately, as it is followed by a non-DML statement.
+    // The BEGIN statement will switch the batch to use an explicit transaction. The first DML
+    // statement will be included as part of that explicit transaction.
     List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
     assertEquals(1, requests.size());
     assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
 
+    // Verify that the DML statements were batched together by PgAdapter.
     List<ExecuteBatchDmlRequest> batchDmlRequests =
         mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class);
     assertEquals(1, requests.size());
@@ -353,11 +368,13 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
     assertEquals(UPDATE_STATEMENT.getSql(), request.getStatements(0).getSql());
     assertEquals(INVALID_DML.getSql(), request.getStatements(1).getSql());
 
+    // The explicit transaction is rolled back by PGAdapter. The
     List<RollbackRequest> rollbackRequests = mockSpanner.getRequestsOfType(RollbackRequest.class);
     assertEquals(1, rollbackRequests.size());
 
+    // The transaction should not be committed as it fails on the invalid DML statement.
     List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
-    assertEquals(1, commitRequests.size());
+    assertEquals(0, commitRequests.size());
   }
 
   @Test
@@ -385,13 +402,14 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
     assertEquals(UPDATE_STATEMENT.getSql(), request.getStatements(0).getSql());
     assertEquals(INVALID_DML.getSql(), request.getStatements(1).getSql());
 
-    // The aborted transaction should be rolled back by pgadapter
+    // The aborted transaction should be rolled back by PGAdapter.
     List<RollbackRequest> rollbackRequests = mockSpanner.getRequestsOfType(RollbackRequest.class);
     assertEquals(1, rollbackRequests.size());
 
-    // BEGIN statement commits the implicit transaction
+    // BEGIN statement converts the implicit transaction to an explicit transaction, but is
+    // otherwise a no-op. We should therefore not receive any commits.
     List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
-    assertEquals(1, commitRequests.size());
+    assertEquals(0, commitRequests.size());
   }
 
   @Test
@@ -409,9 +427,8 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
     }
 
     List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
-    assertEquals(2, requests.size());
+    assertEquals(1, requests.size());
     assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
-    assertEquals(SELECT1.getSql(), requests.get(1).getSql());
 
     List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
     assertEquals(1, commitRequests.size());
@@ -426,7 +443,7 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
   }
 
   @Test
-  public void testErrorHandlingInAutocommitMode() throws SQLException {
+  public void testErrorHandlingInImplicitTransaction() throws SQLException {
     String sql =
         String.format(
             "%s; %s; %s; %s; %s;", INSERT_STATEMENT, "COMMIT", SELECT1, INVALID_SELECT, SELECT2);
@@ -435,19 +452,32 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
         SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
         assertThat(
             exception.getMessage(), containsString("INVALID_ARGUMENT: Statement is invalid."));
+
+        // Verify that the transaction was rolled back and that the connection is usable.
+        assertTrue(statement.execute("show transaction isolation level"));
       }
     }
 
     List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
-    assertEquals(4, requests.size());
+    assertEquals(3, requests.size());
     assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
+    assertTrue(requests.get(0).getTransaction().hasBegin());
+    // The rest of the statements in the batch are all selects, so PGAdapter tries to use a
+    // read-only transaction.
+    assertEquals(1, mockSpanner.getRequestsOfType(BeginTransactionRequest.class).size());
+    assertTrue(
+        mockSpanner
+            .getRequestsOfType(BeginTransactionRequest.class)
+            .get(0)
+            .getOptions()
+            .hasReadOnly());
     assertEquals(SELECT1.getSql(), requests.get(1).getSql());
-    assertTrue(requests.get(1).getTransaction().hasSingleUse());
+    assertTrue(requests.get(1).getTransaction().hasId());
     assertEquals(INVALID_SELECT.getSql(), requests.get(2).getSql());
-    assertTrue(requests.get(2).getTransaction().hasSingleUse());
-    assertEquals(SELECT2.getSql(), requests.get(3).getSql());
-    assertTrue(requests.get(3).getTransaction().hasSingleUse());
+    assertTrue(requests.get(2).getTransaction().hasId());
 
+    // We get one commit for the read/write transaction. The read-only transaction is not committed
+    // or rolled back, as that is not necessary for read-only transactions.
     List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
     assertEquals(1, commitRequests.size());
 
@@ -486,135 +516,16 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
     List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
     assertEquals(2, requests.size());
     assertEquals(SELECT1.getSql(), requests.get(0).getSql());
-    assertTrue(requests.get(0).getTransaction().hasSingleUse());
+    assertTrue(requests.get(0).getTransaction().hasId());
     assertEquals(SELECT2.getSql(), requests.get(1).getSql());
-    assertTrue(requests.get(1).getTransaction().hasSingleUse());
-  }
-
-  @Test
-  public void testMixedBatch() throws SQLException {
-    String sql =
-        "CREATE TABLE foo (id bigint primary key); INSERT INTO FOO VALUES (1); UPDATE FOO SET BAR=1 WHERE BAZ=2;";
-    addDdlResponseToSpannerAdmin();
-    try (Connection connection = DriverManager.getConnection(createUrl())) {
-      try (Statement statement = connection.createStatement()) {
-        // Statement#execute(String) returns false if the result was either an update count or there
-        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
-        assertFalse(statement.execute(sql));
-        assertEquals(0, statement.getUpdateCount());
-
-        // getMoreResults() returns false as the next result is an update count.
-        assertFalse(statement.getMoreResults());
-        assertEquals(1, statement.getUpdateCount());
-
-        // getMoreResults() returns false as the next result is an update count.
-        assertFalse(statement.getMoreResults());
-        assertEquals(2, statement.getUpdateCount());
-
-        // getMoreResults() should now return false. We should also check getUpdateCount() as that
-        // method should return -1 to indicate that there is also no update count available.
-        assertFalse(statement.getMoreResults());
-        assertEquals(-1, statement.getUpdateCount());
-      }
-    }
-
-    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
-        mockDatabaseAdmin.getRequests().stream()
-            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
-            .map(UpdateDatabaseDdlRequest.class::cast)
-            .collect(Collectors.toList());
-    assertEquals(1, updateDatabaseDdlRequests.size());
-    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
-    assertEquals(
-        "CREATE TABLE foo (id bigint primary key)",
-        updateDatabaseDdlRequests.get(0).getStatements(0));
-
-    // Verify that the DML statements were batched together by PgAdapter.
-    List<ExecuteBatchDmlRequest> requests =
-        mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class);
-    assertEquals(1, requests.size());
-    ExecuteBatchDmlRequest request = requests.get(0);
-    assertEquals(2, request.getStatementsCount());
-    assertEquals(INSERT_STATEMENT.getSql(), request.getStatements(0).getSql());
-    assertEquals(UPDATE_STATEMENT.getSql(), request.getStatements(1).getSql());
-
-    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
-    assertEquals(1, commitRequests.size());
-  }
-
-  @Test
-  public void testErrorHandlingOfDdl() throws SQLException {
-    addDdlExceptionToSpannerAdmin();
-    String sql = String.format("%s; %s; %s", INSERT_STATEMENT, INVALID_DDL, UPDATE_STATEMENT);
-    try (Connection connection = DriverManager.getConnection(createUrl())) {
-      try (Statement statement = connection.createStatement()) {
-        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
-        assertThat(
-            exception.getMessage(), containsString("INVALID_ARGUMENT: Statement is invalid."));
-      }
-    }
-
-    // Verify that the execution does not halt and the last statement is executed.
-    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
-    assertEquals(2, requests.size());
-    assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
-    assertEquals(UPDATE_STATEMENT.getSql(), requests.get(1).getSql());
-
-    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
-    assertEquals(2, commitRequests.size());
-  }
-
-  @Test
-  public void testSelectInDdlBatch() throws SQLException {
-    String sql = "CREATE TABLE foo (id bigint primary key); SELECT 1; SELECT 2;";
-    addDdlResponseToSpannerAdmin();
-    try (Connection connection = DriverManager.getConnection(createUrl())) {
-      try (Statement statement = connection.createStatement()) {
-        // Statement#execute(String) returns false if the result was either an update count or there
-        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
-        assertFalse(statement.execute(sql));
-        assertEquals(0, statement.getUpdateCount());
-
-        assertTrue(statement.getMoreResults());
-        try (ResultSet resultSet = statement.getResultSet()) {
-          assertTrue(resultSet.next());
-          assertEquals(1L, resultSet.getLong(1));
-          assertFalse(resultSet.next());
-        }
-
-        assertTrue(statement.getMoreResults());
-        try (ResultSet resultSet = statement.getResultSet()) {
-          assertTrue(resultSet.next());
-          assertEquals(2L, resultSet.getLong(1));
-          assertFalse(resultSet.next());
-        }
-
-        // getMoreResults() should now return false. We should also check getUpdateCount() as that
-        // method should return -1 to indicate that there is also no update count available.
-        assertFalse(statement.getMoreResults());
-        assertEquals(-1, statement.getUpdateCount());
-      }
-    }
-
-    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
-        mockDatabaseAdmin.getRequests().stream()
-            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
-            .map(UpdateDatabaseDdlRequest.class::cast)
-            .collect(Collectors.toList());
-    assertEquals(1, updateDatabaseDdlRequests.size());
-    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
-    assertEquals(
-        "CREATE TABLE foo (id bigint primary key)",
-        updateDatabaseDdlRequests.get(0).getStatements(0));
+    assertTrue(requests.get(1).getTransaction().hasId());
 
-    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
-    assertEquals(2, requests.size());
-    assertEquals(SELECT1.getSql(), requests.get(0).getSql());
-    // Check that DDL statements close the implicit transaction
-    assertTrue(requests.get(0).getTransaction().hasSingleUse());
-    assertEquals(SELECT2.getSql(), requests.get(1).getSql());
-    // Check that DDL statements close the implicit transaction
-    assertTrue(requests.get(1).getTransaction().hasSingleUse());
+    // PGAdapter will use a read-only transaction if it sees that the implicit transaction will only
+    // read.
+    List<BeginTransactionRequest> beginRequests =
+        mockSpanner.getRequestsOfType(BeginTransactionRequest.class);
+    assertEquals(1, beginRequests.size());
+    assertTrue(beginRequests.get(0).getOptions().hasReadOnly());
   }
 
   @Test
@@ -782,11 +693,11 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
       List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
       assertEquals(3, requests.size());
       assertEquals(SELECT1.getSql(), requests.get(0).getSql());
-      assertTrue(requests.get(0).getTransaction().hasSingleUse());
+      assertTrue(requests.get(0).getTransaction().hasBegin());
       assertEquals(INSERT_STATEMENT.getSql(), requests.get(1).getSql());
-      assertTrue(requests.get(1).getTransaction().hasBegin());
+      assertTrue(requests.get(1).getTransaction().hasId());
       assertEquals(SELECT2.getSql(), requests.get(2).getSql());
-      assertTrue(requests.get(2).getTransaction().hasSingleUse());
+      assertTrue(requests.get(2).getTransaction().hasId());
 
       List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
       assertEquals(1, commitRequests.size());
@@ -833,44 +744,41 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
   }
 
   @Test
-  public void testDdlInExplicitTransaction() throws SQLException {
-    addDdlResponseToSpannerAdmin();
-
+  public void testBeginTransactionWithOptions() throws SQLException {
+    String sql = "BEGIN TRANSACTION; SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;";
     try (Connection connection = DriverManager.getConnection(createUrl())) {
-      try (java.sql.Statement statement = connection.createStatement()) {
-        // Start an explicit transaction before executing batch
-        assertFalse(statement.execute("BEGIN; SELECT 1"));
+      try (Statement statement = connection.createStatement()) {
+        // Execute returns false for statements that return no results.
+        assertFalse(statement.execute(sql));
         assertEquals(0, statement.getUpdateCount());
 
-        assertTrue(statement.getMoreResults());
-        try (ResultSet resultSet = statement.getResultSet()) {
-          assertTrue(resultSet.next());
-          assertEquals(1L, resultSet.getLong(1));
-          assertFalse(resultSet.next());
-        }
+        // getMoreResults also returns false for statements that return no results.
+        assertFalse(statement.getMoreResults());
+        assertEquals(0, statement.getUpdateCount());
 
+        // getMoreResults also returns false if there are no more results. If getUpdateCount returns
+        // -1, it is an indication that there are no more results.
         assertFalse(statement.getMoreResults());
         assertEquals(-1, statement.getUpdateCount());
 
-        // Execute batch
+        // Execute a garbled statement. This should abort the explicit transaction.
+        assertThrows(SQLException.class, () -> statement.execute("bork bork bork"));
         SQLException exception =
             assertThrows(
-                SQLException.class,
-                () -> statement.execute("SELECT 2; CREATE TABLE BAR (id bigint primary key);"));
-        assertThat(
-            exception.getMessage(),
-            containsString("DDL statements are only allowed outside explicit transactions."));
+                SQLException.class, () -> statement.execute("show transaction isolation level"));
+        assertTrue(
+            exception.getMessage(), exception.getMessage().contains(TRANSACTION_ABORTED_ERROR));
+
+        // Verify that we can get out of the aborted transaction by committing it. This will
+        // effectively rollback the transaction.
+        assertFalse(statement.execute("commit work and no chain"));
+        assertTrue(statement.execute("show transaction isolation level"));
       }
     }
-
-    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
-    assertEquals(2, requests.size());
-    assertEquals(SELECT1.getSql(), requests.get(0).getSql());
-    assertTrue(requests.get(0).getTransaction().hasBegin());
-    // Verify that the explicit transaction is active while batch starts
-    assertEquals(SELECT2.getSql(), requests.get(1).getSql());
-    assertFalse(requests.get(1).getTransaction().hasBegin());
-    assertTrue(requests.get(1).getTransaction().hasId());
+    // There should be no commit or rollback requests on the server, as PGAdapter never actually
+    // starts a transaction, as all statements are handled in the connection itself.
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
   }
 
   @Test
@@ -934,4 +842,20 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
     List<RollbackRequest> rollbackRequests = mockSpanner.getRequestsOfType(RollbackRequest.class);
     assertEquals(1, rollbackRequests.size());
   }
+
+  @Test
+  public void testStatementTagError() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute("begin"));
+        assertFalse(statement.execute(INSERT_STATEMENT.getSql()));
+        assertFalse(statement.execute("set spanner.statement_tag='foo'"));
+        // Execute an invalid statement.
+        assertThrows(SQLException.class, () -> statement.execute("set statement_timeout=2s"));
+        // Make sure that we actually received a Rollback statement. The rollback was initiated by
+        // PGAdapter when the transaction was aborted.
+        assertEquals(1, mockSpanner.countRequestsOfType(RollbackRequest.class));
+      }
+    }
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java
index 8247ecec..55d343d0 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java
@@ -38,6 +38,7 @@ import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePortalStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement;
@@ -189,6 +190,7 @@ public class StatementTest {
 
   @Test
   public void testBasicNoResultStatement() throws Exception {
+    when(options.getDdlTransactionMode()).thenReturn(DdlTransactionMode.Batch);
     when(statementResult.getResultType()).thenReturn(StatementResult.ResultType.NO_RESULT);
     when(connection.execute(Statement.of("CREATE TABLE users (name varchar(100) primary key)")))
         .thenReturn(statementResult);