perf: re-used build cursors - #3

* # This is a combination of 3 commits.
# This is the 1st commit message:

Add Display for Expr::BinaryExpr

# This is the commit message #2:

Update logical_plan/operators tests

# This is the commit message #3:

rebase and debug display for non binary expr

* Add Display for Expr::BinaryExpr

Update logical_plan/operators tests

rebase and debug display for non binary expr

Add Display for Expr::BinaryExpr

Update logical_plan/operators tests

Updating tests

Update aggregate display

Updating tests without aggregate

More tests

Working on agg/scalar functions

Fix binary_expr in create_name function and attendant tests

More tests

More tests

Doc tests

Rebase and update new tests

* Submodule update

* Restore submodule references from master

Co-authored-by: Andrew Lamb <[email protected]>

* Initial commit

* initial commit

* failing test

* table scan projection

* closer

* test passes, with some hacks

* use DataFrame (#2)

* update README

* update dependency

* code cleanup (#3)

* Add support for Filter operator and BinaryOp expressions (#4)

* GitHub action (#5)

* Split code into producer and consumer modules (#6)

* Support more functions and scalar types (#7)

* Use substrait 0.1 and datafusion 8.0 (#8)

* use substrait 0.1

* use datafusion 8.0

* update datafusion to 10.0 and substrait to 0.2 (#11)

* Add basic join support (#12)

* Added fetch support (#23)

Added fetch to consumer

Added limit to producer

Added unit tests for limit

Added roundtrip_fill_none() for testing when None input can be converted to 0

Update src/consumer.rs

Co-authored-by: Andy Grove <[email protected]>

Co-authored-by: Andy Grove <[email protected]>

* Upgrade to DataFusion 13.0.0 (#25)

* Add sort consumer and producer (#24)

Add consumer

Add producer and test

Modified error string

* Add serializer/deserializer (#26)

* Add plan and function extension support (#27)

* Add plan and function extension support

* Removed unwraps

* Implement GROUP BY (#28)

* Add consumer, producer and tests for aggregate relation

Change function extension registration from absolute to relative anchor
(reference)

Remove operator to/from reference

* Fixed function registration bug

* Add test

* Addressed PR comments

* Changed field reference from mask to direct reference (#29)

* Changed field reference from masked reference to direct reference

* Handle unsupported case (struct with child)

* Handle SubqueryAlias (#30)

Fixed aggregate function register bug

* Add support for SELECT DISTINCT (#31)

Add test case

* Implement BETWEEN (#32)

* Add case (#33)

* Implement CASE WHEN

* Add more case to test

* Addressed comments

* feat: support explicit catalog/schema names in ReadRel (#34)

* feat: support explicit catalog/schema names in ReadRel

Signed-off-by: Ruihang Xia <[email protected]>

* fix: use re-exported expr crate

Signed-off-by: Ruihang Xia <[email protected]>

Signed-off-by: Ruihang Xia <[email protected]>

* move files to subfolder

* RAT

* remove rust.yaml

* revert .gitignore changes

* tomlfmt

* tomlfmt

Signed-off-by: Ruihang Xia <[email protected]>
Co-authored-by: Daniël Heres <[email protected]>
Co-authored-by: JanKaul <[email protected]>
Co-authored-by: nseekhao <[email protected]>
Co-authored-by: Ruihang Xia <[email protected]>

…4544)

* add mut annotation

* fix rust examples

* fix rust examples

* update

* fix first doctest

* fix first doctest

* fix more doctest

* fix more doctest

* fix more doctest

* adopt rustdoc syntax

* adopt rustdoc syntax

* adopt rustdoc syntax

* fix more doctest

* add missing imports

* final udtf

* reenable

* remove dep

* run prettier

* api-health

* update doc

* update doc

* temp fix

* fix doc

* fix async schema provider

* fix async schema provider

* fix doc

* fix doc

* reorder

* refactor

* s

* finish

* minor update

* add missing docs

* add deps (#3)

* fix doctest

* update doc

* fix doctest

* fix doctest

* tweak showkeys

* fix doctest

* fix doctest

* fix doctest

* fix doctest

* update to use user_doc

* add rustdoc preprocessing

* fix dir

* revert to original doc

* add allocator

* mark type

* update

* fix doctest

* add doctest

* add doctest

* fix doctest

* fix doctest

* fix doctest

* fix doctest

* fix doctest

* fix doctest

* fix doctest

* fix doctest

* fix doctest

* prettier format

* revert change to datafusion-testing

* add apache header

* install cmake in setup-builder for ci workflow dependency

* taplo + fix snmalloc

* Update function docs

* preprocess user-guide

* Render examples as sql

* fix intro

* fix docs via script

---------

Co-authored-by: Andrew Lamb <[email protected]>

… default null ordering (apache#16963)

* Allow to set up the default null ordering (#3)

* add default_null_ordering config

* add test for different config

* Update datafusion/sql/src/planner.rs

Co-authored-by: Copilot <[email protected]>

* update doc

* fix sqllogictest

---------

Co-authored-by: Copilot <[email protected]>

* rename config and fix test

* update doc

* fix default

* fix doc

* fix sqllogictests

* address comments

---------

Co-authored-by: Copilot <[email protected]>

…TINCT) group (apache#24857)

A grouped `COUNT(DISTINCT <string>)` over 4,000 groups holding 2 short
strings each needs a 36 MB memory budget. It needs 2.0 MB after this
change.

Every group gets its own hash table, and each table is allocated at
warm-up size before the group holds anything, so the memory the query
needs tracks the number of groups rather than the amount of data. The
query also reports less memory than it holds, so a memory limit does not
stop it at the right point.

## Reproduction

This needs only `datafusion-cli`. There is no patch, no custom allocator
and no data file.

```sql
-- repro.sql
SET datafusion.execution.target_partitions = 1;

-- 4,000 groups with 2 distinct short strings in each.
-- avg() stops SingleDistinctToGroupBy from rewriting the distinct aggregate away.
SELECT g, count(DISTINCT s) AS d, avg(p) AS a
FROM (
  SELECT v % 4000 AS g, 'v' || CAST(v AS VARCHAR) AS s, v AS p
  FROM generate_series(0, 7999) AS t(v)
)
GROUP BY g
ORDER BY g
LIMIT 3;
```

```
datafusion-cli -m 8M -f repro.sql
```

`s` is a `Utf8View` column, so this exercises `ArrowBytesViewMap`. The
8,000 rows arrive in one batch, so the aggregate builds all 4,000
accumulators before it can emit or spill.

Current `main` at `20d1c56761` fails:

```
Resources exhausted: Additional allocation failed for SingleHashAggregateStream[0] with top memory
consumers (across reservations) as:
  DataFusion-Cli#1(can spill: false) consumed 0.0 B, peak 0.0 B,
  SingleHashAggregateStream[0]#2(can spill: true) consumed 0.0 B, peak 48.0 B,
  TopK[0]#3(can spill: false) consumed 0.0 B, peak 0.0 B.
Error: Failed to allocate additional 111.1 MB for SingleHashAggregateStream[0] with 0.0 B already
allocated for this reservation - 8.0 MB remain available for the total memory pool:
greedy(used: 0.0 B, pool_size: 8.0 MB)
```

This branch returns the rows:

```
+---+---+--------+
| g | d | a      |
+---+---+--------+
| 0 | 2 | 2000.0 |
| 1 | 2 | 2001.0 |
| 2 | 2 | 2002.0 |
+---+---+--------+
3 row(s) fetched.
Elapsed 0.006 seconds.
```

Those 4,000 accumulators hold 8,000 short strings, which is about 100 KB
of data. The base asks the pool for 111.1 MB to hold it. This branch
runs the same query inside `-m 3M`. Both builds return the same rows,
and the base does so at `-m 200M`. Each run takes well under a second,
and the outcome repeats exactly over three runs on each side.

## How much it improves

The minimum memory limit at which that query completes, bisected on each
side:

| value column | before | after |
| --- | --- | --- |
| `Utf8` | fails 34 MB, passes 36 MB | fails 1.8 MB, passes 2.0 MB |
| `Utf8View` | fails 120 MB, passes 124 MB | fails 2.4 MB, passes 2.6 MB
|

`clickbench_extended` at `DATAFUSION_RUNTIME_MEMORY_LIMIT: 4G`, pool
peak over six runs:

| query | base | this branch | change |
| --- | --- | --- | --- |
| Q2, grouped, 4 string distincts | 98.1 to 98.6 MiB | 11.7 MiB in all
six runs | -88.1% |
| Q1, ungrouped string distincts | 3.4 MiB | 2.4 MiB | -29.4% |
| Q0, ungrouped, high cardinality | 796.8 to 834.8 MiB | 846.5 to 885.5
MiB | +6.3% |

Q2 is the only query in any benchmark suite that puts a grouped
`COUNT(DISTINCT)` on a non-integer column.

Q0 costs more, and it is the one disclosed cost of this PR. Those extra
bytes are memory Q0 always held and the pool could not see, not new
allocation; appendix A has the three-build decomposition that separates
the two. Latency does not move anywhere, which is what an
allocation-sizing change should do.

## Which issue does this PR close?

No existing issue. I found this when I investigated a production out of
memory. I can file an issue if you want it in the changelog.

## Rationale for this change

Pre-allocating is right for the one long-lived map behind a `GROUP BY`
on a string column. It is wrong for the distinct-count accumulators,
because `GroupsAccumulatorAdapter` creates one accumulator per group and
most groups hold a handful of values. There the warm-up dwarfs the data.

Both maps also under-report the table they hold. `ArrowBytesViewMap`
left the control bytes out. `ArrowBytesMap` charged the table only when
it grew, so a map that stayed inside its pre-allocation reported its
table as free forever. A memory limit acted on a number that was too
small.

`clear_shrink` is the third part. The aggregate stream calls it to hand
memory back before it spills and before a downstream sort. It restored
the warm-up capacity instead of releasing it, so nothing came back.

## What changes are included in this PR?

- `new` on both maps allocates nothing. A new `with_capacity` keeps the
previous pre-allocating behavior. `GroupValuesBytes` and
`GroupValuesBytesView` use `with_capacity`, and the two distinct-count
accumulators use `new`. A map remembers how it was built, so `take`
warms it back up the way it started.
- `size()` reports `HashTable::allocation_size()`, the real hashbrown
allocation including the control bytes, in place of the old estimate.
- A new `clear_and_release` drops every allocation the map holds, and
`clear_shrink` calls it.
- The value buffer rounds each growth up to a power of two. A lazily
grown buffer and a pre-allocated one then sit on one ladder, so a lazy
map is never the larger of the two for the same values. Growth stays
geometric.
- `benches/arrow_bytes_map.rs` moves to `with_capacity` so it keeps
measuring the pre-allocating constructor.

## What is the testing strategy for this PR?

Two tests in `datafusion/core/tests/memory_limit/mod.rs`,
`group_by_count_distinct_utf8` and `group_by_count_distinct_utf8_view`,
turn the headline claim into a pass or a fail rather than a number. They
run the reproduction query over 4,000 groups with spilling disabled and
`target_partitions` pinned to 1, so completing means the query fits the
budget rather than spills out of it. The limits are 8 MB and 16 MB, at
least 4x clear of both cliffs in the table above. Both tests fail on the
merge base and pass here, over five consecutive runs. The `avg(payload)`
in the query is load bearing; appendix C says why.

Unit tests cover what the memory-limit tests cannot see: that `new`
allocates nothing, that `with_capacity` reports a table size bracketed
by an independently derived lower bound, that `take` preserves the
configured capacity, that `clear_shrink` drops the reported size to near
zero, and that a lazily grown buffer never exceeds a pre-allocated one
holding the same values.

Run locally on the rebased head, all passing:
`datafusion-physical-expr-common` (87 lib, 8 doc),
`datafusion-functions-aggregate-common` (49),
`datafusion-functions-aggregate -- count_distinct` (2),
`datafusion-physical-plan -- group_values` (96) and the `memory_limit`
module (39, which includes the `count_distinct_spill` test that arrived
on `main` in apache#24888 and apache#24918). `cargo fmt --check` and `cargo clippy
--all-targets -D warnings` are clean on the changed crates. CI has not
yet run this branch against the new base.

No query results change.

## Are there any user-facing changes?

Yes, in `datafusion-physical-expr-common`. `ArrowBytesMap::new` and
`ArrowBytesViewMap::new` no longer pre-allocate, and callers that want
the previous behavior should use the new `with_capacity`. Both types
also gain `clear_and_release`. This changes an existing public
constructor rather than adding one, so tell me if you would like the
`api change` label.

For users, a grouped `COUNT(DISTINCT)` on string and binary columns uses
much less memory and reports its usage to the `MemoryPool` accurately. A
query that previously hit a memory limit may now succeed.

---

## Appendix A: Query 0 costs 6.3% more

Q0 is `COUNT(DISTINCT)` over three high-cardinality strings with no
`GROUP BY`. It is a handful of maps that each grow to millions of
entries, which is the opposite population from the one this PR targets.
The pre-allocation was never the dominant cost there, so removing it
buys nothing.

Over six runs the base spans 796.8 to 834.8 MiB and this branch spans
846.5 to 885.5 MiB. The ranges do not overlap, so the effect is real and
not run-to-run noise.

Three local builds on a deterministic subset separate the two changes.
The middle build differs from the base only in the accounting, because
restoring the warm-up makes the constructors byte-identical to base:

| build | Q0 pool peak |
| --- | --- |
| base | 48,421,820 |
| this branch with the warm-up restored | 51,048,396 |
| this branch | 51,018,828 |

That decomposes the increase exactly. +2,626,576 is the accounting
correction: `allocation_size()` charges the real hashbrown allocation,
which is `4 * buckets + 5,384` more than the old formula, being the
control bytes plus the 7/8 load-factor slack. -29,568 is the lazy
constructor, which makes Q0 slightly better.

Reverting the accounting would restore an under-report of about 19% on
this path. That under-report is the bug this PR exists to fix, and the
memory-limit result above depends on fixing it.

## Appendix B: what one accumulator costs

One per-group accumulator holding a single 24-byte value:

| | before, actual | before, reported | after |
| --- | --- | --- | --- |
| `BytesDistinctCountAccumulator` | 14,648 B | 8,240 B | 180 B |
| `BytesViewDistinctCountAccumulator` | 33,920 B | 28,792 B | 260 B |

The middle column is the reporting gap. The `Utf8` map really held
14,648 bytes and reported 8,240, because the whole hash table was
invisible to the old accounting.

These are measured directly rather than asserted in a test, since the
exact numbers follow the hashbrown layout.

## Appendix C: notes on the tests and the benchmarks

The memory-limit query uses `avg(payload)`, not `count(*)`. A
non-distinct `count` lets `SingleDistinctToGroupBy` rewrite the distinct
aggregate into a plain two-stage `GROUP BY`. The per-group accumulators
would then never exist, and the tests would pass by construction on the
base commit too. That rule accepts a non-distinct `sum`, `min` or `max`
because each re-aggregates its own partial results correctly over the
deduplicated inner group by. `avg` does not, so the rule can never admit
it under any extension, including the one apache#24859 proposes.

The benchmark figures were measured against the previous merge base
`da89c7c85b`. They have not been re-run against the current base
`20d1c56761`. The commits after `84f07da` on this branch touch only
`datafusion/core/tests/memory_limit/mod.rs`, so nothing on this branch
since then can move a benchmark, but the base itself has moved.

Pool peak is the instrument here, not peak RSS. Pool peak reproduces to
under 1% on a 98 MiB query and exactly on the 3.4 and 11.7 MiB ones.
Peak RSS on this harness has a 4.1% standard deviation over 11 readings
of identical code plus a 2.3% order bias, and shows no effect from this
change once that null is accounted for.

## Follow-ups, not in this PR

- The same undercount remains at five other production
`insert_accounted` call sites: `group_values/row.rs:171`,
`multi_group_by/mod.rs:434,554`, `multi_group_by/dictionary.rs:197,584`
and `array_agg.rs:989`. Each is one map per query, so the absolute error
is bounded, and the fix is the same one-line swap.
- The `count_distinct_groups` benchmarks in
`datafusion/functions-aggregate/benches/count_distinct.rs` cover
`Int64`, `Int32` and `UInt32` only, so this path has no criterion
coverage.

---------

Co-authored-by: Claude Opus 5 <[email protected]>