Merge remote-tracking branch 'upstream/main'

Merge with upstream

Author: tbirch-cyber

.github/pull_request_template.md

@@ -7,8 +7,8 @@ Thanks for opening a pull request!
 
 # Rationale for this change
 
-# Are these changes tested?
+## Are these changes tested?
 
-# Are there any user-facing changes?
+## Are there any user-facing changes?
 
 <!-- In the case of user-facing changes, please add the changelog label. -->

.pre-commit-config.yaml

@@ -39,7 +39,7 @@ repos:
         args:
           [--install-types, --non-interactive, --config=pyproject.toml]
   - repo: https://github.com/igorshubovych/markdownlint-cli
-    rev: v0.43.0
+    rev: v0.45.0
     hooks:
       - id: markdownlint
         args: ["--fix"]

README.md

@@ -21,6 +21,6 @@ PyIceberg is a Python library for programmatic access to Iceberg table metadata
 
 The documentation is available at [https://py.iceberg.apache.org/](https://py.iceberg.apache.org/).
 
-# Get in Touch
+## Get in Touch
 
 - [Iceberg community](https://iceberg.apache.org/community/)

mkdocs/docs/how-to-release.md

@@ -397,7 +397,7 @@ Ensure to update the `PYICEBERG_VERSION` in the [Dockerfile](https://github.com/
 
 ### Set up GPG key and Upload to Apache Iceberg KEYS file
 
-To set up GPG key locally, see the instructions [here](http://www.apache.org/dev/openpgp.html#key-gen-generate-key).
+To set up GPG key locally, see the [instructions](http://www.apache.org/dev/openpgp.html#key-gen-generate-key).
 
 To install gpg on a M1 based Mac, a couple of additional steps are required: <https://gist.github.com/phortuin/cf24b1cca3258720c71ad42977e1ba57>.
 

mkdocs/docs/verify-release.md

@@ -113,7 +113,7 @@ make test-coverage
 
 This will spin up Docker containers to facilitate running test coverage.
 
-# Cast the vote
+## Cast the vote
 
 Votes are cast by replying to the release candidate announcement email on the dev mailing list with either `+1`, `0`, or `-1`. For example :
 

pyiceberg/io/pyarrow.py

@@ -779,6 +779,7 @@ def visit_uuid(self, _: UUIDType) -> pa.DataType:
         return pa.uuid()
 
     def visit_unknown(self, _: UnknownType) -> pa.DataType:
+        """Type `UnknownType` can be promoted to any primitive type in V3+ tables per the Iceberg spec."""
         return pa.null()
 
     def visit_binary(self, _: BinaryType) -> pa.DataType:
@@ -1358,6 +1359,8 @@ def primitive(self, primitive: pa.DataType) -> PrimitiveType:
             primitive = cast(pa.FixedSizeBinaryType, primitive)
             return FixedType(primitive.byte_width)
         elif pa.types.is_null(primitive):
+            # PyArrow null type (pa.null()) is converted to Iceberg UnknownType
+            # UnknownType can be promoted to any primitive type in V3+ tables per the Iceberg spec
             return UnknownType()
         elif isinstance(primitive, pa.UuidType):
             return UUIDType()

pyiceberg/schema.py

@@ -1692,6 +1692,15 @@ def _(file_type: FixedType, read_type: IcebergType) -> IcebergType:
         raise ResolveError(f"Cannot promote {file_type} to {read_type}")
 
 
+@promote.register(UnknownType)
+def _(file_type: UnknownType, read_type: IcebergType) -> IcebergType:
+    # Per V3 Spec, "Unknown" can be promoted to any Primitive type
+    if isinstance(read_type, PrimitiveType):
+        return read_type
+    else:
+        raise ResolveError(f"Cannot promote {file_type} to {read_type}")
+
+
 def _check_schema_compatible(requested_schema: Schema, provided_schema: Schema) -> None:
     """
     Check if the `provided_schema` is compatible with `requested_schema`.
@@ -1761,7 +1770,15 @@ def _is_field_compatible(self, lhs: NestedField) -> bool:
                 self.rich_table.add_row("✅", str(lhs), str(rhs))
                 return True
             except ResolveError:
-                self.rich_table.add_row("❌", str(lhs), str(rhs))
+                # UnknownType can only be promoted to Primitive types
+                if isinstance(rhs.field_type, UnknownType):
+                    if not isinstance(lhs.field_type, PrimitiveType):
+                        error_msg = f"Null type (UnknownType) cannot be promoted to non-primitive type {lhs.field_type}. UnknownType can only be promoted to primitive types (string, int, boolean, etc.) in V3+ tables."
+                    else:
+                        error_msg = f"Null type (UnknownType) cannot be promoted to {lhs.field_type}. This may be due to table format version limitations (V1/V2 tables don't support UnknownType promotion)."
+                    self.rich_table.add_row("❌", str(lhs), f"{str(rhs)} - {error_msg}")
+                else:
+                    self.rich_table.add_row("❌", str(lhs), str(rhs))
                 return False
 
     def schema(self, schema: Schema, struct_result: Callable[[], bool]) -> bool:

pyiceberg/table/inspect.py

@@ -20,7 +20,7 @@
 from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Set, Tuple
 
 from pyiceberg.conversions import from_bytes
-from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestFile, PartitionFieldSummary
+from pyiceberg.manifest import DataFileContent, ManifestContent, ManifestFile, PartitionFieldSummary
 from pyiceberg.partitioning import PartitionSpec
 from pyiceberg.table.snapshots import Snapshot, ancestors_of
 from pyiceberg.types import PrimitiveType
@@ -288,64 +288,86 @@ def partitions(self, snapshot_id: Optional[int] = None) -> "pa.Table":
 
             table_schema = pa.unify_schemas([partitions_schema, table_schema])
 
-        def update_partitions_map(
-            partitions_map: Dict[Tuple[str, Any], Any],
-            file: DataFile,
-            partition_record_dict: Dict[str, Any],
-            snapshot: Optional[Snapshot],
-        ) -> None:
+        snapshot = self._get_snapshot(snapshot_id)
+        executor = ExecutorFactory.get_or_create()
+        local_partitions_maps = executor.map(self._process_manifest, snapshot.manifests(self.tbl.io))
+
+        partitions_map: Dict[Tuple[str, Any], Any] = {}
+        for local_map in local_partitions_maps:
+            for partition_record_key, partition_row in local_map.items():
+                if partition_record_key not in partitions_map:
+                    partitions_map[partition_record_key] = partition_row
+                else:
+                    existing = partitions_map[partition_record_key]
+                    existing["record_count"] += partition_row["record_count"]
+                    existing["file_count"] += partition_row["file_count"]
+                    existing["total_data_file_size_in_bytes"] += partition_row["total_data_file_size_in_bytes"]
+                    existing["position_delete_record_count"] += partition_row["position_delete_record_count"]
+                    existing["position_delete_file_count"] += partition_row["position_delete_file_count"]
+                    existing["equality_delete_record_count"] += partition_row["equality_delete_record_count"]
+                    existing["equality_delete_file_count"] += partition_row["equality_delete_file_count"]
+
+                    if partition_row["last_updated_at"] and (
+                        not existing["last_updated_at"] or partition_row["last_updated_at"] > existing["last_updated_at"]
+                    ):
+                        existing["last_updated_at"] = partition_row["last_updated_at"]
+                        existing["last_updated_snapshot_id"] = partition_row["last_updated_snapshot_id"]
+
+        return pa.Table.from_pylist(
+            partitions_map.values(),
+            schema=table_schema,
+        )
+
+    def _process_manifest(self, manifest: ManifestFile) -> Dict[Tuple[str, Any], Any]:
+        partitions_map: Dict[Tuple[str, Any], Any] = {}
+        for entry in manifest.fetch_manifest_entry(io=self.tbl.io):
+            partition = entry.data_file.partition
+            partition_record_dict = {
+                field.name: partition[pos]
+                for pos, field in enumerate(self.tbl.metadata.specs()[manifest.partition_spec_id].fields)
+            }
+            entry_snapshot = self.tbl.snapshot_by_id(entry.snapshot_id) if entry.snapshot_id is not None else None
+
             partition_record_key = _convert_to_hashable_type(partition_record_dict)
             if partition_record_key not in partitions_map:
                 partitions_map[partition_record_key] = {
                     "partition": partition_record_dict,
-                    "spec_id": file.spec_id,
+                    "spec_id": entry.data_file.spec_id,
                     "record_count": 0,
                     "file_count": 0,
                     "total_data_file_size_in_bytes": 0,
                     "position_delete_record_count": 0,
                     "position_delete_file_count": 0,
                     "equality_delete_record_count": 0,
                     "equality_delete_file_count": 0,
-                    "last_updated_at": snapshot.timestamp_ms if snapshot else None,
-                    "last_updated_snapshot_id": snapshot.snapshot_id if snapshot else None,
+                    "last_updated_at": entry_snapshot.timestamp_ms if entry_snapshot else None,
+                    "last_updated_snapshot_id": entry_snapshot.snapshot_id if entry_snapshot else None,
                 }
 
             partition_row = partitions_map[partition_record_key]
 
-            if snapshot is not None:
-                if partition_row["last_updated_at"] is None or partition_row["last_updated_snapshot_id"] < snapshot.timestamp_ms:
-                    partition_row["last_updated_at"] = snapshot.timestamp_ms
-                    partition_row["last_updated_snapshot_id"] = snapshot.snapshot_id
+            if entry_snapshot is not None:
+                if (
+                    partition_row["last_updated_at"] is None
+                    or partition_row["last_updated_snapshot_id"] < entry_snapshot.timestamp_ms
+                ):
+                    partition_row["last_updated_at"] = entry_snapshot.timestamp_ms
+                    partition_row["last_updated_snapshot_id"] = entry_snapshot.snapshot_id
 
-            if file.content == DataFileContent.DATA:
-                partition_row["record_count"] += file.record_count
+            if entry.data_file.content == DataFileContent.DATA:
+                partition_row["record_count"] += entry.data_file.record_count
                 partition_row["file_count"] += 1
-                partition_row["total_data_file_size_in_bytes"] += file.file_size_in_bytes
-            elif file.content == DataFileContent.POSITION_DELETES:
-                partition_row["position_delete_record_count"] += file.record_count
+                partition_row["total_data_file_size_in_bytes"] += entry.data_file.file_size_in_bytes
+            elif entry.data_file.content == DataFileContent.POSITION_DELETES:
+                partition_row["position_delete_record_count"] += entry.data_file.record_count
                 partition_row["position_delete_file_count"] += 1
-            elif file.content == DataFileContent.EQUALITY_DELETES:
-                partition_row["equality_delete_record_count"] += file.record_count
+            elif entry.data_file.content == DataFileContent.EQUALITY_DELETES:
+                partition_row["equality_delete_record_count"] += entry.data_file.record_count
                 partition_row["equality_delete_file_count"] += 1
             else:
-                raise ValueError(f"Unknown DataFileContent ({file.content})")
-
-        partitions_map: Dict[Tuple[str, Any], Any] = {}
-        snapshot = self._get_snapshot(snapshot_id)
-        for manifest in snapshot.manifests(self.tbl.io):
-            for entry in manifest.fetch_manifest_entry(io=self.tbl.io):
-                partition = entry.data_file.partition
-                partition_record_dict = {
-                    field.name: partition[pos]
-                    for pos, field in enumerate(self.tbl.metadata.specs()[manifest.partition_spec_id].fields)
-                }
-                entry_snapshot = self.tbl.snapshot_by_id(entry.snapshot_id) if entry.snapshot_id is not None else None
-                update_partitions_map(partitions_map, entry.data_file, partition_record_dict, entry_snapshot)
+                raise ValueError(f"Unknown DataFileContent ({entry.data_file.content})")
 
-        return pa.Table.from_pylist(
-            partitions_map.values(),
-            schema=table_schema,
-        )
+        return partitions_map
 
     def _get_manifests_schema(self) -> "pa.Schema":
         import pyarrow as pa

tests/test_schema.py

@@ -55,6 +55,7 @@
     TimestampType,
     TimestamptzType,
     TimeType,
+    UnknownType,
     UUIDType,
 )
 
@@ -920,6 +921,36 @@ def test_promotion(file_type: IcebergType, read_type: IcebergType) -> None:
             promote(file_type, read_type)
 
 
+def test_unknown_type_promotion_to_primitive() -> None:
+    """Test that UnknownType can be promoted to primitive types (V3+ behavior)"""
+    unknown_type = UnknownType()
+
+    assert promote(unknown_type, StringType()) == StringType()
+    assert promote(unknown_type, IntegerType()) == IntegerType()
+    assert promote(unknown_type, BooleanType()) == BooleanType()
+    assert promote(unknown_type, FloatType()) == FloatType()
+
+
+def test_unknown_type_promotion_to_non_primitive_raises_resolve_error() -> None:
+    """Test that UnknownType cannot be promoted to non-primitive types and raises ResolveError"""
+    unknown_type = UnknownType()
+
+    with pytest.raises(ResolveError) as exc_info:
+        promote(unknown_type, ListType(element_id=1, element_type=StringType(), element_required=False))
+
+    assert "Cannot promote unknown to list<string>" in str(exc_info.value)
+
+    with pytest.raises(ResolveError) as exc_info:
+        promote(unknown_type, MapType(key_id=1, key_type=StringType(), value_id=2, value_type=StringType(), value_required=False))
+
+    assert "Cannot promote unknown to map<string, string>" in str(exc_info.value)
+
+    with pytest.raises(ResolveError) as exc_info:
+        promote(unknown_type, StructType(NestedField(field_id=1, name="field", field_type=StringType(), required=False)))
+
+    assert "Cannot promote unknown to struct<1: field: optional string>" in str(exc_info.value)
+
+
 @pytest.fixture()
 def primitive_fields() -> List[NestedField]:
     return [