Spark: Merge new position deletes with old deletes during writing

[amogh-jahagirdar]

This change consumes the updated fanout position delete writers in #11222 to maintain position deletes during writes (minor compaction). The mapping of data files to file scoped deletes is broadcasted to executors, where delete file writers merge the historical position deletes with new position deletes. This behavior is behind a Spark conf maintain-position-deletes and can also be controlled via the write.delete.maintain-during-write table property.
By default this maintenance during write is enabled.

ToDo: UpdateProjectionBenchmark may need to include some changes to use file granularity for MoR cases, so that we can look at the impact of this change before/after. The benchmark should probably be run after #11131 gets in

[amogh-jahagirdar]

We may want to put this behind a spark conf which defaults to loading the previous deletes and doing the merging in the write. Maybe it's better to handle such a conf when building the mapping (if the conf is false, we could return a empty map or don't perform the broadcast).

That way in case users hit some unforseen issue with merging they can disable the behavior dynamically.

Didn't want to introduce more configuration knobs initially to keep it simple but I can see an argument that it's worth it just to have a lever for unforseen issues with performing minor compactions as part of writes.

CC @aokolnychyi @rdblue @nastra

[amogh-jahagirdar]

Some of the tests in TestRewritePositionDeleteFilesProcedure are expectedly failing after this change since those tests have assertions on expected delete files after performing a set of delete operations but now since there's maintenance happening as part of the write, the number of delete files will be expectedly reduced (e.g.)

If we decide to add a conf, that can be set accordingly, otherwise I'll go ahead and update the tests. At the moment, still leaning towards adding a conf for the reasons mentioned earlier but I'll get others input.

[amogh-jahagirdar]

I went ahead with adding the configuration and table property similar to delete file granularity.

[aokolnychyi]

Looks really promising!

[aokolnychyi]

I am not sure we need this configuration. It is pretty clear we want to always maintain position deletes. We will also not support this property in V3. Given that we want to switch to file-scoped position deletes in V2 tables by default, I think we should just always maintain deletes if the granularity is file.

[aokolnychyi]

In other words, switching write.delete.granularity to file should trigger maintenance.
If set to partition, skip it as we can't do it safely.

[singhpk234]

write.delete.granularity to file

[doubt] are any other writers accept from spark respecting this property ? Are other writers also gonna respect this property going forward, if yes how ?

[aokolnychyi]

This is up to an engine in V2. We discuss making it a requirement for V3 on the dev list right now.

[singhpk234]

write.delete.granularity to file

[doubt] are any other writers accept from spark respecting this property ? Are other writers also gonna respect this property going forward, if yes how ?

[singhpk234]

can we put an estimate on the size of the HM ? if it goes very high it can fail the query in that case should we let the query fail ?

[aokolnychyi]

This is a good point. We need to check what's the actual limit on the object size and how Spark would behave.

[nastra]

overall this LGTM, just left a few nits

[nastra]

nit: might be worth updating

iceberg/core/src/main/java/org/apache/iceberg/deletes/SortingPositionOnlyDeleteWriter.java

Line 174 in c8fe01e

private boolean isFileScoped(DeleteFile deleteFile) {

to use this method too

[nastra]

nit: in checkDeleteFileGranularity() we're using TableProperties rather than plain strings, so I think it would be good to do the same here too

[aokolnychyi]

This seems almost ready to me.

Like @singhpk234 mentioned, we need to verify Spark can handle this logic if the map gets large. The table state that we currently send to executors is larger than this map but we need to look into how this particular broadcast is handled. If it is split into chunks by the torrent broadcast, that should probably be OK.

@amogh-jahagirdar, could you check it is going to be safe?

[aokolnychyi]

What about a helper method and using it directly in statements below?

private Map<String, DeleteFileSet> rewritableDeletes() {
  return rewritableDeletesBroadcast != null ? rewritableDeletesBroadcast.getValue() : null;
}

[amogh-jahagirdar]

Good suggestion, done! Note, since it's possible now that rewritable deletes is null in the writer, I explicitly pass in the previous loader to be a function path -> null. Previously I was passing in an empty map so the lookup would return null, but i think your suggestion is better because we can avoid having to do a lookup in the map entirely

[aokolnychyi]

I understand we refactor common logic but this makes tests harder to read. For instance, the reader has no indication we add 4 batches in the init method.

[aokolnychyi]

LGTM. I left some minor comments/suggestions. We will revisit and benchmark this prior to 1.8.

[singhpk234]

LGTM as well, Thanks @amogh-jahagirdar !

I am assuming since we went ahead with broadcasting approach, it sends it chunk by chunk using torrent broadcast as @aokolnychyi mentioned, so OOM not a problem ? Was mostly coming from when doing BHJ spark enforces 8GB limit and if anything more than that is observed spark fails the query.

[amogh-jahagirdar]

I am assuming since we went ahead with broadcasting approach, it sends it chunk by chunk using torrent broadcast as @aokolnychyi mentioned, so OOM not a problem ?

I collected some data points on memory consumption of the broadcast here https://docs.google.com/document/d/1yUObq45kBIwyofJYhurcQrpsdXQJC6NFIPBJugiZNWI/edit?tab=t.0

Torrent broadcast is performed in a chunked way but I wouldn't say that doesn't mean OOMs aren't possible. The TLDR is that we would have to be at pretty large scale (multiple millions of data files) and have a very large multiple of deletes per data file for OOMs to be hit in most environments, and running position delete maintenance to shrink that ratio + increasing memory should be a practical enough solution. As maintenance of position deletes runs, that ratio becomes more 1:1 between data to delete files. In V3, this will actually be a requirement. I did a look at more distributed approaches to compute this (changing the Spark APIs to pass historical deletes just for a particular executor) but there are limitations on that. One thing I'm looking into further is how the Spark Delta DV integration looks like to handle this, and we can perhaps take some inspiration from that, but don't think there's really any need to wait for all of that.

There are relatively simple things we can do to limit the size of the global map, one is removing any unnecessary metadata that executors don't need, for example referenced manifest locations per delete file (that's only needed in the driver for more efficient commits), and also relativizing the paths in memory. That should shrink the total amount of memory that gets used by the paths in all the in-memory structure and has more of an impact the longer file path before the actual data file/delete file name.

Edit: One other aspect I looked into is using Spark's BytesToBytesMap which is an offheap map, this requires a bit more work but is another possible path to shrinking memory usage. Java strings are UTF-16 but in Spark, it's UTF-8 so we could theoretically cut memory usage in half for the equivalent data + delete file combination.

My plan is to work towards having the "simple things we can do" in the 1.8 release, so that we further reduce the chance of OOMs in large scale cases. The long term plan is to look at how Spark + delta DV handles this and if it makes sense for us, incorporate that strategy here.

Was mostly coming from when doing BHJ spark enforces 8GB limit and if anything more than that is observed spark fails the query.

That 8gb limit is specific to broadcast joins, there's no such limit enforced by Spark itself for arbitrary broadcasts (of course there are system limitations that would get hit).

[amogh-jahagirdar]

I'll go ahead and merge, thanks for reviewing @singhpk234 @aokolnychyi . As discussed above, I will work towards shrinking the memory consumption of the broadcast until the 1.8 release!