fix(rust): handle 429 from GCS

[adriangb]

Fixes #2451

This is probably not the best implementation but it at least let me test the outcome.

Here's my test script:

import os
from uuid import uuid4

os.environ['RUST_LOG'] = 'INFO'

from time import time
from concurrent.futures import ProcessPoolExecutor

import polars as pl
from deltalake import DeltaTable, write_deltalake

def worker(uri: str, df: pl.DataFrame) -> None:
    table = DeltaTable(uri, log_buffer_size=1024 * 1024)
    table.update_incremental()

    print('writing in a loop')
    while True:
        try:
            start = time()
            write_deltalake(table, df.to_arrow(), engine='rust', mode='append')
            print('Time taken:', time() - start)
        except Exception as e:
            print('Python error:', e)
            continue

if __name__ == '__main__':    
    df = pl.DataFrame({
        'a': [1, 2, 3],
        'b': [4, 5, 6],
    })

    uri = f'gs://bucket-00d0331/test/delta/{uuid4()}'

    print('initializing delta table')
    write_deltalake(uri, df.to_arrow(), engine='rust', mode='append')

    with ProcessPoolExecutor(32) as executor:
        futures = [
            executor.submit(worker, uri, df)
            for _ in range(32)
        ]
        for future in futures:
            future.result()

With this change I now get:

Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Python error: Generic DeltaTable error: Version mismatch
Time taken: 1.4619898796081543
Attempt 1 failed: Version 1 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 1 already exists
Attempt 2 failed: Version 2 already exists
Python error: Generic DeltaTable error: Version mismatch
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 3 failed: Version 3 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 3 failed: Version 3 already exists
Attempt 1 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 1 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 1 already exists
Attempt 1 failed: Version 1 already exists
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 1 already exists
Python error: Generic DeltaTable error: Version mismatch
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 2 already exists
Attempt 3 failed: Version 3 already exists
Attempt 1 failed: Version 2 already exists
Attempt 1 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 3 failed: Version 3 already exists
Attempt 4 failed: Version 4 already exists
Attempt 3 failed: Version 3 already exists
Attempt 3 failed: Version 3 already exists
Attempt 2 failed: Version 3 already exists
Attempt 3 failed: Version 3 already exists
Attempt 3 failed: Version 3 already exists
Attempt 2 failed: Version 3 already exists
Attempt 3 failed: Version 3 already exists
Attempt 2 failed: Version 3 already exists
Attempt 2 failed: Version 2 already exists
Attempt 2 failed: Version 2 already exists
Attempt 1 failed: Version 2 already exists
Attempt 3 failed: Version 3 already exists
Attempt 3 failed: Version 3 already exists
Attempt 3 failed: Version 3 already exists
Attempt 2 failed: Version 3 already exists
Attempt 2 failed: Version 2 already exists
Attempt 3 failed: Version 3 already exists
Python error: Generic DeltaTable error: Version mismatch

So this is an improvement but the behavior is still problematic: I'm not sure that it's updating to try the next version every time a commit fails. I also don't know if updating to the next version is even the right thing to do, maybe it should be refreshing the table state before retrying in case it's gotten far out of sync (e.g. because writing the data took a long time). Buy maybe these are crate wide or protocol wide issues not related to the particularities of GCS.

[github-actions]

ACTION NEEDED

delta-rs follows the Conventional Commits specification for release automation.

The PR title and description are used as the merge commit message. Please update your PR title and description to match the specification.

[adriangb]

cc @ion-elgreco

[ion-elgreco]

If you can remove this, and fix the linting/fmt, then we are good to go. Also our GCS integration tests are disabled at the moment, so trusting that this works ; P