`add_files` raises `KeyError` if parquet file doe not have column stats

[binayakd]

Apache Iceberg version

0.8.0 (latest release)

Please describe the bug 🐞

Using the NYC taxi data set found here, if I follow the standard way of creating catalog, and table, but instead of doing append, I do add_files:

from pyiceberg.catalog.sql import SqlCatalog
import pyarrow.parquet as pq


warehouse_path = "/tmp/warehouse"
data_file_path = "/tmp/test-data" 

catalog = SqlCatalog(
 "default",
 **{
 "uri": f"sqlite:///{warehouse_path}/pyiceberg_catalog.db",
 "warehouse": f"file://{warehouse_path}",
 }
)

df = pq.read_table(f"{data_file_path}/yellow_tripdata_2024-01.parquet")

catalog.create_namespace("default")

table = catalog.create_table(
 "default.taxi_dataset",
 schema=df.schema,
)

table.add_files([f"{data_file_path}/yellow_tripdata_2024-01.parquet"])

I get a KeyError:

Traceback (most recent call last):
 File "/home/binayak/Dropbox/dev/tests/iceberg-test/main.py", line 42, in <module>
 main()
 File "/home/binayak/Dropbox/dev/tests/iceberg-test/main.py", line 29, in main
 table.add_files([f"{data_file_path}/yellow_tripdata_2024-01.parquet"])
 File "/home/binayak/Dropbox/dev/my-github/iceberg-python/pyiceberg/table/__init__.py", line 1036, in add_files
 tx.add_files(
 File "/home/binayak/Dropbox/dev/my-github/iceberg-python/pyiceberg/table/__init__.py", line 594, in add_files
 for data_file in data_files:
 File "/home/binayak/Dropbox/dev/my-github/iceberg-python/pyiceberg/table/__init__.py", line 1537, in _parquet_files_to_data_files
 yield from parquet_files_to_data_files(io=io, table_metadata=table_metadata, file_paths=iter(file_paths))
 File "/home/binayak/Dropbox/dev/my-github/iceberg-python/pyiceberg/io/pyarrow.py", line 2535, in parquet_files_to_data_files
 statistics = data_file_statistics_from_parquet_metadata(
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/home/binayak/Dropbox/dev/my-github/iceberg-python/pyiceberg/io/pyarrow.py", line 2400, in data_file_statistics_from_parquet_metadata
 del col_aggs[field_id]
 ~~~~~~~~^^^^^^^^^^
KeyError: 1

This is because since this parquet file does not have columns level stats sets, in the source code, it goes into the else block here
So col_aggs and null_value_counts is not updated, but invalidate_col is update. So when the del command is run here, the KeyError is thrown.

As discussed on slack, @kevinjqliu proposed to switch del col_aggs[field_id] with col_aggs.pop(field_id, None).

I will be raising a PR soon.