feat: support transaction timeouts (GoogleCloudPlatform#3831)

PGAdapter now supports setting a transaction timeout for read/write transactions.

Execute the following SQL statement, or set the option in the connection string, to
enable transaction timeouts:

```sql
-- Timeout in milliseconds
set spanner.transaction_timeout=10000;
begin;
insert into my_table (id, value) values (1, 'One');
insert into my_table (id, value) values (2, 'Two');
commit;
```

Author: olavloite

docs/faq.md

@@ -175,3 +175,22 @@ from large_table
 order by my_col;
 ```
 
+
+## How can I set a transaction timeout?
+You can set the transaction timeout that a connection should use for read/write transactions by
+executing a `set spanner.transaction_timeout=<timeout>` statement. The timeout is specified in
+milliseconds.
+
+Example:
+
+```sql
+-- Use a 10 second (10,000 milliseconds) transaction timeout.
+set spanner.transaction_timeout=10000;
+begin;
+-- Execute transaction statements. If the total time needed for these statements exceed
+-- the transaction timeout, then the transaction will fail.
+insert into my_table (id, value) values (1, 'One');
+...
+insert into my_table (id, value) values (10000, 'Ten thousand');
+commit;
+```

src/main/java/com/google/cloud/spanner/pgadapter/error/PGExceptionFactory.java

@@ -85,6 +85,14 @@ public static PGException newQueryCancelledException() {
     return newPGException("Query cancelled", SQLState.QueryCanceled);
   }
 
+  private static PGException newQueryCancelledDueToTimeoutException(SpannerException exception) {
+    return PGException.newBuilder("canceling statement due to statement timeout")
+        .setSQLState(SQLState.QueryCanceled)
+        .setSeverity(Severity.ERROR)
+        .setCause(exception)
+        .build();
+  }
+
   /**
    * Creates a new exception that indicates that the current transaction is in the aborted state.
    */
@@ -145,6 +153,8 @@ public static PGException toPGException(SpannerException spannerException) {
           .setHints(
               "Execute 'set spanner.support_drop_cascade=true' to enable 'drop {table|schema} cascade' statements.")
           .build();
+    } else if (spannerException.getErrorCode() == ErrorCode.DEADLINE_EXCEEDED) {
+      return newQueryCancelledDueToTimeoutException(spannerException);
     }
     return newPGException(extractMessage(spannerException));
   }

src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponse.java

@@ -37,6 +37,7 @@ public class ErrorResponse extends WireOutput {
   private static final byte MESSAGE_FLAG = 'M';
   private static final byte SEVERITY_FLAG = 'S';
   private static final byte HINT_FLAG = 'H';
+  private static final byte DETAIL_FLAG = 'D';
   private static final byte NULL_TERMINATOR = 0;
 
   private final PGException pgException;
@@ -63,6 +64,10 @@ static int calculateLength(PGException pgException, WellKnownClient client) {
             + convertMessageToWireProtocol(pgException).length
             + NULL_TERMINATOR_LENGTH
             + NULL_TERMINATOR_LENGTH;
+    byte[] detail = convertDetailToWireProtocol(pgException);
+    if (detail.length > 0) {
+      length += FIELD_IDENTIFIER_LENGTH + detail.length + NULL_TERMINATOR_LENGTH;
+    }
     byte[] hints = convertHintsToWireProtocol(pgException, client);
     if (hints.length > 0) {
       length += FIELD_IDENTIFIER_LENGTH + hints.length + NULL_TERMINATOR_LENGTH;
@@ -82,6 +87,19 @@ static byte[] convertMessageToWireProtocol(PGException pgException) {
     return pgException.getMessage().getBytes(StandardCharsets.UTF_8);
   }
 
+  static byte[] convertDetailToWireProtocol(PGException pgException) {
+    if (pgException.getCause() != null
+        && !Strings.isNullOrEmpty(pgException.getCause().getMessage())) {
+      if (pgException.getCause().getMessage().equals(pgException.getMessage())) {
+        // Do not repeat the same error message in both the 'message' and 'detail' field if they
+        // are equal.
+        return EMPTY_BYTE_ARRAY;
+      }
+      return pgException.getCause().getMessage().getBytes(StandardCharsets.UTF_8);
+    }
+    return EMPTY_BYTE_ARRAY;
+  }
+
   static byte[] convertHintsToWireProtocol(PGException pgException, WellKnownClient client) {
     if (Strings.isNullOrEmpty(pgException.getHints())
         && client.getErrorHints(pgException).isEmpty()) {
@@ -112,6 +130,12 @@ protected void sendPayload() throws IOException {
     this.outputStream.writeByte(MESSAGE_FLAG);
     this.outputStream.write(convertMessageToWireProtocol(pgException));
     this.outputStream.writeByte(NULL_TERMINATOR);
+    byte[] detail = convertDetailToWireProtocol(pgException);
+    if (detail.length > 0) {
+      this.outputStream.writeByte(DETAIL_FLAG);
+      this.outputStream.write(detail);
+      this.outputStream.writeByte(NULL_TERMINATOR);
+    }
     byte[] hints = convertHintsToWireProtocol(pgException, client);
     if (hints.length > 0) {
       this.outputStream.writeByte(HINT_FLAG);

src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java

@@ -6237,8 +6237,6 @@ public void testIsolationLevelForOneTransaction() throws SQLException {
           ExecuteSqlRequest request = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(0);
           assertTrue(request.hasTransaction());
           assertTrue(request.getTransaction().hasBegin());
-          // TODO: Change this when https://github.com/googleapis/java-spanner/pull/3718 has been
-          // released and merged into PGAdapter.
           assertEquals(
               translateIsolationLevel(isolation),
               request.getTransaction().getBegin().getIsolationLevel());
@@ -6249,6 +6247,23 @@ public void testIsolationLevelForOneTransaction() throws SQLException {
     }
   }
 
+  @Test
+  public void testTransactionTimeout() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.setAutoCommit(false);
+      connection.createStatement().execute("set spanner.transaction_timeout='1ns'");
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class,
+              () -> connection.createStatement().execute(UPDATE_STATEMENT.getSql()));
+      assertTrue(
+          exception.getMessage(),
+          exception.getMessage().startsWith("ERROR: canceling statement due to statement timeout"));
+      assertTrue(exception.getMessage(), exception.getMessage().contains("DEADLINE_EXCEEDED"));
+      assertEquals(SQLState.QueryCanceled.toString(), exception.getSQLState());
+    }
+  }
+
   private static IsolationLevel translateIsolationLevel(int jdbcLevel) {
     switch (jdbcLevel) {
       case Connection.TRANSACTION_SERIALIZABLE: