feat: implement aborted transactions retry mechanism

[IlyaFaer]

This is the first part of the transaction retry mechanism implementation. Here is what it consists of:

Class ResultsChecksum is written according to the design and covered with unit tests.
Every transaction inits a checksum property on creation.
Transaction.execute_update() and Transaction.batch_update() methods are now adding their results into transaction checksum.
StreamedResultSet now have an optional constructor arg - checksum. If checksum is passed result set will be pushing every streamed row into the checksum before returning it to user.

At the next step I'm gonna implement propagation of transaction checksum into the result sets of operations runned in this transaction.

[IlyaFaer]

@mf2199, @AVaksman

[IlyaFaer]

@AVaksman, @mf2199, seems like only adding a big unit test for Session.run_in_transaction() is left. It's a tough thing, requires a lot of mocks, I'm still in progress with it.

[IlyaFaer]

Two tests failed:

FAILED tests/system/test_system.py::TestSessionAPI::test_transaction_query_w_concurrent_updates
FAILED tests/system/test_system.py::TestSessionAPI::test_transaction_read_w_concurrent_updates

Seems to me it's a right behavior: data concurrently changed, so results of the retried transaction are not equal to the results of the original transaction.

[IlyaFaer]

@mf2199, @AVaksman, okay, I've pushed the last part

[c24t]

Some questions about the implementation and a note about failing tests.

[c24t]

The java version actually consumes the results up to the same point that the user did before retrying before doing the comparison: https://github.com/googleapis/java-spanner/blob/059ef1ef1f03e80f4ff2705b45a62c553bb4e83d/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ChecksumResultSet.java#L145-L153. Here we actually yield the rows used in the comparison. How come this doesn't result in the user getting the same rows twice after an internal retry?

It's also a bit confusing to see this do the checksum comparison with each row. I know _compare_checksums passes if it sees fewer retried rows than original, but that's not what I'd expect it to do without reading it.

[c24t]

@olavloite, @skuruppu mentioned you might be able to review and compare this to the java client behavior.

[IlyaFaer]

How come this doesn't result in the user getting the same rows twice after an internal retry?

Seems to me a user will not use the rows read within a transaction in operations outside the transaction until the transaction is finished. It probably sounds hard, so consider this case:

User starts a transaction, reads several rows, then calls his own (not Spanner) API which is using these rows to write some data somewhere. Then he calls more functions in the same transaction and commits it. It's incorrect way of use - one should not do anything with data which the transaction is using at that moment, as transaction can fail and all the data/changes will be rollbacked, but it'll not rollback the records made in the user API, so the user will get the data desynchronization.

With this, the data that was read within transaction are considered as a temporary inexact-until-transaction-commit data, and if we're retrying a transaction, all that data should be re-read and recalculated, that's how I see it.

[IlyaFaer]

It's also a bit confusing to see this do the checksum comparison with each row. I know _compare_checksums passes if it sees fewer retried rows than original, but that's not what I'd expect it to do without reading it.

Yeah, that's something I'm not 100% like too, but retry logic looks too apportioned across classes and functions to me, so I've decided to make many things deep under the hood. Thus, in the transaction class we're not thinking about results number of a result set of one of the operations of this transaction, we're just calling == and it takes care about such a things deeper in the result set itself. Result set object is much more closer to these details than transaction.

[olavloite]

I have absolutely no Python experience, so bear with me if I misread or misunderstood (parts of) the code here, but if I understand it correctly the suggested implementation for Python here does the following (during the initial transaction):

1. The __iter__ method will normally (i.e. in the original transaction) iterate over the PartialResultSets that are received from Spanner, and for each of these iterate over the rows in that PartialResultSet.
2. If the ResultSet has a checksum, it will update the running checksum with the contents of the current row.
3. It then checks whether the current checksum is equal to the checksum of the original transaction. If this is the original transaction and not a retry, the value of _original_results_checksum will be None and the comparison will always succeed.
4. It then returns the row.

If a transaction is retried, the following happens:

1. The __iter__ method will iterate over the PartialResultSets that are received from Spanner for the retried query, and for each of these iterate over the rows in that PartialResultSet.
2. If the ResultSet has a checksum, it will update the running checksum with the contents of the current row.
3. It then checks whether the current checksum is equal to the checksum of the original transaction. As far as I can see, this would always fail if the result set contained more than 1 row, as it is a running checksum.

Assume that a query returns the following rows during the original transaction:

Key	Value
1	One
2	Two

During the original transaction the entire result set was consumed so the running checksum was calculated based on all the rows:

• After row 1 the checksum is C1
• After row 2 the checksum is C2
  So the original checksum is now C2.

During a retry the query will be retried and the checksum will be calculated again (I assume that the results during the retry are equal to the original query):

• After row 1 the checksum is C1. That is compared to the original checksum that is C2 and the comparison will fail.

The checksum should therefore not be compared for each row, but after consuming the same number of rows during the retry as during original query.

(But I get a feeling that there is a more fundamental misunderstanding about this retry logic. I'll elaborate on that in a separate comment.)

[IlyaFaer]

Hi, @olavloite

It then checks whether the current checksum is equal to the checksum of the original transaction. As far as I can see, this would always fail if the result set contained more than 1 row, as it is a running checksum.

Yeah, that's the thing @c24t mentioned as something that we don't expect. In fact I'm redefining == operator for ResultsChecksum object, so that it compares checksums in this way:

if original is not None:
        if retried != original:
            if not retried < original:  # retried has less results

At the first line it checks if we're in retried transaction - if not, than we have nothing to compare.

The second checks if checksums are equal, considering their length (number of consumed results).

And the third line checks if the retried transaction came to the original transaction failure point.

So, if the given checksums have different length, it'll be considered like there are no problems so far, 'cause a) retried transaction is not at the point, where the original failed, so it's too soon to compare checksums; b) retried transaction went farther than the original transaction = the checksums were compared = they were equal at the original transaction failure point).

Looking at this, I think there can be a problem when retried transaction went farther than the original. I'll double check

[IlyaFaer]

@c24t, hm-m, I think the logic here can be little bit simpler, if I'll use a __len__ magic method.

UPDATE: yes, it looks better, PTAL

[c24t]

cc @skuruppu and @larkee, who will understand better than I do how this interacts with the existing python-spanner retry logic.

[olavloite]

@IlyaFaer I get the impression that there is a misunderstanding about when retry logic based on checksums are needed for aborted transactions. The Java client contains two different APIs for interacting with Cloud Spanner:

1. The 'normal' Spanner client. This client requires client applications to define a transaction as a function, and to pass that function in to a TransactionRunner. The Spanner client will start a transaction and then call the function that the client application supplied. The function can then execute queries and updates on the transaction. If the transaction is aborted by Cloud Spanner, queries and/or updates on the transaction will return an Aborted error. This will trigger the Spanner client to retry the transaction by starting a new transaction and calling the transaction function once again. This retry strategy does not require any checksums, as the entire transaction is defined as a function that can be retried in its entirety.
2. The 'Connection API': This is a generic API based on a more traditional database API that has a permanent 'connection' with the database. The client application can execute transactions and statements on a connection, and it is not known in advance what statements will be executed in a transaction, as each transaction can be a random list of statements and is not defined as a function. If a transaction is aborted by Cloud Spanner, it cannot be retried by simply 'calling the transaction function again using a new transaction ID', as that function is simply not there. Instead, the Connection API keeps track of all statements that are executed during a transaction and the results that were returned by those statements. If the transaction is aborted, the Connection API will start a new transaction and execute the exact same statements once more on the new transaction, and compare each the result of each statement with the result of the original statement. If they match, the Aborted error from Cloud Spanner is 'ignored' and the transaction is allowed to proceed. If the results do not match, the Aborted error is propagated to the client application and the transaction is aborted.

The way that the Connection API keeps track of the results per statement depends on the type of statement:

• All statements remember whether the statement returned an error or success. During the retry, the error or success should be the same.
• Update statements and batch update statements remember the update count that was returned by the statement. During a retry, the update counts are compared with each other.
• Queries can potentially return a large number of rows, and it is not feasible to keep all rows that have been seen during a transaction in memory. So instead queries keep track of a running checksum based on each row that has actually been returned to the client application. During a transaction retry, the same query is executed again, and the retry will consume the exact same number of rows as the original transaction. If the checksum that is calculated during that retry matches the checksum during the original transaction, the results are deemed equal and the transaction may proceed.

