Writing v3 Table Metadata

pyiceberg/table/metadata.py

@@ -630,9 +630,6 @@ def serialize_model(self, handler: ModelWrapSerializerWithoutInfo) -> dict[str,
             serialized.pop("encryption-keys", None)
         return serialized
 
-    def model_dump_json(self, exclude_none: bool = True, exclude: Any | None = None, by_alias: bool = True, **kwargs: Any) -> str:
-        raise NotImplementedError("Writing V3 is not yet supported, see: https://github.com/apache/iceberg-python/issues/1551")
-
 
 TableMetadata = Annotated[TableMetadataV1 | TableMetadataV2 | TableMetadataV3, Field(discriminator="format_version")]
 
@@ -724,6 +721,7 @@ def new_table_metadata(
             properties=properties,
             last_partition_id=fresh_partition_spec.last_assigned_field_id,
             table_uuid=table_uuid,
+            next_row_id=INITIAL_ROW_ID,
         )
     else:
         raise ValidationError(f"Unknown format version: {format_version}")

tests/integration/test_writes/test_writes.py

@@ -2374,10 +2374,10 @@ def test_nanosecond_support_on_catalog(
 
     _create_table(session_catalog, identifier, {"format-version": "3"}, schema=arrow_table_schema_with_all_timestamp_precisions)
 
-    with pytest.raises(NotImplementedError, match="Writing V3 is not yet supported"):
-        catalog.create_table(
-            "ns.table1", schema=arrow_table_schema_with_all_timestamp_precisions, properties={"format-version": "3"}
-        )
+    table_v3 = catalog.create_table(
+        "ns.table1", schema=arrow_table_schema_with_all_timestamp_precisions, properties={"format-version": "3"}
+    )
+    assert table_v3.metadata.format_version == 3
 
     with pytest.raises(
         UnsupportedPyArrowTypeException, match=re.escape("Column 'timestamp_ns' has an unsupported type: timestamp[ns]")
@@ -2387,6 +2387,28 @@ def test_nanosecond_support_on_catalog(
         )
 
 
+@pytest.mark.integration
+def test_spark_reads_v3_table_metadata_written_by_pyiceberg(spark: SparkSession, session_catalog_hive: Catalog) -> None:
+    """Spark should be able to load a V3 table whose metadata file PyIceberg wrote.
+
+    The Hive catalog is used rather than REST because REST has the server build the metadata,
+    which would leave PyIceberg's serialization untested. Rows cannot be written to a V3 table
+    yet (`_manifest_writer` rejects version 3), so this covers the metadata itself.
+    """
+    identifier = "default.test_spark_reads_v3_table_metadata_written_by_pyiceberg"
+    tbl = _create_table(session_catalog_hive, identifier, {"format-version": "3"})
+
+    assert tbl.metadata.format_version == 3
+    assert tbl.metadata.next_row_id == 0
+
+    properties = {row.key: row.value for row in spark.sql(f"SHOW TBLPROPERTIES hive.{identifier}").collect()}
+    assert properties["format-version"] == "3"
+
+    df = spark.table(f"hive.{identifier}")
+    assert df.count() == 0
+    assert df.columns == [field.name for field in tbl.schema().fields]
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize("format_version", [1, 2])
 def test_stage_only_delete(

tests/table/test_metadata.py

@@ -186,12 +186,15 @@ def test_serialize_v2(example_table_metadata_v2: dict[str, Any]) -> None:
 
 
 def test_serialize_v3(example_table_metadata_v3: dict[str, Any]) -> None:
-    # Writing will be part of https://github.com/apache/iceberg-python/issues/1551
+    table_metadata = TableMetadataV3(**example_table_metadata_v3).model_dump_json()
+    expected = """{"location":"s3://bucket/test/location","table-uuid":"9c12d441-03fe-4693-9a96-a0705ddf69c1","last-updated-ms":1602638573590,"last-column-id":3,"schemas":[{"type":"struct","fields":[{"id":1,"name":"x","type":"long","required":true}],"schema-id":0,"identifier-field-ids":[]},{"type":"struct","fields":[{"id":1,"name":"x","type":"long","required":true},{"id":2,"name":"y","type":"long","required":true,"doc":"comment"},{"id":3,"name":"z","type":"long","required":true},{"id":4,"name":"u","type":"unknown","required":true},{"id":5,"name":"ns","type":"timestamp_ns","required":true},{"id":6,"name":"nstz","type":"timestamptz_ns","required":true}],"schema-id":1,"identifier-field-ids":[1,2]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[{"source-id":1,"field-id":1000,"transform":"identity","name":"x"}]}],"default-spec-id":0,"last-partition-id":1000,"properties":{"read.split.target.size":"134217728"},"current-snapshot-id":3055729675574597004,"snapshots":[{"snapshot-id":3051729675574597004,"sequence-number":0,"timestamp-ms":1515100955770,"manifest-list":"s3://a/b/1.avro","summary":{"operation":"append"}},{"snapshot-id":3055729675574597004,"parent-snapshot-id":3051729675574597004,"sequence-number":1,"timestamp-ms":1555100955770,"manifest-list":"s3://a/b/2.avro","summary":{"operation":"append"},"schema-id":1}],"snapshot-log":[{"snapshot-id":3051729675574597004,"timestamp-ms":1515100955770},{"snapshot-id":3055729675574597004,"timestamp-ms":1555100955770}],"metadata-log":[{"metadata-file":"s3://bucket/.../v1.json","timestamp-ms":1515100}],"sort-orders":[{"order-id":3,"fields":[{"source-id":2,"transform":"identity","direction":"asc","null-order":"nulls-first"},{"source-id":3,"transform":"bucket[4]","direction":"desc","null-order":"nulls-last"}]}],"default-sort-order-id":3,"refs":{"test":{"snapshot-id":3051729675574597004,"type":"tag","max-ref-age-ms":10000000},"main":{"snapshot-id":3055729675574597004,"type":"branch"}},"statistics":[],"partition-statistics":[],"format-version":3,"last-sequence-number":34,"next-row-id":1}"""
+    assert table_metadata == expected
 
-    with pytest.raises(NotImplementedError) as exc_info:
-        _ = TableMetadataV3(**example_table_metadata_v3).model_dump_json()
 
-    assert "Writing V3 is not yet supported, see: https://github.com/apache/iceberg-python/issues/1551" in str(exc_info.value)
+def test_serialize_v3_round_trip(example_table_metadata_v3: dict[str, Any]) -> None:
+    table_metadata = TableMetadataV3(**example_table_metadata_v3)
+    reparsed = TableMetadataV3(**json.loads(table_metadata.model_dump_json()))
+    assert table_metadata == reparsed
 
 
 def test_migrate_v1_schemas(example_table_metadata_v1: dict[str, Any]) -> None:
@@ -839,6 +842,7 @@ def test_new_table_metadata_with_v3_schema() -> None:
         default_sort_order_id=1,
         refs={},
         format_version=3,
+        next_row_id=0,
     )
 
     assert actual.model_dump() == expected.model_dump()
@@ -847,6 +851,20 @@ def test_new_table_metadata_with_v3_schema() -> None:
     assert actual.sort_orders == [expected_sort_order]
 
 
+def test_new_table_metadata_v3_initializes_next_row_id() -> None:
+    """`next-row-id` is required in V3, so a new table has to start it at 0 rather than leave it unset."""
+    actual = new_table_metadata(
+        schema=Schema(NestedField(field_id=1, name="foo", field_type=StringType(), required=False)),
+        partition_spec=PartitionSpec(),
+        sort_order=SortOrder(),
+        location="s3://some_v3_location/",
+        properties={"format-version": "3"},
+    )
+
+    assert actual.next_row_id == 0
+    assert json.loads(actual.model_dump_json())["next-row-id"] == 0
+
+
 @pytest.mark.parametrize(
     "field_type",
     [
@@ -992,3 +1010,18 @@ def test_v3_metadata_with_encryption_keys_serializes_them(example_table_metadata
     assert table_metadata.model_dump(mode="json")["encryption-keys"] == [
         {"key-id": "a", "encrypted-key-metadata": "a2V5", "properties": {}}
     ]
+
+
+def test_v3_metadata_encryption_keys_round_trip_through_json(example_table_metadata_v3: dict[str, Any]) -> None:
+    encryption_keys = [
+        {"key-id": "table-key-1", "encrypted-key-metadata": "a2V5", "properties": {"a": "b"}},
+        {"key-id": "table-key-2", "encrypted-key-metadata": "a2V5", "encrypted-by-id": "table-key-1"},
+    ]
+    table_metadata = TableMetadataUtil.parse_obj({**example_table_metadata_v3, "encryption-keys": encryption_keys})
+
+    serialized = table_metadata.model_dump_json()
+    assert json.loads(serialized)["encryption-keys"] == [
+        {"key-id": "table-key-1", "encrypted-key-metadata": "a2V5", "properties": {"a": "b"}},
+        {"key-id": "table-key-2", "encrypted-key-metadata": "a2V5", "encrypted-by-id": "table-key-1", "properties": {}},
+    ]
+    assert TableMetadataUtil.parse_raw(serialized) == table_metadata