Read: fetch file_schema directly from pyarrow_to_schema

[HonahX]

#584 (comment)

If we truly read by Field-IDs the names should be irrelevant, so we should probably update our mapping to ensure we correctly project by IDs.

I think we do correctly project by IDs. The real problem is the way that we sanitize the column names.
In #83, we add sanitize the file_schema in _task_to_table with the assumption that the column name of the parquet file follows the Avro Naming spec. However, I think the "sanitization" should be more general here: it should just ensure that the final file_project_schema contains the same column names of the parquet file's schema.

The names in file_schema are different from the actual column names in the parquet file because we first try to load the file schema from the json string stored in the parquet file metadata: link

file_schema = (
            Schema.model_validate_json(schema_raw) if schema_raw is not None else pyarrow_to_schema(physical_schema, name_mapping)
        )

Parquet files written by iceberg java contain this metadata json string. The json string represents the iceberg table schema at the time of writting the file. Therefore, it contains un-sanitized column names.

Since we always need to run a visitor to sanitize/ensure column names match, how about we just get the file_schema directly from the pyarrow physical schema?

file_schema = pyarrow_to_schema(physical_schema, name_mapping)

This way, we can ensure that the column names match, and thus do not need to sanitize the column names later.

I have verified that changing to this can fix both the sanitization issue in #83 and the issue here. Given that we want to align the writing behavior with the java implementation, we should also proceed #590.

Borrowed the integration test from #590

[kevinjqliu]

LGTM!

I want to summarize my understanding, based on the comment from #584 .

When reading the parquet files, we use the projected version of the parquet file's schema, the arrow table created is then
"casted" the Iceberg's schema. This mapping is based on field_id

iceberg-python/pyiceberg/io/pyarrow.py

Line 1026 in 5039b5d

return to_requested_schema(projected_schema, file_project_schema, arrow_table)

[Fokko]

My initial intent was that it was probably faster to deserialize the schema, rather than run the visitor, but this shows is not worth the additional complexity :)

[HonahX]

Thanks @kevinjqliu and @Fokko for reviewing! Thanks @kevinjqliu for the integration test!