Add next-row-id default and Spark integration test

Author: rambleraptor

pyiceberg/table/metadata.py

@@ -64,6 +64,7 @@
 
 INITIAL_SEQUENCE_NUMBER = 0
 INITIAL_SPEC_ID = 0
+INITIAL_ROW_ID = 0
 DEFAULT_SCHEMA_ID = 0
 
 SUPPORTED_TABLE_FORMAT_VERSION = 2
@@ -652,6 +653,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

@@ -2386,6 +2386,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

@@ -840,6 +840,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()
@@ -848,6 +849,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",
     [