Repository navigation
[feature](function) Add MIN_TOPK/MAX_TOPK and ARRAY_SORTED_MERGE functions - #69456
Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
|
run buildall |
|
/review |
|
run buildall |
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Request changes: six verified findings, including incorrect per-row scalar bounds. Reviewed all 22 changed paths at ddda17f using static analysis and independent BE, FE/test and risk-focused passes. There were no existing inline threads to duplicate and no additional user focus points.
Critical checkpoint conclusions:
- Goal and evidence: the PR adds the four top-K foundation functions; FE IVM integration is explicitly deferred. Small-case aggregate/delta semantics and matching FE/BE schemas are covered by added tests, but the findings prevent the implementation from satisfying the full supported-input contract.
- Scope and reuse: registration, shared MIN/MAX implementations, visitor additions and foreach exclusion are focused. No unrelated functional change was found in the string-test API adaptation.
- Concurrency and locks: no new threads or locks; aggregate state belongs to each group/evaluator, scalar temporaries are invocation-local, and function objects have no new mutable execution state. No new lock-order issue found.
- Lifecycle and initialization: create/reset/destroy and generic serialization scratch cleanup follow existing helpers; factory initialization follows existing singletons and constexpr limits add no cross-TU initialization dependency. Count-buffer accounting remains a finding.
- Configuration: no configuration or dynamic-update behavior added.
- Compatibility and serialization: existing function layouts are unchanged and aggregate intermediate data uses the existing string path. The new compression-size assertions and AggState capacity compatibility need correction; no runtime rolling-upgrade validation was performed.
- Shared definitions: no Thrift/protobuf fields or enum IDs changed.
- Parallel paths: checked MIN/MAX, k=0 and positive k, nullable/nonnullable, empty/all-NULL, grouped/ungrouped, partial/final aggregation, windows, foreach and public AggState combinators. Multirow scalar bounds and differing-capacity AggState merges expose findings.
- Special conditions and errors: range, overflow and negative trusted-count failures propagate explicitly. Const columns are materialized before row handling. Malformed lists and heterogeneous states were assessed against the documented homogeneous, sorted, unique producer-state contract; no additional substantiated finding was established there.
- Test coverage: BE and FE unit tests and regression tests exist, including negative k/DISTINCT cases. Missing cases include differing bounds within one block, compressed states, memory attribution, rejected-value cost and incompatible stored-state capacities. Builds and tests were not run, as required by this review contract.
- Test results: reviewed every added expected result against its inputs. Signed-zero string output is schedule-dependent and needs a stable oracle; the other small-case expected values match static reasoning. Output generation was not independently verified.
- Observability: existing expression error reporting provides function/argument context; no new background service needs separate logs or metrics. Missing query memory attribution is reported inline.
- Transactions and persistence: no EditLog, visible-version, delete-bitmap, failover or transactional mutation path changed. Persistable AggState values expose the reported k compatibility problem.
- Data writes and atomicity: these changes compute expression results; they add no storage write/commit protocol.
- FE/BE passing: function names, supported primitive families, struct fields/nullability and intermediate types agree for homogeneous calls; no new transmitted variable or sending point is added.
- Performance and memory: the bounded list still rebuilds unchanged retained data for rejected distinct values, and the count buffers bypass query accounting. Both are separate inline findings.
- Additional issues and limits: all initial risks and subsequent candidates have explicit dispositions in the review ledger. No other substantiated issue or uncovered changed path remains. Validation is static only; execution behavior and full application-level IVM integration were not exercised.
Review converged after 2 round(s).
| } | ||
| value_column = &nullable->get_nested_column(); | ||
| } | ||
| view.value = value_column; |
There was a problem hiding this comment.
[P1] Preserve the row index in bound views
make_bound_view checks the current row's NULL bit but retains only the whole column; effective-bound selection, trusted-range filtering and bound output then all read index 0. For a two-row MIN block with k=2, n=3, dn=0, empty I/D and NULL delta bounds, let S/b be [(1,1),(2,1)]/2 and [(10,1),(20,1)]/20. The second state should stay unchanged, but it becomes an empty list with bound 2. Nonconstant table arguments remain vectorized, so this affects normal multi-group apply calls; MAX is symmetric. Carry and use the bound row index throughout, and cover differing bounds in one block.
There was a problem hiding this comment.
Adding a concrete counterexample for the original implementation at ddda17f849: this can produce a wrong nonempty candidate and suppress a required fallback.
Consider two rows in the same execution block, with k = 1:
| Row | Original group data | Stored S / b | Inserts I / bi | Deletes D / bd |
|---|---|---|---|---|
| 0 | {10,20} |
[(10,1)] / 10 |
[] / NULL |
[] / NULL |
| 1 | {1,3} |
[(1,1)] / 1 |
[(5,1)] / NULL |
[(1,1)] / NULL |
For row 1, n = 2, dn = 0, and the actual post-update data is {3,5}, so MIN is 3. Its old state has discarded 3, and only values <= 1 remain within the trusted range.
Expected: use row 1's bound, bE = 1. The count of 1 becomes 1 + 0 - 1 = 0; 5 is outside the trusted range. Return (values=[], bound=1). Since n' = 2, the empty candidate list tells the caller to fall back to a complete refresh, which finds 3.
Original behavior: the lists are taken from row 1, but the bound is read from row 0, giving bE = 10. After removing 1, the merge incorrectly admits 5 and returns (values=[(5,1)], bound=10). The candidate list is nonempty, so the documented exhaustion check does not trigger fallback; using the first candidate as MIN would return 5, although 3 still exists.
I checked the latest head, e4acad346d: TopKBoundView now carries the row index, and effective-bound selection, range filtering, and bound output all use it. This fixes the identified indexing cause by source inspection. The added multirow regression uses empty insert/delete lists; please also cover this delete-plus-insert case and its MAX counterpart, with both rows in one execution block, to guard against suppressing a required fallback.
This example is derived from the code; I have not executed it against Doris.
There was a problem hiding this comment.
Fixed in 56a6223. TopKBoundView now carries the row index, and the effective-bound selection, the trusted-range comparison and the bound output all read that row instead of index 0. order_qt_merge_multi_row_bounds covers two rows in the same block with different bounds (2 and 20).
| state.k = other.k; | ||
| } else { | ||
| // Both sides are partial states of the same constant k. | ||
| DCHECK_EQ(state.k, other.k); |
There was a problem hiding this comment.
[P2] Reject incompatible k values across AggState merges
The same-k assumption holds for ordinary partial aggregation, but these functions also expose _state/_merge/_union. AggState type identity records argument types and nullability, not literal k, so a UNION ALL of topk_min_count_state(2, CAST(1 AS BIGINT)) and topk_min_count_state(1, CAST(0 AS BIGINT)) can feed topk_min_count_merge(s). With the k=1 state merged first, release builds ignore the k=0 state's extreme and return value 2 as a complete state; reversing the order changes the result. Checked builds hit this DCHECK. Reject mismatched serialized capacities with a query error, or disable AggState support until this case is defined.
There was a problem hiding this comment.
Fixed in 56a6223. merge() now rejects two partial states with different k with INVALID_ARGUMENT instead of assuming they match, and TopKCountTest.MergeRejectsIncompatibleK merges states built with k=2 and k=1.
| buf.resize(size_bytes); | ||
| auto* pos = | ||
| this->argument_types[0]->serialize(column, buf.data(), IAggregateFunction::version); | ||
| DCHECK_EQ(pos, buf.data() + size_bytes); |
There was a problem hiding this comment.
[P2] Distinguish serialized capacity from the encoded length
get_uncompressed_serialized_bytes returns an upper bound when a column exceeds the 256-byte compression threshold, while serialize returns the actual compressed end pointer. A partial INT state containing 0..64 with k=65 therefore fails this equality in debug/ASAN builds; a compressible string longer than 256 bytes can fail even with k=0. read_column repeats the same invalid equality. The existing round-trip test only stores two INT values. Frame the actual encoded length, or explicitly support padded capacity without asserting equality, and cover compressed states.
There was a problem hiding this comment.
Fixed in 56a6223. write_column declares get_uncompressed_serialized_bytes as the capacity and pads the buffer when serialize returns an earlier pointer; read_column skips the same declared capacity, so a compressed payload is not misread. TopKCountTest.SerializeDeserializeRoundTrip now round-trips 65 distinct INTs (260 bytes, above the compression threshold) with k=65.
| bool has_bound = false; | ||
| MutableColumnPtr bound_column; | ||
| MutableColumnPtr value_column; | ||
| std::vector<int64_t> counts; |
There was a problem hiding this comment.
[P2] Track the owned count buffers with a Doris allocator
Both this persistent counts vector and merged_counts use the default allocator, bypassing Doris query memory accounting and allocation-limit checks. At k=65536 the count payload alone is at least 512 KiB per group, with additional merge scratch; many groups can retain substantial unreported memory. The aggregate helper reports only sizeof(Data), and the aggregation sink counts arenas/container/hash-table storage, so neither accounts for these separate heaps. Use DorisVector<int64_t> or another tracked container for both buffers so growth, moves and destruction participate in query memory tracking.
There was a problem hiding this comment.
Fixed in 56a6223. Both the per-group counts and the merged_counts scratch buffer are now DorisVector<int64_t>, so the 512 KiB-per-group payload at k=65536 is charged to the query memory tracker.
| return; | ||
| } | ||
|
|
||
| insert_entry(state, left, column, row_num, 1); |
There was a problem hiding this comment.
[P2] Reject worse values before rebuilding the retained column
Once a MIN state holds k values, every new distinct value larger than its last entry is immediately discarded, yet this call clones the column and copies all k retained values before popping the new entry. MAX has the symmetric path. One million ascending distinct inputs at k=65536 cause over 61 billion retained-value copies after filling the list; string inputs also copy the retained bytes. Detect this full-list rejection before insert_entry, update the truncation flag/bound, and return without rebuilding unchanged state.
There was a problem hiding this comment.
The full-list rejection fast path is fixed at e4acad346d. There is still an avoidable rebuild while filling the candidate list: insert_entry() clones the value column and copies every existing value even when pos == state.value_column->size().
For topk_min_count(x, 65536) with distinct ascending input 1..65536, every insertion is at the end and the list has room. Filling one group's state therefore copies 0 + 1 + ... + 65535 = 2,147,450,880 existing elements, plus repeated column allocations. MAX with descending input has the symmetric case. This is a source-derived operation count, not a measured runtime result.
Please append directly before the clone/copy path:
if (pos == state.value_column->size()) {
state.value_column->insert_from(column, row_num);
state.counts.push_back(count);
return;
}The state already owns a mutable value column, and insert_from() appends to it. The caller already rejects worse values when the list is full, so this handles the remaining tail-insertion case without rebuilding the retained prefix. A benchmark with sorted input and large k would quantify the benefit.
There was a problem hiding this comment.
Fixed as suggested in e7a3dbf: insert_entry() appends in place when pos == state.value_column->size(), before the clone/copy path, and state.counts.push_back(count) keeps the parallel vector in sync. Together with the full-list rejection in add_value(), sorted input at large k only appends while the list fills. I did not add the benchmark; TopKCountTest.* and the regression suite still pass.
| e {"values":[], "bound":null} {"values":[], "bound":null} | ||
|
|
||
| -- !double_special_values -- | ||
| e {"values":[{"value":-0, "count":2}, {"value":1, "count":1}, {"value":NaN, "count":2}], "bound":null} {"values":[{"value":NaN, "count":2}, {"value":1, "count":1}, {"value":-0, "count":2}], "bound":null} |
There was a problem hiding this comment.
[P2] Avoid requiring a particular signed-zero representative
This expected struct requires -0, but the input contains both +0 and -0 on an unordered two-bucket scan. add_value keeps the first representative when CompareFloat reports equality, and merge_values keeps the left representative, so a valid scan/merge order can return +0 with the same count. Number JSON formatting preserves the sign, while the regression framework compares STRUCT cells as exact strings; order_qt only orders result rows. Canonicalize zero representations or project numeric values/counts for comparison, then regenerate the output so this test does not depend on aggregate input order.
There was a problem hiding this comment.
Fixed in 56a6223. The case no longer depends on scan order: it is an explicit topk_merge_min with -0.0 on the left and 0.0 inserted on the right, so the retained representative is deterministic, and the golden pins that single merge result.
|
run buildall |
|
/review |
There was a problem hiding this comment.
Review of exact head 56a6223 across all 22 changed paths. Four new findings: decimal precision lost during signature promotion, lossy list coercion breaking distinct counts, nested NULL fields silently consumed in release, and runtime-NULL aggregate k bypassing validation. The previous six inline findings are fixed at this head; none is reposted, and no existing P0/P1 remains applicable. There was no additional user-provided focus.
The reviewed flow is SQL binding/custom signature -> implicit casts -> BE aggregate add/partial serialization/merge -> ordered values/bound state -> scalar S+I-D within the trusted bound -> result state. IVM plan integration is explicitly a separate PR.
Critical checkpoint conclusions:
- Goal and proof: the four functions implement ordered top-k counts/bounds and a delta apply step; the four findings prevent full correctness. New FE/BE/regression tests cover normal paths but omit these cases.
- Scope and reuse: changes are focused on the functions and registries, use common MIN/MAX implementations, and reuse existing aggregate/column helpers. The incidental string-test edits update existing API calls.
- Concurrency: no new threads, locks or shared mutable function state; aggregate state and scalar scratch follow existing evaluator ownership. No new lock-order or lock-held work issue found.
- Lifecycle/static initialization: nontrivial aggregate state is argument-constructed, destroyed and reset through existing helpers; scratch cleanup follows their exception paths. No new ownership cycle or cross-translation-unit initializer dependency found.
- Configuration: no new configuration or dynamic-update path.
- Compatibility: names are registered on both FE/BE; symbols and the aggregate format are new, and no existing storage/protocol layout changes. Datatype serialization receives the inherited execution version. Decimal precision selection is finding 1; normal analysis recomputes the result type after narrowing its arguments. Mixed-version execution of new functions was not run.
- Thrift/protobuf: no gensrc changes or field-ID changes.
- Parallel paths: MIN/MAX share the reported defects; nullable aggregates, partial/final aggregation, empty input, AggState combinators and foreach exclusion were traced. Cloud/non-cloud use the same function path.
- Conditions/error handling: truncation/exactly-k/zero-count/bounds/overflow branches were checked. Previous incompatible-k and bound-row defects are fixed. Nested NULL validation is finding 3; backend-evaluated NULL k can bypass validation (finding 4).
- Test coverage: regression, BE unit and JUnit 5 signature tests exist for empty/all-NULL groups, types, bounds, k limits, partial merge, compressed numeric serde, mismatch-k and overflow. Mixed-type/decimal scalar, nested-NULL negative and runtime-NULL k cases need coverage.
- Expected results: every added output cell was checked statically against inputs/algorithms; the signed-zero representative is now deterministic. Actual output-generation history and runtime results were not verified.
- Observability: existing local query errors are appropriate for these pure functions; no additional metrics requirement identified.
- Transactions/persistence/failover: no EditLog, transaction, visible-version, rowset or persistence lifecycle changes.
- Data writes/atomicity: no storage-write path changes; no new crash-recovery or atomicity mechanism.
- FE-BE transport: no new variables or send paths; existing expression/function metadata carries the types, with decimal type narrowing noted above.
- Performance/memory: retained and merge counts use tracked DorisVector; rejected worse values now avoid full-list copying. Sorted replacements remain linear in k, overlapping the prior performance topic. No additional distinct performance finding verified.
- Other checks: registrations, arities, visitor dispatch, nullability and CMake discovery were covered. The text-only build hygiene gates passed header rules/budgets/PCH, extern-template pairing and unity coverage.
Validation was static only, as required: no builds, runtime tests or source changes. Review converged in 3 rounds: all three latest reviewers returned NO_NEW_VALUABLE_FINDINGS on the same final four-comment set. All 22 changed paths and initial risk areas are covered; no unresolved candidate or coverage gap remains. The P1 decimal finding warrants changes requested.
Review converged after 3 round(s).
| + toSql()); | ||
| } | ||
| ArrayType listType = ArrayType.of(TopKCountFunction.pairType(valueType)); | ||
| return FunctionSignature.ret(resultType(valueType)).args( |
There was a problem hiding this comment.
[P1] Preserve nested decimal precision when bounds are NULL
The inherited decimal precision pass changes the value type chosen here even for homogeneous input lists. With S=[(1.23,1)] and empty I/D all typed as ARRAY<STRUCTvalue:DECIMAL(4,2),count:BIGINT>, plain NULL bounds, n=1, dn=0 and k=2, extractBySignature skips the STRUCT entries, so only the NULL bounds contribute and select DECIMAL(1,0). defaultDecimalV3PrecisionPromotion then rewrites every nested decimal argument to that type. Coercion changes 1.23 to 1, and normal analysis recomputes the result type from the narrowed children, returning 1 instead of 1.23. Both MIN and MAX are affected. Override precision computation to include the nested value types and ignore untyped NULL bounds, so a complete decimal state retains its values; add a decimal merge regression with plain NULL bounds.
There was a problem hiding this comment.
Fixed in e4acad3. TopKMergeFunction now overrides computePrecision and returns the signature unchanged, so a NULL bound can no longer contribute DECIMAL(1,0) and narrow the nested lists. TopKFunctionSignatureTest.testMergeKeepsDecimalPrecisionWithNullBounds covers S=[(1.23,1)] with NULL bounds, and order_qt_merge_decimal_null_bounds covers it end-to-end.
| if (fields.size() != 2) { | ||
| continue; | ||
| } | ||
| return TopKCountFunction.normalizeValueType(fields.get(0).getDataType()); |
There was a problem hiding this comment.
[P2] Preserve distinct values when coercing the input lists
Taking T from the first typed list forces later lists through potentially lossy casts. With empty INT S/D, sorted DOUBLE I=[(1.1,1),(1.9,1)], NULL bounds, n=0, dn=2 and k=2, FE implicitly casts I to [(1,1),(1,1)]. consume_equal advances a side only once per head, so the scalar returns two entries for value 1 instead of one entry with the summed count. Each source list was ordered, distinct and non-NULL before this function coerced it; MAX has the symmetric case. Choose a lossless common value type across all lists/bounds, or reject incompatible input types before coercion, and add mixed-type merge coverage.
There was a problem hiding this comment.
Fixed in e4acad3. inferValueType now requires the same value type in every typed list and bound and raises an analysis error for mixed types, so no list is coerced to a narrower type. TopKFunctionSignatureTest.testMergeRejectsMixedValueTypes covers empty INT lists with a DOUBLE list, and the regression asserts the same error.
| } | ||
| // The top-K producers never emit a NULL entry, a NULL value or a NULL count. | ||
| DCHECK(view.null_map == nullptr || (*view.null_map)[index] == 0); | ||
| DCHECK(view.value_null_map == nullptr || (*view.value_null_map)[index] == 0); |
There was a problem hiding this comment.
[P2] Reject NULL state fields before consuming their nested payload
These DCHECKs are the only checks for actual NULL entries/fields, but the public scalar signature does not guarantee their absence: FE complex coercion preserves source struct-field nullability. For example, S=array(named_struct('value', CAST(NULL AS INT), 'count', CAST(1 AS BIGINT))), empty typed INT I/D, NULL bounds, n=1, dn=0 and k=1 is accepted. In release the value NULL map is ignored and its default INT payload is emitted as a non-NULL value 0 with count 1; checked builds hit the assertion. NULL counts/elements have the same unchecked path, and MAX shares it. Return a query error for actual NULL state entries/fields before comparing or consuming their payload, while continuing to allow NULL top-level lists and nullable wrappers containing non-NULL data.
There was a problem hiding this comment.
Fixed in e4acad3. check_entry() runs before any comparison or consumption and returns INVALID_ARGUMENT when an entry, its value or its count is NULL. NULL top-level lists are still accepted as empty states (order_qt_merge_all_null), and the NULL-value case is in the regression.
| checkLegalityBeforeTypeCoercion(); | ||
| Expression count = getArgument(1); | ||
| // A constant FE cannot fold is left to BE, which reads the value from the count column. | ||
| if (!(count instanceof Literal)) { |
There was a problem hiding this comment.
[P2] Validate evaluated k before nullable argument filtering
Deferring validation here does not ensure BE ever reads the count. With BE constant folding disabled (the default), SELECT g, topk_min_count(v, bitmap_min(bitmap_from_string(''))) FROM (SELECT 1 AS g, 1 AS v UNION ALL SELECT 1 AS g, 2 AS v) t GROUP BY g passes the constant check and remains nonliteral in FE, but its count evaluates to NULL. The variadic nullable aggregate wrapper then skips every input row before AggregateFunctionTopKCount::check_k is called, so this nonempty group silently returns an empty state instead of the required non-null-count error. MAX shares this path. Validate the evaluated constant k independently of value-row null filtering, and cover a backend-evaluated NULL count as well as a literal NULL.
There was a problem hiding this comment.
Fixed in e4acad3. The aggregate is now constructed directly instead of through the nullable-argument adapter, which dropped rows before add() ran, so add() calls check_k for every row, including rows whose x is NULL, and a NULL or out-of-range evaluated k fails the query. The bitmap_min regression covers the non-foldable NULL count.
There was a problem hiding this comment.
Fixed in 1c73d49: TopKFunction no longer implements NullIgnoringAggregateFunction. The marker's contract (an input consisting only of rows whose argument is NULL behaves like an empty input) does not hold while a backend-evaluated count is validated per input row, so InferAggNotNull must not filter those rows away for these aggregates.
Without the marker the global aggregate keeps its rows, add validates the count before skipping a NULL value, and an all-NULL input with a non-literal NULL count now reports the count of min_topk and max_topk must be a non-null constant instead of returning an empty array. NULL values of x are still ignored by add itself, so the aggregate's result is unchanged for valid counts.
Covered in query_p0/aggregate/min_max_topk: the all-NULL global input with bitmap_min(bitmap_from_string('')) fails for both aggregates, and the same shape with the valid non-literal count bitmap_min(bitmap_from_string('1')) still returns the empty array.
|
run buildall |
|
/review |
There was a problem hiding this comment.
Static review of e4acad3, covering all 22 changed paths and relevant callers/callees. Two new P2 findings are inline: NULL scalar counters reach a fatal invariant path in checked builds, and a valid insert/delete cancellation can falsely overflow. No P0/P1 finding remains independently confirmed for this head.
Existing review context: both prior P1 concerns are repaired (per-row bound indexes and nested decimal precision). The earlier incompatible-k, column framing, tracked buffers, rejected-value copies, mixed-type coercion, NULL state-field repro and signed-zero expectation concerns are also repaired. The evaluated-NULL aggregate k concern remains partially applicable: for a global aggregate over a nonempty table whose x values are all NULL, InferAggNotNull can insert x IS NOT NULL and discard every row before check_k. This is already covered by thread 4347517695 and is not reposted. Its grouped/non-NULL-x repro is repaired. No additional user focus was supplied.
Critical checkpoint conclusions:
- Goal and proof: the four functions implement counted top-K state and delta apply; registrations and nested return metadata agree. Supplied tests cover core behavior, but the two inline edge cases remain untested. FE IVM integration is outside this PR's stated scope.
- Scope and reuse: shared MIN/MAX implementations and type helpers keep the change focused; the string unit-test API adjustments are small and unrelated to runtime behavior.
- Concurrency and locking: per-group aggregate state and stateless scalar functions introduce no new threads, shared mutable state, locks or lock ordering.
- Lifecycle and initialization: helpers destroy/recreate state on reset; columns and DorisVector buffers own tracked memory. No new cross-translation-unit initialization dependency was found.
- Configuration: none added.
- Compatibility: new function names and a new aggregate-state format; existing function/storage formats are unchanged. The codecs use the negotiated execution version. Newly introduced builtins require supporting BEs.
- Thrift/protobuf: no changes.
- Parallel paths: MIN/MAX, nullable/nonnullable, constants, batch/selected/single-place, streaming, window and AggState paths were traced. Constants are materialized before aggregate row access; foreach is explicitly excluded.
- Conditions and errors: truncation/bound branches have concrete explanations. The NULL scalar assertion and cancellable intermediate overflow need the inline fixes; residual aggregate validation is tracked in the existing thread.
- Test coverage: BE and FE unit tests plus regression SQL cover empty/all-NULL input, k boundaries, truncation, partial merges, compressed serialization, metadata, precision, mixed types, NULL state fields and multirow bounds. Nullable scalar counters and count-limit cancellation need coverage.
- Expected results: all supplied output entries were reconciled with the SQL and algorithms; signed zero is now deterministic. Execution and output-generation provenance were not verified.
- Observability: existing function/argument-specific errors are adequate for these local expressions; no new service metrics are needed.
- Transactions/persistence/failover: no transaction, EditLog or persisted catalog behavior changes; partial-state transport follows the existing aggregate infrastructure.
- Data writes and atomicity: no table-write path changes; propagated query errors discard local state.
- FE/BE passing: function names, arity, supported value families, BIGINT counts and nested nullability align; no new wire variables.
- Performance: full-list worse-value rejection avoids the known redundant copy, scalar merge scans only the required prefix, and count buffers are tracked. Accepted distinct insertions retain the sorted-column O(k) rebuild cost covered by existing review context.
- Other issues: malformed struct arities are rejected by common analysis checks; decimal/time precision, NULL coercion, NaN/zero comparison, ownership and special execution paths were rechecked without another distinct verified issue.
Review converged in two rounds after all three reviewers rechecked the unchanged final findings, duplicate dispositions and complete assigned scope. No unresolved static-review candidate or uncovered changed path remains. Validation is static only: no builds, tests, formatting changes or source modifications were performed, as required by this review contract.
Review converged after 2 round(s).
| const IColumn* nested = &column; | ||
| if (const auto* nullable = check_and_get_column<ColumnNullable>(&column)) { | ||
| // The contract declares these arguments as non-NULL BIGINT. | ||
| DORIS_CHECK(!nullable->is_null_at(row)) << argument_name << " must not be NULL"; |
There was a problem hiding this comment.
[P2] Reject NULL scalar counters with a query error
The FE signature accepts nullable BIGINT arguments, so SELECT topk_merge_min(CAST(NULL AS ARRAY<STRUCT<value:INT,count:BIGINT>>),NULL,NULL,NULL,NULL,NULL,0,0,CAST(NULL AS BIGINT)) reaches this check. DORIS_CHECK calls Status::FatalError, which executes LOG(FATAL) in builds without NDEBUG, terminating the BE for invalid SQL input; release builds instead return FATAL_ERROR. NULL n/dn and topk_merge_max follow the same path. Return INVALID_ARGUMENT for actual NULL counter values, including nullable nonliteral inputs, and cover all three counters.
There was a problem hiding this comment.
Please replace the DORIS_CHECK in read_int64() with an explicit INVALID_ARGUMENT exception:
if (nullable->is_null_at(row)) {
throw Exception(ErrorCode::INVALID_ARGUMENT,
"{} must not be NULL", argument_name);
}NULL n, dn, or k can reach this helper through the public SQL signature. They are invalid argument values and should fail the query with a normal parameter error. The current assertion reaches LOG(FATAL) in builds without NDEBUG, terminating the BE; release builds report FATAL_ERROR. Throwing INVALID_ARGUMENT also follows the existing check_k() and overflow-checking error-handling pattern.
Please cover NULL values for all three counters in both MIN/MAX variants, including nullable column inputs.
There was a problem hiding this comment.
Fixed in e7a3dbf: read_int64 returns a Status and reports INVALID_ARGUMENT for a NULL n, dn or k. b0ae93f adds the coverage requested here: NULL for all three counters on both topk_merge_min and topk_merge_max, plus a table whose nullable BIGINT columns feed the counters, so the nullable-column input path is exercised as well.
| RETURN_IF_ERROR(consume_equal(i_view, i_index, *head, head_row, type, &count_i)); | ||
| RETURN_IF_ERROR(consume_equal(d_view, d_index, *head, head_row, type, &count_d)); | ||
|
|
||
| int64_t post = checked_add(count_s, count_i, name); |
There was a problem hiding this comment.
[P2] Check overflow after insert/delete cancellation
With complete S=[(1,9223372036854775807)], I=D=[(1,1)], NULL bounds, n=9223372036854775807, dn=0 and k=1, a valid delta replaces one occurrence with the identical value and the result should remain S. This addition overflows before the deletion is subtracted, although every input and final count fits BIGINT. MAX shares the path. Evaluate S+I-D with a wider intermediate or cancellation-safe ordering, range-check the final count, and cover this boundary cancellation.
There was a problem hiding this comment.
Fixed in e7a3dbf. S + I - D is evaluated in __int128 and range-checked before narrowing to int64, so the INT64_MAX cancel pair returns the original state instead of overflowing. order_qt_merge_min_cancel_at_int64_max pins it.
|
run buildall |
|
/review |
|
run buildall |
|
/review |
There was a problem hiding this comment.
Static review of b0ae93f: all 22 changed paths reviewed. No new unique inline findings. All three reviewers returned NO_NEW_VALUABLE_FINDINGS; all candidates and risk focuses have evidence-backed dispositions.
Both existing P1 findings are corrected: row-specific bounds (#4346834936) and nested decimal precision with NULL bounds (#4347517663). Neither remains a blocking finding for this head. The other supplied fixes were revalidated, including tracked count buffers, serialization framing, incompatible-k rejection, append/rejection fast paths, mixed-type rejection, NULL field/counter errors, cancellation-safe arithmetic, and deterministic signed-zero coverage.
One residual path is already covered by existing P2 #4347517695, so it is not reposted: with only NULL values in a nullable INT x column, a global topk_min_count(x, bitmap_min(bitmap_from_string(''))) can acquire an x IS NOT NULL filter through InferAggNotNull. The reduced plan is Aggregate(topk_min_count(x,k)) -> Filter(x IS NOT NULL) -> Scan; removing all rows bypasses add/check_k, even though k evaluates to NULL. MAX shares this path. Direct aggregate creation fixes the originally reported BE nullable-adapter instance, but validation can still be suppressed upstream.
Critical checkpoint conclusions:
- Goal/correctness: counted MIN/MAX prefixes and trusted-range S+I-D apply preserve exact retained values/counts for valid producer states; IVM plan integration is a separate PR.
- Scope/reuse: shared MIN/MAX templates and FE result-type helpers keep the feature focused; the string-test changes only update existing API usage.
- Concurrency/locks: mutable aggregate data is per group; scalar state is local to execution. No new threads, shared mutable state, locks or lock-order risk identified.
- Lifecycle/static initialization: create/reset/destroy, streaming temporaries, deserialization and window buffering were traced; owned columns/vectors clean up through existing helpers. No new cross-TU initialization dependency.
- Configuration: no new configuration or dynamic-update path.
- Compatibility: new function names and state encoding only; no existing function/storage format replaced. Negotiated execution versions flow through aggregate/state serialization; older binaries do not implement these new functions.
- Thrift/protobuf: no definitions changed.
- Parallel paths: MIN/MAX, k=0/positive k, grouped/global/window, partial/final and AggState paths checked; unsupported foreach is rejected.
- Conditions/errors: reachable NULL fields/counters, invalid k, mismatched state k and overflow now fail through normal errors. Constant-column assumptions are supported by upstream materialization; the known NULL-k filtering exception is described above.
- Tests: BE state/merge/serialization and FE signature tests plus regression success/error cases were inspected. No builds or tests were run.
- Expected results: all added output blocks reconcile statically with the inputs and merge rules; signed-zero representative selection is controlled in the explicit scalar case.
- Observability: existing function-specific query errors/profiles are adequate for these local expressions; no new service lifecycle requiring metrics was identified.
- Persistence/failover: no EditLog, transaction, rowset/version or delete-bitmap change; temporary/public aggregate-state serialization was reviewed separately.
- Data writes/atomicity: no new durable write or transactional mutation path.
- FE/BE propagation: registered names, supported families, intermediate string state and nested STRUCT/ARRAY nullability agree through existing metadata/translation paths; no new wire variable.
- Performance/memory: significant count/column buffers are tracked; full-list rejection and tail append remove the prior redundant copies. Interior sorted insertion remains linear in retained size; scalar merge stops after k+1 positive candidates.
- Other concerns/user focus: no additional user focus was supplied. No unresolved new candidate or unreviewed changed path remains.
Limits: this is static validation, not evidence of passing builds, runtime tests or benchmarks. Generated-output provenance was not independently established. Plain untyped NULL decimal bounds are covered by the FE signature test, while the SQL regression uses typed NULLs; full supported-type and multi-step IVM runtime coverage was not executed.
Review converged after 1 round(s).
b0ae93f to
a742080
Compare
yujun777
changed the title
|
run buildall |
|
/review |
There was a problem hiding this comment.
Static review of all 26 changed paths at a7420809fc69a4ea822cfd1f29ed9f8b03afbb3a found two new P1 issues in array_sorted_merge: captured columns can produce incorrect results, and nonnullable merge bodies reach a fatal nullable-column cast. Both are reported inline. Request changes.
Review checkpoint conclusions:
- Goal and proof: TOPK_MIN_COUNT/TOPK_MAX_COUNT implement the intended distinct values/counts/bound shape; the sorted-merge invocation chain is wired, but the two accepted lambda cases above are broken. Added tests cover normal behavior and miss these cases.
- Scope and reuse: changes are focused on the functions, registrations, lambda extra-input plumbing and tests; the string-test changes are mechanical API spelling repairs. Existing aggregate helpers and column machinery are reused.
- Concurrency and locks: no new workers, shared query state or lock ordering; registration uses call_once and states are local to aggregation places/execution calls.
- Lifecycle and initialization: reset destroys/reconstructs owned aggregate state; temporary columns and lambda frames have bounded synchronous lifetimes. No cross-translation-unit initializer dependency or ownership cycle found.
- Configuration: none added; no dynamic-update obligation.
- Compatibility: no changes to existing storage/wire layouts or existing function symbols; the new aggregate state uses existing datatype serialization and execution-version plumbing. Mixed-version execution was not exercised.
- Thrift/protobuf: no changes.
- Parallel paths: MIN/MAX share implementation; k=0, partial/final, serialized and state-combinator paths were traced. Both merge directions share the same machinery; unsupported foreach variants are explicitly excluded.
- Conditions, errors and nulls: aggregate k/overflow checks and ordinary NULL handling were traced. The new lambda result-shape error is inline; captured-input resolution is independently inline.
- Test coverage: FE signatures/legality, BE aggregate tests and both regression suites were inspected. Missing coverage includes captures/nested lambdas, nonnullable bodies, broad scalar types, window/reset and compressed STRING/MAX states.
- Expected results: all added golden results agree with the supplied test inputs by source inspection. The current special-float input has only positive zero, removing the earlier sign-dependent expected result. Golden generation was not verified by execution.
- Observability: existing query errors suffice for these local functions; the fatal error on valid input is reported. No new distributed diagnostic mechanism is needed.
- Transactions and persistence: none changed; no edit-log/replay or failover obligation introduced.
- Data writes and atomicity: no persistent write path changed.
- FE/BE transport: the ascending flag survives binding and translation; result schema and column shapes were traced. Capture transport and nonnullable body output are the two reported gaps.
- Performance and memory: count buffers use DorisVector; prior full-list rejection and tail-append fixes are present. No additional independent performance finding; no benchmark or release-artifact ABI audit performed.
- Other concerns: every candidate/risk has an accepted, duplicate or evidence-backed dismissed disposition. No unresolved concrete review question remains.
Existing threads were revalidated and not reposted. The two earlier P1 findings concern scalar topk_merge APIs absent from this head, so no existing P0/P1 blocker IDs remain applicable. Earlier aggregate mismatch/serde/accounting/copy issues are fixed. The existing evaluated-NULL-k P2 thread remains relevant when global optimization filters all-NULL x before add(); that continuation is already covered by discussion r4347517695.
No additional user focus was supplied. Review converged after two rounds: all three latest reviewers returned NO_NEW_VALUABLE_FINDINGS on the unchanged final comment set, and all changed paths and risk focuses were covered. This is static validation only: no builds, tests, runtime queries, formatting/hygiene execution, benchmarks or golden regeneration were performed under the review contract.
Review converged after 2 round(s).
| argument_columns[1]->insert_from(*data[1], right); | ||
|
|
||
| int lambda_res_id = lambda_result_base; | ||
| RETURN_IF_ERROR(children[0]->execute(context, &lambda_block, &lambda_res_id)); |
There was a problem hiding this comment.
[P1] Bind captured inputs before executing the merge lambda
The FE accepts enclosing slots in the merge lambda, but this executes them against lambda_block, which contains only a and b. For a reduced input plan Project(array_sorted_merge((a,b)->IF(a=b AND a<lim,a,NULL),l,r,true)) -> Scan(lim INT,l ARRAY<INT>,r ARRAY<INT>), the row lim=2,l=[1],r=[1] should return [1]. The captured nullable INT lim retains input column index 0, so VSlotRef reads candidate a=1 instead; the predicate becomes a=b AND a<a, and the result is [1,1]. With lim at index 2 the same path instead reports a missing-column error. Materialize and rebind current-row captures and enclosing lambda bindings, or reject unsupported captures during prepare as array_sort does. Add table-column and outer-lambda capture regressions.
There was a problem hiding this comment.
Fixed in 42efc96: captured input columns are materialized at their original positions in the one-row lambda block for the current row, so an enclosing slot reads the correct value. A nested lambda that captures one of the merge arguments is rejected in prepare with NotSupported instead of resolving against the wrong column. Regression covers a table-column capture at index 0 (previously resolved to candidate a) and one beyond the two candidates (previously a missing column), plus the rejected nested capture.
There was a problem hiding this comment.
Fixed in 6ac1f1a: an enclosing lambda's argument now resolves correctly. The function collects the named bindings visible at the current execution point (LambdaExecutionContext::collect_visible_bindings) and publishes their values under those names in its one-row merge block, before the two merge arguments so that a merge argument with the same name still shadows an outer one. The values are read at the same row as the arrays, so short-circuit evaluation and batched lambda execution keep reading the right value.
order_qt_capture_enclosing_lambda in test_array_sorted_merge covers array_map(x -> array_sorted_merge((a, b) -> IF(a = b AND a < x, a, NULL), [1], [1], true), [0, 2]) and returns [[1, 1], [1]] (x=0 keeps both values, x=2 merges them). A nested lambda that captures one of the merge arguments is still rejected during prepare, because those arguments are not published into nested frames.
There was a problem hiding this comment.
Fixed in 51bd4e1: the merge frame now rebases every binding it can see onto the private block instead of hiding them. It publishes the enclosing bindings with their positions in that block, then the two merge arguments, so a same-named a/b shadows an outer binding. parent_bindings_visible stays false, because the parent frame positions do not describe this block.
A lambda nested inside the merge body therefore resolves x through the merge frame: the nested function collects that binding position and materializes it from the one-row merge block, and the value is cut at the same source row as the arrays. The reduced plan
SELECT array_map(x -> array_sorted_merge(
(a, b) -> IF(a = b AND a < element_at(array_map(y -> x, [1]), 1), a, NULL),
CAST([1] AS ARRAY<INT>), CAST([1] AS ARRAY<INT>), true), [0, 2])now returns [[1, 1], [1]], pinned by order_qt_capture_enclosing_lambda_nested in test_array_sorted_merge. Nested captures of the merge-local a/b are still rejected during prepare, and the legacy positional frame is unchanged when the lambda carries no argument names.
| merged_is_null = nullable->is_null_at(0); | ||
| } | ||
| if (!merged_is_null) { | ||
| result_nested->insert_from(*merged_column, 0); |
There was a problem hiding this comment.
[P1] Wrap nonnullable lambda results before inserting array elements
SELECT array_sorted_merge((a,b)->COALESCE(a,b,CAST(0 AS INT)), CAST([1,2] AS ARRAY<INT>), CAST([1,2] AS ARRAY<INT>), true) passes FE type validation and should return [1,2]. The fallback makes the body nonnullable, so VectorizedCoalesceExpr returns a raw ColumnInt32; VLambdaFunctionExpr preserves that body type, and materializing constants does not add a nullable wrapper. However, result_nested is ColumnNullable(Int32) because DataTypeArray always makes its elements nullable. This call therefore reaches ColumnNullable::insert_from with a nonnullable source and fails assert_cast (FATAL_ERROR, or LOG(FATAL) without NDEBUG). Normalize the lambda result to the nullable item representation before inserting it, and cover nonnullable COALESCE/IFNULL or constant bodies.
There was a problem hiding this comment.
Fixed in 42efc96: the lambda result is normalized to the nullable array element representation before insertion -- a non-nullable body inserts its payload into the nested column and appends a 0 to the null map. A regression case with COALESCE(a, b, 0) now returns [1, 2].
TPC-H: Total hot run time: 28308 ms |
ClickBench: Total hot run time: 24.27 s |
|
Correction to the memory table above, and a re-measurement. The table above varied the number of rows per group at a fixed total row count (10M rows), so the 8-rows-per-group case also had eight times fewer groups (1.25M instead of 10M). That is not the shape the change is about. The correct comparison keeps the group count fixed and grows the rows per group, i.e. 10M groups with 1 row each (10M rows) against 10M groups with 8 rows each (80M rows). Setup for the numbers below: release build, single BE restarted before every query (cold page cache on both sides), BIGINT (10M groups)
VARCHAR (10M groups)
State kept on top of
What this says: in a release build the change is at the edge of, or below, what this method resolves. At eight distinct values per group it changes nothing measurable, because the slots dominate and a slot costs the same either way (24-byte metadata plus the 8-byte value against one 32-byte slot carrying the value inline). At one value per group the state is ~24 bytes and one heap allocation smaller, which is the whole effect and is about the size of the measurement noise. The larger gain reported earlier (1959 MB to 606 MB at 10M groups) was measured on a debug/ASAN build, where every allocation carries a redzone, so the second vector per group is what dominated. The change is worth keeping for that build, and because it removes a container, but it should not be sold as a release-memory win. |
|
/review |
|
Codex automated review failed and did not complete. Error: Review recovery stopped: Command '['gh', 'api', 'repos/apache/doris/pulls/69456']' returned non-zero exit status 1. Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
|
Codex automated review failed and did not complete. Error: Selected model is at capacity. Please try a different model. Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
The state carried a separate has_k flag and 64-bit slot counters next to the value payload. k is the aggregate's non-negative count argument, so -1 can stand in for "this partial state has seen no row", and both counters are bounded by the slot count that the int32_t treap links already cap. The priority seed only feeds 32-bit priorities, so a 32-bit splitmix covers it. Key changes: - Drop has_k: k == -1 means unset; the serialized form keeps the historical flag byte, derived from k - live/dead and the priority seed become 32-bit; the state is 56 bytes instead of 72 for fixed-width values - Keep k and the per-value row count 64-bit: an int32 count would cap a value at 2^31 rows per group, which is legal data Unit Test: - be/test/exprs/aggregate/agg_topk_test.cpp (18 tests, including the serialized-format round trip) - regression query_p0/aggregate/min_max_topk
|
Memory of the final top-K state, for Setup: release build, single BE restarted before every query (cold page cache on both sides),
"state per group" is the Query profile cross-check (10M groups, 1 row per group, per-operator What the state costs: 56 bytes of fixed part per group (k, the slot vector header, the cached worst slot, the live/dead counters and the priority seed) plus 32 bytes per retained value for BIGINT and 40 bytes for VARCHAR (count, both treap links, the priority and the value or its span). A |
|
run buildall |
|
/review |
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
|
Codex automated review failed and did not complete. Error: Selected model is at capacity. Please try a different model. Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
FE UT Coverage ReportIncrement line coverage |
TPC-H: Total hot run time: 28694 ms |
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
TPC-DS: Total hot run time: 151336 ms |
ClickBench: Total hot run time: 24.38 s |
FE Regression Coverage ReportIncrement line coverage |
|
/review |
|
Codex automated review failed and did not complete. Error: Selected model is at capacity. Please try a different model. Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
|
/review |
HappenLee
left a comment
HappenLee
left a comment
There was a problem hiding this comment.
Reviewed exact head 29fb3c4. No remaining blocking code issue found.
Checked FE registration, signatures and lambda binding/translation through BE aggregation and sorted-array execution, including count overflow, incompatible k, compressed state serialization, string padding/reclamation, singleton merges, candidate alias lifetimes, selected/virtual/enclosing captures and nullable result insertion. The previous relevant correctness and performance findings are addressed in this implementation. Shared state is query-local; MIN/MAX reuse the same implementation, and existing lambda bindings retain their default behavior. No existing storage/protocol/configuration or transaction path changes are introduced.
Validation: local build-hygiene checks passed; the current head's compile, FE/BE UT and main regression CI statuses are successful. This review did not rerun a full local build, tests or performance benchmarks. Automated code-review status is still pending.
Nonblocking documentation nit: please synchronize the PR description's k <= 65536 statement with the current code/tests, which allow larger nonnegative BIGINT k values (subject to the representable state-size limit).
|
Codex automated review failed and did not complete. Error: Selected model is at capacity. Please try a different model. Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
github-actions
Bot
added
the
approved
|
PR approved by at least one committer and no changes requested. |
What problem does this PR solve?
Issue Number: N/A (tracked in #65418)
Related PR: #69286
Problem Summary:
Adds the generic building blocks that incremental MIN/MAX maintenance builds on; the IVM
integration itself is a separate PR.
min_topk(x, k)/max_topk(x, k): an aggregate returningARRAY<STRUCT<value T, count BIGINT>>— the k smallest (largest) distinct non-NULL valueswith their exact row counts, ordered by value. The array carries no truncation flag: a
caller that needs one infers it from the length, treating a list of exactly k entries as
possibly truncated. With
k = 0the array holds at most the single(extreme, count)pair.
kis a constantBIGINTwith0 <= k <= 65536.array_sorted_merge(merge_fn, arr1, arr2, is_asc): a generic higher-order function thatmerges two arrays already sorted by the elements' natural order.
merge_fnreceives thetwo current candidates and returns the combined element, or NULL when the two elements
must be kept separately.
is_ascstates the sort direction, so one function coversascending and descending inputs, and calls can be chained to merge more than two arrays.
The aggregates keep a sorted
(value, count)state with checked int64 counting, so partialaggregation and merge stay exact (
topK(A ∪ B) ⊆ topK(A) ∪ topK(B)). FLOAT/DOUBLE order NaNas greater than every number so the distinct list keeps a strict order.
Returning the array alone keeps the aggregate in the same shape as the other array
aggregates and leaves the truncation flag to the caller that actually persists a bound:
incremental MIN/MAX maintenance still stores one, but as its own state column rather than
inside the aggregate result.
array_sorted_mergekeeps its non-lambdais_ascargument through the higher-order binding:the existing pipeline passed only the lambda arguments to the bound function, so the binding
contract gained a way to return extra inputs.
Release note
Add the
min_topk/max_topkaggregate functions and thearray_sorted_mergescalar function.
Check List (For Author)
regression-test/suites/query_p0/aggregate/min_max_topk.groovy: k = 0/1/2/8/65536/65537, empty input, all-NULL groups, NOT NULL input types, string/decimal/datetime, -0.0/NaN, agreement with MIN/MAX, the array shape (cardinalityandelement_at(...).value/count), two-phase partial+final merge and the error paths.regression-test/suites/query_p0/sql_functions/array_functions/test_array_sorted_merge.groovy: ascending and descending merges, struct keys, empty and NULL arrays, and the analysis errors.TopKFunctionSignatureTest.agg_topk_test.cpp(k = 0, truncation, MIN/MAX merge, serialize/deserialize round-trip, count overflow, NULL count).