Fix duplicates on update call

[mosyp]

Description

Every time the update is called, the Add action from the latest checkpoints are being applied once more, leading into duplicates.
As it turned out, it's a regression from #124 changes.

First commit shows the repro steps, second revers mention changes and resolves the issue

[rtyler]

@dispanser If you're available, I think it would be helpful for you to chime in on this thread 😄

Need to take a closer inspection

[houqp]

I think the optimization that @dispanser introduced is valid, the bug is caused by incorrect use of apply_logs_from_current_version :)

[dispanser]

@rtyler : just coming back from vacation today, I'll take a closer look tomorrow.

[dispanser]

Indeed, calling updatewith the fixes from #124 produces a duplicate file. Nice catch @mosyp !

However, I believe the fix proposed in this PR only solves this as a side effect of not doing the proper thing in the first place: restore_checkpoint clears the state, which has the same cost (and code path) as loading a table from scratch, which is IMHO undesired for a table without any updates..

What the condition basically says is "if the previously loaded latest checkpoint is the same as the currently seen checkpoint, reload from the checkpoint", which leads to the checkpoints being only used if it was current anyway, so doesn't contain anything not already loaded. If the version has advanced past the next checkpoint, we don't use it, which is not what we want because these checkpoints are supposed to allow you skipping all those jsons. Hence the rationale for #124.

As @hougp already mentioned, the root cause for the duplication is calling apply_logs_from_current_version with the current version instead of the next version. The "restoring from checkpoint" code path works around that by increasing the version number self.apply_logs_from_current_version, which we could also do in an else branch:

            Ok(last_check_point) => {
                if self.last_check_point != Some(last_check_point) {
                    self.last_check_point = Some(last_check_point);
                    self.restore_checkpoint(last_check_point).await?;
                    self.version = last_check_point.version + 1;
                } else {
                    self.version += 1;
                }
            }

Of course, now we basically do a +1 in all non-error cases which is a bit odd so it's probably a good idea to move that out. Also, there's update_incremental, maybe we just call that?

[houqp]

Yep, i think it would be cleaner to use update_incremental wherever applicable, then remove all the self.version += 1 code. apply_logs_from_current_version is very easy to be missed use based on its name, so we should probably only use it in update_incremental as an internal method.

[mosyp]

@dispanser @houqp I got rid of apply_logs_from_current_version (it was basically an incomplete version of update_incremental) and applied your suggestion.
So the update is pretty straightforward one.

• If the latest checkpoint is the same - then just call update_incremental
• otherwise - clear state and apply latest checkpoint then call update_incremental.

It also lead to rework the load function. But I've done a little trick there. Before calling the update_incremental, I clear the state and set version to -1 and then call update_incremental. This was we have the same behavior as before but with much less code. (the -1 is required for update_incremental to start from applying the version 0)

[houqp]

nice clean up @mosyp !