This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch revert-69512-fix/hll-deserialize-bounds in repository https://gitbox.apache.org/repos/asf/doris.git
commit baa88f8678739989a6f610258b037b826e0652fc Author: yiguolei <[email protected]> AuthorDate: Wed Nov 4 16:53:50 2026 +0800 Revert "[fix](be) Validate serialized HLL payloads and row counts (#69512)" This reverts commit c1d91b1cc22e427b24279f3256e29c1e294d27b8. --- 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, 37 insertions(+), 457 deletions(-) diff --git a/be/src/core/block/block.cpp b/be/src/core/block/block.cpp index 930a639ab16..63a19544b7c 100644 --- a/be/src/core/block/block.cpp +++ b/be/src/core/block/block.cpp @@ -151,7 +151,6 @@ 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 @@ -168,29 +167,20 @@ 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)); - if (decompressed_slice.size != uncompressed_size) { - return Status::Corruption("Decompressed block size mismatch: expected {}, got {}", - uncompressed_size, decompressed_slice.size); - } + DCHECK(uncompressed_size == decompressed_slice.size); } else { bool success = snappy::GetUncompressedLength(compressed_data, compressed_size, &uncompressed_size); - if (!success) { - return Status::Corruption("snappy::GetUncompressedLength failed"); - } + DCHECK(success) << "snappy::GetUncompressedLength failed"; compression_scratch.resize(uncompressed_size); success = snappy::RawUncompress(compressed_data, compressed_size, compression_scratch.data()); - if (!success) { - return Status::Corruption("snappy::RawUncompress failed"); - } + DCHECK(success) << "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()) { @@ -198,10 +188,7 @@ 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, buf_end, &data_column, pblock.be_exec_version())); - if (buf == nullptr || buf > buf_end) { - return Status::Corruption("Column deserializer advanced beyond the input buffer"); - } + buf = type->deserialize(buf, &data_column, pblock.be_exec_version())); 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 48542a430be..8948b6072bc 100644 --- a/be/src/core/data_type/data_type.h +++ b/be/src/core/data_type/data_type.h @@ -157,12 +157,6 @@ 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 a37f6b749a9..88b0f48d188 100644 --- a/be/src/core/data_type/data_type_hll.cpp +++ b/be/src/core/data_type/data_type_hll.cpp @@ -28,7 +28,6 @@ #include "core/column/column_const.h" #include "core/string_buffer.hpp" #include "util/slice.h" -#include "util/unaligned.h" namespace doris { @@ -70,52 +69,10 @@ 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); @@ -128,21 +85,8 @@ const char* DataTypeHLL::deserialize(const char* buf, const char* end, MutableCo buf += sizeof(size_t) * real_have_saved_num; for (int i = 0; i < real_have_saved_num; ++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; + data[i].deserialize(Slice(buf, hll_size_array[i])); + buf += hll_size_array[i]; } 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 56c9f2340fc..fb172e7e2b3 100644 --- a/be/src/core/data_type/data_type_hll.h +++ b/be/src/core/data_type/data_type_hll.h @@ -59,8 +59,6 @@ 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 fd6a6398d08..77250f5cdbe 100644 --- a/be/src/core/data_type/data_type_nullable.cpp +++ b/be/src/core/data_type/data_type_nullable.cpp @@ -103,91 +103,11 @@ 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); - 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 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. @@ -204,18 +124,12 @@ const char* DataTypeNullable::deserialize(const char* buf, const char* end, size_t encode_size = unaligned_load<size_t>(buf); buf += sizeof(size_t); // Throw exception if mem_size is large than UINT32_MAX - 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); - } + streamvbyte_decode((const uint8_t*)buf, (uint32_t*)(null_map_data.data()), + cast_set<UInt32>(upper_int32(mem_size))); buf += encode_size; } - buf = nested_data_type->deserialize(buf, end, &nested, be_exec_version); + buf = nested_data_type->deserialize(buf, &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 beaeac6b869..a506fb9fb56 100644 --- a/be/src/core/data_type/data_type_nullable.h +++ b/be/src/core/data_type/data_type_nullable.h @@ -68,8 +68,6 @@ 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 b72147806ae..17151ad4335 100644 --- a/be/src/core/value/hll.cpp +++ b/be/src/core/value/hll.cpp @@ -211,56 +211,32 @@ size_t HyperLogLog::serialize(uint8_t* dst) const { } bool HyperLogLog::is_valid(const Slice& slice) { - if (slice.data == nullptr || slice.size < 1) { + if (slice.size < 1) { return false; } - const auto* ptr = reinterpret_cast<const uint8_t*>(slice.data); - const auto* end = ptr + slice.size; + const uint8_t* ptr = (uint8_t*)slice.data; + const uint8_t* end = (uint8_t*)slice.data + slice.size; auto type = (HllDataType)*ptr++; switch (type) { case HLL_DATA_EMPTY: break; case HLL_DATA_EXPLICIT: { - if (static_cast<size_t>(end - ptr) < 1) { + if ((ptr + 1) > end) { return false; } uint8_t num_explicits = *ptr++; - 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; + ptr += num_explicits * 8; break; } case HLL_DATA_SPARSE: { - if (static_cast<size_t>(end - ptr) < sizeof(uint32_t)) { + if ((ptr + 4) > end) { return false; } uint32_t num_registers = decode_fixed32_le(ptr); - 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; - } + ptr += 4 + 3 * num_registers; break; } case HLL_DATA_FULL: { - if (HLL_REGISTERS_COUNT > static_cast<size_t>(end - ptr)) { - return false; - } ptr += HLL_REGISTERS_COUNT; break; } @@ -317,10 +293,6 @@ 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 f4057ce727f..f285ba2f2f1 100644 --- a/be/src/core/value/hll.h +++ b/be/src/core/value/hll.h @@ -228,8 +228,9 @@ public: static HyperLogLog empty() { return HyperLogLog {}; } - // 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. + // 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). 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 ec606c6f79c..5b9f14048a1 100644 --- a/be/src/exec/exchange/vdata_stream_recvr.cpp +++ b/be/src/exec/exchange/vdata_stream_recvr.cpp @@ -103,27 +103,22 @@ 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(); - 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(); - } - } + 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); - }}; + } - BlockUPtr next_block; - RETURN_IF_ERROR(block_item.get_block(next_block)); + block_item.call_done(_recvr); 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 f7267d683a2..d08e73df369 100644 --- a/be/src/format/arrow/arrow_block_convertor.cpp +++ b/be/src/format/arrow/arrow_block_convertor.cpp @@ -537,22 +537,17 @@ 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 declared_rows = block.rows(); - const size_t actual_end = end_row == 0 ? declared_rows : end_row; + const size_t actual_end = end_row == 0 ? block.rows() : end_row; // Validate endpoints before subtraction: an inverted unsigned range otherwise wraps. - if (start_row > actual_end || actual_end > declared_rows) { + if (start_row > actual_end || actual_end > block.rows()) { return Status::InvalidArgument("Row range out of bounds: start={}, end={}, block_rows={}", - start_row, actual_end, declared_rows); + start_row, actual_end, block.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 79cf723fcd2..b68a5e60d0f 100644 --- a/be/test/core/data_type/data_type_hll_test.cpp +++ b/be/test/core/data_type/data_type_hll_test.cpp @@ -21,7 +21,6 @@ #include <gtest/gtest-test-part.h> #include <gtest/gtest.h> -#include <cstring> #include <iostream> #include "agent/be_exec_version_manager.h" @@ -132,14 +131,6 @@ 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; } @@ -165,100 +156,6 @@ 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 +} // namespace doris \ No newline at end of file diff --git a/be/test/core/value/hll_test.cpp b/be/test/core/value/hll_test.cpp index 84cacd9d43e..f4685f6048c 100644 --- a/be/test/core/value/hll_test.cpp +++ b/be/test/core/value/hll_test.cpp @@ -20,12 +20,7 @@ #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" @@ -211,48 +206,4 @@ 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 844ee7ad165..6355f03e6a5 100644 --- a/be/test/exec/pipeline/vdata_stream_recvr_test.cpp +++ b/be/test/exec/pipeline/vdata_stream_recvr_test.cpp @@ -27,14 +27,12 @@ #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 { @@ -347,25 +345,6 @@ 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; @@ -548,33 +527,6 @@ 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 b6fa589570b..b02b60cb7a7 100644 --- a/be/test/format/arrow/arrow_block_convertor_test.cpp +++ b/be/test/format/arrow/arrow_block_convertor_test.cpp @@ -144,24 +144,6 @@ 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]
