This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c1d91b1cc22 [fix](be) Validate serialized HLL payloads and row counts
(#69512)
c1d91b1cc22 is described below
commit c1d91b1cc22e427b24279f3256e29c1e294d27b8
Author: daidai <[email protected]>
AuthorDate: Mon Oct 26 12:29:52 2026 +0800
[fix](be) Validate serialized HLL payloads and row counts (#69512)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
Add bounds and consistency validation while deserializing HLL and
nullable HLL columns. Pass the actual buffer end through block decoding,
validate StreamVByte control and data lengths for compressed nullable
null maps, and verify row counts before Arrow conversion. When block
deserialization returns an error, release exchange queue accounting and
invoke the deferred completion callback.
### Release note
None
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- 22 focused BE tests passed with ASAN, including a 257-row compressed
nullable HLL case and a rejected block under backpressure.
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes.
- Invalid serialized HLL values, compressed null maps, and inconsistent
row counts are rejected; exchange accounting and completion callbacks
are released on the error path.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/core/block/block.cpp | 21 ++++-
be/src/core/data_type/data_type.h | 6 ++
be/src/core/data_type/data_type_hll.cpp | 60 +++++++++++-
be/src/core/data_type/data_type_hll.h | 2 +
be/src/core/data_type/data_type_nullable.cpp | 94 +++++++++++++++++-
be/src/core/data_type/data_type_nullable.h | 2 +
be/src/core/value/hll.cpp | 42 +++++++--
be/src/core/value/hll.h | 5 +-
be/src/exec/exchange/vdata_stream_recvr.cpp | 31 +++---
be/src/format/arrow/arrow_block_convertor.cpp | 11 ++-
be/test/core/data_type/data_type_hll_test.cpp | 105 ++++++++++++++++++++-
be/test/core/value/hll_test.cpp | 49 ++++++++++
be/test/exec/pipeline/vdata_stream_recvr_test.cpp | 48 ++++++++++
.../format/arrow/arrow_block_convertor_test.cpp | 18 ++++
14 files changed, 457 insertions(+), 37 deletions(-)
diff --git a/be/src/core/block/block.cpp b/be/src/core/block/block.cpp
index 63a19544b7c..930a639ab16 100644
--- a/be/src/core/block/block.cpp
+++ b/be/src/core/block/block.cpp
@@ -151,6 +151,7 @@ Status Block::deserialize(const PBlock& pblock, size_t*
uncompressed_bytes,
RETURN_IF_ERROR(BeExecVersionManager::check_be_exec_version(be_exec_version));
const char* buf = nullptr;
+ const char* buf_end = nullptr;
std::string compression_scratch;
if (pblock.compressed()) {
// Decompress
@@ -167,20 +168,29 @@ Status Block::deserialize(const PBlock& pblock, size_t*
uncompressed_bytes,
Slice decompressed_slice(compression_scratch);
RETURN_IF_ERROR(codec->decompress(Slice(compressed_data,
compressed_size),
&decompressed_slice));
- DCHECK(uncompressed_size == decompressed_slice.size);
+ if (decompressed_slice.size != uncompressed_size) {
+ return Status::Corruption("Decompressed block size mismatch:
expected {}, got {}",
+ uncompressed_size,
decompressed_slice.size);
+ }
} else {
bool success = snappy::GetUncompressedLength(compressed_data,
compressed_size,
&uncompressed_size);
- DCHECK(success) << "snappy::GetUncompressedLength failed";
+ if (!success) {
+ return Status::Corruption("snappy::GetUncompressedLength
failed");
+ }
compression_scratch.resize(uncompressed_size);
success = snappy::RawUncompress(compressed_data, compressed_size,
compression_scratch.data());
- DCHECK(success) << "snappy::RawUncompress failed";
+ if (!success) {
+ return Status::Corruption("snappy::RawUncompress failed");
+ }
}
*uncompressed_bytes = uncompressed_size;
buf = compression_scratch.data();
+ buf_end = buf + compression_scratch.size();
} else {
buf = pblock.column_values().data();
+ buf_end = buf + pblock.column_values().size();
}
for (const auto& pcol_meta : pblock.column_metas()) {
@@ -188,7 +198,10 @@ Status Block::deserialize(const PBlock& pblock, size_t*
uncompressed_bytes,
MutableColumnPtr data_column = type->create_column();
// Here will try to allocate large memory, should return error if
failed.
RETURN_IF_CATCH_EXCEPTION(
- buf = type->deserialize(buf, &data_column,
pblock.be_exec_version()));
+ buf = type->deserialize(buf, buf_end, &data_column,
pblock.be_exec_version()));
+ if (buf == nullptr || buf > buf_end) {
+ return Status::Corruption("Column deserializer advanced beyond the
input buffer");
+ }
data.emplace_back(data_column->get_ptr(), type, pcol_meta.name());
}
diff --git a/be/src/core/data_type/data_type.h
b/be/src/core/data_type/data_type.h
index 8948b6072bc..48542a430be 100644
--- a/be/src/core/data_type/data_type.h
+++ b/be/src/core/data_type/data_type.h
@@ -157,6 +157,12 @@ public:
virtual char* serialize(const IColumn& column, char* buf, int
be_exec_version) const = 0;
virtual const char* deserialize(const char* buf, MutableColumnPtr* column,
int be_exec_version) const = 0;
+ // Variable-length types can override this overload to validate embedded
lengths
+ // against end. Fixed-size types can use the default forwarding.
+ virtual const char* deserialize(const char* buf, const char* end,
MutableColumnPtr* column,
+ int be_exec_version) const {
+ return deserialize(buf, column, be_exec_version);
+ }
virtual void to_pb_column_meta(PColumnMeta* col_meta) const;
diff --git a/be/src/core/data_type/data_type_hll.cpp
b/be/src/core/data_type/data_type_hll.cpp
index 88b0f48d188..a37f6b749a9 100644
--- a/be/src/core/data_type/data_type_hll.cpp
+++ b/be/src/core/data_type/data_type_hll.cpp
@@ -28,6 +28,7 @@
#include "core/column/column_const.h"
#include "core/string_buffer.hpp"
#include "util/slice.h"
+#include "util/unaligned.h"
namespace doris {
@@ -69,10 +70,52 @@ char* DataTypeHLL::serialize(const IColumn& column, char*
buf, int be_exec_versi
// second: hll1 | hll2 | ...
const char* DataTypeHLL::deserialize(const char* buf, MutableColumnPtr* column,
int be_exec_version) const {
+ return deserialize(buf, nullptr, column, be_exec_version);
+}
+
+const char* DataTypeHLL::deserialize(const char* buf, const char* end,
MutableColumnPtr* column,
+ int be_exec_version) const {
auto* origin_column = column->get();
+ constexpr size_t header_size = sizeof(bool) + 2 * sizeof(size_t);
+ if (end != nullptr) {
+ if (end < buf || static_cast<size_t>(end - buf) < header_size) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed HLL column: missing serialized header");
+ }
+ uint8_t is_const_flag = unaligned_load<uint8_t>(buf);
+ if (is_const_flag > 1) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed HLL column: invalid const flag {}",
is_const_flag);
+ }
+ bool is_const = is_const_flag != 0;
+ size_t row_num = unaligned_load<size_t>(buf + sizeof(bool));
+ size_t saved_num = unaligned_load<size_t>(buf + sizeof(bool) +
sizeof(size_t));
+ if (saved_num != (is_const ? 1 : row_num)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed HLL column: inconsistent row count {},
saved count {}",
+ row_num, saved_num);
+ }
+ }
+
size_t real_have_saved_num = 0;
buf = deserialize_const_flag_and_row_num(buf, column,
&real_have_saved_num);
+ if (end != nullptr) {
+ size_t remaining = static_cast<size_t>(end - buf);
+ if (real_have_saved_num > remaining / sizeof(size_t)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed HLL column: length array exceeds
input");
+ }
+ size_t size_array_bytes = real_have_saved_num * sizeof(size_t);
+ size_t data_bytes = remaining - size_array_bytes;
+ // Every serialized HLL occupies at least one byte. Check this before
resizing
+ // the column.
+ if (real_have_saved_num > data_bytes) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed HLL column: row count exceeds input");
+ }
+ }
+
auto& data_column = assert_cast<ColumnHLL&>(*origin_column);
auto& data = data_column.get_data();
data.resize(real_have_saved_num);
@@ -85,8 +128,21 @@ const char* DataTypeHLL::deserialize(const char* buf,
MutableColumnPtr* column,
buf += sizeof(size_t) * real_have_saved_num;
for (int i = 0; i < real_have_saved_num; ++i) {
- data[i].deserialize(Slice(buf, hll_size_array[i]));
- buf += hll_size_array[i];
+ size_t hll_size = hll_size_array[i];
+ if (hll_size > HLL_COLUMN_DEFAULT_LEN) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed HLL column: value {} is too large: {}",
i, hll_size);
+ }
+ if (end != nullptr && hll_size > static_cast<size_t>(end - buf)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed HLL column: value {} exceeds input: {}
> {}", i, hll_size,
+ end - buf);
+ }
+ if (!data[i].deserialize(Slice(buf, hll_size))) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed HLL column: failed to deserialize value
{}", i);
+ }
+ buf += hll_size;
}
return buf;
}
diff --git a/be/src/core/data_type/data_type_hll.h
b/be/src/core/data_type/data_type_hll.h
index fb172e7e2b3..56c9f2340fc 100644
--- a/be/src/core/data_type/data_type_hll.h
+++ b/be/src/core/data_type/data_type_hll.h
@@ -59,6 +59,8 @@ public:
char* serialize(const IColumn& column, char* buf, int be_exec_version)
const override;
const char* deserialize(const char* buf, MutableColumnPtr* column,
int be_exec_version) const override;
+ const char* deserialize(const char* buf, const char* end,
MutableColumnPtr* column,
+ int be_exec_version) const override;
MutableColumnPtr create_column() const override;
Status check_column(const IColumn& column) const override;
diff --git a/be/src/core/data_type/data_type_nullable.cpp
b/be/src/core/data_type/data_type_nullable.cpp
index 77250f5cdbe..fd6a6398d08 100644
--- a/be/src/core/data_type/data_type_nullable.cpp
+++ b/be/src/core/data_type/data_type_nullable.cpp
@@ -103,11 +103,91 @@ 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,
+ MutableColumnPtr* column, int
be_exec_version) const {
auto* origin_column = column->get();
+ constexpr size_t header_size = sizeof(bool) + 2 * sizeof(size_t);
+ if (end != nullptr) {
+ if (end < buf || static_cast<size_t>(end - buf) < header_size) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: missing serialized
header");
+ }
+ uint8_t is_const_flag = unaligned_load<uint8_t>(buf);
+ if (is_const_flag > 1) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: invalid const flag
{}", is_const_flag);
+ }
+ bool is_const = is_const_flag != 0;
+ size_t row_num = unaligned_load<size_t>(buf + sizeof(bool));
+ size_t saved_num = unaligned_load<size_t>(buf + sizeof(bool) +
sizeof(size_t));
+ if (saved_num != (is_const ? 1 : row_num)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: inconsistent row count
{}, saved count {}",
+ row_num, saved_num);
+ }
+ }
+
size_t real_have_saved_num = 0;
buf = deserialize_const_flag_and_row_num(buf, column,
&real_have_saved_num);
- auto mem_size = real_have_saved_num * sizeof(bool);
+ if (end != nullptr) {
+ size_t remaining = static_cast<size_t>(end - buf);
+ size_t null_map_size = real_have_saved_num * sizeof(bool);
+ if (null_map_size <= SERIALIZED_MEM_SIZE_LIMIT) {
+ if (null_map_size > remaining) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: truncated null
map");
+ }
+ } else {
+ if (remaining < sizeof(size_t)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: missing null map
length");
+ }
+ size_t encode_size = unaligned_load<size_t>(buf);
+ size_t payload_remaining = remaining - sizeof(size_t);
+ if (encode_size > payload_remaining ||
+ payload_remaining - encode_size < STREAMVBYTE_PADDING) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: encoded null map
exceeds input");
+ }
+ const uint8_t* encoded_data = reinterpret_cast<const uint8_t*>(buf
+ sizeof(size_t));
+ uint32_t value_count =
cast_set<UInt32>(upper_int32(null_map_size));
+ size_t control_bytes = (static_cast<size_t>(value_count) + 3) / 4;
+ if (control_bytes > encode_size) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: null map control
bytes exceed input");
+ }
+
+ // StreamVByte stores one control byte per four integers, and each
+ // two-bit code selects one to four data bytes for its integer.
+ size_t expected_encode_size = control_bytes;
+ size_t remaining_values = value_count;
+ for (size_t i = 0; i < control_bytes; ++i) {
+ uint8_t control = encoded_data[i];
+ size_t values_in_group = std::min<size_t>(remaining_values, 4);
+ for (size_t j = 0; j < values_in_group; ++j) {
+ size_t value_bytes = ((control >> (j * 2)) & 0x03) + 1;
+ if (value_bytes > encode_size - expected_encode_size) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: null map
data exceeds input");
+ }
+ expected_encode_size += value_bytes;
+ }
+ remaining_values -= values_in_group;
+ }
+ if (expected_encode_size != encode_size) {
+ throw Exception(
+ ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: encoded null map size
mismatch: {} != {}",
+ expected_encode_size, encode_size);
+ }
+ }
+ }
+
+ size_t mem_size = real_have_saved_num * sizeof(bool);
auto* col = assert_cast<ColumnNullable*>(origin_column);
// A nullable column can be exclusive while its subcolumns are still shared
// after a shallow COW clone. Detach both owner slots before writing into
them.
@@ -124,12 +204,18 @@ const char* DataTypeNullable::deserialize(const char*
buf, MutableColumnPtr* col
size_t encode_size = unaligned_load<size_t>(buf);
buf += sizeof(size_t);
// Throw exception if mem_size is large than UINT32_MAX
- streamvbyte_decode((const uint8_t*)buf,
(uint32_t*)(null_map_data.data()),
- cast_set<UInt32>(upper_int32(mem_size)));
+ uint32_t value_count = cast_set<UInt32>(upper_int32(mem_size));
+ size_t decoded_bytes = streamvbyte_decode((const uint8_t*)buf,
+
(uint32_t*)(null_map_data.data()), value_count);
+ if (decoded_bytes != encode_size) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Malformed nullable column: decoded null map size
mismatch: {} != {}",
+ decoded_bytes, encode_size);
+ }
buf += encode_size;
}
- buf = nested_data_type->deserialize(buf, &nested, be_exec_version);
+ buf = nested_data_type->deserialize(buf, end, &nested, be_exec_version);
col->replace_columns(std::move(nested), std::move(null_map));
return buf;
}
diff --git a/be/src/core/data_type/data_type_nullable.h
b/be/src/core/data_type/data_type_nullable.h
index a506fb9fb56..beaeac6b869 100644
--- a/be/src/core/data_type/data_type_nullable.h
+++ b/be/src/core/data_type/data_type_nullable.h
@@ -68,6 +68,8 @@ public:
char* serialize(const IColumn& column, char* buf, int be_exec_version)
const override;
const char* deserialize(const char* buf, MutableColumnPtr* column,
int be_exec_version) const override;
+ const char* deserialize(const char* buf, const char* end,
MutableColumnPtr* column,
+ int be_exec_version) const override;
void to_pb_column_meta(PColumnMeta* col_meta) const override;
MutableColumnPtr create_column() const override;
diff --git a/be/src/core/value/hll.cpp b/be/src/core/value/hll.cpp
index 17151ad4335..b72147806ae 100644
--- a/be/src/core/value/hll.cpp
+++ b/be/src/core/value/hll.cpp
@@ -211,32 +211,56 @@ size_t HyperLogLog::serialize(uint8_t* dst) const {
}
bool HyperLogLog::is_valid(const Slice& slice) {
- if (slice.size < 1) {
+ if (slice.data == nullptr || slice.size < 1) {
return false;
}
- const uint8_t* ptr = (uint8_t*)slice.data;
- const uint8_t* end = (uint8_t*)slice.data + slice.size;
+ const auto* ptr = reinterpret_cast<const uint8_t*>(slice.data);
+ const auto* end = ptr + slice.size;
auto type = (HllDataType)*ptr++;
switch (type) {
case HLL_DATA_EMPTY:
break;
case HLL_DATA_EXPLICIT: {
- if ((ptr + 1) > end) {
+ if (static_cast<size_t>(end - ptr) < 1) {
return false;
}
uint8_t num_explicits = *ptr++;
- ptr += num_explicits * 8;
+ size_t values_size = static_cast<size_t>(num_explicits) *
sizeof(uint64_t);
+ if (num_explicits > HLL_EXPLICIT_INT64_NUM ||
+ values_size > static_cast<size_t>(end - ptr)) {
+ return false;
+ }
+ ptr += values_size;
break;
}
case HLL_DATA_SPARSE: {
- if ((ptr + 4) > end) {
+ if (static_cast<size_t>(end - ptr) < sizeof(uint32_t)) {
return false;
}
uint32_t num_registers = decode_fixed32_le(ptr);
- ptr += 4 + 3 * num_registers;
+ ptr += sizeof(uint32_t);
+ // Reject counts outside the range produced by serialize() before
computing
+ // the payload size.
+ if (num_registers > HLL_SPARSE_THRESHOLD) {
+ return false;
+ }
+ size_t entries_size = static_cast<size_t>(num_registers) * 3;
+ if (entries_size > static_cast<size_t>(end - ptr)) {
+ return false;
+ }
+ for (uint32_t i = 0; i < num_registers; ++i) {
+ // A SPARSE entry index must address one of the register array
entries.
+ if (decode_fixed16_le(ptr) >= HLL_REGISTERS_COUNT) {
+ return false;
+ }
+ ptr += 3;
+ }
break;
}
case HLL_DATA_FULL: {
+ if (HLL_REGISTERS_COUNT > static_cast<size_t>(end - ptr)) {
+ return false;
+ }
ptr += HLL_REGISTERS_COUNT;
break;
}
@@ -293,6 +317,10 @@ bool HyperLogLog::deserialize(const Slice& slice) {
// 1 byte: register value
uint16_t register_idx = decode_fixed16_le(ptr);
ptr += 2;
+ // Keep the index check next to the register assignment.
+ if (register_idx >= HLL_REGISTERS_COUNT) {
+ return false;
+ }
_registers[register_idx] = *ptr++;
}
break;
diff --git a/be/src/core/value/hll.h b/be/src/core/value/hll.h
index f285ba2f2f1..f4057ce727f 100644
--- a/be/src/core/value/hll.h
+++ b/be/src/core/value/hll.h
@@ -228,9 +228,8 @@ public:
static HyperLogLog empty() { return HyperLogLog {}; }
- // Check if input slice is a valid serialized binary of HyperLogLog.
- // This function only check the encoded type in slice, whose complex
- // function is O(1).
+ // Check that the serialized fields fit in the slice and every SPARSE
register
+ // index is in range. SPARSE validation is linear in its number of entries.
static bool is_valid(const Slice& slice);
// only for debug
diff --git a/be/src/exec/exchange/vdata_stream_recvr.cpp
b/be/src/exec/exchange/vdata_stream_recvr.cpp
index f54baf2221d..f248549e222 100644
--- a/be/src/exec/exchange/vdata_stream_recvr.cpp
+++ b/be/src/exec/exchange/vdata_stream_recvr.cpp
@@ -90,22 +90,27 @@ 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);
- 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();
+ }
+ }
}
- }
+ block_item.call_done(_recvr);
+ }};
- block_item.call_done(_recvr);
+ BlockUPtr next_block;
+ RETURN_IF_ERROR(block_item.get_block(next_block));
DCHECK(block->empty());
block->swap(*next_block);
diff --git a/be/src/format/arrow/arrow_block_convertor.cpp
b/be/src/format/arrow/arrow_block_convertor.cpp
index 596d0bcdc1b..bfa4e08ad6e 100644
--- a/be/src/format/arrow/arrow_block_convertor.cpp
+++ b/be/src/format/arrow/arrow_block_convertor.cpp
@@ -359,17 +359,22 @@ Status ArrowBlockConvertor::convert_to_arrow(const Block&
block, arrow::MemoryPo
if (block.columns() != num_fields) {
return Status::InvalidArgument("number fields not match");
}
- const size_t actual_end = end_row == 0 ? block.rows() : end_row;
+ const size_t declared_rows = block.rows();
+ const size_t actual_end = end_row == 0 ? declared_rows : end_row;
// Validate endpoints before subtraction: an inverted unsigned range
otherwise wraps.
- if (start_row > actual_end || actual_end > block.rows()) {
+ if (start_row > actual_end || actual_end > declared_rows) {
return Status::InvalidArgument("Row range out of bounds: start={},
end={}, block_rows={}",
- start_row, actual_end, block.rows());
+ start_row, actual_end, declared_rows);
}
const size_t actual_rows = actual_end - start_row;
std::vector<std::shared_ptr<arrow::Array>> arrays(num_fields);
for (int idx = 0; idx < num_fields; ++idx) {
const auto& entry = block.get_by_position(idx);
auto column = entry.column->convert_to_full_column_if_const();
+ if (column->size() != declared_rows) {
+ return Status::InvalidArgument("Column {} has {} rows, but block
declares {} rows", idx,
+ column->size(), declared_rows);
+ }
auto builder_arrow_type = schema->field(idx)->type();
if (builder_arrow_type->id() == arrow::Type::STRING &&
column->byte_size() >= MAX_ARROW_UTF8) {
diff --git a/be/test/core/data_type/data_type_hll_test.cpp
b/be/test/core/data_type/data_type_hll_test.cpp
index b68a5e60d0f..79cf723fcd2 100644
--- a/be/test/core/data_type/data_type_hll_test.cpp
+++ b/be/test/core/data_type/data_type_hll_test.cpp
@@ -21,6 +21,7 @@
#include <gtest/gtest-test-part.h>
#include <gtest/gtest.h>
+#include <cstring>
#include <iostream>
#include "agent/be_exec_version_manager.h"
@@ -131,6 +132,14 @@ TEST_P(DataTypeHLLTest, SerializeDeserializeTest) {
auto* column_res = assert_cast<ColumnHLL*>(column2.get());
ASSERT_EQ(column->get_data()[i].to_string(),
column_res->get_data()[i].to_string());
}
+
+ auto column3 = dt_hll->create_column();
+ dt_hll->deserialize(buf.get(), buf.get() + size, &column3,
+ BeExecVersionManager::get_newest_version());
+ for (size_t i = 0; i != rows_value; ++i) {
+ auto* column_res = assert_cast<ColumnHLL*>(column3.get());
+ ASSERT_EQ(column->get_data()[i].to_string(),
column_res->get_data()[i].to_string());
+ }
helper->serialize_deserialize_assert(hll_cols, {dt_hll});
std::cout << "finish serialize deserialize test" << std::endl;
}
@@ -156,6 +165,100 @@ TEST_P(DataTypeHLLTest, SerializeDeserializeAsStreamTest)
{
std::cout << "finish serialize deserialize as stream test" << std::endl;
}
+TEST(DataTypeHLLBoundsTest, RejectTruncatedValue) {
+ auto dt_hll =
+
DataTypeFactory::instance().create_data_type(FieldType::OLAP_FIELD_TYPE_HLL, 0,
0);
+ std::vector<char> buf(sizeof(bool) + 2 * sizeof(size_t) + 2 *
sizeof(size_t) + 1);
+ char* ptr = buf.data();
+ bool is_const = false;
+ memcpy(ptr, &is_const, sizeof(is_const));
+ ptr += sizeof(is_const);
+ size_t row_num = 1;
+ memcpy(ptr, &row_num, sizeof(row_num));
+ ptr += sizeof(row_num);
+ size_t saved_num = 1;
+ memcpy(ptr, &saved_num, sizeof(saved_num));
+ ptr += sizeof(saved_num);
+ size_t hll_size = HLL_COLUMN_DEFAULT_LEN;
+ memcpy(ptr, &hll_size, sizeof(hll_size));
+ ptr += sizeof(hll_size);
+ *ptr = HLL_DATA_FULL;
+
+ auto column = dt_hll->create_column();
+ EXPECT_THROW(dt_hll->deserialize(buf.data(), buf.data() + buf.size(),
&column,
+
BeExecVersionManager::get_newest_version()),
+ doris::Exception);
+}
+
+TEST(DataTypeHLLBoundsTest, RejectInconsistentRowCount) {
+ auto dt_hll =
+
DataTypeFactory::instance().create_data_type(FieldType::OLAP_FIELD_TYPE_HLL, 0,
0);
+ std::vector<char> buf(sizeof(bool) + 2 * sizeof(size_t));
+ char* ptr = buf.data();
+ bool is_const = false;
+ memcpy(ptr, &is_const, sizeof(is_const));
+ ptr += sizeof(is_const);
+ size_t row_num = 2;
+ memcpy(ptr, &row_num, sizeof(row_num));
+ ptr += sizeof(row_num);
+ size_t saved_num = 1;
+ memcpy(ptr, &saved_num, sizeof(saved_num));
+
+ auto column = dt_hll->create_column();
+ EXPECT_THROW(dt_hll->deserialize(buf.data(), buf.data() + buf.size(),
&column,
+
BeExecVersionManager::get_newest_version()),
+ doris::Exception);
+}
+
+TEST(DataTypeHLLBoundsTest, AcceptCompressedNullableHLL) {
+ auto dt_hll =
+
DataTypeFactory::instance().create_data_type(FieldType::OLAP_FIELD_TYPE_HLL, 0,
0);
+ auto dt_nullable = std::make_shared<DataTypeNullable>(dt_hll);
+
+ auto nested = dt_nullable->get_nested_type()->create_column();
+ nested->insert_many_defaults(257);
+ auto nullable = ColumnNullable::create(std::move(nested),
ColumnUInt8::create(257, 0));
+
+ std::vector<char> buf(dt_nullable->get_uncompressed_serialized_bytes(
+ *nullable, BeExecVersionManager::get_newest_version()));
+ char* end = dt_nullable->serialize(*nullable, buf.data(),
+
BeExecVersionManager::get_newest_version());
+
+ auto decoded = dt_nullable->create_column();
+ dt_nullable->deserialize(buf.data(), end, &decoded,
BeExecVersionManager::get_newest_version());
+ EXPECT_EQ(decoded->size(), 257);
+}
+
+TEST(DataTypeHLLBoundsTest, RejectInvalidCompressedNullMap) {
+ constexpr size_t nested_header_size = sizeof(bool) + 2 * sizeof(size_t);
+ constexpr size_t encoded_prefix_size = sizeof(size_t) + 1;
+ auto dt_hll =
+
DataTypeFactory::instance().create_data_type(FieldType::OLAP_FIELD_TYPE_HLL, 0,
0);
+ auto dt_nullable = std::make_shared<DataTypeNullable>(dt_hll);
+
+ std::vector<char> buf(
+ 2 * sizeof(size_t) + sizeof(bool) + encoded_prefix_size +
nested_header_size, 0);
+ char* ptr = buf.data();
+ bool is_const = false;
+ memcpy(ptr, &is_const, sizeof(is_const));
+ ptr += sizeof(is_const);
+ size_t row_num = 257;
+ memcpy(ptr, &row_num, sizeof(row_num));
+ ptr += sizeof(row_num);
+ size_t saved_num = 257;
+ memcpy(ptr, &saved_num, sizeof(saved_num));
+ ptr += sizeof(saved_num);
+ size_t encode_size = 1;
+ memcpy(ptr, &encode_size, sizeof(encode_size));
+ ptr += sizeof(encode_size);
+ *ptr = 0;
+
+ auto column = dt_nullable->create_column();
+ EXPECT_THROW(dt_nullable->deserialize(buf.data(), buf.data() + buf.size(),
&column,
+
BeExecVersionManager::get_newest_version()),
+ doris::Exception);
+}
+
INSTANTIATE_TEST_SUITE_P(Params, DataTypeHLLTest, ::testing::Values(0, 1, 10,
100));
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/test/core/value/hll_test.cpp b/be/test/core/value/hll_test.cpp
index f4685f6048c..84cacd9d43e 100644
--- a/be/test/core/value/hll_test.cpp
+++ b/be/test/core/value/hll_test.cpp
@@ -20,7 +20,12 @@
#include <gtest/gtest-message.h>
#include <gtest/gtest-test-part.h>
+#include <cstdint>
+#include <limits>
+#include <vector>
+
#include "gtest/gtest_pred_impl.h"
+#include "util/coding.h"
#include "util/hash_util.hpp"
#include "util/slice.h"
@@ -206,4 +211,48 @@ TEST_F(TestHll, InvalidPtr) {
}
}
+TEST_F(TestHll, SparseRegisterIndexBoundary) {
+ uint8_t buf[8];
+ uint8_t* ptr = buf;
+ *ptr++ = HLL_DATA_SPARSE;
+ encode_fixed32_le(ptr, 1);
+ ptr += sizeof(uint32_t);
+ encode_fixed16_le(ptr, HLL_REGISTERS_COUNT - 1);
+ ptr += sizeof(uint16_t);
+ *ptr++ = 1;
+
+ Slice valid(reinterpret_cast<const char*>(buf), static_cast<size_t>(ptr -
buf));
+ EXPECT_TRUE(HyperLogLog::is_valid(valid));
+ HyperLogLog valid_hll;
+ EXPECT_TRUE(valid_hll.deserialize(valid));
+
+ encode_fixed16_le(buf + 5, HLL_REGISTERS_COUNT);
+ Slice invalid(reinterpret_cast<const char*>(buf), static_cast<size_t>(ptr
- buf));
+ EXPECT_FALSE(HyperLogLog::is_valid(invalid));
+ HyperLogLog invalid_hll;
+ EXPECT_FALSE(invalid_hll.deserialize(invalid));
+ EXPECT_EQ(0, invalid_hll.estimate_cardinality());
+}
+
+TEST_F(TestHll, RejectSparseCountOverflowAndOversizedPayload) {
+ std::vector<uint8_t> oversized(5 + 3 * (HLL_SPARSE_THRESHOLD + 1));
+ oversized[0] = HLL_DATA_SPARSE;
+ encode_fixed32_le(oversized.data() + 1, HLL_SPARSE_THRESHOLD + 1);
+ EXPECT_FALSE(HyperLogLog::is_valid(
+ Slice(reinterpret_cast<const char*>(oversized.data()),
oversized.size())));
+
+ uint8_t buf[5] = {HLL_DATA_SPARSE};
+ encode_fixed32_le(buf + 1, std::numeric_limits<uint32_t>::max());
+ EXPECT_FALSE(HyperLogLog::is_valid(Slice(reinterpret_cast<const
char*>(buf), sizeof(buf))));
+}
+
+TEST_F(TestHll, RejectTruncatedFullPayload) {
+ uint8_t buf[HLL_REGISTERS_COUNT] = {HLL_DATA_FULL};
+ Slice truncated(reinterpret_cast<const char*>(buf), sizeof(buf));
+ EXPECT_FALSE(HyperLogLog::is_valid(truncated));
+
+ HyperLogLog hll(truncated);
+ EXPECT_EQ(0, hll.estimate_cardinality());
+}
+
} // namespace doris
diff --git a/be/test/exec/pipeline/vdata_stream_recvr_test.cpp
b/be/test/exec/pipeline/vdata_stream_recvr_test.cpp
index 6355f03e6a5..844ee7ad165 100644
--- a/be/test/exec/pipeline/vdata_stream_recvr_test.cpp
+++ b/be/test/exec/pipeline/vdata_stream_recvr_test.cpp
@@ -27,12 +27,14 @@
#include <thread>
#include <vector>
+#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_number.h"
#include "exec/exchange/vdata_stream_mgr.h"
#include "exec/operator/multi_cast_data_streamer.h"
#include "exec/pipeline/dependency.h"
#include "testutil/column_helper.h"
#include "testutil/mock/mock_runtime_state.h"
+#include "util/unaligned.h"
namespace doris {
@@ -345,6 +347,25 @@ void to_pblock(Block& block, PBlock* pblock) {
segment_v2::CompressionTypePB::NO_COMPRESSION));
}
+std::unique_ptr<PBlock> make_invalid_hll_pblock() {
+ constexpr size_t hll_header_size = sizeof(bool) + 2 * sizeof(size_t);
+ constexpr uint8_t invalid_hll_type = 0x04;
+ auto hll_type =
+
DataTypeFactory::instance().create_data_type(FieldType::OLAP_FIELD_TYPE_HLL, 0,
0);
+ auto column = hll_type->create_column();
+ column->insert_default();
+
+ auto block = Block({ColumnWithTypeAndName(std::move(column),
std::move(hll_type), "hll")});
+ auto pblock = std::make_unique<PBlock>();
+ to_pblock(block, pblock.get());
+
+ auto* values = pblock->mutable_column_values();
+ DCHECK_GE(values->size(), hll_header_size + sizeof(size_t) + 1);
+ unaligned_store<size_t>(values->data() + hll_header_size, 1);
+ (*values)[hll_header_size + sizeof(size_t)] =
static_cast<char>(invalid_hll_type);
+ return pblock;
+}
+
void run_concurrent_remote_add_and_drain(VDataStreamRecvr::SenderQueue* sender,
MockVDataStreamRecvr* recvr, bool
use_cancel) {
constexpr int num_threads = 4;
@@ -527,6 +548,33 @@ TEST_F(DataStreamRecvrTest, TestRemoteMemLimitSender) {
sender->close();
}
+TEST_F(DataStreamRecvrTest, RejectedRemoteBlockReleasesCallbackAndAccounting) {
+ create_recvr(1, false);
+ recvr->always_exceeds_limit = true;
+ auto* sender = recvr->sender_queues().back();
+
+ int callback_count = 0;
+ MockClosure closure;
+ closure._cb = [&]() { ++callback_count; };
+ google::protobuf::Closure* done = &closure;
+
+ auto st = sender->add_block(make_invalid_hll_pblock(), 1, 1, &done, 0, 0);
+ ASSERT_TRUE(st) << st.msg();
+ ASSERT_EQ(done, nullptr);
+ ASSERT_GT(_mock_counter->current_value(), 0);
+
+ Block block;
+ bool eos = false;
+ st = sender->get_batch(&block, &eos);
+ EXPECT_FALSE(st.ok());
+ EXPECT_EQ(callback_count, 1);
+ EXPECT_EQ(_mock_counter->current_value(), 0);
+
+ recvr->close();
+ EXPECT_EQ(callback_count, 1);
+ EXPECT_EQ(_mock_counter->current_value(), 0);
+}
+
TEST_F(DataStreamRecvrTest, TestRemoteMultiSender) {
create_recvr(3, false);
EXPECT_EQ(recvr->sender_queues().size(), 1);
diff --git a/be/test/format/arrow/arrow_block_convertor_test.cpp
b/be/test/format/arrow/arrow_block_convertor_test.cpp
index f3db16d59de..16cd9c3146c 100644
--- a/be/test/format/arrow/arrow_block_convertor_test.cpp
+++ b/be/test/format/arrow/arrow_block_convertor_test.cpp
@@ -143,6 +143,24 @@ TEST_F(ArrowBlockConvertorTest,
PythonBuildsSchemaAndConvertsSlicesInternally) {
EXPECT_EQ(2, output.rows());
}
+TEST_F(ArrowBlockConvertorTest, RejectsColumnsWithDifferentRowCounts) {
+ auto type = std::make_shared<DataTypeInt32>();
+ auto first = ColumnInt32::create();
+ first->get_data().assign({1, 2});
+ auto second = ColumnInt32::create();
+ second->get_data().push_back(3);
+ Block block;
+ block.insert({std::move(first), type, "first"});
+ block.insert({std::move(second), type, "second"});
+
+ auto schema = arrow::schema(
+ {arrow::field("first", arrow::int32()), arrow::field("second",
arrow::int32())});
+ ArrowFlightArrowBlockConvertor converter(schema, cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ EXPECT_FALSE(converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch).ok());
+ EXPECT_EQ(nullptr, batch);
+}
+
TEST_F(ArrowBlockConvertorTest,
PaimonSerializedSchemaKeepsTimestampBindingsPerInstance) {
cctz::time_zone shanghai;
ASSERT_TRUE(cctz::load_time_zone("Asia/Shanghai", &shanghai));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
