feat(Spanner): Inline Begin Transaction

Spanner/src/Database.php

@@ -903,6 +903,9 @@ public function runTransaction(callable $operation, array $options = [])
         $startTransactionFn = function ($session, $options) use (&$attempt) {
             if ($attempt > 0) {
                 $options['isRetry'] = true;
+            } elseif (!isset($options['transactionOptions']['partitionedDml'])) {
+                // Make the begin options
+                $options['begin'] = $options['transactionOptions'];
             }
 
             $transaction = $this->operation->transaction($session, $options);
@@ -1784,7 +1787,7 @@ public function executePartitionedUpdate($statement, array $options = [])
         try {
             return $this->operation->executeUpdate($session, $transaction, $statement, [
                 'statsItem' => 'rowCountLowerBound'
-            ] + $options);
+            ] + $options, $transaction->getIdReference());
         } finally {
             $session->setExpiration();
         }

Spanner/src/Operation.php

@@ -48,6 +48,7 @@ class Operation
     const OP_INSERT_OR_UPDATE = 'insertOrUpdate';
     const OP_REPLACE = 'replace';
     const OP_DELETE = 'delete';
+    const DEFAULT_RETRIES = 3;
 
     /**
      * @var ConnectionInterface
@@ -187,6 +188,11 @@ public function commitWithResponse(Session $session, array $mutations, array $op
      */
     public function rollback(Session $session, $transactionId, array $options = [])
     {
+        // TODO: Check with team on this and decide
+        if (is_array($transactionId) and isset($transactionId['begin'])) {
+            return;
+            // throw new \RuntimeException('Streaming calls must invoke rows() function');
+        }
         $this->connection->rollback([
             'transactionId' => $transactionId,
             'session' => $session->name(),
@@ -210,7 +216,7 @@ public function rollback(Session $session, $transactionId, array $options = [])
      * }
      * @return Result
      */
-    public function execute(Session $session, $sql, array $options = [])
+    public function execute(Session $session, $sql, array $options = [], &$transactionId = null)
     {
         $options += [
             'parameters' => [],
@@ -224,7 +230,13 @@ 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 generated it will get populated from Result->rows()
+        // Incase of stream failure, the transactionId is preserved here to reuse.
+        $call = function ($resumeToken = null) use ($session, $sql, $options, &$transactionId) {
+            if ( !is_null($transactionId) and !is_array($transactionId) ) {
+                $options['transaction'] = ['id' => $transactionId];
+            }
             if ($resumeToken) {
                 $options['resumeToken'] = $resumeToken;
             }
@@ -236,7 +248,15 @@ public function execute(Session $session, $sql, array $options = [])
             ] + $options);
         };
 
-        return new Result($this, $session, $call, $context, $this->mapper);
+        return new Result(
+            $this,
+            $session,
+            $call,
+            $context,
+            $this->mapper,
+            self::DEFAULT_RETRIES,
+            $transactionId
+        );
     }
 
     /**
@@ -261,11 +281,13 @@ public function executeUpdate(
         Session $session,
         Transaction $transaction,
         $sql,
-        array $options = []
+        array $options = [],
+        &$transactionId = null
     ) {
-        $res = $this->execute($session, $sql, [
-            'transactionId' => $transaction->id()
-        ] + $options);
+        if ( !is_null($transactionId) and !is_array($transactionId) ) {
+            $options += ['transactionId' => $transactionId];
+        }
+        $res = $this->execute($session, $sql, $options, $transactionId);
 
         // Iterate through the result to ensure we have query statistics available.
         iterator_to_array($res->rows());
@@ -324,7 +346,8 @@ public function executeUpdateBatch(
         Session $session,
         Transaction $transaction,
         array $statements,
-        array $options = []
+        array $options = [],
+        &$transactionId = null
     ) {
         $stmts = [];
         foreach ($statements as $statement) {
@@ -339,13 +362,23 @@ public function executeUpdateBatch(
             ] + $this->mapper->formatParamsForExecuteSql($parameters, $types);
         }
 
+        if ( !is_null($transactionId) and !is_array($transactionId) ) {
+            $options += ['transactionId' => $transactionId];
+        }
+
         $res = $this->connection->executeBatchDml([
             'session' => $session->name(),
             'database' => $this->getDatabaseNameFromSession($session),
-            'transactionId' => $transaction->id(),
             'statements' => $stmts
         ] + $options);
 
+        if (
+            isset($res['resultSets'][0]['metadata']['transaction']['id']) and
+            $res['resultSets'][0]['metadata']['transaction']['id']
+        ) {
+            $transactionId = $res['resultSets'][0]['metadata']['transaction']['id'];
+        }
+
         $errorStatement = null;
         if (isset($res['status']) && $res['status']['code'] !== Code::OK) {
             $errIndex = count($res['resultSets']);
@@ -376,8 +409,14 @@ public function executeUpdateBatch(
      * }
      * @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 = [],
+        &$transactionId = null
+    ) {
         $options += [
             'index' => null,
             'limit' => null,
@@ -401,7 +440,15 @@ public function read(Session $session, $table, KeySet $keySet, array $columns, a
             ] + $options);
         };
 
-        return new Result($this, $session, $call, $context, $this->mapper);
+        return new Result(
+            $this,
+            $session,
+            $call,
+            $context,
+            $this->mapper,
+            self::DEFAULT_RETRIES,
+            $transactionId
+        );
     }
 
     /**
@@ -436,12 +483,27 @@ public function transaction(Session $session, array $options = [])
         }
 
         if (!$options['singleUse']) {
-            $res = $this->beginTransaction($session, $options);
+            if (
+                isset($options['begin']) and
+                !isset($options['transactionOptions']['partitionedDml'])
+            ) {
+                $res = ['id' => ['begin' => $options['begin']]];
+            } else {
+                $res = $this->beginTransaction($session, $options);
+            }
         } else {
             $res = [];
         }
 
-        return $this->createTransaction($session, $res, ['tag' => $transactionTag]);
+        if (isset($options['fromTransaction'])) {
+            return $res;
+        }
+
+        return $this->createTransaction(
+            $session,
+            $res,
+            ['tag' => $transactionTag, 'isRetry' => $options['isRetry'], 'options' => $options]
+        );
     }
 
     /**
@@ -458,12 +520,20 @@ public function createTransaction(Session $session, array $res = [], array $opti
             'id' => null
         ];
         $options += [
-            'tag' => null
+            'tag' => null,
+            'options' => 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['options']
+        );
     }
 
     /**

Spanner/src/Result.php

@@ -121,6 +121,11 @@ class Result implements \IteratorAggregate
      */
     private $call;
 
+    /**
+     * @var string|array
+     */
+    private $transactionId;
+
     /**
      * @param Operation $operation Runs operations against Google Cloud Spanner.
      * @param Session $session The session used for any operations executed.
@@ -136,14 +141,16 @@ public function __construct(
         callable $call,
         $transactionContext,
         ValueMapper $mapper,
-        $retries = 3
+        $retries = 3,
+        &$transactionId = null
     ) {
         $this->operation = $operation;
         $this->session = $session;
         $this->call = $call;
         $this->transactionContext = $transactionContext;
         $this->mapper = $mapper;
         $this->retries = $retries;
+        $this->transactionId = &$transactionId;
     }
 
     /**
@@ -467,6 +474,7 @@ private function setResultData(array $result, $format)
         }
 
         if (isset($result['metadata']['transaction']['id']) && $result['metadata']['transaction']['id']) {
+            $this->transactionId = $result['metadata']['transaction']['id'];
             if ($this->transactionContext === SessionPoolInterface::CONTEXT_READ) {
                 $this->snapshot = $this->operation->createSnapshot(
                     $this->session,

Spanner/src/Transaction.php

@@ -81,6 +81,11 @@ class Transaction implements TransactionalReadInterface
      */
     private $isRetry = false;
 
+    /**
+     * @var array
+     */
+    private $options = [];
+
     /**
      * @param Operation $operation The Operation instance.
      * @param Session $session The session to use for spanner interactions.
@@ -96,7 +101,8 @@ public function __construct(
         Session $session,
         $transactionId = null,
         $isRetry = false,
-        $tag = null
+        $tag = null,
+        $options = null
     ) {
         $this->operation = $operation;
         $this->session = $session;
@@ -115,6 +121,7 @@ public function __construct(
         $this->tag = $tag;
 
         $this->context = SessionPoolInterface::CONTEXT_READWRITE;
+        $this->options = $options;
     }
 
     /**
@@ -440,8 +447,18 @@ public function executeUpdate($sql, array $options = [])
         $options['seqno'] = $this->seqno;
         $this->seqno++;
 
+        $options['transactionType'] = $this->context;
+        if (is_array($this->transactionId) and isset($this->transactionId['begin'])) {
+            $options['begin'] = $this->transactionId['begin'];
+        } else {
+            $options['transactionId'] = $this->transactionId;
+        }
+        $selector = $this->transactionSelector($options);
+
+        $options['transaction'] = $selector[0];
+
         return $this->operation
-            ->executeUpdate($this->session, $this, $sql, $options);
+            ->executeUpdate($this->session, $this, $sql, $options, $this->transactionId);
     }
 
     /**
@@ -538,9 +555,25 @@ public function executeUpdateBatch(array $statements, array $options = [])
         }
         $options['seqno'] = $this->seqno;
         $this->seqno++;
+
+        $options['transactionType'] = $this->context;
+        if (is_array($this->transactionId) and isset($this->transactionId['begin'])) {
+            $options['begin'] = $this->transactionId['begin'];
+        } else {
+            $options['transactionId'] = $this->transactionId;
+        }
+        $selector = $this->transactionSelector($options);
+
+        $options['transaction'] = $selector[0];
 
         return $this->operation
-            ->executeUpdateBatch($this->session, $this, $statements, $options);
+            ->executeUpdateBatch(
+                $this->session,
+                $this,
+                $statements,
+                $options,
+                $this->transactionId
+            );
     }
 
     /**
@@ -615,6 +648,19 @@ public function commit(array $options = [])
             throw new \BadMethodCallException('The transaction cannot be committed because it is not active');
         }
 
+        if (is_array($this->transactionId) and isset($this->transactionId['begin'])) {
+            unset($this->options['begin']);
+            $res = $this->operation->transaction($this->session, $this->options+[
+                'fromTransaction' => true
+            ]);
+            if (is_array($res) and isset($res['id'])) {
+                $this->transactionId = $res['id'];
+            } else {
+                // singleUse transaction
+                $this->transactionId = null;
+            }
+        }
+
         if (!$this->singleUseState()) {
             $this->state = self::STATE_COMMITTED;
         }