feat(Spanner): Inline Begin Transaction

Spanner/src/Database.php

@@ -901,7 +901,14 @@ public function runTransaction(callable $operation, array $options = [])
 
         $attempt = 0;
         $startTransactionFn = function ($session, $options) use (&$attempt) {
-            if ($attempt > 0) {
+
+            // Initial attempt requires to set `begin` options (ILB).
+            if ($attempt === 0) {
+                // Partitioned DML does not support ILB.
+                if (!isset($options['transactionOptions']['partitionedDml'])) {
+                    $options['begin'] = $options['transactionOptions'];
+                }
+            } else {
                 $options['isRetry'] = true;
             }
 

Spanner/src/Operation.php

@@ -26,6 +26,7 @@
 use Google\Cloud\Spanner\Session\Session;
 use Google\Cloud\Spanner\V1\SpannerClient as GapicSpannerClient;
 use Google\Rpc\Code;
+use InvalidArgumentException;
 
 /**
  * Common interface for running operations against Cloud Spanner. This class is
@@ -184,9 +185,13 @@ public function commitWithResponse(Session $session, array $mutations, array $op
      * @param string $transactionId The transaction to roll back.
      * @param array $options [optional] Configuration Options.
      * @return void
+     * @throws InvalidArgumentException If the transaction is not yet initialized.
      */
     public function rollback(Session $session, $transactionId, array $options = [])
     {
+        if (empty($transactionId)) {
+            throw new InvalidArgumentException('Rollback failed: Transaction not initiated.');
+        }
         $this->connection->rollback([
             'transactionId' => $transactionId,
             'session' => $session->name(),
@@ -207,6 +212,9 @@ public function rollback(Session $session, $transactionId, array $options = [])
      *         [the upstream documentation](https://cloud.google.com/spanner/docs/reference/rest/v1/RequestOptions).
      *         Please note, if using the `priority` setting you may utilize the constants available
      *         on {@see \Google\Cloud\Spanner\V1\RequestOptions\Priority} to set a value.
+     *     @type array $transaction Transaction selector options.
+     *     @type array $transaction.begin The begin transaction options.
+     *           [Refer](https://cloud.google.com/spanner/docs/reference/rpc/google.spanner.v1#transactionoptions)
      * }
      * @return Result
      */
@@ -224,7 +232,17 @@ public function execute(Session $session, $sql, array $options = [])
 
         $context = $this->pluck('transactionContext', $options);
 
-        $call = function ($resumeToken = null) use ($session, $sql, $options) {
+        // Initially with begin, transactionId will be null.
+        // Once transaction is generated, even in the case of stream failure,
+        // transaction will be passed to this callable by the Result class.
+        $call = function ($resumeToken = null, $transaction = null) use (
+            $session,
+            $sql,
+            $options
+        ) {
+            if ($transaction && !empty($transaction->id())) {
+                $options['transaction'] = ['id' => $transaction->id()];
+            }
             if ($resumeToken) {
                 $options['resumeToken'] = $resumeToken;
             }
@@ -253,26 +271,33 @@ public function execute(Session $session, $sql, array $options = [])
      *         [the upstream documentation](https://cloud.google.com/spanner/docs/reference/rest/v1/RequestOptions).
      *         Please note, if using the `priority` setting you may utilize the constants available
      *         on {@see \Google\Cloud\Spanner\V1\RequestOptions\Priority} to set a value.
+     *     @type array $transaction Transaction selector options.
+     *     @type array $transaction.begin The begin transaction options.
+     *           [Refer](https://cloud.google.com/spanner/docs/reference/rpc/google.spanner.v1#transactionoptions)
      * }
      * @return int
-     * @throws \InvalidArgumentException If the SQL string isn't an update operation.
+     * @throws InvalidArgumentException If the SQL string isn't an update operation.
      */
     public function executeUpdate(
         Session $session,
         Transaction $transaction,
         $sql,
         array $options = []
     ) {
-        $res = $this->execute($session, $sql, [
-            'transactionId' => $transaction->id()
-        ] + $options);
+        if (!isset($options['transaction']['begin'])) {
+            $options['transaction'] = ['id' => $transaction->id()];
+        }
+        $res = $this->execute($session, $sql, $options);
+        if (empty($transaction->id()) && $res->transaction()) {
+            $transaction->setId($res->transaction()->id());
+        }
 
         // Iterate through the result to ensure we have query statistics available.
         iterator_to_array($res->rows());
 
         $stats = $res->stats();
         if (!$stats) {
-            throw new \InvalidArgumentException(
+            throw new InvalidArgumentException(
                 'Partitioned DML response missing stats, possible due to non-DML statement as input.'
             );
         }
@@ -316,9 +341,12 @@ public function executeUpdate(
      *         [the upstream documentation](https://cloud.google.com/spanner/docs/reference/rest/v1/RequestOptions).
      *         Please note, if using the `priority` setting you may utilize the constants available
      *         on {@see \Google\Cloud\Spanner\V1\RequestOptions\Priority} to set a value.
+     *     @type array $transaction Transaction selector options.
+     *     @type array $transaction.begin The begin transaction options.
+     *           [Refer](https://cloud.google.com/spanner/docs/reference/rpc/google.spanner.v1#transactionoptions)
      * }
      * @return BatchDmlResult
-     * @throws \InvalidArgumentException If any statement is missing the `sql` key.
+     * @throws InvalidArgumentException If any statement is missing the `sql` key.
      */
     public function executeUpdateBatch(
         Session $session,
@@ -329,7 +357,7 @@ public function executeUpdateBatch(
         $stmts = [];
         foreach ($statements as $statement) {
             if (!isset($statement['sql'])) {
-                throw new \InvalidArgumentException('Each statement must contain a SQL key.');
+                throw new InvalidArgumentException('Each statement must contain a SQL key.');
             }
 
             $parameters = $this->pluck('parameters', $statement, false) ?: [];
@@ -339,13 +367,23 @@ public function executeUpdateBatch(
             ] + $this->mapper->formatParamsForExecuteSql($parameters, $types);
         }
 
+        if (!isset($options['transaction']['begin'])) {
+            $options['transaction'] = ['id' => $transaction->id()];
+        }
+
         $res = $this->connection->executeBatchDml([
             'session' => $session->name(),
             'database' => $this->getDatabaseNameFromSession($session),
-            'transactionId' => $transaction->id(),
             'statements' => $stmts
         ] + $options);
 
+        if (empty($transaction->id())) {
+            // Get the transaction from array of ResultSets.
+            // ResultSet contains transaction in the metadata.
+            // @see https://cloud.google.com/spanner/docs/reference/rest/v1/ResultSet
+            $transaction->setId($res['resultSets'][0]['metadata']['transaction']['id'] ?? null);
+        }
+
         $errorStatement = null;
         if (isset($res['status']) && $res['status']['code'] !== Code::OK) {
             $errIndex = count($res['resultSets']);
@@ -373,11 +411,19 @@ public function executeUpdateBatch(
      *         [the upstream documentation](https://cloud.google.com/spanner/docs/reference/rest/v1/RequestOptions).
      *         Please note, if using the `priority` setting you may utilize the constants available
      *         on {@see \Google\Cloud\Spanner\V1\RequestOptions\Priority} to set a value.
+     *     @type array $transaction Transaction selector options.
+     *     @type array $transaction.begin The begin transaction options.
+     *           [Refer](https://cloud.google.com/spanner/docs/reference/rpc/google.spanner.v1#transactionoptions)
      * }
      * @return Result
      */
-    public function read(Session $session, $table, KeySet $keySet, array $columns, array $options = [])
-    {
+    public function read(
+        Session $session,
+        $table,
+        KeySet $keySet,
+        array $columns,
+        array $options = []
+    ) {
         $options += [
             'index' => null,
             'limit' => null,
@@ -387,7 +433,16 @@ public function read(Session $session, $table, KeySet $keySet, array $columns, a
 
         $context = $this->pluck('transactionContext', $options);
 
-        $call = function ($resumeToken = null) use ($table, $session, $columns, $keySet, $options) {
+        $call = function ($resumeToken = null, $transaction = null) use (
+            $table,
+            $session,
+            $columns,
+            $keySet,
+            $options
+        ) {
+            if ($transaction && !empty($transaction->id())) {
+                $options['transaction'] = ['id' => $transaction->id()];
+            }
             if ($resumeToken) {
                 $options['resumeToken'] = $resumeToken;
             }
@@ -420,6 +475,8 @@ public function read(Session $session, $table, KeySet $keySet, array $columns, a
      *     @type bool $isRetry If true, the resulting transaction will indicate
      *           that it is the result of a retry operation. **Defaults to**
      *           `false`.
+     *     @type array $begin The begin transaction options.
+     *           [Refer](https://cloud.google.com/spanner/docs/reference/rpc/google.spanner.v1#transactionoptions)
      * }
      * @return Transaction
      */
@@ -435,13 +492,23 @@ public function transaction(Session $session, array $options = [])
             $options['requestOptions']['transactionTag'] = $transactionTag;
         }
 
-        if (!$options['singleUse']) {
+        if (!$options['singleUse'] && (!isset($options['begin']) ||
+            isset($options['transactionOptions']['partitionedDml']))
+        ) {
             $res = $this->beginTransaction($session, $options);
         } else {
             $res = [];
         }
 
-        return $this->createTransaction($session, $res, ['tag' => $transactionTag]);
+        return $this->createTransaction(
+            $session,
+            $res,
+            [
+                'tag' => $transactionTag,
+                'isRetry' => $options['isRetry'],
+                'transactionOptions' => $options
+            ]
+        );
     }
 
     /**
@@ -450,6 +517,8 @@ public function transaction(Session $session, array $options = [])
      * @param Session $session The session the transaction belongs to.
      * @param array $res [optional] The createTransaction response.
      * @param array $options [optional] Options for the transaction object.
+     *     @type array $begin The begin transaction options.
+     *           [Refer](https://cloud.google.com/spanner/docs/reference/rpc/google.spanner.v1#transactionoptions)
      * @return Transaction
      */
     public function createTransaction(Session $session, array $res = [], array $options = [])
@@ -458,12 +527,20 @@ public function createTransaction(Session $session, array $res = [], array $opti
             'id' => null
         ];
         $options += [
-            'tag' => null
+            'tag' => null,
+            'transactionOptions' => null
         ];
 
         $options['isRetry'] = $options['isRetry'] ?? false;
 
-        return new Transaction($this, $session, $res['id'], $options['isRetry'], $options['tag']);
+        return new Transaction(
+            $this,
+            $session,
+            $res['id'],
+            $options['isRetry'],
+            $options['tag'],
+            $options['transactionOptions']
+        );
     }
 
     /**
@@ -741,6 +818,7 @@ private function partitionOptions(array $options)
      *
      * @param Session $session The session to start the snapshot in.
      * @param array $options [optional] Configuration options.
+     *
      * @return array
      */
     private function beginTransaction(Session $session, array $options = [])