Refactor Iceberg table statistics to be deterministic

[alexjo2144]

Fixes #9716

Existing table statistics were non-deterministic because they depended on the order that data files were loaded from the Iceberg API. This hopefully cleans the code up a bit and makes it more consistent.

[findepi]

@alexjo2144 the build is red

[alexjo2144]

@findepi finally had some time to get back to this. Mind taking a look?

[findepi]

@Immutable, final

[findepi]

defensive copy

[findepi]

defensive copy

[findepi]

add empty line between immutable and mutable state

[findepi]

acceptDataFile ?

[findepi]

computeIfAbsent (without another get) gives you 0.5x map access and skips IcebergStatisticsBuilder allocation.

[findepi]

Functions.compose -> plan lambda entry -> ...

[findepi]

add // invalidate

[findepi]

You're doing 3 map lookups (contains, get, merge).
You can do this in one shot:

nullCounts.merge(id, nullCount, (existingCount, newCount) ->
        existingCount.isPresent() && newCount.isPresent() ? Optional.of(existingCount.get() + newCount.get()) : Optional.empty());

(you can extract lambda body to a method like sumOfOptionals)

[jirassimok]

This method is also equivalent to this:

nullCounts.merge(id, nullCount, (oldCount, newCount) -> 
        oldCount.flatMap(oldValue ->
                newCount.map(newValue -> newValue + oldValue)));

Though after looking at this, it's definitely less clear about the intent, though I would use this form if extracting a method to add optionals.

Actually, instead of just addition, maybe mergeOptionals would be better (I see somewhere else in this file it could be used). Here, it would be used like mergeOptionals(oldCount, newCount, Long::sum).

/**
 * Apply a function to the values in two optionals, returning an optional containing the result.
 * If either argument is empty, return empty.
 */
<A, B, C> Optional<C> mergeOptionals(Optional<A> a, Optional<B> b, BiFunction<A, B, C> mergeFunction)
{
    return a.flatMap(aa -> b.map(bb -> mergeFunction.apply(aa, bb)));
}

[findepi]

return a.flatMap(aa -> b.map(bb -> mergeFunction.apply(aa, bb)));

i thought about that, but i don't find it readable, that's why i suggested ?: use

[findepi]

where did these go to?

[alexjo2144]

Before my changes the ORC version of this table was deemed to have invalid column metrics, thus the NULL min/max/null count rows. I think that was a bug though, and the stats are the same now for both Parquet and ORC, with a few exceptions below.

[jirassimok]

I don't fully understand how this works, but overall it looks pretty good.

[jirassimok]

Should the builder's method return this?

[jirassimok]

Do these need to be declared so long before they're used?

[alexjo2144]

Any closer and they'd be inside the loop

[jirassimok]

I see partitionValue.ifPresentOrElse here to avoid Optional.get.

Or maybe just an ifPresent, leaving the updateNullCountStats call outside:

partitionValues.get(id).ifPresent(partition -> {
    // ...
    updateMinMaxStats(...);
});
updateNullCountStats(id, partitionValue.map(v -> 0).orElseGet(dataFile::recordCount));

[alexjo2144]

I see what you're getting at. I kinda like the separation as it is because there's a "this partition value is non-null" block, and a "this partition value is null" block.

[jirassimok]

I'm not a fan of these null checks.

Why not inline upperBounds and lowerBounds here as optionals (changing the signatureof convertBounds)?

