github-actions[bot] commented on code in PR #66820:
URL: https://github.com/apache/doris/pull/66820#discussion_r3794098624
##########
be/src/util/thrift_util.cpp:
##########
@@ -56,6 +60,118 @@ class TProtocol;
#include <thread>
namespace doris {
+namespace {
+
+constexpr size_t DECODED_THRIFT_STRUCT_RESERVATION_BYTES = 1024;
+
+size_t decoded_thrift_value_reservation(apache::thrift::protocol::TType type) {
+ using apache::thrift::protocol::T_BOOL;
+ using apache::thrift::protocol::T_BYTE;
+ using apache::thrift::protocol::T_DOUBLE;
+ using apache::thrift::protocol::T_I16;
+ using apache::thrift::protocol::T_I32;
+ using apache::thrift::protocol::T_I64;
+ using apache::thrift::protocol::T_LIST;
+ using apache::thrift::protocol::T_MAP;
+ using apache::thrift::protocol::T_SET;
+ using apache::thrift::protocol::T_STRING;
+ using apache::thrift::protocol::T_STRUCT;
+ switch (type) {
+ case T_BOOL:
+ case T_BYTE:
+ return 1;
+ case T_I16:
+ return sizeof(int16_t);
+ case T_I32:
+ return sizeof(int32_t);
+ case T_I64:
+ case T_DOUBLE:
+ return sizeof(int64_t);
+ case T_STRING:
+ return sizeof(std::string);
+ case T_LIST:
+ case T_SET:
+ case T_MAP:
+ return sizeof(std::vector<uint8_t>);
+ case T_STRUCT:
+ // Generated structs vary in size. Reserving a conservative inline
object budget keeps
+ // their eager vector resize inside task admission; the
serialized-size reservation covers
+ // their dynamic field payloads.
+ return DECODED_THRIFT_STRUCT_RESERVATION_BYTES;
+ default:
+ return 1;
+ }
+}
+
+class MemoryBudgetProtocol final : public
apache::thrift::protocol::TProtocolDecorator {
+public:
+ MemoryBudgetProtocol(std::shared_ptr<apache::thrift::protocol::TProtocol>
protocol,
+ int32_t serialized_size)
+ : TProtocolDecorator(std::move(protocol)),
+ _memory_manager(thread_context()->thread_mem_tracker_mgr.get()),
Review Comment:
[P1] Keep this utility valid on contextless service threads
`MemoryBudgetProtocol` now calls `thread_context()` before reading anything,
but the backend `open_scanner` service uses `ThriftServer`'s default `THREADED`
mode. Those workers come from Apache Thrift's `ThreadFactory`, and the server
event hooks install only `_session_key`, not a Doris `ThreadLocalHandle`.
Consequently every valid `open_scanner()` plan reaches this line without a
context, throws, and is returned as an invalid scanner open. Please make
absent-context calls use a legitimate limiter (merely creating TLS leaves the
orphan tracker and still fails the reservation invariant), preserve any
attached task tracker, and add a real contextless threaded-service regression.
##########
be/src/util/thrift_util.cpp:
##########
@@ -56,6 +60,118 @@ class TProtocol;
#include <thread>
namespace doris {
+namespace {
+
+constexpr size_t DECODED_THRIFT_STRUCT_RESERVATION_BYTES = 1024;
+
+size_t decoded_thrift_value_reservation(apache::thrift::protocol::TType type) {
+ using apache::thrift::protocol::T_BOOL;
+ using apache::thrift::protocol::T_BYTE;
+ using apache::thrift::protocol::T_DOUBLE;
+ using apache::thrift::protocol::T_I16;
+ using apache::thrift::protocol::T_I32;
+ using apache::thrift::protocol::T_I64;
+ using apache::thrift::protocol::T_LIST;
+ using apache::thrift::protocol::T_MAP;
+ using apache::thrift::protocol::T_SET;
+ using apache::thrift::protocol::T_STRING;
+ using apache::thrift::protocol::T_STRUCT;
+ switch (type) {
+ case T_BOOL:
+ case T_BYTE:
+ return 1;
+ case T_I16:
+ return sizeof(int16_t);
+ case T_I32:
+ return sizeof(int32_t);
+ case T_I64:
+ case T_DOUBLE:
+ return sizeof(int64_t);
+ case T_STRING:
+ return sizeof(std::string);
+ case T_LIST:
+ case T_SET:
+ case T_MAP:
+ return sizeof(std::vector<uint8_t>);
+ case T_STRUCT:
+ // Generated structs vary in size. Reserving a conservative inline
object budget keeps
+ // their eager vector resize inside task admission; the
serialized-size reservation covers
+ // their dynamic field payloads.
+ return DECODED_THRIFT_STRUCT_RESERVATION_BYTES;
+ default:
+ return 1;
+ }
+}
+
+class MemoryBudgetProtocol final : public
apache::thrift::protocol::TProtocolDecorator {
+public:
+ MemoryBudgetProtocol(std::shared_ptr<apache::thrift::protocol::TProtocol>
protocol,
+ int32_t serialized_size)
+ : TProtocolDecorator(std::move(protocol)),
+ _memory_manager(thread_context()->thread_mem_tracker_mgr.get()),
+ _prior_reservation(_memory_manager->take_reserved_memory()) {
+ reserve_or_throw(static_cast<size_t>(serialized_size),
/*restore_prior_on_failure=*/true);
Review Comment:
[P1] Do not reserve the caller's whole readable window
`len` is only an upper bound on the encoded object (the helper replaces it
with the bytes actually consumed), but both Parquet page-cache hit paths pass a
cache entry containing the tiny Thrift header plus level data and the full
compressed/decompressed payload. Reserving all `serialized_size` bytes here can
therefore reject a valid cached page before parsing its header whenever the
query has enough memory for the header but less than the already-cached payload
size. Please separate the exact decoded-allocation budget from the readable
window, or have these callers pass a bounded header-only view, and cover a
valid small header followed by a large cached payload under a low task limit.
##########
be/test/format/parquet/parquet_statistics_test.cpp:
##########
@@ -17,16 +17,151 @@
#include <gtest/gtest.h>
+#include <algorithm>
+#include <cstring>
+#include <memory>
#include <regex>
+#include <vector>
#include "format/parquet/parquet_predicate.h"
+#include "util/thrift_util.h"
namespace doris {
+namespace {
+
+class BloomFilterFileReader final : public io::FileReader {
+public:
+ explicit BloomFilterFileReader(std::vector<uint8_t> data, size_t
logical_size = 0)
+ : _data(std::move(data)),
+ _logical_size(logical_size == 0 ? _data.size() : logical_size) {}
+
+ Status close() override {
+ _closed = true;
+ return Status::OK();
+ }
+
+ const io::Path& path() const override { return _path; }
+ size_t size() const override { return _logical_size; }
+ bool closed() const override { return _closed; }
+ int64_t mtime() const override { return 0; }
+ bool returned_short_nonzero_offset_read() const { return
_returned_short_nonzero_offset_read; }
+
+protected:
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const io::IOContext* io_ctx) override {
+ if (offset > _data.size()) {
+ return Status::IOError("Out of bounds");
+ }
+ *bytes_read = std::min(result.size, _data.size() - offset);
+ memcpy(result.data, _data.data() + offset, *bytes_read);
+ _returned_short_nonzero_offset_read |= offset > 0 && *bytes_read !=
result.size;
+ return Status::OK();
+ }
+
+private:
+ std::vector<uint8_t> _data;
+ size_t _logical_size;
+ io::Path _path = "parquet_bloom_filter_test";
+ bool _closed = false;
+ bool _returned_short_nonzero_offset_read = false;
+};
+
+Status read_test_bloom_filter(int32_t header_payload_size, size_t
actual_payload_size,
+ int32_t declared_length_adjustment = 0,
+ size_t logical_payload_size = 0, bool*
returned_short_read = nullptr,
+ bool* installed_bloom_filter = nullptr) {
+ tparquet::BloomFilterAlgorithm algorithm;
+ algorithm.__set_BLOCK(tparquet::SplitBlockAlgorithm());
+ tparquet::BloomFilterHash hash;
+ hash.__set_XXHASH(tparquet::XxHash());
+ tparquet::BloomFilterCompression compression;
+ compression.__set_UNCOMPRESSED(tparquet::Uncompressed());
+ tparquet::BloomFilterHeader header;
+ header.__set_numBytes(header_payload_size);
+ header.__set_algorithm(algorithm);
+ header.__set_hash(hash);
+ header.__set_compression(compression);
+
+ std::vector<uint8_t> file_bytes;
+ ThriftSerializer serializer(/*compact=*/true, /*initial_buffer_size=*/64);
+ RETURN_IF_ERROR(serializer.serialize(&header, &file_bytes));
+ const size_t header_size = file_bytes.size();
+ file_bytes.resize(header_size + actual_payload_size);
+
+ tparquet::ColumnMetaData metadata;
+ metadata.__set_bloom_filter_offset(0);
+ metadata.__set_bloom_filter_length(static_cast<int32_t>(file_bytes.size())
+
+ declared_length_adjustment);
+ const size_t logical_size =
+ logical_payload_size == 0 ? file_bytes.size() : header_size +
logical_payload_size;
+ auto reader =
std::make_shared<BloomFilterFileReader>(std::move(file_bytes), logical_size);
+ ParquetPredicate::ColumnStat stat;
+ Status status = ParquetPredicate::read_bloom_filter(metadata, reader,
nullptr, &stat);
+ if (returned_short_read != nullptr) {
+ *returned_short_read = reader->returned_short_nonzero_offset_read();
+ }
+ if (installed_bloom_filter != nullptr) {
+ *installed_bloom_filter = stat.bloom_filter != nullptr;
+ }
+ return status;
+}
+
+} // namespace
+
class ParquetStatisticsTest : public testing::Test {
public:
ParquetStatisticsTest() = default;
};
+TEST_F(ParquetStatisticsTest, reject_truncated_bloom_filter_payload) {
+ // The reader may legally return a short read at EOF, so accepting it
would initialize a
+ // Bloom filter whose missing bytes came from zero-filled process memory.
+ bool returned_short_read = false;
+ bool installed_bloom_filter = true;
+ EXPECT_FALSE(read_test_bloom_filter(/*header_payload_size=*/64,
/*actual_payload_size=*/32,
+ /*declared_length_adjustment=*/32,
+ /*logical_payload_size=*/64,
&returned_short_read,
+ &installed_bloom_filter)
+ .ok());
+ EXPECT_TRUE(returned_short_read);
+ EXPECT_FALSE(installed_bloom_filter);
+}
+
+TEST_F(ParquetStatisticsTest, reject_bloom_filter_range_beyond_file) {
+ bool returned_short_read = false;
+ EXPECT_FALSE(read_test_bloom_filter(/*header_payload_size=*/64,
/*actual_payload_size=*/32,
+ /*declared_length_adjustment=*/0,
+ /*logical_payload_size=*/0,
&returned_short_read)
+ .ok());
+ EXPECT_FALSE(returned_short_read);
+}
+
+TEST_F(ParquetStatisticsTest, reject_declared_bloom_filter_length_mismatch) {
+ // A present length describes exactly one header and payload. Treating it
as an upper bound can
+ // reinterpret a multi-block filter as a smaller filter and cause
false-negative pruning.
+ EXPECT_FALSE(
+ read_test_bloom_filter(/*header_payload_size=*/32,
/*actual_payload_size=*/64).ok());
+}
+
+TEST_F(ParquetStatisticsTest, reject_invalid_bloom_filter_block_sizes) {
+ EXPECT_FALSE(
+ read_test_bloom_filter(/*header_payload_size=*/16,
/*actual_payload_size=*/16).ok());
+ EXPECT_FALSE(
+ read_test_bloom_filter(/*header_payload_size=*/33,
/*actual_payload_size=*/33).ok());
+}
+
+TEST_F(ParquetStatisticsTest, reject_nonpositive_bloom_filter_declared_length)
{
+ const int32_t declared_length_adjustment = -1000;
+ EXPECT_FALSE(read_test_bloom_filter(/*header_payload_size=*/32,
/*actual_payload_size=*/32,
+ declared_length_adjustment)
+ .ok());
+}
+
+TEST_F(ParquetStatisticsTest, accept_valid_bloom_filter_layout) {
Review Comment:
[P2] Cover Bloom metadata without the optional length
The local Parquet definition says `bloom_filter_length` was added in 2.10
and may be absent in older files, and both changed readers retain a separate
compatibility branch for that case. Every v1/v2 Bloom fixture sets the field,
including this valid control, so an accidental dependency on it can break
old/external files while the suite stays green. Please add row-group cases with
a valid offset, no declared length, and trailing file bytes, proving both a
present value is retained and an absent value can still prune.
##########
be/src/util/thrift_util.cpp:
##########
@@ -56,6 +60,118 @@ class TProtocol;
#include <thread>
namespace doris {
+namespace {
+
+constexpr size_t DECODED_THRIFT_STRUCT_RESERVATION_BYTES = 1024;
+
+size_t decoded_thrift_value_reservation(apache::thrift::protocol::TType type) {
+ using apache::thrift::protocol::T_BOOL;
+ using apache::thrift::protocol::T_BYTE;
+ using apache::thrift::protocol::T_DOUBLE;
+ using apache::thrift::protocol::T_I16;
+ using apache::thrift::protocol::T_I32;
+ using apache::thrift::protocol::T_I64;
+ using apache::thrift::protocol::T_LIST;
+ using apache::thrift::protocol::T_MAP;
+ using apache::thrift::protocol::T_SET;
+ using apache::thrift::protocol::T_STRING;
+ using apache::thrift::protocol::T_STRUCT;
+ switch (type) {
+ case T_BOOL:
+ case T_BYTE:
+ return 1;
+ case T_I16:
+ return sizeof(int16_t);
+ case T_I32:
+ return sizeof(int32_t);
+ case T_I64:
+ case T_DOUBLE:
+ return sizeof(int64_t);
+ case T_STRING:
+ return sizeof(std::string);
+ case T_LIST:
+ case T_SET:
+ case T_MAP:
+ return sizeof(std::vector<uint8_t>);
+ case T_STRUCT:
+ // Generated structs vary in size. Reserving a conservative inline
object budget keeps
+ // their eager vector resize inside task admission; the
serialized-size reservation covers
+ // their dynamic field payloads.
+ return DECODED_THRIFT_STRUCT_RESERVATION_BYTES;
+ default:
+ return 1;
+ }
+}
+
+class MemoryBudgetProtocol final : public
apache::thrift::protocol::TProtocolDecorator {
+public:
+ MemoryBudgetProtocol(std::shared_ptr<apache::thrift::protocol::TProtocol>
protocol,
+ int32_t serialized_size)
+ : TProtocolDecorator(std::move(protocol)),
+ _memory_manager(thread_context()->thread_mem_tracker_mgr.get()),
+ _prior_reservation(_memory_manager->take_reserved_memory()) {
+ reserve_or_throw(static_cast<size_t>(serialized_size),
/*restore_prior_on_failure=*/true);
+ }
+
+ ~MemoryBudgetProtocol() override {
+ _memory_manager->shrink_reserved();
+ _memory_manager->adopt_reserved_memory(std::move(_prior_reservation));
+ }
+
+ uint32_t readMapBegin_virt(apache::thrift::protocol::TType& key_type,
+ apache::thrift::protocol::TType& value_type,
+ uint32_t& size) override {
+ const uint32_t consumed =
TProtocolDecorator::readMapBegin_virt(key_type, value_type, size);
+ const uint32_t count = size;
+ const size_t element_size = decoded_thrift_value_reservation(key_type)
+
+
decoded_thrift_value_reservation(value_type) +
+ 4 * sizeof(void*);
+ reserve_container(count, element_size);
+ return consumed;
+ }
+
+ uint32_t readListBegin_virt(apache::thrift::protocol::TType& element_type,
+ uint32_t& size) override {
+ const uint32_t consumed =
TProtocolDecorator::readListBegin_virt(element_type, size);
+ reserve_container(size,
decoded_thrift_value_reservation(element_type));
Review Comment:
[P1] Move decoded admission to the actual allocation site
This hook cannot infer decoded storage from the wire tag. Generated readers
resize their statically declared C++ vector regardless of the returned
`element_type`, so `list<TPlanNode>` can advertise `T_BOOL`, reserve one byte
per element here, and allocate `count * sizeof(TPlanNode)` before parsing. The
opposite path also fails: Thrift 0.16's generic `skip()` reaches this virtual
hook for unknown fields but constructs no container, so a valid
forward-compatible `list<empty struct>` is charged 1 KiB per element and can be
rejected for phantom memory. Please enforce admission where the actual
generated allocation/target type is known (or at allocator resize), leave skip
traversal charged only for storage it creates, and test both a forged tag with
a real large struct and a large unknown container.
##########
be/src/util/thrift_util.cpp:
##########
@@ -56,6 +60,118 @@ class TProtocol;
#include <thread>
namespace doris {
+namespace {
+
+constexpr size_t DECODED_THRIFT_STRUCT_RESERVATION_BYTES = 1024;
+
+size_t decoded_thrift_value_reservation(apache::thrift::protocol::TType type) {
+ using apache::thrift::protocol::T_BOOL;
+ using apache::thrift::protocol::T_BYTE;
+ using apache::thrift::protocol::T_DOUBLE;
+ using apache::thrift::protocol::T_I16;
+ using apache::thrift::protocol::T_I32;
+ using apache::thrift::protocol::T_I64;
+ using apache::thrift::protocol::T_LIST;
+ using apache::thrift::protocol::T_MAP;
+ using apache::thrift::protocol::T_SET;
+ using apache::thrift::protocol::T_STRING;
+ using apache::thrift::protocol::T_STRUCT;
+ switch (type) {
+ case T_BOOL:
+ case T_BYTE:
+ return 1;
+ case T_I16:
+ return sizeof(int16_t);
+ case T_I32:
+ return sizeof(int32_t);
+ case T_I64:
+ case T_DOUBLE:
+ return sizeof(int64_t);
+ case T_STRING:
+ return sizeof(std::string);
+ case T_LIST:
+ case T_SET:
+ case T_MAP:
+ return sizeof(std::vector<uint8_t>);
+ case T_STRUCT:
+ // Generated structs vary in size. Reserving a conservative inline
object budget keeps
+ // their eager vector resize inside task admission; the
serialized-size reservation covers
+ // their dynamic field payloads.
+ return DECODED_THRIFT_STRUCT_RESERVATION_BYTES;
+ default:
+ return 1;
+ }
+}
+
+class MemoryBudgetProtocol final : public
apache::thrift::protocol::TProtocolDecorator {
+public:
+ MemoryBudgetProtocol(std::shared_ptr<apache::thrift::protocol::TProtocol>
protocol,
+ int32_t serialized_size)
+ : TProtocolDecorator(std::move(protocol)),
+ _memory_manager(thread_context()->thread_mem_tracker_mgr.get()),
+ _prior_reservation(_memory_manager->take_reserved_memory()) {
+ reserve_or_throw(static_cast<size_t>(serialized_size),
/*restore_prior_on_failure=*/true);
+ }
+
+ ~MemoryBudgetProtocol() override {
+ _memory_manager->shrink_reserved();
+ _memory_manager->adopt_reserved_memory(std::move(_prior_reservation));
+ }
+
+ uint32_t readMapBegin_virt(apache::thrift::protocol::TType& key_type,
+ apache::thrift::protocol::TType& value_type,
+ uint32_t& size) override {
+ const uint32_t consumed =
TProtocolDecorator::readMapBegin_virt(key_type, value_type, size);
+ const uint32_t count = size;
+ const size_t element_size = decoded_thrift_value_reservation(key_type)
+
+
decoded_thrift_value_reservation(value_type) +
+ 4 * sizeof(void*);
+ reserve_container(count, element_size);
+ return consumed;
+ }
+
+ uint32_t readListBegin_virt(apache::thrift::protocol::TType& element_type,
+ uint32_t& size) override {
+ const uint32_t consumed =
TProtocolDecorator::readListBegin_virt(element_type, size);
+ reserve_container(size,
decoded_thrift_value_reservation(element_type));
+ return consumed;
+ }
+
+ uint32_t readSetBegin_virt(apache::thrift::protocol::TType& element_type,
+ uint32_t& size) override {
+ const uint32_t consumed =
TProtocolDecorator::readSetBegin_virt(element_type, size);
+ reserve_container(size, decoded_thrift_value_reservation(element_type)
+ 4 * sizeof(void*));
+ return consumed;
+ }
+
+private:
+ void reserve_container(uint32_t count, size_t element_size) {
+ if (count > std::numeric_limits<size_t>::max() / element_size) {
+ throw apache::thrift::protocol::TProtocolException(
+ apache::thrift::protocol::TProtocolException::SIZE_LIMIT,
+ "Decoded Thrift container size overflows");
+ }
+ reserve_or_throw(static_cast<size_t>(count) * element_size,
+ /*restore_prior_on_failure=*/false);
+ }
+
+ void reserve_or_throw(size_t bytes, bool restore_prior_on_failure) {
+ const Status status =
_memory_manager->try_reserve(static_cast<int64_t>(bytes));
+ if (status.ok()) {
+ return;
+ }
+ if (restore_prior_on_failure) {
+
_memory_manager->adopt_reserved_memory(std::move(_prior_reservation));
+ }
+ throw apache::thrift::protocol::TProtocolException(
Review Comment:
[P2] Preserve memory-limit failures across this boundary
`try_reserve()` returns specific query, workload-group, or process memory
errors, but wrapping only `status.to_string()` in a `TProtocolException` makes
`deserialize_thrift_msg()` return `InternalError`. Both Parquet page-header
loops then treat that as an incomplete parse, retry with progressively larger
reads, and finally replace it with `IOError`; other mandatory callers also lose
the actionable status. Please carry the original Doris `Status` through the
protocol boundary and return it unchanged, reserving deserialization errors for
actual wire failures, with a regression that proves page readers do not retry
memory-pressure failures.
--
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]
