Spark 3.2: Use SnapshotSummary to deduce latestOffset in Spark SS

[singhpk234]

• At present we iterate the iterator of addedFiles to find the no of added files in the snapshot, this info is already present in snapshot summary. We should use this rather than iterating the addedFiles iterator to determine files added.

Came across this while working on

• Spark 3.3: support rate limit in Spark Streaming #4479

---

cc @rdblue, @RussellSpitzer, @jackye1995, @kbendick

[SreeramGarlapati]

@singhpk234 - we explicitly opted out of reading the SnapshotSummary - based on discussion with @aokolnychyi, @RussellSpitzer & @rdblue.

rationale: SnapshotSummary is a free form dictionary & ADDED_FILES_PROP as a key in this dictionary - is NOT added to Iceberg Spec & is not NOT populated by all engines yet.

Pl. read thru this comment thread for full context.
#2660 (comment)

[singhpk234]

Thanks @SreeramGarlapati !!! for sharing the #2660 (comment) thread, it really helps. Glad to know it was already considered while the implementation.

It looks like we were on the fence to establish the reliability of this metric and hence decided to re-visit it in future, refering this comment #2660 (comment)

SnapshotSummary is a free form dictionary & ADDED_FILES_PROP as a key in this dictionary - is NOT added to Iceberg Spec

Agree with you on this.

is not NOT populated by all engines yet

As per my understanding engines don't own the responsibility of calculating / updating the snapshot summary stats, IIUC it's done entirely by core library. The flow for ex : commit() [from engine] -> apply() in SnapshotProducer (core) -> Summary calculation (core).
Am I missing something here ?

[rdblue]

@singhpk234, this is going to get the size of latestSnapshot.addedFiles() either way because it needs to calculate that value to pass it into the property check.

@SreeramGarlapati makes a good point, but if the added files property is present, then it seems to me like we could use it and fall back to reading the snapshot.

[jackye1995]

ah I see, good point we cannot just put Iterables.size(latestSnapshot.addedFiles()) as the input to the last method otherwise it will be evaluated.

[jackye1995]

looks good to me!

[jackye1995]

@SreeramGarlapati @rdblue any additional comments? If not I think it is good to be merged.

[jackye1995]

I think this is good to go, and there is no further comment for quite a while now. I will merge it, thanks for the reviews! @rdblue @SreeramGarlapati