hassaanch23 opened a new pull request, #25654: URL: https://github.com/apache/datafusion/pull/25654
## Which issue does this PR close? - Closes #25619. Part of #25610. ## Rationale for this change A TopK that preserves partitioning keeps up to `fetch` rows from **each** partition, but its overall row estimate is capped at one partition's `fetch`. In the issue's reproducer it is estimated at 10 rows and emits 40. The cause is shared by every operator that applies its fetch per partition. Each computes overall statistics as `with_fetch(self.fetch, 0, 1)` on the child's *overall* statistics: | Operator | Fetch applies to | Overall estimate on `main` | | --- | --- | --- | | `SortExec` TopK, `preserve_partitioning=[true]` | each partition | 1 × fetch | | `LocalLimitExec` | each partition | 1 × fetch, reported as **`Exact`** | | `HashJoinExec` with a fetch | each output partition | 1 × fetch | | `CoalesceBatchesExec` (deprecated) with a fetch | each partition | 1 × fetch | This predates the current statistics API. Before #11875, `LocalLimitExec` already reported `fetch` as exact (with a comment saying it *"is not actually exact, but will be when GlobalLimit is applied"*), and multiplied by the partition count only when the input had no row estimate. ## What changes are included in this PR? A crate-private helper, `with_per_partition_fetch` in `physical-plan/src/statistics.rs`, used by the four operators: - **One partition, or one output partition:** unchanged, it is `with_fetch(fetch, 0, 1)`. - **Overall, across `n` partitions:** estimates `min(input rows, fetch × n)`. - That count is **inexact** unless the fetch cannot drop any row (input ≤ fetch). How the input is spread over the partitions is unknown: 400 rows over 4 partitions under a limit of 10 gives 40 when spread evenly, and 10 when they all sit in one partition. I didn't reuse `with_fetch`'s `n_partitions` parameter. It treats `self` as *one partition's* statistics and multiplies them; given overall statistics it would report `Exact(40)` for a skewed split, and 40 rows out of a 25-row input. The merge above a per-partition TopK still caps the estimate at the global limit: `SortPreservingMergeExec` stays at 10 (see the `explain.slt` case). ### One trade-off to call out On `main`, `SELECT * FROM t LIMIT 5` over a two-partition table with exact statistics reports `Exact(5)` at the top. That exactness only came from `LocalLimitExec` claiming `Exact(5)` for output that is really 5–10 rows. With honest local estimates, the global limit reports **`Inexact(5)`**: the count is still right, but statistics can't express the lower bound needed to prove it exact. `custom_sources_cases::statistics::sql_limit` pinned the old value; I updated it with a comment. My reasoning is that an exact count should only be claimed when it's provable, since rules such as the `COUNT(*)`-from-statistics shortcut rely on it. If you'd rather keep `LocalLimitExec` as it is for now and scope this PR to TopK and joins, that's a one-line revert and I'm happy to make it. ## What is the testing strategy for this PR? - `statistics::tests::per_partition_fetch_counts_every_partition`: every branch of the helper. - `sorts::sort::tests::test_partitioned_topk_statistics_count_every_partition`: runs a per-partition TopK over 4 × 100 rows and checks the estimate against what it emits. - `joins::hash_join::exec::tests::join_fetch_statistics_count_every_output_partition`: a `CollectLeft` join with 4 output partitions and `fetch=10` emits 40 rows, and is now estimated at `Inexact(40)`. - `limit::tests::test_row_number_statistics_for_local_limit`: asserted `Exact(10)` for 4 partitions × 100 rows under a limit of 10; now `Inexact(40)`, plus the no-drop and single-partition cases. - `explain.slt`: the issue's plan shape (grouped aggregate → per-partition TopK → merge) with `show_statistics`. TopK is `Inexact(40)`, the merge is `Inexact(10)`. All of these fail on `main`: I checked by reverting only the helper's logic and keeping the tests. A note on the TopK test: `SortExec` with a fetch builds a dynamic filter that its partitions share, so a partition can stop short of `fetch` once another partition's heap is full. The test emits 37–40 rows depending on scheduling, so it checks that range rather than exactly 40. The estimate is an upper bound on it, and the issue's reproducer disables dynamic filtering for the same reason. Full runs pass: `datafusion-physical-plan` (with `test_utils`), `datafusion-physical-optimizer`, `datafusion`'s `core_integration` and `tpcds_planning`, and all 521 sqllogictest files. ## Are there any user-facing changes? Only statistics estimates: larger row estimates for per-partition TopK, local limits, and hash joins with a fetch, plus the `Exact` → `Inexact` change described above. No public API changes; the helper is `pub(crate)`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
