[fix](be) Support complex array_agg state serialization

[HappenLee]

What problem does this PR solve?

Issue Number: close #65976

Related PR: None

Problem Summary: A two-phase array_agg_foreach query over nested arrays in an ordinary OLAP table failed with array_agg not support write. The generic complex-type array_agg state did not implement serialization, deserialization, or merge, so it could not cross aggregation phases or grow the ForEach state array. This change serializes complex states with the native DataType format using the active BE execution version and merges state columns by value. The query now completes without using Table Stream.

Release note

Fix array_agg and array_agg_foreach failures for complex element types in multi-phase aggregation.

Check List (For Author)

• Test: Unit Test and Regression test
  • Unit Test: GLIBC_COMPATIBILITY=OFF ./run-be-ut.sh -j 48 --run --filter='AggregateFunctionArrayAggTest.*' (7 tests passed)
  • Regression test: ./run-regression-test.sh --run -d function_p0 -s test_agg_foreach (1 suite passed)
  • ASAN BE build passed
  • FE build and Checkstyle passed
• Behavior changed: Yes. Complex array_agg states now support serialization and merge.
• Does this need documentation: No

[hello-stephen]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?

[HappenLee]

/review

[github-actions]

Request changes: one blocking issue remains in complex array_agg state deserialization. The exact-size std::string copy does not preserve StreamVByte's required readable tail and separately imposes a 1 GiB ceiling on otherwise supported aggregate states.

Critical checkpoint conclusions:

• Goal and proof: The change correctly replaces the unsupported complex-state path with native DataType serialization, forwards the negotiated BE execution version, and adds array/map/struct/NULL plus ForEach growth coverage. The forced agg_phase=2 regression reaches the production partial/global boundary. The goal is not fully achieved because ordinary states above the StreamVByte threshold can still be deserialized from an unsafe representation; see the inline P1.
• Scope: The production and test changes are small, clear, and focused on complex array_agg/array_agg_foreach state handling.
• Concurrency: Aggregate states remain task/local-state owned and use the existing staged merge flow. No new threads, shared mutable state, locks, atomics, lock-order concern, or memory-tracking context is introduced.
• Lifecycle and static initialization: DataTypePtr ownership, aggregate state construction/reset/destruction, COW detachment, and RHS merge ownership were traced. No cycle, premature release, non-intuitive static lifetime, or cross-TU initialization dependency was introduced.
• Configuration: No product configuration was added. The agg_phase=2 and pipeline-task settings are test-local plan controls.
• Compatibility: The new path was previously unsupported; it captures the negotiated be_exec_version, and nullable and ForEach wrappers forward it before state creation. No persisted storage format or function-symbol compatibility issue was found.
• Parallel paths: Ordinary, bucketed, streaming, analytic, nullable, ForEach/ForEachV2, sort, and distinct paths were checked. Sort/distinct serialize their own buffered state rather than this nested state; no distinct parallel-path change is missing.
• Conditions and error handling: The new DORIS_CHECKs enforce real framing/state invariants, and no ignored Status or exception boundary issue was found. They run after the unsafe exact-size representation is created, so they do not resolve the inline issue.
• Tests and results: Unit tests compare complete nested values and NULL maps across serialize/deserialize/merge and repeated ForEach growth; the regression output remains deterministic and the forced phase reaches the changed production contract. The cases remain below the 65-Int32 StreamVByte trigger and do not cover the 1 GiB boundary. I did not run builds or tests because the review-runner prompt prohibits them. At submission time, formatter, checkstyle, license, dependency, title, and secret checks pass; macOS BE UT remains pending.
• Observability: This local intermediate-state path does not need new logs or metrics; existing pipeline error propagation is sufficient.
• Transactions, persistence, and data writes: No transaction, EditLog, persistent metadata/storage, table data-write, atomicity, failover, or crash-recovery path is modified.
• FE-BE variables: No new cross-layer variable or thrift field is added.
• Performance and memory: The new full-state temporary and copy increase peak memory. More importantly, the same read overload loses required decoder padding and rejects states above 1 GiB; both are covered by the inline issue. No separate time-complexity regression survived review.
• Data correctness and nullable handling: Empty, NULL-bearing, array/map/struct native layouts, exact logical framing, insert_range_from ownership, and nullable/const reachability were checked. No distinct data-loss, visibility, MoW, or nullable-shape issue was found beyond the inline finding.
• Other issues and focus: No additional review focus was supplied. Three rounds covered all changed files, and every final-round implementation, runtime/test, and risk-focused audit returned NO_NEW_VALUABLE_FINDINGS; the review converged on this single merged comment.

[HappenLee]

/review

[github-actions]

Found one blocking exception-safety issue in the newly enabled complex array_agg merge. Successful native framing, execution-version propagation, null/value semantics, and normal-path test reachability look sound, but the foreach growth path can turn a recoverable query memory-limit error into a BE crash/corruption.

