Data: Add partition stats writer and reader

[ajantha-bhat]

Introduce APIs to write the partition stats into files in table default format using Iceberg generic writers and readers.

PartitionStatisticsFile partitionStatisticsFile =
        PartitionStatsHandler.computeAndWriteStatsFile(testTable, "b1");

testTable.updatePartitionStatistics().setPartitionStatistics(partitionStatisticsFile).commit();

[ajantha-bhat]

There was no option to pass the table properties before.
Needed to pass different file format for paramterized test.

[ajantha-bhat]

@aokolnychyi: This PR is ready. But as we discussed previously, this PR wraps the PartitionStats into a Record as the writers cannot work with Iceberg internal objects yet.

I will explore adding the internal writers for Parquet and Orc. Similar to #11108.
If we fail to have it ready by 1.7.0, I think it makes sense to merge this PR and introduce the optimized writer in the next version by deprecating this writer.

[ajantha-bhat]

@RussellSpitzer: It would be good to have this in 1.7.0.
I am waiting from a month for a review.

[aokolnychyi]

I think we should try to use "internal" writers. @rdblue added "internal" readers recently.

Any guidance on how to add a writer, @rdblue? We can start with Avro for now. We will also need such readers/writers for Parquet.

[ajantha-bhat]

@aokolnychyi, @rdblue:

I already tried POC for internal writers on another branch,
c209bc9

The problems:
a) I am using PartitionData instead of Record for partition value, but the PartitionData get() method wraps the byte array to the byte buffer, which is a problem for internal writers, they expect byte[]. So, I didn't felt like using a new class instead of PartitionData just for this.

b) Also, Using partitionData in StructLikeMap is not working fine. Some keys are missing in the map (looks like equals() logic), If I use Record, it is fine.

Maybe in the next version we can have optimized writer and reader (without converter using internal reader and writers).
For end user it doesn't make any difference as new readers can also read the old partition stats parquet file and old readers can read the new partition stats parquet file. So, can we merge this?

[RussellSpitzer]

Moving out of 1.7.0 since we still have a bit of discussion here

[ajantha-bhat]

@RussellSpitzer: I have added the Assertion for Partition type as you suggested and replied to #11216 (comment), do you have anymore comments for this PR?

[aokolnychyi]

I had a conversation with @rdblue today about internal writers. Ryan should have a bit of time to help/guide.
I will check the current implementation today too.

[jbonofre]

@RussellSpitzer @aokolnychyi I'm reviewing the stale PRs, and this one is open for month. Do we have a way to move forward ? I can do a new review, but at the end of the day, it won't help for the merge (as only committers can merge PR).

[deniskuzZ]

Thanks @ajantha-bhat for your work on partition stats support in Iceberg! That could be reused in Hive as a building block for apache/hive#5498

[jbonofre]

@danielcweeks @RussellSpitzer @aokolnychyi would you have some time to take a look on this PR and my proposal (previous comment) ?

[regadas]

I just found this PR as I'm desperately looking for this functionality. Thanks @ajantha-bhat! Let's see if the review gets wrapped soon 🤞

[deniskuzZ]

Wouldn't it be better not to have fat methods with multiple responsibilities? What if we introduce the write method that gets the stats iterator as an argument.
We might not need to execute the complete stats rebuild for all the registered partitions, but only for those that were changed in the current snapshot.
Snapshot summary already has a metric for the number of changed partitions, maybe we could extend it with the partition list and re-compute stats only for them. Then generate a new stats file based on the prev snapshot stats with updates to the changed partitions.

[ajantha-bhat]

What if we introduce the write method that gets the stats iterator as an argument.

We do have a package-private write method for that writePartitionStatsFile, just have to make it public.
Didn't do because what if the users don't sort the stats before, it will violate the spec.

[deniskuzZ]

hi @ajantha-bhat, could you please check below:

it seems that Date, time, timestamp partition values are not properly serialized
see Type.TypeID

DATE(Integer.class),
TIME(Long.class),
TIMESTAMP(Long.class),

PartitionSpec.partitionToPath(PartitionStatsRecord.unwrap().partition()) thows an exception

Caused by: java.lang.IllegalStateException: Not an instance of java.lang.Integer: 1999-12-31
	at org.apache.iceberg.data.GenericRecord.get(GenericRecord.java:123)
	at org.apache.iceberg.PartitionSpec.get(PartitionSpec.java:195)
	at org.apache.iceberg.PartitionSpec.partitionToPath(PartitionSpec.java:213)

I think, instead of Record(1999-12-31) it should be Record{10956}

