Read ORC file based Iceberg table and Integration test for the same

Author: MehulBatra

dev/provision.py

@@ -346,6 +346,8 @@
     spark.sql(
         f"""
     CREATE TABLE {catalog_name}.default.test_read_orc (
+        dt     date,
+        ts     timestamp,
         number integer,
         letter string
     )
@@ -357,4 +359,11 @@
     """
     )
 
-    spark.sql(f"INSERT INTO {catalog_name}.default.test_read_orc VALUES (1,'hello')")
+    spark.sql(f"""INSERT INTO {catalog_name}.default.test_read_orc
+            VALUES
+            (CAST('2022-03-01' AS date), CAST('2022-03-01 01:22:00' AS timestamp), 1, 'a'),
+            (CAST('2022-03-02' AS date), CAST('2022-03-02 02:22:00' AS timestamp), 2, 'b'),
+            (CAST('2022-03-03' AS date), CAST('2022-03-02 02:22:00' AS timestamp), 3, 'c')
+            """)
+
+    spark.sql(f"DELETE FROM {catalog_name}.default.test_read_orc WHERE number = 3")

pyiceberg/io/pyarrow.py

@@ -168,6 +168,7 @@
 ICEBERG_SCHEMA = b"iceberg.schema"
 # The PARQUET: in front means that it is Parquet specific, in this case the field_id
 PYARROW_PARQUET_FIELD_ID_KEY = b"PARQUET:field_id"
+PYARROW_ORC_FIELD_ID_KEY = b"iceberg.id"
 PYARROW_FIELD_DOC_KEY = b"doc"
 LIST_ELEMENT_NAME = "element"
 MAP_KEY_NAME = "key"
@@ -801,11 +802,12 @@ def primitive(self, primitive: pa.DataType) -> T:
 
 
 def _get_field_id(field: pa.Field) -> Optional[int]:
-    return (
-        int(field_id_str.decode())
-        if (field.metadata and (field_id_str := field.metadata.get(PYARROW_PARQUET_FIELD_ID_KEY)))
-        else None
-    )
+    if field.metadata and (field_id_str := field.metadata.get(PYARROW_ORC_FIELD_ID_KEY)):
+        return int(field_id_str.decode())
+    elif field.metadata and (field_id_str := field.metadata.get(PYARROW_PARQUET_FIELD_ID_KEY)):
+        return int(field_id_str.decode())
+    else:
+        return None
 
 
 class _HasIds(PyArrowSchemaVisitor[bool]):
@@ -914,6 +916,9 @@ def primitive(self, primitive: pa.DataType) -> PrimitiveType:
                     return TimestamptzType()
                 elif primitive.tz is None:
                     return TimestampType()
+            if primitive.unit == "ns":
+                if primitive.tz == "UTC":
+                    return TimestamptzType()
         elif pa.types.is_binary(primitive) or pa.types.is_large_binary(primitive):
             return BinaryType()
         elif pa.types.is_fixed_size_binary(primitive):

tests/integration/test_reads.py

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 # pylint:disable=redefined-outer-name
-
+import datetime
 import math
 import time
 import uuid
@@ -541,8 +541,10 @@ def another_task() -> None:
 
 @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:
+def test_pyarrow_read_orc(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
+    arrow_table = table_orc.scan(row_filter="number > -1", selected_fields=("number", "letter", "dt")).to_arrow()
+    assert len(arrow_table) == 2
     assert arrow_table["number"][0].as_py() == 1
+    assert arrow_table["letter"][1].as_py() == "b"
+    assert arrow_table["dt"][0].as_py() == datetime.date(2022, 3, 1)