[github-actions]

Found one blocking exception-safety issue in the newly enabled complex array_agg merge. Successful native framing, execution-version propagation, null/value semantics, and normal-path test reachability look sound, but the foreach growth path can turn a recoverable query memory-limit error into a BE crash/corruption.

[github-actions]

Found one blocking exception-safety issue in the newly enabled complex array_agg merge. Successful native framing, execution-version propagation, null/value semantics, and normal-path test reachability look sound, but the foreach growth path can turn a recoverable query memory-limit error into a BE crash/corruption.

[github-actions]

Found one blocking exception-safety issue in the newly enabled complex array_agg merge. Successful native framing, execution-version propagation, null/value semantics, and normal-path test reachability look sound, but the foreach growth path can turn a recoverable query memory-limit error into a BE crash/corruption.

[github-actions]

Found one blocking exception-safety issue in the newly enabled complex array_agg merge. Successful native framing, execution-version propagation, null/value semantics, and normal-path test reachability look sound, but the foreach growth path can turn a recoverable query memory-limit error into a BE crash/corruption.

[HappenLee]

/review

[github-actions]

Request changes: two P2 memory/performance regressions remain in the generic complex array_agg state path. Immutable serializer metadata is duplicated in every group/foreach-position state, and foreach partial serialization allocates an untracked full-state scratch before copying into the tracked destination.

Critical checkpoint conclusions:

• Goal and proof: The patch now enables native binary write/read/merge for complex nullable array_agg beneath array_agg_foreach. Logical framing, padded zero-copy input, exact cursor consumption, and value-copy merge are sound, and the changed tests reach the production local/global boundary. The functional goal is demonstrated, but the two inline P2 memory regressions remain.
• Scope: The production and test changes are focused on generic complex state serialization and the required foreach relocation repair.
• Concurrency and thread safety: Aggregate states remain query/operator-local. No new thread, shared mutable state, lock, atomic synchronization contract, or lock-order issue was introduced; shared-pointer refcount traffic is only part of the MAIN-2 overhead.
• Error handling: The head-SHA foreach fix constructs and merges all replacements before publication, destroys every constructed replacement on failure, and leaves the old state intact. The new DORIS_CHECKs enforce real internal frame invariants. A direct-output fix for MAIN-1 should also roll the destination back if native serialization throws.
• Memory safety: Reader padding, exact frame advancement, COW ownership, and the previous double-destroy/leak path are fixed on the head. No remaining out-of-bounds, leak, or partial-publication defect was found. The remaining memory problems are the untracked full-payload scratch and the persistent 24-byte per-state metadata increase described inline.
• Data correctness: Empty and NULL-bearing Array, Map, Struct, JSONB/Varbinary, Variant V1/V2, Bitmap, HLL, QuantileState, AggState/fixed-object, TIMEV2, and TIMESTAMPTZ paths were traced. Native deserializers replace detached children, merge copies rhs values, and no distinct row-count, framing, or value-loss issue survived.
• Observability: This local intermediate-state path does not need a new log or metric; existing query and process memory reporting is sufficient once allocations use tracked storage.
• BE null and nullable handling: Nullable and nullable-v2 wrappers propagate the active version before state creation/reset. NULL-literal routing materializes UInt8 rather than DataTypeNothing, and the tests verify nullable nested values and NULL positions.
• Lifecycle and static initialization: Function-owned type/version metadata outlives all states; create, reset, destroy, merge, and exception cleanup are sound. No ownership cycle, static/global initialization, or cross-translation-unit lifetime dependency was introduced.
• Configuration and compatibility: No product configuration was added; agg_phase=2 and the pipeline-task hint are test-local. The complex foreach binary frame previously always threw, so it has no successful legacy representation to preserve. Active execution-version propagation is sound, and current TIMEV2/TIMESTAMPTZ native serializers do not branch on their defaulted state version.
• Parallel paths: Direct typed intermediate output, outer-foreach binary state, grouped/without-key/streaming materialization, nullable wrappers, reset, deserialize/merge, and successful/failed growth were checked. No additional parallel-path patch is missing.
• Tests and results: The units compare exact nullable Array/Struct/Map values, foreach growth and binary round trip, a 65-element StreamVByte leaf, and second-merge failure cleanup; the regression forces agg_phase=2 with deterministic cardinalities. I did not run builds or tests because the authoritative review prompt prohibits them. The author reports the focused ASAN run passes 11/11 tests; at submission time formatter, checkstyle, license, dependency, large-file, title/review-status, and secret checks pass, while macOS BE UT remains in progress.
• Transactions, persistence, and FE-BE variables: No transaction, EditLog, persistent storage format, table data-write, failover/recovery behavior, thrift field, or cross-layer variable is modified.
• Performance and memory: MAIN-1 adds an avoidable upper-bound std::string allocation outside Doris query accounting while source and destination coexist. MAIN-2 grows each generic state from 8 to 32 bytes, adding about 24 MiB per million groups plus shared-pointer refcount traffic. The separate exact-growth/deep-copy concern was dismissed because that capacity algorithm already applied to supported scalar array_agg_foreach before this PR.
• Other issues and focus: No additional user focus was supplied. Three complete rounds covered all five changed files; every maximum-round implementation, type/test, and memory-focused audit returned NO_NEW_VALUABLE_FINDINGS, so the review converged on these two distinct comments.