From what I see in this PR, the checksum retry logic is being added to the 'normal' client. The run_in_transaction method takes a function that contains all transaction logic. If the transaction is aborted by Cloud Spanner, the client will retry the transaction by starting a new one and calling the same transaction function once more with the new transaction. There is no need to add any checksum checking to this API, as this retry logic already works as intended.

Retries based on checksums are only needed if client applications can interact with Cloud Spanner in a more traditional way through some kind of 'connection' object (or similar) that does not require you to define a transaction as a function.

Consider the following (pseudo-)code example based on a simple example on how to work with MySQL in Python:

try:
    connection = mysql.connector.connect(host='spanner.googleapis.com',
                                         database='Electronics',
                                         user='pynative',
                                         password='pynative@#29')
    if connection.is_connected():
        connection.autocommit = false
        cursor = connection.cursor()
        cursor.execute("SELECT * FROM Foo;")
        for row in cursor:
            print(row)
        cursor.execute("""UPDATE Foo SET Bar = 'baz' WHERE Id = 2""")
        connection.commit()
        
except Error as e:
    connection.rollback()
    print("Error ", e)
finally:
    if (connection.is_connected()):
        cursor.close()
        connection.close()
        print("Spanner connection is closed")

Cloud Spanner could for example abort the transaction before/during the statement cursor.execute("""UPDATE Foo SET Bar = 'baz' WHERE Id = 2"""). This will cause this statement to return an Aborted error. The underlying connection API should in that case catch the Aborted error, start a new transaction, retry the SELECT * FROM Foo query and check whether the results of that query is still equal to the results of the original attempt. If they are, then the transaction can proceed and the UPDATE Foo SET Bar = 'baz' WHERE Id = 2 can be executed again.

All the above should be transparent to the client application (except for a delay during the execution of UPDATE Foo SET Bar = 'baz' WHERE Id = 2, as the entire retry has be executed and checked before the result of that statement can be returned to the client application).

Please don't hesitate to ask if you have additional questions on this (or let me know if I completely misunderstood how the current PR works).

[IlyaFaer]

@olavloite,

From what I see in this PR, the checksum retry logic is being added to the 'normal' client.

Yes, that's correct. The thing is that the connection API we have for now is actually delegating to the original Spanner client, and exactly run_in_transaction() (which takes a function and runs it like a transaction, like you mentioned), for example:

https://github.com/googleapis/python-spanner-django/blob/e6f6179b4100c7c995849dc521a02c63aea1d791/google/cloud/spanner_dbapi/cursor.py#L116

where in_transaction() is:

https://github.com/googleapis/python-spanner-django/blob/e6f6179b4100c7c995849dc521a02c63aea1d791/google/cloud/spanner_dbapi/connection.py#L59

where db_handle is Database object from the original client.

In fact our Python connection API always works in autocommit mode:
https://github.com/googleapis/python-spanner-django#transaction-management-isnt-supported

Thus, our way of use is little bit different (at least in some parts) from Java - our transactions are always functions, and commit/rollback are called under the hood. We probably can write several wrappers in the connection API, to use checksums comparisons only in connection API, not in traditional Spanner client. But it looks like an overkilling way, as we will have to inherit and then override several original Spanner methods from several classes, and these new methods will be very-very similar to the original ones, like 15-20% defference. It's probably easier to make checksums comparison enablable (and always enabled when using connection API), if it's critical not to compare checksums in the original Spanner client.

For now I don't see simpler way to add checksums comparison into connection API without adding it into original Spanner, as to make comparison work it takes to make changes in several classes, that are embedding each other on different levels.

[olavloite]

Yes, that's correct. The thing is that the connection API we have for now is actually delegating to the original Spanner client, and exactly run_in_transaction() (which takes a function and runs it like a transaction, like you mentioned), for example:

Thanks for the clarification, that makes sense.

In fact our Python connection API always works in autocommit mode:

Uhm... If it is not the intention to change that, then you don't need retry logic for transactions, as there are no transactions.

