[C++] write_dataset: max_open_files does not close least recently used file

[xWaita]

Describe the bug, including details regarding any error messages, version, and platform.

On the latest version of pyarrow (18.0.0), according to the docs of pyarrow.dataset.write_dataset, when max_open_files is reached, the least recently used file will be closed. (https://arrow.apache.org/docs/python/generated/pyarrow.dataset.write_dataset.html)

However in the C++ code of dataset writer, what actually happens is the largest file is closed. (https://github.com/apache/arrow/blob/main/cpp/src/arrow/dataset/dataset_writer.cc#L656)

For long running dataset writes with many partitions, this means that once max open files is saturated, over time the set of open files will trend towards being smaller and smaller in size. Recently opened files can begin to be closed prematurely, while old open files can hang around and take up a file slot.

It would probably make sense for C++ behaviour to be brought in line with the docs, which should be to close the least recently used file rather than the largest file, with the rationale that the least recently used file is more likely to be finished permanently.

A solution could be to store the last write time whenever DatasetWriterDirectoryQueue::StartWrite() is called (maybe someone more knowledgeable about the codebase can suggest somewhere better), and use that to replace DatasetWriter::TryCloseLargestFile() with something like DatasetWriter::TryCloseLeastRecentlyUsedFile().

Component(s)

C++, Python

[raulcd]

FYI @mapleFU @pitrou

[pitrou]

It would probably make sense for C++ behaviour to be brought in line with the docs, which should be to close the least recently used file rather than the largest file, with the rationale that the least recently used file is more likely to be finished permanently.

Thanks for the report, and yes, I agree this would be a good idea.

[pitrou]

Hmm, the current behavior seems to be deliberate according to the PR description:

FileSystemDatasetWriteOptions now has a max_open_files setting (ARROW-12321) which prevents opening too many files. Instead the writer will apply backpressure on the scanner while also closing the open file with the greatest # of rows already written (then resume writing once the file is closed).

@westonpace Do you happen to remember why the option description doesn't match the intended behavior? Is it just a mistake in the docstring?

[mapleFU]

This idea LGTM, would you like to implement this? Or I'll check it then

[xWaita]

Thanks for the replies! I can work on it

[mapleFU]

why the option description doesn't match the intended behavior? Is it just a mistake in the docstring?

I guess maybe using same strategy when memory overlimit and fd over limit? Since when we want to reclaim memory, close the largest file ( or k largest file) is usally a good way?

[pitrou]

Before working on anything, we should decide which behavior is most desirable. If you want your dataset files to have relatively similar sizes, then it makes sense to close the largest files first.

[xWaita]

There is a secondary write_dataset parameter in max_rows_per_file that can be used to control how large each file will get.

With the current implementation of max_open_files and writing a dataset with many partitions, rather than seeing file sizes remaining similar, you'd actually see file sizes steadily trend downwards due to small partition files that remain open, while newer partition files compete to stay under the file size of those smaller partitions.

[pitrou]

With the current implementation of max_open_files and writing a dataset with many partitions, rather than seeing file sizes remaining similar, you'd actually see file sizes steadily trend downwards due to small partition files that remain open, while newer partition files compete to stay under the file size of those smaller partitions.

Ah, that may be the case if you have many partitions (more than max_open_files) that are very rarely hit?

[xWaita]

yeah exactly, and I believe this would be a very common usecase, i.e. if you were to process/write a large file where the partition column is sorted (take a date partitioned file for example), new partitions would constantly be created while old partitions would never be touched again