1. Block optimization when OFFSET > 0 — the optimizer now checks
   `LimitStep::getOffset` and bails out, preventing incorrect results
   for `LIMIT K OFFSET N` queries.

2. Eliminate per-row arena growth in key serialization — replaced
   `serializeValueIntoArena` (which allocated into the arena on every row)
   with a pure `SipHash`-128 key. Arena is now only used for aggregate
   state allocation, bounded by number of groups.

3. Preserve full `SortColumnDescription` — the optimizer now copies the
   original `nulls_direction` and `collator` from the `SortingStep`
   instead of rebuilding with defaults. Mode 2's `compareAt` uses the
   correct `nulls_direction` for NULL ordering.

4. Document mode-2 as intentional v1 simplification — the header comment
   now states that threshold-based pruning is not yet implemented.

5. Add test coverage for OFFSET (negative case) and NULLS FIRST/LAST
   (correctness comparison between optimized and unoptimized paths).

Made-with: Cursor

1. Collision-safe group keys: Replace hash-only `std::string` key with
   `UInt128` SipHash + exact column comparison via `unordered_multimap`.
   On hash match, compares actual key column values against stored groups
   in `result_columns`, eliminating the theoretical collision risk.

2. Collation-aware compare in Mode 2: `generateMode2` now checks for a
   `collator` in `SortColumnDescription` and uses `compareAtWithCollation`
   when present, ensuring correct ordering for string ORDER BY with
   explicit collation (e.g., COLLATE 'en').

3. Mode 1 safety for argMin/argMax/any: Add `output_ordered_by_sort_key`
   flag to `TopKAggregateInfo`. Only min/max set it to true (their output
   is monotonically related to the sort key). The optimizer now only
   enables Mode 1 (early termination with sorted input) when this flag is
   true, preventing incorrect results for argMin/argMax where the output
   column differs from the sort key.

4. Extended test coverage:
   - Collation-sensitive ordering (COLLATE 'en') optimized vs reference
   - Mode 1 EXPLAIN PLAN + correctness with small LIMIT
   - argMin/argMax ASC/DESC correctness parity
   - Tie-heavy dataset (100 groups, deterministic values)
   - Multi-aggregate (max + argMax) correctness
   - EXPLAIN PLAN for argMin optimization

Made-with: Cursor

The `.cpp` was out of sync with the `.h` header due to a file-write
tool failure in the previous session. The header declared the new
collision-safe API (`hashGroupKey`, `findGroupIndex`,
`unordered_multimap<UInt128, size_t, UInt128Hash>`), but the `.cpp`
still contained the old `serializeGroupKey` returning `std::string`
with hash-only identity.

This commit makes `.cpp` consistent with `.h`:

- `hashGroupKey`: returns `UInt128` SipHash digest of key columns.
- `findGroupIndex`: on hash match, performs exact column-by-column
  comparison against stored keys in `result_columns` via `compareAt`,
  eliminating the hash-collision correctness risk.
- Both `consumeMode1` and `consumeMode2` now use the new
  `unordered_multimap::emplace` / `equal_range` API.

Made-with: Cursor

Made-with: Cursor

1. Collision-safe group identity: replace `HashMap<UInt128, size_t>`
   (SipHash-only key) with `HashMapWithSavedHash<std::string_view, size_t>`
   backed by `IColumn::serializeValueIntoArena`. This gives exact key
   comparison with automatic arena rollback for non-new keys via
   `SerializedKeyHolder`, eliminating the theoretical hash-collision
   correctness risk.

2. Safe threshold pruning: remove the per-chunk row-level threshold
   update from `consumeMode2`. Row-level K-th values can be stricter
   than the true group-level K-th aggregate, which would incorrectly
   filter rows needed by groups in the final top K. The threshold is
   now updated only from group-level aggregates in
   `generateMode2Partial`, which is provably safe (a partial's K-th
   group aggregate is always <= the global K-th).

3. Fix Mode 2 gating: remove the `group_by_matches_sorting_prefix`
   check that blocked Mode 2 even when Mode 1 was inapplicable (e.g.
   GROUP BY matches the sorting key but the ORDER BY aggregate argument
   is a different column). The `!sorted_input` guard already suffices.

4. Add documentation comments explaining why Mode 2 is intentionally
   not applied as a generic unsorted fallback, and noting that Mode 2
   memory is O(unique groups) without in-stream eviction.

5. Add stress test for large composite keys, skewed ties, and parallel
   Mode 2 threshold pruning.

Made-with: Cursor

The test uses `COLLATE 'en'` which requires ICU. The fast test
build is compiled without ICU, causing `SUPPORT_IS_DISABLED` error.

