github-actions[bot] commented on code in PR #66601:
URL: https://github.com/apache/doris/pull/66601#discussion_r3747684978
##########
be/src/exprs/aggregate/aggregate_function_array_agg.h:
##########
@@ -265,15 +268,28 @@ struct AggregateFunctionArrayAggData<T> {
}
void write(BufferWritable& buf) const {
- throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, "array_agg not
support write");
+ const auto serialized_bytes =
+ column_type->get_uncompressed_serialized_bytes(*column_data,
be_exec_version);
+ std::string serialized_buffer(serialized_bytes, '\0');
+ const auto* end =
+ column_type->serialize(*column_data, serialized_buffer.data(),
be_exec_version);
+ DORIS_CHECK_LE(end, serialized_buffer.data() + serialized_bytes);
+ serialized_buffer.resize(end - serialized_buffer.data());
+ buf.write_binary(serialized_buffer);
}
void read(BufferReadable& buf) {
- throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, "array_agg not
support read");
+ DORIS_CHECK(column_data->empty());
+ UInt64 serialized_bytes = 0;
+ buf.read_var_uint(serialized_bytes);
+ const auto* serialized_data = buf.data();
+ const auto* end = column_type->deserialize(serialized_data,
&column_data, be_exec_version);
+ DORIS_CHECK_EQ(end, serialized_data + serialized_bytes);
+ buf.add_offset(serialized_bytes);
}
void merge(const Self& rhs) {
- throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, "array_agg not
support merge");
+ column_data->insert_range_from(*rhs.column_data, 0,
rhs.column_data->size());
Review Comment:
**[P1] Make foreach growth exception-safe before allocating here**
`insert_range_from()` can throw `MEM_ALLOC_FAILED` while
`AggregateFunctionForEach::ensure_aggregate_data()` is relocating a
multi-position state. That loop merges and immediately destroys each old
position, but it does not publish `new_state` until every position finishes. If
position 0 succeeds and position 1 runs out of memory, the live foreach state
still points to the old buffer and old size even though position 0 was already
destroyed; ordinary query-error cleanup then destroys position 0 again, while
the constructed new states also miss destruction. The old complex merge always
failed on the first position, so this partial-success/double-destroy state is
newly reachable with this implementation. Please keep all old states alive
until every merge succeeds, destroy every constructed new state on failure, and
only publish the new buffer and destroy the old states after successful
relocation. A failure-injection test should throw after at least one position
has migrated.
##########
be/src/exprs/aggregate/aggregate_function_array_agg.h:
##########
@@ -265,15 +268,28 @@ struct AggregateFunctionArrayAggData<T> {
}
void write(BufferWritable& buf) const {
- throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, "array_agg not
support write");
+ const auto serialized_bytes =
+ column_type->get_uncompressed_serialized_bytes(*column_data,
be_exec_version);
+ std::string serialized_buffer(serialized_bytes, '\0');
+ const auto* end =
+ column_type->serialize(*column_data, serialized_buffer.data(),
be_exec_version);
+ DORIS_CHECK_LE(end, serialized_buffer.data() + serialized_bytes);
+ serialized_buffer.resize(end - serialized_buffer.data());
+ buf.write_binary(serialized_buffer);
}
void read(BufferReadable& buf) {
- throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, "array_agg not
support read");
+ DORIS_CHECK(column_data->empty());
+ UInt64 serialized_bytes = 0;
+ buf.read_var_uint(serialized_bytes);
+ const auto* serialized_data = buf.data();
+ const auto* end = column_type->deserialize(serialized_data,
&column_data, be_exec_version);
+ DORIS_CHECK_EQ(end, serialized_data + serialized_bytes);
+ buf.add_offset(serialized_bytes);
}
void merge(const Self& rhs) {
- throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, "array_agg not
support merge");
Review Comment:
**[P1] Make foreach growth exception-safe before allocating here**
`insert_range_from()` can throw `MEM_ALLOC_FAILED` while
`AggregateFunctionForEach::ensure_aggregate_data()` is relocating a
multi-position state. That loop merges and immediately destroys each old
position, but it does not publish `new_state` until every position finishes. If
position 0 succeeds and position 1 runs out of memory, the live foreach state
still points to the old buffer and old size even though position 0 was already
destroyed; ordinary query-error cleanup then destroys position 0 again, while
the constructed new states also miss destruction. The old complex merge always
failed on the first position, so this partial-success/double-destroy state is
newly reachable with this implementation. Please keep all old states alive
until every merge succeeds, destroy every constructed new state on failure, and
only publish the new buffer and destroy the old states after successful
relocation. A failure-injection test should throw after at least one position
has migrated.
##########
be/src/exprs/aggregate/aggregate_function_array_agg.h:
##########
@@ -285,15 +301,24 @@ class AggregateFunctionArrayAgg final
UnaryExpression,
NotNullableAggregateFunction {
public:
+ using Base = IAggregateFunctionDataHelper<Data,
AggregateFunctionArrayAgg<Data>, true>;
+
AggregateFunctionArrayAgg(const DataTypes& argument_types_)
- : IAggregateFunctionDataHelper<Data,
AggregateFunctionArrayAgg<Data>, true>(
- {argument_types_}),
+ : Base({argument_types_}),
return_type(std::make_shared<DataTypeArray>(make_nullable(argument_types_[0])))
{}
std::string get_name() const override { return "array_agg"; }
Review Comment:
**[P1] Make foreach growth exception-safe before allocating here**
`insert_range_from()` can throw `MEM_ALLOC_FAILED` while
`AggregateFunctionForEach::ensure_aggregate_data()` is relocating a
multi-position state. That loop merges and immediately destroys each old
position, but it does not publish `new_state` until every position finishes. If
position 0 succeeds and position 1 runs out of memory, the live foreach state
still points to the old buffer and old size even though position 0 was already
destroyed; ordinary query-error cleanup then destroys position 0 again, while
the constructed new states also miss destruction. The old complex merge always
failed on the first position, so this partial-success/double-destroy state is
newly reachable with this implementation. Please keep all old states alive
until every merge succeeds, destroy every constructed new state on failure, and
only publish the new buffer and destroy the old states after successful
relocation. A failure-injection test should throw after at least one position
has migrated.
##########
be/src/exprs/aggregate/aggregate_function_array_agg.h:
##########
@@ -285,15 +301,24 @@ class AggregateFunctionArrayAgg final
UnaryExpression,
NotNullableAggregateFunction {
public:
+ using Base = IAggregateFunctionDataHelper<Data,
AggregateFunctionArrayAgg<Data>, true>;
+
AggregateFunctionArrayAgg(const DataTypes& argument_types_)
- : IAggregateFunctionDataHelper<Data,
AggregateFunctionArrayAgg<Data>, true>(
Review Comment:
**[P1] Make foreach growth exception-safe before allocating here**
`insert_range_from()` can throw `MEM_ALLOC_FAILED` while
`AggregateFunctionForEach::ensure_aggregate_data()` is relocating a
multi-position state. That loop merges and immediately destroys each old
position, but it does not publish `new_state` until every position finishes. If
position 0 succeeds and position 1 runs out of memory, the live foreach state
still points to the old buffer and old size even though position 0 was already
destroyed; ordinary query-error cleanup then destroys position 0 again, while
the constructed new states also miss destruction. The old complex merge always
failed on the first position, so this partial-success/double-destroy state is
newly reachable with this implementation. Please keep all old states alive
until every merge succeeds, destroy every constructed new state on failure, and
only publish the new buffer and destroy the old states after successful
relocation. A failure-injection test should throw after at least one position
has migrated.
##########
be/src/exprs/aggregate/aggregate_function_array_agg.h:
##########
@@ -285,15 +301,24 @@ class AggregateFunctionArrayAgg final
UnaryExpression,
NotNullableAggregateFunction {
public:
+ using Base = IAggregateFunctionDataHelper<Data,
AggregateFunctionArrayAgg<Data>, true>;
+
AggregateFunctionArrayAgg(const DataTypes& argument_types_)
- : IAggregateFunctionDataHelper<Data,
AggregateFunctionArrayAgg<Data>, true>(
- {argument_types_}),
+ : Base({argument_types_}),
Review Comment:
**[P1] Make foreach growth exception-safe before allocating here**
`insert_range_from()` can throw `MEM_ALLOC_FAILED` while
`AggregateFunctionForEach::ensure_aggregate_data()` is relocating a
multi-position state. That loop merges and immediately destroys each old
position, but it does not publish `new_state` until every position finishes. If
position 0 succeeds and position 1 runs out of memory, the live foreach state
still points to the old buffer and old size even though position 0 was already
destroyed; ordinary query-error cleanup then destroys position 0 again, while
the constructed new states also miss destruction. The old complex merge always
failed on the first position, so this partial-success/double-destroy state is
newly reachable with this implementation. Please keep all old states alive
until every merge succeeds, destroy every constructed new state on failure, and
only publish the new buffer and destroy the old states after successful
relocation. A failure-injection test should throw after at least one position
has migrated.
--
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]
