Read orc file

Author: MehulBatra

dev/provision.py

@@ -342,3 +342,19 @@
            (array(), map(), array(struct(1)))
     """
     )
+
+    spark.sql(
+        f"""
+    CREATE TABLE {catalog_name}.default.test_read_orc (
+        number integer,
+        letter string
+    )
+    USING iceberg
+    TBLPROPERTIES (
+        'format-version'='2',
+        'write.format.default'='orc'
+    );
+    """
+    )
+
+    spark.sql(f"INSERT INTO {catalog_name}.default.test_read_orc VALUES (1,'hello')")

pyiceberg/io/pyarrow.py

@@ -627,6 +627,8 @@ def expression_to_pyarrow(expr: BooleanExpression) -> pc.Expression:
 def _get_file_format(file_format: FileFormat, **kwargs: Dict[str, Any]) -> ds.FileFormat:
     if file_format == FileFormat.PARQUET:
         return ds.ParquetFileFormat(**kwargs)
+    elif file_format == FileFormat.ORC:
+        return ds.OrcFileFormat()
     else:
         raise ValueError(f"Unsupported file format: {file_format}")
 
@@ -972,8 +974,11 @@ def _task_to_table(
     name_mapping: Optional[NameMapping] = None,
 ) -> Optional[pa.Table]:
     _, _, path = PyArrowFileIO.parse_location(task.file.file_path)
-    arrow_format = ds.ParquetFileFormat(pre_buffer=True, buffer_size=(ONE_MEGABYTE * 8))
     with fs.open_input_file(path) as fin:
+        if task.file.file_format == FileFormat.PARQUET:
+            arrow_format = ds.ParquetFileFormat(pre_buffer=True, buffer_size=(ONE_MEGABYTE * 8))
+        if task.file.file_format == FileFormat.ORC:
+            arrow_format = ds.OrcFileFormat()  # currently ORC doesn't support any fragment scan options
         fragment = arrow_format.make_fragment(fin)
         physical_schema = fragment.physical_schema
         file_schema = pyarrow_to_schema(physical_schema, name_mapping)

tests/integration/test_reads.py

@@ -537,3 +537,12 @@ def another_task() -> None:
 
         table.transaction().set_properties(lock="xxx").commit_transaction()
         assert table.properties.get("lock") == "xxx"
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("session_catalog")])
+def test_pyarrow_read(catalog: Catalog) -> None:
+    table_orc = catalog.load_table("default.test_read_orc")
+    arrow_table = table_orc.scan(row_filter="number > -1", selected_fields=("number", "letter")).to_arrow()
+    assert len(arrow_table) == 1
+    assert arrow_table["number"][0].as_py() == 1