matanper opened a new issue, #70383: URL: https://github.com/apache/doris/issues/70383
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/doris/issues?q=is%3Aissue) and found no similar issues. ### Version 4.1.4. master (`41cd2b473e9`), `branch-4.1` and `4.1.5-release` have the same code: - `be/src/exprs/aggregate/aggregate_function_collect.h:137` (`AggregateFunctionCollectSetData<string>::add`) and `:153` (`::merge`) - `be/src/exprs/aggregate/aggregate_function_distinct.h:155` (`AggregateFunctionDistinctGenericData::merge`), `:199` (`AggregateFunctionDistinctSingleGenericData::add`) and `:238` (`AggregateFunctionDistinctMultipleGenericData::add`) ### Description Several aggregate states keep a hash set of `StringRef`s whose bytes live in the aggregate `Arena`. They copy each input value into the arena first and only then insert it into the set: ```cpp // aggregate_function_collect.h:135-139 (master) void add(const IColumn& column, size_t row_num, Arena& arena) { auto key = column.get_data_at(row_num); key.data = arena.insert(key.data, key.size); // copy every row data_set.insert(key); // usually a duplicate } ``` The arena can only free memory when it is destroyed, together with the aggregate states. So every duplicate row leaves a dead copy behind until the query ends. Memory grows with the number of input rows, not the number of distinct values, even though the result has one value per distinct key. Affected: | Function (SQL) | State / method | master `file:line` | |---|---|---| | `collect_set(s)`, `collect_set(s, max_size)`, `group_uniq_array` on strings | `AggregateFunctionCollectSetData<string>::add` | `aggregate_function_collect.h:137` | | same, merge / `deserialize_and_merge` | `AggregateFunctionCollectSetData<string>::merge` | `aggregate_function_collect.h:153` | | `multi_distinct_<fn>` combinator with one non-numeric argument, e.g. `multi_distinct_group_concat(s)` | `AggregateFunctionDistinctSingleGenericData::add` (stable and unstable) | `aggregate_function_distinct.h:199` | | `multi_distinct_<fn>` with several arguments, e.g. `multi_distinct_group_concat(s, sep)` | `AggregateFunctionDistinctMultipleGenericData::add` (`serialize_value_into_arena`, never released) | `aggregate_function_distinct.h:238` | | merge of either `multi_distinct_<fn>` state | `AggregateFunctionDistinctGenericData::merge` | `aggregate_function_distinct.h:155` | The `max_size` cap of `collect_set(s, max_size)` does not help: only distinct values count toward the cap, so with fewer distinct values than `max_size` every row is still copied. Other dedup states already do this right and show the pattern: `map_agg` (`aggregate_function_map.h`) calls `find` before copying, and `AggregateFunctionUniqVariant::add` (`aggregate_function_uniq.h:300-305`) serializes the key and calls `arena.rollback` when the key is a duplicate. ### Solution Look up first, copy into the arena only on insert: - single-value states (`collect_set` add/merge, `AggregateFunctionDistinctSingleGenericData::add`, `AggregateFunctionDistinctGenericData::merge`): `lazy_emplace` with a key that points at the input bytes, and make the arena copy in the constructor callback. Duplicates then cost one hash probe and no allocation; new keys still cost one probe. - multi-column keys (`AggregateFunctionDistinctMultipleGenericData::add`): the key has to be serialized to be looked up, so keep `serialize_value_into_arena`, and `arena.rollback(key.size)` when `emplace` reports a duplicate, as `AggregateFunctionUniqVariant::add` does. Results are unchanged. We have a patch with unit tests and can open a PR. ### How to Reproduce? ```sql CREATE TABLE t (k BIGINT, g INT, s STRING) DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 16 PROPERTIES ('replication_num' = '1'); INSERT INTO t SELECT number, number % 1000, concat(repeat('x', 198), lpad(cast(number % 10 AS STRING), 2, '0')) FROM numbers('number' = '5000000'); -- 5M rows, 10 distinct 200-byte strings SET parallel_pipeline_task_num = 8; SELECT size(collect_set(s, 100)) FROM t; SELECT size(collect_set(s)) FROM t; SELECT count(*) FROM (SELECT g, collect_set(s, 100) FROM t GROUP BY g) x; SELECT count(*) FROM (SELECT g, group_concat(DISTINCT s) x FROM t GROUP BY g) y; -- planned as multi_distinct_group_concat SELECT count(*) FROM (SELECT g, group_concat(DISTINCT s, '|') x FROM t GROUP BY g) y; -- multi-column distinct state ``` and compare `PeakMemoryBytes` in `fe.audit.log` (or the query profile) with the ~2 KB of distinct data. A unit-level check: feed 20,000 rows cycling over 10 distinct 201-byte strings into `AggregateFunctionCollectSetData<TYPE_STRING, false>::add` and read `arena.used_size()`: it is 4,020,000 bytes today and 2,010 bytes (one copy per distinct value) with the change. Merging the same 10-value state 1,000 times gives 2,010,000 vs 2,010 bytes. ### Anything Else? Single BE, release build, aarch64, the data above, 3 runs each (`PeakMemoryBytes` / `CpuTimeMS` from the audit log): | Query | Peak memory today | Peak memory with the change | CPU ms today -> with the change | |---|---|---|---| | `collect_set(s, 100)`, no GROUP BY | 1.61-1.76 GB | 234-242 MB | 1631-1786 -> 1341-1838 | | `collect_set(s)`, no GROUP BY | 1.48-1.55 GB | 232-248 MB | 1551-1618 -> 1169-1234 | | `collect_set(s, 100)` GROUP BY g (1000 groups) | 1.70-1.77 GB | 590-607 MB | 1589-1676 -> 1348-1411 | | `group_concat(DISTINCT s)` GROUP BY g (`multi_distinct_group_concat`) | 1.36-1.57 GB | 584-629 MB | 444-499 -> 277-345 | | `group_concat(DISTINCT s, '\|')` GROUP BY g | 1.67-1.80 GB | 595-754 MB | wall 116-128 ms -> 72-78 ms | Worst case for the change (no duplicates to save: 5M all-unique ~200-byte strings, high-cardinality GROUP BY with 1M groups, and the order-keeping `group_concat(DISTINCT s ORDER BY s)`): median CPU and peak memory are within ±5% of today, and results are identical. The remaining peak after the change is the scan and the hash table, not the aggregate states. The waste scales with rows × value size, so with longer strings or more rows per group it reaches the query memory limit long before the result is large. Checked and not affected: `multi_distinct_count` (stores 128-bit hashes), `map_agg` (find first), `collect_list`/`array_agg` with `max_size` (cap checked before the copy), `min`/`max`/`any_value`/`min_by`/`max_by` on strings (copy only on a better value, into one reusable buffer), `approx_count_distinct` (HLL). `group_array_union`/`group_array_intersect`, `topn` and `histogram` build a temporary `std::string` per row, which costs CPU but is freed immediately. ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