[HappenLee]

/review

[github-actions]

Review result: no new blocking issue remains on head 4d28bde19dd6073cc8127274c237c27149c0de2f. The four previously raised issues are fixed in the current code, and this review converged with zero new inline comments.

Critical checkpoint conclusions:

• Goal and proof: The PR now gives generic complex array_agg states a complete write/read/merge contract, so nested array_agg_foreach can cross a partial/global aggregation boundary. agg_phase=2 forces the production local INPUT_TO_BUFFER plus global BUFFER_TO_RESULT path. Unit coverage verifies exact nullable Array/Map/Struct values, multi-position foreach growth, round-trip and merge behavior, a StreamVByte-sized leaf, serialization rollback, tracked large-state output, and relocation failure after one successful merge.
• Scope and clarity: The two production changes and three test changes are focused on the unsupported complex-state contract and the exception-safety requirement exposed when its merge became fallible. Existing scalar/string specializations retain their prior format through the compile-time use_native_serde split.
• Concurrency and thread safety: Aggregate states remain operator/task-local. No new thread, shared mutable state, lock, atomic, lock ordering, or memory-tracking context is introduced; the changed relocation runs within the existing staged aggregate-state lifecycle.
• Lifecycle and error handling: Replacement foreach states are fully created and merged before publication. Any create or merge exception destroys every constructed replacement while leaving the published old pointer, size, and states live; success destroys old states exactly once before publishing. Native serialization rolls back only the current frame and rethrows. The new checks enforce real empty-state, size-bound, and exact-consumption invariants rather than silently continuing.
• Memory safety and ownership: Native serialization writes directly into MemTracker-accounted padded ColumnString storage, preserves the 16-byte readable tail needed by StreamVByte, and no longer stores invariant type/version metadata in each group state. Generic merge copies values from the const RHS; mutable columns remain state-owned. The prior out-of-bounds, untracked-scratch, state-size, and partial-destruction concerns are resolved.
• Configuration and compatibility: No product configuration is added; agg_phase=2 and parallel_pipeline_task_num=1 are test-local plan controls. Complex binary state serialization previously always failed, so no successful legacy state representation is replaced. Producer and consumer use the same runtime BE execution version, propagated through nullable, foreach/foreachv2, and nested array_agg wrappers. No persisted storage format or function symbol changes.
• Parallel paths and conditions: Standalone typed intermediate output, outer foreach string transport, grouped/without-key/streaming/bucketed aggregation setup, nullable wrappers, reset, add/merge/deserialize growth, and foreachv2 inheritance were checked. The generic-type predicate was rechecked across complex and fallback datatype families; no missing parallel implementation or unjustified condition survived.
• Tests and results: The regression output is deterministic and its 3,5,2,1 cardinalities match the inserted rows; exact semantic values and NULL placement are covered by BE unit tests. I did not run builds or tests because the review-runner instructions prohibit them. At submission time, Clang Formatter and CheckStyle are green, while the macOS BE UT job is still in progress; the author also reports the focused ASAN aggregate tests passing in the resolved threads.
• Observability: This local intermediate-state fix does not require a new log or metric. Existing query error propagation and memory tracking cover the relevant failures.
• Transactions, persistence, and data writes: No transaction, EditLog, metadata persistence, table write, visible-version, delete-bitmap, atomicity, failover, or crash-recovery path is modified.
• FE-BE propagation: No new thrift field or cross-layer variable is introduced. The existing BE execution version reaches all producer/consumer paths that use the new native state serializer.
• Performance: The current head removes the full untracked scratch copy and restores the generic state to one MutableColumnPtr. Keeping old foreach states alive until replacement publication is the required peak-memory tradeoff for strong exception safety; no distinct time-complexity or hot-path regression remained.
• Other issues and user focus: No additional user focus was supplied, so the full PR was reviewed. Two normal full-review tracks and a separate risk-focused track all returned NO_NEW_VALUABLE_FINDINGS; the main-agent final changed-file, live-thread, and unresolved-candidate sweep agreed. From the code-review perspective, the current head is ready to proceed once the repository's normal CI policy is satisfied.