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": "phreddy@example.com"},
 {"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