[core] Add required Iceberg Field IDs to support ID-based column pruning

[muttcg]

Purpose

Linked issue: close #5875

Apache Iceberg specification for Avro requires to have field-ids in order to support ID-based column pruning.

For example Google Big Query engine as well as PyIceberg don't work at all without field-ids causing critical issues:

Error while reading data, error message: The Apache Avro library failed to read data with the following error: Cannot resolve: { "type": "record", "name": "r2_null_value_counts", "fields": [ { "name": "key", "type": "int" }, { "name": "value", "type": "long" } ] } with { "type": "record", "name": "k121_v122", "fields": [ { "name": "key", "type": "int", "field-id": 121 }, { "name": "value", "type": "long", "field-id": 122 } ] }; Failed to dispatch pruner query for ***.example.flair_paimon.meta original query id: *** with new query_id: *** File: bigstore/*****-test/table_v2/iceberg/example/flair_paimon/metadata/7d8b21d4-e536-49cd-bc1c-ee68b154c178-m8.avro

The change affects only Iceberg Avro schema creation, bringing new Iceberg custom properties following the specification, as well as fixing the wrong ID for partitions values.

Tests

All existing tests must work. Added org.apache.paimon.iceberg.IcebergCompatibilityTest.testIcebergAvroFieldIds test to cover all required ids.

API and Format

No

Documentation

[LsomeYeah]

Thanks a lot for your contribution! Looks good to me!

Additionally, I was wondering if you could try reading the data using PyIceberg after this change? I ran some tests locally and found the following:

1. compress strategy

• snappy: When the manifest's compression strategy is set to snappy, I encountered ValueError: Checksum failure when reading manifests.
• null: When the manifest's compression strategy is null, the manifests can be read correctly without any issues.

2. file format

• avro: I ran into errors when trying to read Avro-formatted data files. This seems to be due to PyArrow not supporting the reading of Avro files.
• parquet: when the manifest's compression strategy is null and the data file's format is Parquet, I was able to successfully read the data using PyIceberg.

[muttcg]

@LsomeYeah Thank you for checking
We found and fixed another bug in PyIceberg apache/iceberg-python#2252
Initially we tested with 'deflate', as I understand by default nobody uses avro compression for Iceberg manifests or so

[LsomeYeah]

@LsomeYeah Thank you for checking We found and fixed another bug in PyIceberg apache/iceberg-python#2252 Initially we tested with 'deflate', as I understand by default nobody uses avro compression for Iceberg manifests or so

Thanks for your reply.
The default compression for Iceberg manifest files in Paimon is Snappy. Perhaps we should consider changing the default compression for iceberg manifest file later. CC @JingsongLi

[LsomeYeah]

+1

[muttcg]

@LsomeYeah
I have re-checked Google BQ and Polars(uses PyIceberg and PyArrow under the hood) using fix and deflate, and everything works as I expect. Examples I used:

BQ query:

SELECT * 
FROM `***.example.flair_paimon_iceberg_v7`
WHERE partition_date = "2024-07-01"
AND partition_hh = "17"
AND app_version = '10.335.0'
ORDER BY country DESC
LIMIT 1000

Polars query:

import polars as pl

table_path = "gs://***/table_v2/iceberg/example/flair_paimon_iceberg_v7/metadata/v3.metadata.json"

df = (
  pl.scan_iceberg(table_path)
  .filter(
    (pl.col("partition_date") == "2024-07-01") &
    (pl.col("partition_hh") == "17") &
    (pl.col("app_version") == "10.335.0")
  )
  .sort("country", descending=True)
  .limit(1000)
  .collect()
)

print(df)

[brunsgaard]

I am excited about this PR, thanks @muttcg and @LsomeYeah for looking into this ❤️ 🥇

[JingsongLi]

Thanks @muttcg and @LsomeYeah , +1