Use only latest Puffin files for Iceberg stats

[homar]

Description

The goal of this change is to improve performance of reading stats, also older stats files may have skewed data which is not optimal for CBO.

Additional context and related issues

Release notes

( ) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
(x) Release notes are required, with the following suggested text:

# Iceberg
* Use only latest puffin file for statistics. ({issue}`issuenumber`)

[electrum]

Why create this here, if it's not needed until the loop below?

[electrum]

Why remove here? The map isn't used if the value is found, so I don't see the point in modifying it.

Seems this is leftover from before, and we can change to the three-argument version of toMap() that returns an unmodifiable map.

[findinpath]

This is not anymore the result of previous analyze right?
ANALYZE " + tableName + " WITH (columns = ARRAY['nationkey', 'regionkey']) has replaced the stats for comment and name as well.
Does the comment still make sense?

[findinpath]

Could you please add in the commit comment also an explanation about why is this change being done?

[findepi]

(x) Release notes are required, with the following suggested text:

# Iceberg
* Use only latest puffin file for statistics. ({issue}`issuenumber`)

I think we should skip this from release notes.
Per our knowledge, this should have no end-user visible impact for currently existing tables.

[findepi]

getOnlyElement(statsFileBySnapshot.values()

more importantly, this special-case path can return a stat file that the longer version below wouldn't, when there is stats file for a newer snapshot)

let's remove the == 1 special case

[findepi]

// No statistics files

?

[findepi]

getLatestStatisticsFile

[findepi]

unused?

[homar]

Used now;)

[findepi]

reanalyze with a subset of columns should carry on unchanged stats to the new stats file

that's the basic idea behind this whole change

[findepi]

what would happen if we used INCREMENTAL_UPDATE always?

[homar]

You would call unnecessarily call code that merges changes instead of just replacing them i think nothing else

[findepi]

INCREMENTAL_UPDATE facilitates stats update during inserts
we cannot update stats during deletions though. That's why we need REPLACE mode. ANALYZE uses REPLACE mode to "erase" values that are no longer in the table. (this is a challenge for incremental analyze, but that's another story).

now, ANALYZE wants to use REPLACE mode regardless of whether all data columns got analyzed, or only the subset. Otherwise it wouldn't be able to fix stats drifting over time. see #16889

[homar]

I cherry-picked your pr and the tests passed so I'd assume we are fine. Or what am I missing ?

[findepi]

That test doesn't cover the case where subset of columns is being analyzed. Can modify the test accordingly?

[findepi]

I've updated #16889 to include analyze with a subset of columns.
The current version of this PR will break that test.

[findepi]

i am not sure we need to capture that in a handle, as the finish* method can know the analyzed vs all column names

[homar]

So I did it for 2 reasons, and probably both are invalid :|
Initially I wanted to do that based on column IDs but that is not feasible because they come as Strings as ANALYZE properties.
When I realised that I didn't revert this because I know I will need this for INCREMENTAL ANALYZE anyway. But as it is premature here I will do what you suggest.

[alexjo2144]

Overall a good change, just a couple style nitpicks and things

[alexjo2144]

This is the same as columns.size() right?

[alexjo2144]

You can use Stream.distinct() rather than collecting to a set.

[homar]

yes but then I will have to collect to List and get size of that list so I thought that this is simpler ;)

[alexjo2144]

Reminder to clean this up, why are they commented out?

[homar]

oh, silly, I just commented them out for the debugging purpose..

[findepi]

I've updated #16889 to include analyze with a subset of columns

[findepi]

redundant?

[findepi]

I've updated #16889 to include analyze with a subset of columns.
The current version of this PR will break that test.

[findepi]

I don't see need to distinguisih between the two cases.

mergeStatisticsIfNecessary gets some statistics to write.
in REPLACE mode, it should write those new statistics, and should carry forward all statistics that hasn't been replaced (other columns, or other stats).

[alexjo2144]

Good point, we can remove the extra case and always run what's in the REPLACE_PARTIALLY block now, right?. It may just be worth refactoring the updateStats method slightly to make the reading of the previous Puffin file lazy.

[findepi]

remove, the class is used

the right way to deal with error-prone reporting problems is do add record compact constructor making a defensive copy and null checks on the provided parameters. thank you, error-prone, for reminding us about that :)

[findepi]

that looks like validation belonging to the (compact) constructor?

[findepi]

what is .map(entry -> Map.entry(entry.getKey(), entry.getValue())) for?

[homar]

I didn't come up with a better way to make a one stream of type Stream<Map.Entry<Integer, Object>> out of these two...

[findepi]

what is oldSketches for?

why isn't io.trino.plugin.iceberg.TableStatisticsWriter#copyRetainedStatistics covering it?