Made-with: Cursor

The static analyzer cannot prove that `enable_threshold_pruning` being
true guarantees `order_arg_col` is non-null. Check the pointer directly
instead of the boolean flag to satisfy clang-analyzer-core.NonNullParamChecker.

Made-with: Cursor

- Setting description for `topn_aggregation_pruning_level` now explicitly
  states that level 2 requires `use_top_k_dynamic_filtering` to inject
  the prewhere, and that level 0 is not recommended (slower than baseline).

- Mode 2 now requires `pruning_level >= 1` to activate. Level 0 (direct
  compute without threshold) is known to regress vs the standard Aggregator
  pipeline, so it should not silently activate.

- EXPLAIN reference for level 0 updated: now shows the standard
  Aggregating + Sorting + Limit pipeline (no TopNAggregating).

Made-with: Cursor

Pass 0 (no limit) to `MergingSortedTransform` instead of the user's
LIMIT K. K refers to the number of distinct groups, not total rows.
With the previous code, if K consecutive rows belonged to the same group,
MergingSorted would stop after K rows and the downstream transform would
see fewer than K distinct groups.

The backpressure from `TopNAggregatingTransform::consumeMode1` calling
`finishConsume` is sufficient to stop reading once K groups are found.

Made-with: Cursor

Header: regroup members into logical sections (configuration, column
indices, state layout, per-group state, Mode 1/2 columns, threshold).

Implementation: extract shared helpers to eliminate duplication:
- `computeStateLayout`: state offset/alignment (was duplicated in both
  constructors)
- `findOrderByAggIndex`: ORDER BY aggregate lookup (was duplicated in
  both constructors and `initColumnIndices`)
- `sortAndLimitColumns`: sort-permute-limit pattern (was triplicated in
  `generateMode2`, `generateMode2Partial`, merge `generate`)
- `prepareArgColumnPtrs`: arg column pointer setup (was duplicated in
  `consumeMode1` and `consumeMode2`)

Also make `consumeMode1` use the shared aggregate state helpers
(`createAggregateStates`/`addRowToAggregateStates`/etc.) instead of
inline state management, and deduplicate the merge loop in
`TopNAggregatingMergeTransform::consume`.

Net: -20 lines, fewer copy-paste surfaces for future changes.
Made-with: Cursor

Avoid large-`LIMIT` regressions by adding a conservative `topn_aggregation_max_limit` gate, and add `EXPLAIN` coverage that verifies default fallback and explicit override behavior. Also reduce threshold refresh overhead in `TopNAggregatingTransform` with adaptive cadence and document `GroupState` intent for future per-group metadata.

Made-with: Cursor

Add `buildThresholdKeepMask` to `TopNAggregatingTransform` which uses
`IColumn::compareColumn` for vectorized threshold filtering in Mode 2
instead of per-row `isBelowThreshold` calls. This amortizes virtual
call overhead and enables SIMD-friendly comparison paths.

Add "Design tradeoffs" and "Future improvements" sections to
`topngroupby.md` covering: serialized-key HashMap vs type-dispatched
tables, threshold refresh cost and adaptive backoff, dynamic filter
effectiveness by data distribution, partial top-K correctness, and
planned improvements (type-dispatched tables, group eviction,
mixed-aggregate two-pass, distributed support, etc.).

Made-with: Cursor

Made-with: Cursor

- Fix `03470_topn_aggregation_parallel_replicas` test failure in "old
  analyzer" CI config: move `allow_experimental_analyzer = 1` from
  subquery SETTINGS to session-level SET, because changing this setting
  inside a subquery is disallowed when the top-level value differs.

- Fix potential overflow in `maybeRefreshThreshold`: replace
  `groups > limit * 10000` with `groups / 10000 > limit` to avoid
  wrapping when `limit` is large.

ClickHouse#98607

Co-Authored-By: Claude <noreply@anthropic.com>

…into murphy_issue_75098

The monolithic `TopNAggregatingTransform` handled both sorted-input
early-termination (Mode 1) and hash-based direct aggregation (Mode 2)
with runtime dispatch.  These modes share little logic beyond column
index mapping and aggregate state lifecycle.

Split into three classes:

- `TopNAggregatingTransformBase` -- base class with shared infrastructure
  (column indices, key serialization, aggregate state helpers)
- `TopNSortedAggregatingTransform` -- Mode 1: sorted input, stops after
  K distinct groups
- `TopNDirectAggregatingTransform` -- Mode 2: hash aggregation with
  optional threshold pruning, supports partial output for parallel workers

