Add integration tests format version 3 for files metadata tables

Author: soumya-ghosh

pyiceberg/table/inspect.py

@@ -635,16 +635,12 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
     def _files(self, snapshot_id: Optional[int] = None, data_file_filter: Optional[Set[DataFileContent]] = None) -> "pa.Table":
         import pyarrow as pa
 
-        files_table: list[pa.Table] = []
-
         if not snapshot_id and not self.tbl.metadata.current_snapshot():
-            return pa.Table.from_pylist(
-                [],
-                schema=self._get_files_schema(),
-            )
-        snapshot = self._get_snapshot(snapshot_id)
+            return self._get_files_schema().empty_table()
 
+        snapshot = self._get_snapshot(snapshot_id)
         io = self.tbl.io
+        files_table: list[pa.Table] = []
         for manifest_list in snapshot.manifests(io):
             files_table.append(self._get_files_from_manifest(manifest_list, data_file_filter))
 

tests/integration/test_inspect_table.py

@@ -100,10 +100,8 @@ def _inspect_files_asserts(df: pa.Table, spark_df: DataFrame) -> None:
             assert isinstance(value.as_py(), int)
 
     for split_offsets in df["split_offsets"]:
-        assert isinstance(split_offsets.as_py(), list)
-
-    for file_format in df["file_format"]:
-        assert file_format.as_py() == "PARQUET"
+        if split_offsets.as_py() is not None:
+            assert isinstance(split_offsets.as_py(), list)
 
     for file_path in df["file_path"]:
         assert file_path.as_py().startswith("s3://")
@@ -985,3 +983,49 @@ def test_inspect_all_files(
     _inspect_files_asserts(all_files_df, spark.table(f"{identifier}.all_files"))
     _inspect_files_asserts(all_data_files_df, spark.table(f"{identifier}.all_data_files"))
     _inspect_files_asserts(all_delete_files_df, spark.table(f"{identifier}.all_delete_files"))
+
+
+@pytest.mark.integration
+def test_inspect_files_format_version_3(spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table) -> None:
+    identifier = "default.table_metadata_files"
+
+    tbl = _create_table(
+        session_catalog,
+        identifier,
+        properties={
+            "format-version": "3",
+            "write.delete.mode": "merge-on-read",
+            "write.update.mode": "merge-on-read",
+            "write.merge.mode": "merge-on-read",
+        },
+    )
+
+    insert_data_sql = f"""INSERT INTO {identifier} VALUES
+        (false, 'a', 'aaaaaaaaaaaaaaaaaaaaaa', 1, 1, 0.0, 0.0, TIMESTAMP('2023-01-01 19:25:00'), TIMESTAMP('2023-01-01 19:25:00+00:00'), DATE('2023-01-01'), X'01', X'00000000000000000000000000000000'),
+        (NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL),
+        (true, 'z', 'zzzzzzzzzzzzzzzzzzzzzz', 9, 9, 0.9, 0.9, TIMESTAMP('2023-03-01 19:25:00'), TIMESTAMP('2023-03-01 19:25:00+00:00'), DATE('2023-03-01'), X'12', X'11111111111111111111111111111111');
+    """
+
+    spark.sql(insert_data_sql)
+    spark.sql(insert_data_sql)
+    spark.sql(f"UPDATE {identifier} SET int = 2 WHERE int = 1")
+    spark.sql(f"DELETE FROM {identifier} WHERE int = 9")
+    spark.table(identifier).show(20, False)
+
+    tbl.refresh()
+
+    files_df = tbl.inspect.files()
+    data_files_df = tbl.inspect.data_files()
+    delete_files_df = tbl.inspect.delete_files()
+
+    all_files_df = tbl.inspect.all_files()
+    all_data_files_df = tbl.inspect.all_data_files()
+    all_delete_files_df = tbl.inspect.all_delete_files()
+
+    _inspect_files_asserts(files_df, spark.table(f"{identifier}.files"))
+    _inspect_files_asserts(data_files_df, spark.table(f"{identifier}.data_files"))
+    _inspect_files_asserts(delete_files_df, spark.table(f"{identifier}.delete_files"))
+
+    _inspect_files_asserts(all_files_df, spark.table(f"{identifier}.all_files"))
+    _inspect_files_asserts(all_data_files_df, spark.table(f"{identifier}.all_data_files"))
+    _inspect_files_asserts(all_delete_files_df, spark.table(f"{identifier}.all_delete_files"))