full code snippet

Types.StructType partitionType = Partitioning.partitionType(table);
Schema schema = PartitionStatsHandler.schema(partitionType);

CloseableIterable<PartitionStatsRecord> partitionStatsRecords = PartitionStatsHandler.readPartitionStatsFile(
    schema, table.io().newInputFile(statsFile.path()));

try (Closeable toClose = partitionStatsRecords) {
  PartitionStats partitionStats = Iterables.tryFind(partitionStatsRecords, stats -> {
        PartitionSpec spec = table.specs().get(stats.unwrap().specId());
        return spec.partitionToPath(stats.unwrap().partition()).equals(partish.getPartition().getName());
      })
      .transform(PartitionStatsRecord::unwrap)
      .orNull();

  if (partitionStats != null) {
    Map<String, String> stats = ImmutableMap.of(
        TOTAL_DATA_FILES_PROP, String.valueOf(partitionStats.dataFileCount()),
        TOTAL_RECORDS_PROP, String.valueOf(partitionStats.dataRecordCount()),
        TOTAL_EQ_DELETES_PROP, String.valueOf(partitionStats.equalityDeleteRecordCount()),
        TOTAL_POS_DELETES_PROP, String.valueOf(partitionStats.positionDeleteRecordCount()),
        TOTAL_FILE_SIZE_PROP, String.valueOf(partitionStats.totalDataFileSizeInBytes())
    );
    return stats;
}

[ajantha-bhat]

@deniskuzZ: Thanks for testing out. We are working on Internal parquet/Avro/orc readers and writes. partition stats will use them. So, we don't need to go through these converters. I will retest all the data types once I use internal writers for partition stats.

PRs:
#11919
#11904

[ajantha-bhat]

Cannot have Equals and HashCode for PartitionStats class as StructLike need to have comparator for equals() which forces that class extends StructLike to hold some more things. Setting comparator while serializing and deserializing that class will be a mess.

Hence, added this util method. Currently used only by tests. But can be useful for developers when they integrate partition stats to engines, they can use it for their tests. So, kept as a util.

[ajantha-bhat]

@aokolnychyi, @rdblue, @RussellSpitzer: I have reworked on the PR to use Internal writers and readers. PR is much simpler and no need to handle those conversions now. I can rebase it once the Parquet internal writer PR is merged.

@deniskuzZ : Feel free to test the latest state. It doesn't have conversion layer. So, should behave as expected now.

[deniskuzZ]

@deniskuzZ : Feel free to test the latest state. It doesn't have conversion layer. So, should behave as expected now.

hi @ajantha-bhat, i need to include #11919, anything else?

[ajantha-bhat]

@aokolnychyi, @rdblue, @RussellSpitzer: I have worked on Internal writers, readers for Avro, parquet and PRs got merged.
I have rebased this PR to use the internal writers and readers.

So, this PR is very simple now (no converter logic) and it just writes stats to a file.

I think if we get a good review support it can be merged for 1.8.0 itself. Please take a look.
It was already reviewed before internal writers. So, I don't think much effort is needed. Thanks in advance.

[deniskuzZ]

hi @ajantha-bhat, what is the purpose of PartitionStats.totalRecordCount? it's always 0 and there is no external setter either.
Also SnapshotSummary.TOTAL_FILE_SIZE_PROP tracks all files (data + delete, see https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/SnapshotSummary.java#L288), whereas PartitionStats only data files totalDataFileSizeInBytes.
Could we extends the PartitionStats with totalFileSizeInBytes metric? I can open a PR with the change if that's ok.

[ajantha-bhat]

@deniskuzZ: While designing the spec (https://iceberg.apache.org/spec/#partition-statistics-file), we have added totalRecordCount to represent the record count after applying the delete file. It is optional field and hence not computed at the moment as it requires scanning all the data files and it can be expensive operation.

Could we extends the PartitionStats with totalFileSizeInBytes metric?

Let us wait for the merge of this PR. After that we can open the discussion to add additional stats for partition stats spec. For example some folks want min max stats also #11083.

[ajantha-bhat]

Ping.

[deniskuzZ]

test would be broken if default format changes, for example with avro format non-zero file would be created

[ajantha-bhat]

But it won't be flaky. So, we can update the test if the behavior changes. This is as per current behavior.

[deniskuzZ]

true, I've already added +1

[ajantha-bhat]

Thanks for your reviews. I hope we ship this feature soon and glad to know Hive, Trino are waiting for this feature.

[deniskuzZ]

LGTM, could we please get this merged? @pvary, would you be able to help