`TopNAggregatingStep` now dispatches to the appropriate transform class
at construction time, which is natural since it already knows the mode.
`TopNAggregatingMergeTransform` is unchanged.

Made-with: Cursor

- Fix NaN direction hint: use `nulls_direction` from `sort_description`
  consistently in `getSortPermutation`, `isBelowThreshold`, and
  `buildThresholdKeepMask` instead of `sort_direction`, matching the
  final output sorting semantics.

- Add Mode 1 collation guard: reject sorted-input early termination
  when `sort_desc[0].collator` is set, because physical MergeTree
  order may not match collation order.

- Clarify `topn_aggregation_pruning_level` docs: level 0 disables
  Mode 2 entirely (not "direct compute only"), only Mode 1 can apply.

- Split `03467_topn_aggregation` test into correctness file (allows
  random settings) and EXPLAIN file (keeps `no-random-settings`);
  drop `no-fasttest` tag.

Made-with: Cursor

- Add `chassert` to verify column type consistency between input and
  boundary columns in `isBelowThreshold` and `buildThresholdKeepMask`.

- Guard `refreshThresholdFromStates` against `limit==0` to prevent
  unsigned underflow in `std::min(limit, n) - 1`.

- Add missing "max + any" reference queries in all three test files
  to verify companion aggregate correctness.

- Add negative EXPLAIN cases for `min + DESC` (wrong direction) and
  multi-column `ORDER BY` to improve optimizer coverage.

- Reset `allow_suspicious_low_cardinality_types` after table creation
  to avoid session pollution in test files.

- Fix misleading "ties" comment (aggregate values are actually distinct).

Made-with: Cursor

- Fix `TopKThresholdTracker::compareFields` to treat NaN the same as
  column-level code: NaN is ordered via `nulls_direction` instead of
  being treated as equal to all values by `Field::operator<`. This
  prevents the shared threshold from disagreeing with per-row pruning
  in `TopNDirectAggregatingTransform` and `FunctionTopKFilter`.

- Reject Mode 1 (sorted-input early termination) when the determining
  aggregate argument is nullable. MergeTree physical NULL ordering may
  differ from the query's `NULLS FIRST`/`NULLS LAST`, so early
  termination after K groups could return wrong results.

- Expand `optimize_topn_aggregation` setting description to document
  Mode 1 vs Mode 2 behavior, eligibility requirements, and current
  limitations (nullable, collation, float/NaN caveats).

- Add regression tests for Float64 NaN with explicit `NULLS FIRST`
  and `NULLS LAST` under Mode 2, and for nullable-primary-key tables
  under Mode 1.

Made-with: Cursor

Route floating-point types through the `executeGeneral` path in
`FunctionTopKFilter` instead of the vectorized `greaterOrEquals` /
`lessOrEquals` fast path. The vectorized comparisons return `false`
for NaN, which can incorrectly drop rows that the transform-level
pruning (using `compareColumn` with `nulls_direction`) would keep.

Add regression tests for Float64 NaN at `topn_aggregation_pruning_level = 2`
with `use_top_k_dynamic_filtering = 1`.

Made-with: Cursor

Document the TopN aggregation optimization in the GROUP BY reference
page: Mode 1 vs Mode 2 behavior, eligibility requirements, current
limitations, example query, and related settings.

Made-with: Cursor

The correctness test uses `COLLATE 'en'` which requires ICU, not
available in the fast test environment.

Made-with: Cursor

… columns

The flaky check with random settings (e.g. `max_block_size=1`) can produce
`ColumnConst` or other wrapped column types that aggregate functions cannot
handle via `assert_cast`. Use `convertToFullIfNeeded` (which handles Const,
Replicated, Sparse, and LowCardinality wrappers) instead of only converting
LowCardinality, matching what the standard `Aggregator` does.

Made-with: Cursor

- Unwrap key columns with `convertToFullIfNeeded` (like aggregate arg
  columns) so that `insertFrom` receives concrete column types instead
  of `ColumnConst`/`ColumnSparse` wrappers that cause `assert_cast`
  failures. Applied to Mode 1, Mode 2, and merge transform.

- Add try/catch rollback in `createAggregateStates` matching the
  standard `Aggregator` pattern: if `create` throws for aggregate j,
  destroy already-created states 0..j-1.

- Add try/catch in Mode 1 `consume` around temporary state usage so
  `destroyAggregateStates` is called even if `addRowToAggregateStates`
  or `insertResultsFromStates` throws.

- Move `++num_groups` before `addRowToAggregateStates` in Mode 2 so
  `accumulated_keys.size()` and `num_groups` stay consistent if `add`
  throws after `group_states.push_back`.

Made-with: Cursor