[ghoshp83]

Closes #4145.

Rationale for this change

PyArrowFileIO writes local (file://) files through pyarrow.fs.LocalFileSystem, whose output stream never calls fsync. The catalog then commits a pointer to the new metadata.json, so after a power loss the pointer can survive while the files it references are empty, and the table no longer loads. #4145 has the full trace and a power-cut reproduction.

This change makes a local write durable when its stream is closed, which is before any catalog commit can reference it:

• PyArrowLocalFileSystem.open_output_stream now returns _SyncOnCloseOutputStream, a pyarrow.BufferedOutputStream over a pyarrow.OSFile. On close() it flushes, fsyncs the file, and then fsyncs the parent directory so that the new entry is durable too.
• When open_output_stream has to create parent directories (for example metadata/ or data/ on the first write), it also fsyncs the parent of each directory it created.

The stream stays a native pyarrow stream with the same buffer size, so the Parquet, Avro and metadata writers are unchanged. Every caller already closes its stream explicitly or through with.

Scope, kept deliberately narrow:

• Only PyArrowFileIO's local file system changes. FsspecFileIO's local file system has the same gap and could follow separately.
• The SQLite catalog's own rollback-journal behaviour, which Local warehouse (file://): data files, manifests and metadata.json are never fsynced before the catalog commit; after a power loss the table cannot be loaded #4145 also mentions, is not touched.
• On Windows, directories cannot be opened, so only the file is synced there.
• If a caller passes compression or metadata to open_output_stream, it still gets pyarrow's own stream, unsynced. PyIceberg itself only passes buffer_size.

The cost is two fsync calls per written file, and only on a local warehouse. Object stores are not affected. I have not added a property to turn this off. If you would like one, I can add it.

Are these changes tested?

Yes. test_pyarrow_local_fs_syncs_file_and_new_directories_on_close writes through PyArrowFileIO into a path whose parent directories do not exist yet. It then checks that the file and the three directories whose entries changed were synced, by comparing the inodes of the fsynced descriptors.

• With the change reverted, the test fails on its assertion (assert [] == [...]: no fsync at all).
• tests/io/test_pyarrow.py, tests/avro, tests/catalog/test_sql.py, tests/test_serializers.py and tests/utils/test_manifest.py give the same results with and without the change, apart from the new test.
• I also traced the reproduction script from Local warehouse (file://): data files, manifests and metadata.json are never fsynced before the catalog commit; after a power loss the table cannot be loaded #4145 (create_table plus one append on a SQLite catalog) with strace. On main there is no fsync under the warehouse. With this change, the Parquet file, manifest, manifest list and metadata.json are each synced, together with their directory, before SQLite commits the new metadata_location.

Are there any user-facing changes?

There is no API change. Writes to a local warehouse are now durable once a commit returns, at the cost of the fsync calls above.

[rambleraptor]

Are you able to give any benchmarks on performance for this? This seems like a bit of an edge case and I'm worried that we're introducing latency on a per-file basis.