[RLlib; Offline RL] Validate episodes before adding them to the buffer.

[simonsays1980]

Why are these changes needed?

At the moment the OfflinePreLearner samples recorded episodes or SampleBatches from a ray.data dataset and then adds them to a buffer which corrdinates the time step sampling. It could potentially happen that

1. The sampled episodes are duplicates (either in the batch or in the buffer)
2. The sampled episodes are not terminated nor truncated and therefore could and certainly will be fragmented in time order (i.e. we maybe sample first an episode chunk that contains timesteps 11 to 21 before we sample 0 to 11).

In both cases the buffer would raise an error as soon as SingleAgentEpisode.concat is called.

This PR introduces a _validate_episodes method to the OfflinePreLearner to check episodes for duplicates and fragments and returns only unique episodes that are not in the buffer, yet. It disallows uncompleted episodes and thereby ensures that no fragments are added. Users are responsible to record only full episodes.

Furthermore, this PR adds a new configuration parameter to the AlgorithmConfig: input_read_batch_size that controls for the size of the raw data batch pulled from an offline dataset. This can be used in cases of EpisodeType or BatchType row formats that contain usually multiple timesteps and therefore even a single row might contain enough timesteps for train_batch_size_per_learner.

Warning The main chorus for users is: Do not record episodes or sample batches that are not done (batch_mode="truncated_episodes") as this would necessary lead to problems with replay buffers b/c episode chunks to not arrive in order.

Note This might become obsolete in the future, if we simply provide sampled episodes with new unique IDs such that an episode buffer considers chunks of the same episode as standalone ones.

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 :(

[sven1977]

dumb question: Why would we expect duplicates at all here? Any algorithm that would be in danger of running into something like this?

[simonsays1980]

Great question: This is tricky. I just ran into it because in my own projects I was using small datasets for testing. It could be that your data is so small that pulling a batch contains mjultiple duplicates. Let's say you have a dataset of size 10 and you pull a batch of size 100 (silly maybe, but if you have episodes one row could be multiple timesteps) now 90 rows would be duplicates in the batch.

[sven1977]

remove these?

[sven1977]

Looks good. Thanks @simonsays1980 for the PR!
Just a few nits.