Ensure FileInfo can be serialized (#9025)

Author: fjetter

distributed/protocol/arrow.py

@@ -47,3 +47,14 @@ def deserialize_table(header, frames):
     blob = frames[0]
     reader = pyarrow.RecordBatchStreamReader(pyarrow.BufferReader(blob))
     return reader.read_all()
+
+
+@dask_serialize.register(pyarrow.fs.FileInfo)
+def serialize_fileinfo(fileinfo):
+    return {}, [(fileinfo.path, fileinfo.size, fileinfo.mtime_ns)]
+
+
+@dask_deserialize.register(pyarrow.fs.FileInfo)
+def serialize_filesystem(header, frames):
+    path, size, mtime_ns = frames[0]
+    return pyarrow.fs.FileInfo(path=path, size=size, mtime_ns=mtime_ns)

distributed/protocol/tests/test_arrow.py

@@ -46,3 +46,15 @@ def test_dumps_compression():
     msg = {"op": "update", "data": to_serialize(t)}
     result = distributed.protocol.loads(distributed.protocol.dumps(msg))
     assert result["data"].equals(t)
+
+
+def test_dump_fileinfo():
+    from pyarrow.fs import FileInfo
+
+    finfo = FileInfo(path="path", size=1, mtime_ns=2)
+    dtfinfo = deserialize(*serialize(finfo))
+    # Does not implement __equal__
+    assert type(dtfinfo) == FileInfo
+    assert dtfinfo.path == finfo.path
+    assert dtfinfo.size == finfo.size
+    assert dtfinfo.mtime_ns == finfo.mtime_ns