Add support for AutoCommit(false) (#3429)

Add support for complex transaction operations in the gRPC server:
- Add state for a connection (so that it can handle a transactional
state) to server side (JdbcService)
- Add stateful connection from client to server (bidi connection stream)

Resolves #3453

---------

Co-authored-by: ohad <[email protected]>

Author: jjezra

fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/AbstractEmbeddedStatement.java

@@ -109,15 +109,15 @@ public boolean executeInternal(String sql) throws SQLException, RelationalExcept
                         resultSetRetrieved = true;
                         //ddl statements are updates that don't return results, so they get 0 for row count
                         currentRowCount = 0;
-                        if (conn.canCommit()) {
+                        if (conn.shouldCommit()) {
                             conn.commitInternal();
                         }
                         return false;
                     }
                 }
             } catch (RelationalException | SQLException | RuntimeException ex) {
                 try {
-                    if (conn.inActiveTransaction() && conn.canCommit()) {
+                    if (conn.inActiveTransaction() && conn.shouldCommit()) {
                         conn.rollbackInternal();
                     }
                 } catch (SQLException e) {
@@ -170,7 +170,7 @@ public void close() throws SQLException {
             closeOpenResultSets();
             closed = true;
         } catch (RuntimeException ex) {
-            if (conn.canCommit()) {
+            if (conn.shouldCommit()) {
                 try {
                     conn.rollbackInternal();
                 } catch (SQLException e) {
@@ -209,12 +209,12 @@ private int countUpdates(@Nonnull ResultSet resultSet) throws SQLException {
             while (resultSet.next()) {
                 count++;
             }
-            if (conn.canCommit()) {
+            if (conn.shouldCommit()) {
                 conn.commitInternal();
             }
             return count;
         } catch (SQLException | RuntimeException ex) {
-            if (conn.canCommit()) {
+            if (conn.shouldCommit()) {
                 conn.rollbackInternal();
             }
             throw ExceptionUtil.toRelationalException(ex).toSqlException();

fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/EmbeddedRelationalConnection.java

@@ -84,7 +84,7 @@
  *     to be used to execute all statements and procedure. For consumer perspective, this is equivalent to {@code autoCommit}
  *     being set to {@code true} since the {@link Connection#commit()} and {@link Connection#rollback()} wont be applicable.
  *     However, all statements run within the external transaction. For internal usage, the consumer should check
- *     {@link EmbeddedRelationalConnection#canCommit()} to see if they are allowed to manage a transaction.</li>
+ *     {@link EmbeddedRelationalConnection#shouldCommit()} to see if they are allowed to manage a transaction.</li>
  * </ul>
  */
 @API(API.Status.EXPERIMENTAL)
@@ -185,7 +185,7 @@ public boolean getAutoCommit() throws SQLException {
      * @return {@code true} if the transaction can be committed by internal stakeholders, else {@code false}.
      * @throws SQLException if the connection is closed.
      */
-    boolean canCommit() throws SQLException {
+    boolean shouldCommit() throws SQLException {
         checkOpen();
         return !usingAnExternalTransaction && this.autoCommit;
     }
@@ -492,7 +492,7 @@ public AbstractDatabase getRecordLayerDatabase() {
      */
     boolean ensureTransactionActive() throws RelationalException, SQLException {
         if (inActiveTransaction()) {
-            if (canCommit()) {
+            if (shouldCommit()) {
                 // canCommit() = true means that there is no external transaction and autoCommit is enabled, meaning
                 // that the internal stakeholders can manage the transactions when required to.
                 // As this implies that autoCommit is enabled, if there is an existing transaction in such a case, that

fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/EmbeddedRelationalStatement.java

@@ -335,7 +335,7 @@ private <T> T ensureTransaction(Supplier<T> operation) throws SQLException {
                 } catch (SQLException sqle) {
                     exception.addSuppressed(sqle);
                 }
-            } else if (conn.canCommit()) {
+            } else if (conn.shouldCommit()) {
                 conn.commitInternal();
             }
         }

fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/RecordLayerResultSet.java

@@ -94,7 +94,7 @@ public void close() throws SQLException {
         } catch (RelationalException e) {
             throw e.toSqlException();
         }
-        if (connection != null && connection.canCommit() && connection.inActiveTransaction()) {
+        if (connection != null && connection.shouldCommit() && connection.inActiveTransaction()) {
             connection.commitInternal();
         }
         this.closed = true;

fdb-relational-grpc/src/main/java/com/apple/foundationdb/relational/jdbc/grpc/GrpcSQLExceptionUtil.java

@@ -125,6 +125,11 @@ public static Status create(SQLException sqlException) {
     @Nullable
     public static SQLException map(StatusRuntimeException statusRuntimeException) {
         Status status = StatusProto.fromThrowable(statusRuntimeException);
+        return map(status);
+    }
+
+    @Nullable
+    public static SQLException map(final Status status) {
         if (status.getCode() != CODE_ON_SQLEXCEPTION) {
             return null;
         }

fdb-relational-grpc/src/main/proto/grpc/relational/jdbc/v1/jdbc.proto

@@ -29,10 +29,12 @@ option java_outer_classname = "RelationalJDBCService";
 
 import "grpc/relational/jdbc/v1/result_set.proto";
 import "grpc/relational/jdbc/v1/column.proto";
+import "google/protobuf/any.proto";
 
 service JDBCService {
   rpc getMetaData(DatabaseMetaDataRequest) returns (DatabaseMetaDataResponse) {}
   rpc execute(StatementRequest) returns (StatementResponse) {}
+  rpc handleAutoCommitOff(stream TransactionalRequest) returns (stream TransactionalResponse) {}
   rpc update(StatementRequest) returns (StatementResponse) {}
   rpc insert(InsertRequest) returns (InsertResponse) {}
   rpc get(GetRequest) returns (GetResponse) {}
@@ -112,6 +114,26 @@ message Options {
   optional int32 maxRows = 1;
 }
 
+// Request that is part of a transactional (stateful) exchange
+message TransactionalRequest {
+  oneof TransactionalMessage {
+      StatementRequest executeRequest = 1;
+      InsertRequest insertRequest = 2;
+      CommitRequest commitRequest = 3;
+      RollbackRequest rollbackRequest = 4;
+  }
+}
+
+message TransactionalResponse {
+  oneof TransactionalMessage {
+      StatementResponse executeResponse = 1;
+      InsertResponse insertResponse = 2;
+      CommitResponse commitResponse = 3;
+      RollbackResponse rollbackResponse = 4;
+      google.protobuf.Any errorResponse = 5;
+  }
+}
+
 message StatementRequest {
   optional string sql = 1;
   optional string database = 2;
@@ -128,6 +150,18 @@ message StatementResponse {
   grpc.relational.jdbc.v1.ResultSet result_set = 2;
 }
 
+message CommitRequest {
+}
+
+message CommitResponse {
+}
+
+message RollbackRequest {
+}
+
+message RollbackResponse {
+}
+
 message DatabaseMetaDataRequest {}
 message DatabaseMetaDataResponse {
   string url = 1;