PyIceberg: ORC file format support

[alaturqua]

Feature Request / Improvement

We use ORC file format to store our iceberg tables on azure storage.

Currently PyIceberg supports parquet format but not ORC.

This is a request to have ORC file format support in PyIceberg.

tbl.location()
---> [18](vscode-notebook-cell:/c%3A/Projects/pandas_snowflake/notebooks/pyiceberg_test.ipynb#W0sZmlsZQ%3D%3D?line=17) tbl.scan().to_pandas()

File C:\Projects\incubator-iceberg\python\pyiceberg\table\__init__.py:409, in DataScan.to_pandas(self, **kwargs)
    408 def to_pandas(self, **kwargs: Any) -> pd.DataFrame:
--> 409     return self.to_arrow().to_pandas(**kwargs)

File C:\Projects\incubator-iceberg\python\pyiceberg\table\__init__.py:404, in DataScan.to_arrow(self)
    401 def to_arrow(self) -> pa.Table:
    402     from pyiceberg.io.pyarrow import project_table
--> 404     return project_table(
    405         self.plan_files(), self.table, self.row_filter, self.projection(), case_sensitive=self.case_sensitive
    406     )

File C:\Projects\incubator-iceberg\python\pyiceberg\io\pyarrow.py:558, in project_table(tasks, table, row_filter, projected_schema, case_sensitive)
    551 projected_field_ids = {
    552     id for id in projected_schema.field_ids if not isinstance(projected_schema.find_type(id), (MapType, ListType))
    553 }.union(extract_field_ids(bound_row_filter))
    555 with ThreadPool() as pool:
    556     tables = [
    557         table
...
File c:\Python\Python39\lib\site-packages\pyarrow\_parquet.pyx:1227, in pyarrow._parquet.ParquetReader.open()

File c:\Python\Python39\lib\site-packages\pyarrow\error.pxi:100, in pyarrow.lib.check_status()

ArrowInvalid: Parquet magic bytes not found in footer. Either the file is corrupted or this is not a parquet file.

Query engine

None

[Fokko]

@alaturqua thanks for raising this issue.

Some steps on how I would add ORC support. First I would refactor the code:

iceberg/python/pyiceberg/io/pyarrow.py

Lines 484 to 519 in 35692a3

	with fs.open_input_file(path) as fout:
	parquet_schema = pq.read_schema(fout)
	schema_raw = None
	if metadata := parquet_schema.metadata:
	schema_raw = metadata.get(ICEBERG_SCHEMA)
	if schema_raw is None:
	raise ValueError(
	"Iceberg schema is not embedded into the Parquet file, see https://github.com/apache/iceberg/issues/6505"
	)
	file_schema = Schema.parse_raw(schema_raw)
	pyarrow_filter = None
	if bound_row_filter is not AlwaysTrue():
	translated_row_filter = translate_column_names(bound_row_filter, file_schema, case_sensitive=case_sensitive)
	bound_file_filter = bind(file_schema, translated_row_filter, case_sensitive=case_sensitive)
	pyarrow_filter = expression_to_pyarrow(bound_file_filter)
	file_project_schema = prune_columns(file_schema, projected_field_ids, select_full_types=False)
	if file_schema is None:
	raise ValueError(f"Missing Iceberg schema in Metadata for file: {path}")
	arrow_table = pq.read_table(
	source=fout,
	schema=parquet_schema,
	pre_buffer=True,
	buffer_size=8 * ONE_MEGABYTE,
	filters=pyarrow_filter,
	columns=[col.name for col in file_project_schema.columns],
	)
	# If there is no data, we don't have to go through the schema
	if len(arrow_table) > 0:
	return to_requested_schema(projected_schema, file_project_schema, arrow_table)
	else:
	return None

By removing the pq package from there, and read in a PyArrow fragment instead:

import pyarrow.dataset as ds
from pyarrow.fs import S3FileSystem

with fs.open_input_file(path) as fout:
    # First get the fragment
    fragment = parquet_format.make_fragment(fout, None)
    print(f"Schema: {fragment.physical_schema}")
    arrow_table = ds.Scanner.from_fragment(
        fragment=fragment,
        ... # add other arguments as well
    ).to_table()

We need to add logic to decide the Format based on the extension of the file, we can read in different fragments:

• https://arrow.apache.org/docs/python/generated/pyarrow.dataset.ParquetFileFormat.html
• https://arrow.apache.org/docs/python/generated/pyarrow.dataset.OrcFileFormat.html

The fragment has this option to say physicial_schema that will return the schema in PyArrow.

Note: For Parquet we actually fetch the Iceberg schema in JSON format from the metadata currently. #6505 Will provide the logic to convert a PyArrow schema into an Iceberg schema.

[JonasJ-ap]

I would like to share a quick test on reading field metadata from ORC file (from an iceberg table created on AWS Athena engine v3) by pyarrow. It seems that the pa.Schema of the ORC file does not contain any field_id information even though they do exist in the ORC file:

I first use orc-tools to inspect the metadata of the ORC file and find that the field id is stored in iceberg.id:

orc-tools meta /Users/jonasjiang/.CMVolumes/gluetestjonas/warehouse/iceberg_ref/athena_orc_test/data/30f712c8/creation_date=2023-03-11/user_id_bucket=2/20230311_203248_00007_2vuin-316269af-62cb-4bef-8f52-ca2d46bc6397.orc
log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Processing data file /Users/jonasjiang/.CMVolumes/gluetestjonas/warehouse/iceberg_ref/athena_orc_test/data/30f712c8/creation_date=2023-03-11/user_id_bucket=2/20230311_203248_00007_2vuin-316269af-62cb-4bef-8f52-ca2d46bc6397.orc [length: 1104]
Structure for /Users/jonasjiang/.CMVolumes/gluetestjonas/warehouse/iceberg_ref/athena_orc_test/data/30f712c8/creation_date=2023-03-11/user_id_bucket=2/20230311_203248_00007_2vuin-316269af-62cb-4bef-8f52-ca2d46bc6397.orc
File Version: 0.12 with TRINO_ORIGINAL by Trino 
Rows: 1
Compression: ZSTD
Compression size: 262144
Calendar: Julian/Gregorian
Type: struct<user_id:int,user_name:string,status:string,cost:double,quantity:int,quantity_big:int,creation_date:date,created_at:timestamp,inserted_at:timestamp>
Attributes on root.user_id
  iceberg.id: 1
  iceberg.required: false
Attributes on root.user_name
  iceberg.id: 2
  iceberg.required: false
Attributes on root.status
  iceberg.id: 3
  iceberg.required: false
Attributes on root.cost
  iceberg.id: 4
  iceberg.required: false
Attributes on root.quantity
  iceberg.id: 5
  iceberg.required: false

However, when I used the similar way in #7033 to get the pyarrow schema, the field id info get lost:

orc_test_path = "/Users/jonasjiang/.CMVolumes/gluetestjonas/warehouse/iceberg_ref/athena_orc_test/data/30f712c8/creation_date=2023-03-11/user_id_bucket=2/20230311_203248_00007_2vuin-316269af-62cb-4bef-8f52-ca2d46bc6397.orc"
fs = LocalFileSystem()
with fs.open_input_file(orc_test_path) as f:
    # print(po.ORCFile(f).read().schema)
    print(ds.OrcFileFormat().make_fragment(f).physical_schema)

user_id: int32
user_name: string
status: string
cost: double
quantity: int32
quantity_big: int32
creation_date: date32[day]
created_at: timestamp[ns]
inserted_at: timestamp[ns]
-- schema metadata --
presto_query_id: '20230311_203248_00007_2vuin'
trino.writer.version: '0.215-16024-g6eec71f'
presto_version: '0.215-16024-g6eec71f'

The pyarrow schema containing field id info should look like:

user_id: int32
  -- field metadata --
  PARQUET:field_id: '1'
user_name: string
  -- field metadata --
  PARQUET:field_id: '2'
status: string
  -- field metadata --
  PARQUET:field_id: '3'
...

It seems even with #6997 merged, we still cannot fully support the ORC file format if we use pyarrow to read it. The name mapping feature may need to be implemented here for conversion from pyarrow schema to iceberg schema.

[Fokko]

@JonasJ-ap I came to the same conclusion. I think this will require some upstream PyArrow patches.

[Fokko]

I also noticed the following:

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../pyiceberg/table/__init__.py:404: in to_arrow
    return project_table(
../pyiceberg/io/pyarrow.py:586: in project_table
    for table in pool.starmap(
/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/pool.py:372: in starmap
    return self._map_async(func, iterable, starmapstar, chunksize).get()
/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/pool.py:771: in get
    raise self._value
/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/pool.py:125: in worker
    result = (True, func(*args, **kwds))
/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/pool.py:51: in starmapstar
    return list(itertools.starmap(args[0], args[1]))
../pyiceberg/io/pyarrow.py:511: in _file_to_table
    physical_schema = fragment.physical_schema
pyarrow/_dataset.pyx:1004: in pyarrow._dataset.Fragment.physical_schema.__get__
    ???
pyarrow/error.pxi:144: in pyarrow.lib.pyarrow_internal_check_status
    ???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

>   ???
E   pyarrow.lib.ArrowTypeError: Unknown Orc type kind: timestamp with local time zone

pyarrow/error.pxi:123: ArrowTypeError

Some polishing to do for ORC

[Fokko]

Moved this issue to the new repository: apache/iceberg-python#20 This was blocked by a change on the Arrow side, but that's in now.