Repository: dask-contrib/dask-expr (python)

Issue:
Implement hashing for FileMetaData object from Arrow
cc @rjzamora if you have thoughts

Pull request description:
Add normalize_token for pyarrow.parquet.ParquetSchema/FileMetaData
Will close #703 

Patch:
diff --git a/dask_expr/io/parquet.py b/dask_expr/io/parquet.py
index a103e60..8cd9e8c 100644
--- a/dask_expr/io/parquet.py
+++ b/dask_expr/io/parquet.py
@@ -77,6 +77,33 @@ def normalize_pa_schema(schema):
     return schema.to_string()
 
 
+@normalize_token.register(pq.ParquetSchema)
+def normalize_pq_schema(schema):
+    try:
+        return hash(schema)
+    except TypeError:  # pyarrow version not supporting ParquetSchema hash
+        return hash(repr(schema))
+
+
+@normalize_token.register(pq.FileMetaData)
+def normalize_pq_filemetadata(meta):
+    try:
+        return hash(meta)
+    except TypeError:
+        # pyarrow version not implementing hash for FileMetaData
+        # use same logic as implemented in version that does support hashing
+        # https://github.com/apache/arrow/blob/bbe59b35de33a0534fc76c9617aa4746031ce16c/python/pyarrow/_parquet.pyx#L853
+        return hash(
+            (
+                repr(meta.schema),
+                meta.num_rows,
+                meta.num_row_groups,
+                meta.format_version,
+                meta.serialized_size,
+            )
+        )
+
+
 class ToParquet(Expr):
     _parameters = [
         "frame",


Test patch:
diff --git a/dask_expr/io/tests/test_io.py b/dask_expr/io/tests/test_io.py
index f6b8e98..3bea37f 100644
--- a/dask_expr/io/tests/test_io.py
+++ b/dask_expr/io/tests/test_io.py
@@ -4,6 +4,7 @@ import os
 import dask.array as da
 import dask.dataframe as dd
 import numpy as np
+import pyarrow.parquet as pq
 import pytest
 from dask.array.utils import assert_eq as array_assert_eq
 from dask.dataframe.utils import assert_eq
@@ -22,7 +23,7 @@ from dask_expr import (
 )
 from dask_expr._expr import Expr, Lengths, Literal, Replace
 from dask_expr._reductions import Len
-from dask_expr.io import FromArray, FromMap, ReadCSV, ReadParquet
+from dask_expr.io import FromArray, FromMap, ReadCSV, ReadParquet, parquet
 from dask_expr.tests._util import _backend_library
 
 # Set DataFrame backend for this module
@@ -501,3 +502,21 @@ def test_from_dict():
     expected = pd.DataFrame(data)
     assert_eq(result, expected)
     assert_eq(DataFrame.from_dict(data), expected)
+
+
+@pytest.mark.parametrize("normalizer", ("filemetadata", "schema"))
+def test_normalize_token_parquet_filemetadata_and_schema(tmpdir, normalizer):
+    df = pd.DataFrame({c: range(10) for c in "abcde"})
+    path = tmpdir / "df.parquet"
+    df.to_parquet(path)
+
+    meta = pq.read_metadata(path)
+
+    if normalizer == "filemetadata":
+        normalizer = parquet.normalize_pq_filemetadata
+        obj = meta
+    else:
+        normalizer = parquet.normalize_pq_schema
+        obj = meta.schema
+
+    assert normalizer(obj) == normalizer(obj)