[air] pyarrow.fs persistence: Don't automatically delete the local checkpoint

[justinvyu]

Why are these changes needed?

This PR updates the persist_current_checkpoint logic to not delete the user's input checkpoint upon uploading. This is instead the job of the user to clean up if they want to. This is in line with making things more explicit.

The new pattern for reporting a checkpoint (that gets cleaned up automatically) is:

import tempfile

from ray import train
from ray.train._checkpoint import Checkpoint  # soon to be ray.train.Checkpoint

def train_fn(config):
    for i in range(epochs):
        should_checkpoint = ...
        if should_checkpoint:
            with tempfile.TemporaryDirectory() as tempdir:
                # write files into tempdir...
                train.report({...}, Checkpoint.from_directory(tempdir))
        else:
            train.report({...})

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[woshiyyya]

I love this change.

Previously, users had to create a tmpdir, copy existing ckpts to that tmp directory, and report it to Train. If the checkpoint is large, the copy step is unnecessary and takes a long time.

[justinvyu]

Ah yeah I didn't even think about that! Saves 1 copy step if they already have a dir created by some integration library.

[woshiyyya]

Thank you @justinvyu!