2010YOUY01 opened a new pull request, #15355: URL: https://github.com/apache/datafusion/pull/15355
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> Related to https://github.com/apache/datafusion/pull/14975 ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> ### What's the inefficiency Let's walkthrough an example, there is one external sort query with 1 partition, and sort exec has: - 10MB memory limit - 1MB `sort_spill_reservation_bytes` (see configuration explanation in https://datafusion.apache.org/user-guide/configs.html) During the execution: 1. `SortExec` will read in batches, and 10MB memory limit is reached 2. It will sorted all buffered batches in-place, and merge them at once. Note only the 1MB buffer from 10MB total limit is pre-reserved for merging, so there is only 1MB available to store the merged output. 3. After we have collected 1MB of merged batch, one spill will be triggered. And this 1MB space will be cleared, the merging can continue. **Inefficency:** Now `ExternalSorter` will create a new spill file for those 1MB merged batches, after spilling all intermediates, all spilled files will be merged at once, then there are too many files to merge. **Ideal case:** All batches in a single sorted run can be incrementally appended to a single file. #### Reproducer Execute datafusion-cli with `cargo run --profile release-nonlto -- --mem-pool-type fair -m 10M` ```sql set datafusion.execution.sort_spill_reservation_bytes = 1000000; set datafusion.execution.target_partitions = 4; explain analyze select * from generate_series(1, 1000000) as t1(v1) order by v1; ``` Main: 10 spills PR: 2 spills ### Rationale for the fix Introduced a new spill interface `SpillManager` with the ability of incrementally appending batches to a written file. 1. `SpillManager` is designed to do `RecordBatch <---> raw file`, and configurations can be put inside `SpillManager` to control how do we do the serialization physically for future optimizations. Example configurations: - General purpose compression like `lz4` - Specialized encoding other than the current `Arrow IPC` , or configurations to change the `IPC Writer` behavior - See `datafusion-comet`'s proprietary serde implementation in https://github.com/apache/datafusion-comet/pull/1190 - One example of extra configuration can be https://github.com/apache/datafusion/issues/15320 2. `SpillManager` is not responsible for holding spilled files inside, because the logical representation of those files can vary, I think it's clearer to place those raw files inside spilling operators. For example, `vec<RefCountedTempFile>` is managed inside `SortExec`, the implicit rule is within each file all entries are sorted by the sort keys, also in `Comet`'s `ShuffleWriterExec`, each partition should maintain one in-progress file. If we keep those tempfiles inside `SpillManager`, it's hard to clearly define those implicit requirements. 3. Additionally, `SpillManager` is responsible for updating related statistics, the spill-related metrics should be the same across operators, so this part of of code can also be reused. Also, total disk usage limit for spilled files can be easily implemented upon it. ### Why refactor and introduce `SpillManager` This fix can be implemented without a major refactor. However, this change is included to prepare for supporting disk limits for spilling queries, as described in https://github.com/apache/datafusion/pull/14975 ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> 1. Group spilling related metrics into one struct 2. Introduce `SpillManager` 3. Update `SortExec` to use the new `SpillManager` interface ### TODO: - [ ] There are two extra operators that can be changed to this new interface (`Aggregate` and `SortMergeJoin`), they're planned to be included in this PR. I plan to do it after getting some review feedback. ## Are these changes tested? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 5. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> For the too-many-spills issue: one test case is updated, and more comment is added above the assertion to prevent regression. For `SpillManager`: unit tests are included. ## Are there any user-facing changes? No. <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> <!-- If there are any breaking changes to public APIs, please add the `api change` label. --> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org For additional commands, e-mail: github-h...@datafusion.apache.org
