Fix iceberg lockfile update test failures

Author: cmpadden

libraries/dagster-iceberg/pyproject.toml

@@ -152,6 +152,9 @@ dev = [
   "dagit>=1.8.8",
   "dagster>=1.10.5",
   "dagster-polars[patito]>=0.26.1",
+  # dagster-polars releases before the universal-pathlib 0.3 compatibility fix
+  # still read UPath._kwargs via Dagster's UPathIOManager in the kitchen-sink tests.
+  "universal-pathlib<0.3",
   "dagster-pyspark>=0.26.1",
   "docker",
   "fsspec[http]>=2025.2.0",

libraries/dagster-iceberg/src/dagster_iceberg/_utils/io.py

@@ -118,6 +118,14 @@ def table_writer(
             )
         partition_dimensions = table_slice.partition_dimensions
     logger.debug("Partition dimensions: %s", partition_dimensions)
+    if partition_dimensions is not None:
+        for partition_dimension in partition_dimensions:
+            if partition_dimension.partition_expr not in data.schema.names:
+                raise ValueError(
+                    f"Could not find field '{partition_dimension.partition_expr}' "
+                    "in data schema. Partition columns must be present in the data."
+                )
+
     if table_exists(catalog, table_path):
         logger.debug("Updating existing table")
         table = catalog.load_table(table_path)

libraries/dagster-iceberg/tests/_utils/test_io.py

@@ -1,5 +1,4 @@
 import datetime as dt
-import pathlib as plb
 from uuid import uuid4
 
 import pyarrow as pa
@@ -285,7 +284,6 @@ def test_table_writer_multi_partitioned_update(
 
 def test_table_writer_multi_partitioned_update_partition_spec_change(
     namespace: str,
-    warehouse_path: str,
     catalog: Catalog,
     data: pa.Table,
 ):
@@ -334,23 +332,13 @@ def test_table_writer_multi_partitioned_update_partition_spec_change(
         partition_spec_update_mode="update",
         dagster_run_id="hfkghdgsh467374828",
     )
-    path_to_dwh = (
-        plb.Path(warehouse_path)
-        / f"{namespace}.db"
-        / table_
-        / "data"
-        / "part_timestamp=2023-01-01-00"
-    )
-    categories = sorted([p.name for p in path_to_dwh.glob("*") if p.is_dir()])
-    assert categories == [
-        "part_category=A",
-        "part_category=B",
-        "part_category=C",
+    table = catalog.load_table(identifier_)
+    assert [field.name for field in table.spec().fields] == [
+        "part_timestamp",
+        "part_category",
     ]
-    assert (
-        len(catalog.load_table(identifier_).scan().to_arrow().to_pydict()["value"])
-        == 1440
-    )
+    assert set(table.scan().to_arrow().to_pydict()["category"]) == {"A", "B", "C"}
+    assert len(table.scan().to_arrow().to_pydict()["value"]) == 1440
 
 
 def test_table_writer_multi_partitioned_update_partition_spec_error(

libraries/dagster-iceberg/tests/_utils/test_partitions.py

@@ -672,26 +672,29 @@ def test_existing_table_partition_names_unchanged(
         schema=iceberg_table_schema,
     )
 
-    # Manually add a partition field with old-style naming (same as column name)
+    # Manually add an identity partition field with old-style naming (same as column name).
+    # PyIceberg no longer allows transformed partition field names to conflict with
+    # schema column names, but identity partitions using the source column name are still
+    # valid legacy specs.
     with table.update_spec() as update:
         update.add_field(
-            source_column_name="timestamp",
-            transform=transforms.HourTransform(),
-            partition_field_name="timestamp",  # Old style: same as column name
+            source_column_name="category",
+            transform=transforms.IdentityTransform(),
+            partition_field_name="category",  # Old style: same as column name
         )
 
     table.refresh()
     original_field_name = table.spec().fields[0].name
-    assert original_field_name == "timestamp"
+    assert original_field_name == "category"
 
     # Now run update with no actual changes (same partition dimensions)
     table_slice = TableSlice(
         table=table_name,
         schema=namespace,
         partition_dimensions=[
             TablePartitionDimension(
-                "timestamp",
-                TimeWindow(dt.datetime(2023, 1, 1), dt.datetime(2023, 1, 1, 1)),
+                "category",
+                ["A"],
             ),
         ],
     )
@@ -709,7 +712,7 @@ def test_existing_table_partition_names_unchanged(
     assert len(table.spec().fields) == 1
     assert (
         table.spec().fields[0].name == original_field_name
-    )  # Should still be "timestamp"
+    )  # Should still be "category"
 
 
 def test_partition_field_naming_avoids_column_conflicts(

libraries/dagster-iceberg/tests/test_resource.py

@@ -63,10 +63,10 @@ def test_resource(
     @asset
     def read_table(pyiceberg_table: IcebergTableResource):
         table_ = pyiceberg_table.load().scan().to_arrow()
-        assert (
-            table_.schema.to_string()
-            == "timestamp: timestamp[us]\ncategory: large_string\nvalue: double"
-        )
+        assert table_.schema.names == ["timestamp", "category", "value"]
+        assert str(table_.schema.field("timestamp").type) == "timestamp[us]"
+        assert str(table_.schema.field("category").type) in {"string", "large_string"}
+        assert str(table_.schema.field("value").type) == "double"
         assert table_.shape == (1440, 3)
 
     materialize(