Spanner: 'test_transaction_batch_update_w_syntax_error' systest flakes with 409.

[tseaver]

From this Kokoro run:

_________ TestSessionAPI.test_transaction_batch_update_w_syntax_error __________
args = (session: "projects/precise-truck-742/instances/google-cloud-python-systest/databases/test_sessions_1561421696334/sess...hARQno60JnjCZosiqjNrh1xW69seY6xZOPqwi7sR9qy5cr-kXFHxiGpx3w"
transaction_id: "AJSwhAR6DqYBotfoiNM7cVnSLV0pgvC3eEPc3g"
,)
kwargs = {'metadata': [('google-cloud-resource-prefix', 'projects/precise-truck-742/instances/google-cloud-python-systest/datab...x3w'), ('x-goog-api-client', 'gl-python/3.7.0b3 grpc/1.22.0rc1 gax/1.13.0 gapic/1.9.0 gccl/1.9.0')], 'timeout': 3599.0}
    @six.wraps(callable_)
    def error_remapped_callable(*args, **kwargs):
        try:
>           return callable_(*args, **kwargs)
../api_core/google/api_core/grpc_helpers.py:57:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <grpc._channel._UnaryUnaryMultiCallable object at 0x7f02eb1a9c18>
request = session: "projects/precise-truck-742/instances/google-cloud-python-systest/databases/test_sessions_1561421696334/sessi...SwhARQno60JnjCZosiqjNrh1xW69seY6xZOPqwi7sR9qy5cr-kXFHxiGpx3w"
transaction_id: "AJSwhAR6DqYBotfoiNM7cVnSLV0pgvC3eEPc3g"
timeout = 3599.0
metadata = [('google-cloud-resource-prefix', 'projects/precise-truck-742/instances/google-cloud-python-systest/databases/test_ses...wi7sR9qy5cr-kXFHxiGpx3w'), ('x-goog-api-client', 'gl-python/3.7.0b3 grpc/1.22.0rc1 gax/1.13.0 gapic/1.9.0 gccl/1.9.0')]
credentials = None, wait_for_ready = None, compression = None
    def __call__(self,
                 request,
                 timeout=None,
                 metadata=None,
                 credentials=None,
                 wait_for_ready=None,
                 compression=None):
        state, call, = self._blocking(request, timeout, metadata, credentials,
                                      wait_for_ready, compression)
>       return _end_unary_response_blocking(state, call, False, None)
.nox/system-3-7/lib/python3.7/site-packages/grpc/_channel.py:565:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
state = <grpc._channel._RPCState object at 0x7f02eb11cb38>
call = <grpc._cython.cygrpc.SegregatedCall object at 0x7f02eb11ad88>
with_call = False, deadline = None
    def _end_unary_response_blocking(state, call, with_call, deadline):
        if state.code is grpc.StatusCode.OK:
            if with_call:
                rendezvous = _Rendezvous(state, call, None, deadline)
                return state.response, rendezvous
            else:
                return state.response
        else:
>           raise _Rendezvous(state, None, None, deadline)
E           grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with:
E           	status = StatusCode.ABORTED
E           	details = "Transaction aborted. Database schema probably changed during transaction, retry may succeed."
E           	debug_error_string = "{"created":"@1561421764.604780791","description":"Error received from peer ipv4:74.125.195.95:443","file":"src/core/lib/surface/call.cc","file_line":1052,"grpc_message":"Transaction aborted. Database schema probably changed during transaction, retry may succeed.","grpc_status":10}"
E           >
.nox/system-3-7/lib/python3.7/site-packages/grpc/_channel.py:467: _Rendezvous
The above exception was the direct cause of the following exception:
self = <tests.system.test_system.TestSessionAPI testMethod=test_transaction_batch_update_w_syntax_error>
    def test_transaction_batch_update_w_syntax_error(self):
        retry = RetryInstanceState(_has_all_ddl)
        retry(self._db.reload)()
        session = self._db.session()
        session.create()
        self.to_delete.append(session)
        with session.batch() as batch:
            batch.delete(self.TABLE, self.ALL)
        insert_statement = list(self._generate_insert_statements())[0]
        update_statement = (
            "UPDTAE contacts SET email = @email " "WHERE contact_id = @contact_id;",
            {"contact_id": 1, "email": "[email protected]"},
            {"contact_id": Type(code=INT64), "email": Type(code=STRING)},
        )
        delete_statement = (
            "DELETE contacts WHERE contact_id = @contact_id;",
            {"contact_id": 1},
            {"contact_id": Type(code=INT64)},
        )
        with session.transaction() as transaction:
            rows = list(transaction.read(self.TABLE, self.COLUMNS, self.ALL))
            self.assertEqual(rows, [])
            status, row_counts = transaction.batch_update(
>               [insert_statement, update_statement, delete_statement]
            )
tests/system/test_system.py:894:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
google/cloud/spanner_v1/transaction.py:268: in __exit__
    self.commit()
google/cloud/spanner_v1/transaction.py:127: in commit
    metadata=metadata,
google/cloud/spanner_v1/gapic/spanner_client.py:1461: in commit
    request, retry=retry, timeout=timeout, metadata=metadata
../api_core/google/api_core/gapic_v1/method.py:143: in __call__
    return wrapped_func(*args, **kwargs)
../api_core/google/api_core/retry.py:273: in retry_wrapped_func
    on_error=on_error,
../api_core/google/api_core/retry.py:182: in retry_target
    return target()
../api_core/google/api_core/timeout.py:214: in func_with_timeout
    return func(*args, **kwargs)
../api_core/google/api_core/grpc_helpers.py:59: in error_remapped_callable
    six.raise_from(exceptions.from_grpc_error(exc), exc)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = None
from_value = <_Rendezvous of RPC that terminated with:
	status = StatusCode.ABORTED
	details = "Transaction aborted. Database schem...ge":"Transaction aborted. Database schema probably changed during transaction, retry may succeed.","grpc_status":10}"
>
>   ???
E   google.api_core.exceptions.Aborted: 409 Transaction aborted. Database schema probably changed during transaction, retry may succeed.
<string>:3: Aborted