1. From what I'm getting from the code, any query that is executed on a connection will be handled here: https://github.com/googleapis/python-spanner-django/blob/e6f6179b4100c7c995849dc521a02c63aea1d791/google/cloud/spanner_dbapi/cursor.py#L103
2. That will use a single-use read-only transaction (a 'read snapshot'): https://github.com/googleapis/python-spanner-django/blob/e6f6179b4100c7c995849dc521a02c63aea1d791/google/cloud/spanner_dbapi/cursor.py#L180
3. Such a query will never abort (read-only transactions never abort), so any retry logic in the result set is not necessary.

Update statements will also not abort, even though they do use transactions, as any aborted transaction is handled by the normal retry logic:

1. Any update on a connection will be handled here: https://github.com/googleapis/python-spanner-django/blob/e6f6179b4100c7c995849dc521a02c63aea1d791/google/cloud/spanner_dbapi/cursor.py#L116
2. That calls https://github.com/googleapis/python-spanner-django/blob/e6f6179b4100c7c995849dc521a02c63aea1d791/google/cloud/spanner_dbapi/connection.py#L59 which again calls the normal run_in_transaction with a standard function that only executes a single update statement. If that statement is aborted, the single statement will be retried until it succeeds.

[skuruppu]

@olavloite thanks for the detailed explanation. That is also my understanding. This fix relates to googleapis#34 and based on the discussion and example, it seems that the intention is to change the retry mechanism in run_in_transaction. I think I'm missing something here. If the cased discussed in the example was an issue, then we would've changed the implementation in all the client library a long time ago. @olavloite do you have any thoughts on Ben's example in that issue?

[olavloite]

Yeah, this getting a little bit confusing as there are a number of different topics getting mixed up a little bit.

Issue googleapis#34 describes a feature request for AUTOCOMMIT=off. A copy-paste from the issue:

In AUTOCOMMIT=off mode, we need to hold a Transaction for perhaps an indefinitely long time.
Cloud Spanner will abort:
a) Transactions when not used for 10seconds or more -- we can periodically send a SELECT 1=1 to keep it active
b) Transactions even when refreshed, can and will abort. This is because Cloud Spanner has a high abort rate

Thus we need to retry Transactions!

So:

1. The current retry logic in the 'normal' client library for Python works as intended, and is similar to the retry logic in most other client libraries as well: It has a run_in_transaction function that takes another function as a parameter. The run_in_transaction function starts a transaction, calls the function that was given as an input parameter, and commits the transaction. If the transaction is aborted, the run_in_transaction function will start a new transaction, call the input function once again, and try to commit again. This strategy will always succeed in the end.
2. The current Connection API implementation for Python always uses AUTOCOMMIT=on. All queries in the Python Connection API are currently executed using single-use read-only transactions. These transactions will never be aborted by Cloud Spanner and do not need any retry mechanism in the client library. All DML statements in the Python Connection API are executed as a single statement transaction. That means that a very simple function is defined that only executes the single statement, and that statement is passed in to the run_in_transaction method mentioned in 1. These statements therefore also do not need any additional retry logic. This AUTOCOMMIT=on behavior is equal to the behavior of the Java Connection API (and JDBC driver) when that is used in AUTOCOMMIT=on.
3. The feature request that seems to be the base of this PR is only relevant if the Python Connection API is to support AUTOCOMMIT=off, and the additional retry strategy should also only be used for the Connection API when using AUTOCOMMIT=off.

As explained above, the normal retry strategy that uses a run_in_transaction cannot be used for the Connection API in AUTOCOMMIT=off, as a transaction is not defined as a function. It would instead require the API to keep track of all statements that have been executed on a read/write transaction and the results these statements returned, and if the transaction is aborted, re-execute the statements and verify that these returned the same results as during the initial attempt. If the results are equal, the transaction should be allowed to continue. If they are not, the Aborted error should be propagated to the client application.

[skuruppu]

Ok that makes more sense to me now @olavloite, thanks so much for making sense of it. So if the intention is to support AUTOCOMMIT=off, it wouldn't help to change the logic in run_in_transaction. It has to be done in the Connection API.