[amengnew]

What does this PR do?

Fixes #8743: ParquetDatasetWriter._write only closes the pyarrow.parquet.ParquetWriter on the successful completion path. If a batch query/write or metadata persistence raises, the writer is left unclosed while the outer file handle is closed by write() — leaking the writer and its underlying file handles.

Change

Move the write loop and metadata persistence into a try/finally block so writer.close() always runs, then return written. No behavior change on the success path.

Tests

Added test_parquet_writer_closes_writer_on_failure, which injects a mid-batch write_table failure and asserts both that:

• the original exception propagates, and
• writer.close() is still called.

Verified locally: 55 passed in tests/io/test_parquet.py (the 2 remaining errors are pre-existing HF-token/network fixture requirements, unrelated to this change).

AI assistance disclosure: this contribution was prepared with AI assistance (fix design, tests, and description); I reviewed and verified the result locally.

[amengnew]

I found a problem with the regression test I added in this PR and pushed a fix for it, so it's worth flagging.

pyarrow's ParquetWriter.__del__ calls close() during garbage collection. My mock recorded close() calls, so once the failed writer became garbage the mock reported a close anyway — the test passed even with the fix reverted. In other words, it was not testing anything.

The updated test keeps the created writer instances alive so __del__ cannot run before the assertion, and it now covers both variants of write():

• path — the writer opens the file via fsspec
• borrowed_buffer — the caller passes a buffer

Verified both directions locally: with the fix the test passes, with src/datasets/io/parquet.py reverted to the previous version both parametrizations fail on the missing close().

tests/io/test_parquet.py: 58 passed, ruff check/ruff format --check clean.