Spanner: 'test_transaction_batch_update_and_execute_dml' systest flakes with unexpected status code.

[emar-kar]

While resolving #9534, I was able to reproduce the previous variant (#7504) of a flaky error.

===================================================== FAILURES =====================================================
___________________________ TestSessionAPI.test_transaction_batch_update_and_execute_dml ___________________________

self = <tests.system.test_system.TestSessionAPI testMethod=test_transaction_batch_update_and_execute_dml>

    def test_transaction_batch_update_and_execute_dml(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_statements = list(self._generate_insert_statements())
        update_statements = [
            (
                "UPDATE 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 TRUE;"

        def unit_of_work(transaction, self):
            rows = list(transaction.read(self.TABLE, self.COLUMNS, self.ALL))
            self.assertEqual(rows, [])

            status, row_counts, response = transaction.batch_update(
                insert_statements + update_statements
            )
            if self._check_for_the_flaky(status.code):
                print(response)
                raise exceptions.from_grpc_status(status.code, "batch_update failed")
            # self._check_batch_status(status.code)
            self.assertEqual(len(row_counts), len(insert_statements) + 1)
            for row_count in row_counts:
                self.assertEqual(row_count, 1)

            row_count = transaction.execute_update(delete_statement)

            self.assertEqual(row_count, len(insert_statements))

>       session.run_in_transaction(unit_of_work, self)

test_system.py:884:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
..\..\google\cloud\spanner_v1\session.py:299: in run_in_transaction
    return_value = func(txn, *args, **kw)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

transaction = <google.cloud.spanner_v1.transaction.Transaction object at 0x0000013A993550C8>,
self = <tests.system.test_system.TestSessionAPI testMethod=test_transaction_batch_update_and_execute_dml>

    def unit_of_work(transaction, self):
        rows = list(transaction.read(self.TABLE, self.COLUMNS, self.ALL))
        self.assertEqual(rows, [])

        status, row_counts, response = transaction.batch_update(
            insert_statements + update_statements
        )
        if self._check_for_the_flaky(status.code):
            print(response)
>           raise exceptions.from_grpc_status(status.code, "batch_update failed")
E           google.api_core.exceptions.GoogleAPICallError: None batch_update failed

test_system.py:874: GoogleAPICallError
----------------------------------------------- Captured stdout call -----------------------------------------------
status {
  code: 10
  message: "Transaction was aborted."
  details {
    type_url: "type.googleapis.com/google.rpc.DebugInfo"
    value: "\022VTransaction already aborted: 0x7d0d99a723e44d6c@0x59607c30290a0: Transaction not found"
  }
}

[emar-kar]

One more strange error during the execution:

============================================================================================================= FAILURES ==============================================================================================================
___________________________________________________________________________________ TestSessionAPI.test_transaction_batch_update_and_execute_dml ____________________________________________________________________________________

self = <tests.system.test_system.TestSessionAPI testMethod=test_transaction_batch_update_and_execute_dml>

    def test_transaction_batch_update_and_execute_dml(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_statements = list(self._generate_insert_statements())
        update_statements = [
            (
                "UPDATE 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 TRUE;"

        def unit_of_work(transaction, self):
            rows = list(transaction.read(self.TABLE, self.COLUMNS, self.ALL))
            self.assertEqual(rows, [])

            status, row_counts, response = transaction.batch_update(
                insert_statements + update_statements
            )
            if self._check_for_the_flaky(status.code):
                print(response)
                raise exceptions.from_grpc_status(status.code, "batch_update failed")
            # self._check_batch_status(status.code)
            self.assertEqual(len(row_counts), len(insert_statements) + 1)
            for row_count in row_counts:
                self.assertEqual(row_count, 1)

            row_count = transaction.execute_update(delete_statement)

            self.assertEqual(row_count, len(insert_statements))

>       session.run_in_transaction(unit_of_work, self)

test_system.py:884:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
..\..\google\cloud\spanner_v1\session.py:299: in run_in_transaction
    return_value = func(txn, *args, **kw)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

transaction = <google.cloud.spanner_v1.transaction.Transaction object at 0x00000237FEEA4248>, self = <tests.system.test_system.TestSessionAPI testMethod=test_transaction_batch_update_and_execute_dml>

    def unit_of_work(transaction, self):
        rows = list(transaction.read(self.TABLE, self.COLUMNS, self.ALL))
        self.assertEqual(rows, [])

        status, row_counts, response = transaction.batch_update(
            insert_statements + update_statements
        )
        if self._check_for_the_flaky(status.code):
            print(response)
>           raise exceptions.from_grpc_status(status.code, "batch_update failed")
E           google.api_core.exceptions.GoogleAPICallError: None batch_update failed

test_system.py:874: GoogleAPICallError
------------------------------------------------------------------------------------------------------- Captured stdout call --------------------------------------------------------------------------------------------------------
status {
  code: 10
  message: "Transaction aborted. Database schema probably changed during transaction, retry may succeed."
  details {
    type_url: "type.googleapis.com/google.rpc.DebugInfo"
    value: "\022?Schema changed (locked at:1, now: 2);  from Flush(g27_57954906)"
  }
}

[emar-kar]

One more during the tests:

result_sets {
  metadata {
    row_type {
    }
  }
  stats {
    row_count_exact: 1
  }
}
status {
  code: 10
  message: "Transaction was aborted."
  details {
    type_url: "type.googleapis.com/google.rpc.DebugInfo"
    value: "\022^Reads performed on different tablets, see http://go/spanner-aborts: z29_57532591, z27_57969867"
  }
}

[tseaver]

This is actually the same issue as #9534: we have a mismatch between the actual grpc.Status enum and the expected google.rpc.Status enum. I have a fix coming for that problem.