Object lowerBound = convertBounds(idToTypeMapping, dataFile.lowerBounds())
    .map(bounds -> convertIcebergValueToTrino(column.type(), bounds)
    .orElse(null);

[jirassimok]

Actually, maybe the bounds should actually be Optional themselves, rather than nullable.

[jirassimok]

This method is also equivalent to this:

nullCounts.merge(id, nullCount, (oldCount, newCount) -> 
        oldCount.flatMap(oldValue ->
                newCount.map(newValue -> newValue + oldValue)));

Though after looking at this, it's definitely less clear about the intent, though I would use this form if extracting a method to add optionals.

Actually, instead of just addition, maybe mergeOptionals would be better (I see somewhere else in this file it could be used). Here, it would be used like mergeOptionals(oldCount, newCount, Long::sum).

/**
 * Apply a function to the values in two optionals, returning an optional containing the result.
 * If either argument is empty, return empty.
 */
<A, B, C> Optional<C> mergeOptionals(Optional<A> a, Optional<B> b, BiFunction<A, B, C> mergeFunction)
{
    return a.flatMap(aa -> b.map(bb -> mergeFunction.apply(aa, bb)));
}

[jirassimok]

This could be a stream.

idToMetricMap.entrySet().stream().map(...).collect(toImmutableMap(Entry::getKey, Entry::getValue))

[jirassimok]

Or make the method return an Optional (also noted in an earlier comment).

[jirassimok]

If it is, ImmutableMap will throw an exception.

[jirassimok]

This could use ifPresent, or even better, it could use map or mergeOptionals (suggested above).

this.min = this.min.map(currentMin ->
        min != null && compareTrinoValue(min, currentMin) < 0 ? newValue : currentMin);

this.min = mergeOptionals(this.min, Optional.ofNullable(min), (currentValue, newValue) ->
        compareTrinoValue(newValue, currentMin) < 0 ? newValue : currentValue);

(If you make the bound variables Optionals as I suggested in an earlier comment, then I think mergeOptionals is best here. Otherwise, I think the map version is better.)

[alexjo2144]

These miss a case where we want to invalidate the stats. For example, if the first file has stats for a column and then the second file does not, we should treat that the same as if the order is reversed.

[alexjo2144]

Comments addressed in the fixup, however I realized this doesn't work with tables that have gone through some schema evolution. Need to work on a fix for that before re-reviewing.

[findepi]

comments to the second fixup

[findepi]

Are Iceberg Types safe to use as map keys?

(my initial thought was to have List<> columnTrinoType and correlate with columns based on list index.)

[findepi]

Mapping them here is useful when you have many columns of same type, with stats.
And it only matters when # files isn't huge.

Doing this in ColumnStatistics::new is simpler and IMO sufficient, as you do lookup per column.
(originally you had lookup per column x file)

[findepi]

Why Optional?
How does empty() differ from empty map?

[findepi]

This is what breaks the schema evolution.
I notice that you didn't have this check and deemed OK.

Maybe we want if (identityPartitionFieldIds.contains(id) && partitionValues.containsKey(id)), so that we still try to take min/max from file stats?

Actually, is identityPartitionFieldIds.contains(id) important?
if partitionValues.containsKey(id) should be enough. The current table partitioning is not important when calculating the stats.

[alexjo2144]

Actually, is identityPartitionFieldIds.contains(id) important?

I was using that to proxy checking if the partition has a transform. If the partitioning is on hour(ts) we can't use the partition information to calculate max(ts), but you're right we can't use the current partitioning it needs to be the spec for that file.

[findepi]

lowerBounds, upperBounds are unnecessarily Optional. "No entry" (null) is treated the same as "no map at all".

[findepi]

Do typeToComparisonHandle.get(type) only if constructing new ColumnStatistics

[findepi]

i missed that previously -- the ColumnStatistics captures initial bounds, so on the first round the call to updateMinMax is noop.

You can skip the call in a somewhat verbose manner with Map.compute

columnStatistics.compute(id, (ignored, columnStatistics) -> {
    if (columnStatistics == null) {
        columnStatistics = new ColumnStatistics(lowerBound, upperBound, comparisonHandle);
    }
    else {
        columnStatistics.updateMinMax(lowerBound, upperBound);
    }
    return columnStatistics;
});

(or document that you're doing what you're doing currently)

[findepi]

comparisonHandle is immutable state, so fits better as first arg (as you put the field order)

[findepi]

i'd remove Optional.of().
also Optional.empty() -> "Empty"

(the fact that value is wrapped in an Optional is obvious, and doesn't need to be talked about)

[alexjo2144]

Linking a thread I started on the iceberg slack channel on how to deal with schema evolution vs missing metrics, still don't have a clear answer for how to tell the two apart though https://apache-iceberg.slack.com/archives/C025PH0G1D4/p1642177083095300

[alexjo2144]

AC thanks

[findepi]

@nineinchnick what is Pull Request Labeler / Test Report failure?
https://github.com/trinodb/trino/pull/9906/checks?check_run_id=4862862822 / https://github.com/trinodb/trino/runs/4862862822

[findepi]

@alexjo2144 please squash (sans rebase)

[findepi]

Fix bug in Iceberg partition schema evolution

this will be squashed, right?

also, should we have a test, with two partitioning transformations over a column?

[alexjo2144]

Yeah, I'll put it in a separate PR though

[alexjo2144]

#10739

[nineinchnick]

@nineinchnick what is Pull Request Labeler / Test Report failure? https://github.com/trinodb/trino/pull/9906/checks?check_run_id=4862862822 / https://github.com/trinodb/trino/runs/4862862822

@findepi the Test Report is a check added by the separate workflow that's triggered by but not associated with the PR, hence it doesn't have its own workflow header and looks like it belongs to the Pull Request Labeler workflow. This is Github's UI issue. This check contains annotations with a summary of failures in jobs of the ci workflow, so you don't have to go into single jobs and search for failures.

It's marked as failed, because there were other failures. Marking it as successful would be a false positive if it contained error annotations.

[findepi]

@nineinchnick could it be attached to the ci flow somehow?
it's confusing to see Pull Request Labeler fail.

[alexjo2144]

Squashed. Thanks @findepi

[vincentpoon]

We have millions of files and find SELECT * from table$partitions hangs in the acceptDataFile here.

[findepi]

@vincentpoon this PR probably didn't change how the stats are calculated, just made the code saner & "more correct"
or, do you mean this is a behavioral regression what you're observing?

in any case, let's have an issue