Non-linear performance for large size of delta log

[dispanser]

In an attempt to create a delta table for #425, I attempted to load a relatively large delta log containing ~ 700k add actions and 9.5million remove actions.

Load performance is extremely slow, and it gets worse the farther inside the load we progress. I've logged the time it takes to handle each separate checkpoint file:

  47.85s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000001.0000000021.parquet
 226.31s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000002.0000000021.parquet
 541.34s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000003.0000000021.parquet
 840.58s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000004.0000000021.parquet
1123.62s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000005.0000000021.parquet
1402.43s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000006.0000000021.parquet
1688.98s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000007.0000000021.parquet

It's seems that the load becomes slower and slower the more actions are already loaded. As this is not the case with a table that only includes add actions (see #435: 2 minutes for 5million files, no tombstones), the culprit must be in the remove actions.

fn process_action(state: &mut DeltaTableState, action: Action) -> Result<(), ApplyLogError> {
    match action {
        Action::add(v) => {
            state.files.push(v);
        }
        Action::remove(v) => {
            state.files.retain(|a| *a.path != v.path);
            state.tombstones.push(v);
        }
	...
	}
	...
}

This brings non-linearity: for each add action we seem to make a linear scan through the entire set of adds. The more add actions we have, the longer that takes.

One possible solution would be to collect all actions, build a set of all files that must be removed, and do that in one pass at the very end.

Also, it seems counter-intuitive for a checkpoint to contain both an add and a remove operation for the same file; I suspect that the behavior can be entirely skipped when loading a checkpoint, and only needs to be applied to (relatively small) incremental update operations.

``

[dispanser]

I'm willing to look into this tomorrow, this seems relatively easy to fix (based on the assumption that my analysis above is correct).

[dispanser]

I disabled the line containing the retain just to make sure I"m not staring at the wrong line.

2.21s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000001.0000000021.parquet
2.17s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000002.0000000021.parquet
2.16s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000003.0000000021.parquet
2.16s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000004.0000000021.parquet
2.17s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000005.0000000021.parquet
2.18s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000006.0000000021.parquet
2.20s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000007.0000000021.parquet
2.21s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000008.0000000021.parquet
2.24s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000009.0000000021.parquet
2.25s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000010.0000000021.parquet
2.25s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000011.0000000021.parquet
2.27s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000012.0000000021.parquet
2.25s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000013.0000000021.parquet
2.23s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000014.0000000021.parquet
2.23s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000015.0000000021.parquet
2.27s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000016.0000000021.parquet
2.25s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000017.0000000021.parquet
2.27s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000018.0000000021.parquet
2.27s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000019.0000000021.parquet
2.25s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000020.0000000021.parquet
2.25s: loading checkpoint data from f=<delta_root>/_delta_log/00000000000000002710.checkpoint.0000000021.0000000021.parquet

[houqp]

I was thinking about the same thing while reviewing #431. We should definitely apply action state change in batch instead of one at a time.

[houqp]

Also, it seems counter-intuitive for a checkpoint to contain both an add and a remove operation for the same file; I suspect that the behavior can be entirely skipped when loading a checkpoint, and only needs to be applied to (relatively small) incremental update operations.

I think you just found a bug in our checkpoint implementation ;) Based on https://github.com/delta-io/delta/blob/master/PROTOCOL.md#action-reconciliation, this should never happen. cc @xianwill @mosyp this unnecessarily increased our checkpoint size.

[dispanser]

I think you just found a bug in our checkpoint implementation ;) Based on https://github.com/delta-io/delta/blob/master/PROTOCOL.md#action-reconciliation, this should never happen. cc @xianwill @mosyp this unnecessarily increased our checkpoint size.

A collection of remove actions with unique paths. The intersection of the paths in the add collection and remove collection must be empty. That means a file cannot exist in both the remove and add collections. The remove actions act as tombstones.

If add / remove for the same file could be in one checkpoint, the final state would depend on the order of observing them. I'm glad the spec does not allow that.

I completely missed #431 btw, good that there's already something done. I repeated my tests with the newest revision, but it doesn't seem to change anything, probably because none of the remove action actually matches any of the add actions so we walk the vec to the very end for each tombstone anyway. I suspect that it improves performance for other use cases, though 👍

[houqp]

but it doesn't seem to change anything, probably because none of the remove action actually matches any of the add actions

That's quite strange. This means we commit a remove action immediately after a file has been added? Anyway, I think this won't be a problem anymore once we move to batch update.

BTW, @dispanser I am not actively working on these performance issues because I assume you are interested in solving them. If that's not the case, please let me know, I will be more than happy to help.

[dispanser]

That's quite strange. This means we commit a remove action immediately after a file has been added? Anyway, I think this won't be a problem anymore once we move to batch update.

I try to formulate better: when loading a checkpoint, we never have a "remove" that matches an "add", because they'd be part of the same checkpoint, so the optimization from #431 can't be applied. Every add is kept, so the search (retain, or index computation in the newest revision) does not find a matching entry so it has to search state.files until the very end and does not modify it.

BTW, @dispanser I am not actively working on these performance issues because I assume you are interested in solving them. If that's not the case, please let me know, I will be more than happy to help.

Indeed. I'm having a day off today, and I don't see a better way to spend my time than learning some rust ;). I'll try to come up with something today.

[houqp]

I try to formulate better: when loading a checkpoint, we never have a "remove" that matches an "add", because they'd be part of the same checkpoint,

Oh, i see. Yeah, for a valid checkpoints, this optimization would have no effect. It only works for incremental version upgrades.