(i think i found out -- the writer should use getLatestStatisticsFile and then reading oldSketches isn't needed here)

[findepi]

This should use getLatestStatisticsFile

getLatestStatisticsFile(table, snapshotId)
                        .ifPresent(previousStatisticsFile -> copyRetainedStatistics

[findepi]

My IDE shows a warning here. Is it because of my specific configuration, or something to address?

[findepi]

readability: orElse for an optional created over previous 37 lines isn't readable to me

correctness: in INCREMENTAL_UPDATE, the collectedNdvSketches need to be merged with existing stats, otherwise they are incorrect.
i'd expect #17227 to fail (but it doesn't). Can you explain to me where the logic applied ensuring we don't write NDV(1) after inserting one row to a table that already has million of rows, but doesn't have stats?

[findepi]

found the code, it's here

trino/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

Lines 1791 to 1795 in 3db96bb

	Set<String> columnsWithExtendedStatistics = tableStatistics.getColumnStatistics().entrySet().stream()
	.filter(entry -> !entry.getValue().getDistinctValuesCount().isUnknown())
	.map(entry -> ((IcebergColumnHandle) entry.getKey()).getName())
	.collect(toImmutableSet());
	return getStatisticsCollectionMetadata(tableMetadata, Optional.of(columnsWithExtendedStatistics), availableColumnNames -> {});

still, the stats writer needs to be smarter, as if that code didn't exist.

[homar]

1. I added that test.
2. I will refactor that orElse
3. It is merged here

 if (finalUpdateMode == INCREMENTAL_UPDATE) {
                                newSketch = SetOperation.builder().buildUnion().union(readPreviousSketch(read), newSketch);
                            }

[findepi]

per the comment elsewhere, the else { sholdn't be needed, the old values should be taken care of using copyRetainedStatistics

[findepi]

Make sure TestIcebergStatistics passes also when

• you include test added in #17227
• apply the below change

Index: plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java
--- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java	(revision 734bc7f5a61c503044fac6ce73aa432751a3550b)
+++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java	(date 1682514123649)
@@ -1789,7 +1789,7 @@
             return getStatisticsCollectionMetadata(tableMetadata, Optional.empty(), availableColumnNames -> {});
         }
         Set<String> columnsWithExtendedStatistics = tableStatistics.getColumnStatistics().entrySet().stream()
-                .filter(entry -> !entry.getValue().getDistinctValuesCount().isUnknown())
+                // .filter(entry -> !entry.getValue().getDistinctValuesCount().isUnknown()) -- this is for optimization only, not correctness, so let's see what happens without it
                 .map(entry -> ((IcebergColumnHandle) entry.getKey()).getName())
                 .collect(toImmutableSet());
         return getStatisticsCollectionMetadata(tableMetadata, Optional.of(columnsWithExtendedStatistics), availableColumnNames -> {});

[findepi]

Why read anything in REPLACE mode?
i think we gonna not use the read results anyway?

[findepi]

what if updateMode is something else than REPLACE or INCREMENTAL_UPDATE? (future-proofing the code)

[findepi]

I am looking at this diff and i don't understand why this is being removed.
Wasn't the previous structure OK?

same for mergeStatisticsIfNecessary return type. What was wrong about it?
if, in the future, we collect anything else besides NDV (CollectedStatistics changes), we want the mergeStatisticsIfNecessary to be updated appropriately, so having the method return CollectedStatistics looked right to me

[findepi]

Why do this only if no previous statistics file? Shouldn't REPLACE mean always replace?

[homar]

rebasing

[findepi]

is needed only in one case, so should inside switch below

i pushed a fixup with this (and some small other changes)

[findepi]

this should be same as hasUsefulData above, i.e.

.filter(blobMetadata -> columnsWithRecentlyComputedStats.contains(getOnlyElement(blobMetadata.fields())))

[findepi]

i fail to understand why this .filter(blobMetadata -> pendingPreviousNdvSketches.contains(getOnlyElement(blobMetadata.inputFields()))) is removed.
can you please explain?

[homar]

I probably lost it while changing the implementation few times. Thanks for catching it

[findepi]

remove if (columnsWithRecentlyComputedStats.remove(fieldId)), here we shouldn't be reading anything that we're not joining with computed NDVs

everything else should go with copyRetainedStatistics path

[findepi]

columnsWithRecentlyComputedStats will become immutable, so no need for initializing it with new HashSet<>(

[findepi]

This means "for any column we have only partial data (previously had no NDV information), treat this partial data as complete information".
So if i insert 1 row into a table with 1 M rows, i can have NDV of 1 for such column.

Am i reading this right?

[homar]

yes you are right

[findepi]

you can inline it back now

[findepi]

style

CompactSketch union = SetOperation.builder().buildUnion().union(
                                        readPreviousSketch(read),
                                        requireNonNull(collectedNdvSketches.get(fieldId), "collectedNdvSketches.get(fieldId) is null"));

however, if you look at the PR diff and want to avoid unnecessary changes, the code should look like this

Memory memory = Memory.wrap(ByteBuffers.getBytes(read.second())); // Memory.wrap(ByteBuffer) results in a different deserialized state
                                CompactSketch previousSketch = CompactSketch.wrap(memory);
                                CompactSketch newSketch = requireNonNull(collectedNdvSketches.get(fieldId), "collectedNdvSketches.get(fieldId) is null");
                                ndvSketches.put(fieldId, SetOperation.builder().buildUnion().union(previousSketch, newSketch));

You can view the PR diff at https://github.com/trinodb/trino/pull/16745/files

[findepi]

pushed a small change to the writer & commit message

[findepi]

thank you