github-actions[bot] commented on code in PR #69682:
URL: https://github.com/apache/doris/pull/69682#discussion_r4387164661
##########
be/src/core/data_type/data_type_nullable.cpp:
##########
@@ -145,7 +200,39 @@ char* DataTypeNullable::serialize(const IColumn& column,
char* buf, int be_exec_
const char* DataTypeNullable::deserialize(const char* buf, MutableColumnPtr*
column,
int be_exec_version) const {
+ return deserialize(buf, nullptr, column, be_exec_version);
+}
+
+const char* DataTypeNullable::deserialize(const char* buf, const char* end,
Review Comment:
[P1] Forward the buffer end to nullable child deserializers
`Block::deserialize` calls this overload with `buf_end`, but both branches
still call `nested_data_type->deserialize(buf, &nested, be_exec_version)`
(lines 281 and 331). That selects `DataTypeHLL`'s three-argument overload,
which resets `end` to `nullptr`. A one-row `Nullable(HLL)` PBlock with valid
outer/inner headers can therefore declare a value length of 16385 but contain
only the `HLL_DATA_FULL` tag. The nested decoder attempts a 16384-byte copy
from truncated input before Block's final cursor check can reject it. Missing
child headers/length arrays also reach unchecked reads. Pass `end` to the child
decoder in both wire layouts and add truncated nullable-HLL coverage.
##########
be/src/exec/exchange/vdata_stream_recvr.cpp:
##########
@@ -96,34 +96,40 @@ Status VDataStreamRecvr::SenderQueue::get_batch(Block*
block, bool* eos) {
block_item = std::move(_block_queue.front());
_block_queue.pop_front();
}
- BlockUPtr next_block;
- RETURN_IF_ERROR(block_item.get_block(next_block));
+
size_t block_byte_size = block_item.block_byte_size();
- COUNTER_UPDATE(_recvr->_deserialize_row_batch_timer,
block_item.deserialize_time());
- COUNTER_UPDATE(_recvr->_decompress_timer, block_item.decompress_time());
- COUNTER_UPDATE(_recvr->_decompress_bytes, block_item.decompress_bytes());
- _recvr->_memory_used_counter->update(-(int64_t)block_byte_size);
- INJECT_MOCK_SLEEP(std::lock_guard<std::mutex> l(_lock));
- sub_blocks_memory_usage(block_byte_size);
- _record_debug_info();
- if (_block_queue.empty() && _source_dependency) {
- if (!_is_cancelled && _num_remaining_senders > 0) {
- _source_dependency->block();
+ Defer release_block {[&]() {
+ COUNTER_UPDATE(_recvr->_deserialize_row_batch_timer,
block_item.deserialize_time());
+ COUNTER_UPDATE(_recvr->_decompress_timer,
block_item.decompress_time());
+ COUNTER_UPDATE(_recvr->_decompress_bytes,
block_item.decompress_bytes());
+ _recvr->_memory_used_counter->update(-(int64_t)block_byte_size);
+ {
+ INJECT_MOCK_SLEEP(std::lock_guard<std::mutex> l(_lock));
+ sub_blocks_memory_usage(block_byte_size);
+ if (_block_queue.empty() && _source_dependency) {
+ if (!_is_cancelled && _num_remaining_senders > 0) {
+ _source_dependency->block();
+ }
+ }
}
- }
+ _record_debug_info();
- if (!_pending_closures.empty()) {
- auto closure_pair = _pending_closures.front();
- closure_pair.first->Run();
- int64_t elapse_time = closure_pair.second.elapsed_time();
- if (_recvr->_max_wait_to_process_time->value() < elapse_time) {
- _recvr->_max_wait_to_process_time->set(elapse_time);
+ if (!_pending_closures.empty()) {
Review Comment:
[P1] Detach pending callbacks while holding the queue lock
The new inner scope releases `_lock` before
`_pending_closures.empty/front/Run/pop_front`, although remote `add_block` and
`cancel` still mutate this deque under that lock. A pipeline consumer can copy
a retained closure here, then a concurrent error-EOS RPC can reach
`remove_sender -> cancel_stream -> cancel`, run that closure and clear the
deque. The consumer subsequently runs the same self-deleting BRPC closure again
and pops the cleared deque, causing use-after-free/undefined behavior. This
also affects successful block reads. Select and remove the callback under
`_lock`, then invoke the detached callback outside it; keep
`_record_debug_info` under its protecting lock as well.
##########
be/src/core/data_type/data_type_nullable.cpp:
##########
@@ -182,17 +294,37 @@ const char* DataTypeNullable::deserialize(const char*
buf, MutableColumnPtr* col
// row num
uint32_t mem_size = unaligned_load<uint32_t>(buf);
buf += sizeof(uint32_t);
+ size_t encode_size = 0;
+ if (end != nullptr) {
+ size_t remaining = static_cast<size_t>(end - buf);
+ if (mem_size > remaining) {
Review Comment:
[P1] Check the encoded length for compressed legacy null maps
For `be_exec_version < USE_CONST_SERDE`, `mem_size` is the uncompressed
null-map size, while `remaining` contains compressed bytes. A valid
single-column `Nullable(TINYINT)` block with 512 zero values and zero null
flags encodes each 512-byte stream into 160 bytes. After the outer row count,
only `8 + 160 + 4 + 8 + 160 + 16 = 356` bytes remain, including Block's
padding, so this check rejects the valid block as a truncated null map (`512 >
356`). This breaks supported legacy decoding even though both serializers
produced valid data. Apply this raw-size check only when `mem_size <=
SERIALIZED_MEM_SIZE_LIMIT`; otherwise use the encoded-map validator, as the
current-format branch already does.
--
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]
