github-actions[bot] commented on code in PR #66820:
URL: https://github.com/apache/doris/pull/66820#discussion_r3793721406
##########
be/src/format/parquet/parquet_predicate.h:
##########
@@ -441,44 +441,76 @@ class ParquetPredicate {
static Status read_bloom_filter(const tparquet::ColumnMetaData&
column_meta_data,
io::FileReaderSPtr file_reader,
io::IOContext* io_ctx,
ColumnStat* ans_stat) {
- size_t size;
if (!column_meta_data.__isset.bloom_filter_offset) {
return Status::NotSupported("Can not use this parquet bloom
filter.");
}
+ if (column_meta_data.bloom_filter_offset < 0 ||
+ (column_meta_data.__isset.bloom_filter_length &&
+ column_meta_data.bloom_filter_length <= 0)) {
+ return Status::Corruption("Invalid Parquet bloom filter offset or
declared length");
+ }
- if (column_meta_data.__isset.bloom_filter_length &&
- column_meta_data.bloom_filter_length > 0) {
- size = column_meta_data.bloom_filter_length;
- } else {
- size = BLOOM_FILTER_MAX_HEADER_LENGTH;
+ const uint64_t bloom_offset =
static_cast<uint64_t>(column_meta_data.bloom_filter_offset);
+ if (bloom_offset >= file_reader->size()) {
+ return Status::Corruption("Parquet bloom filter offset exceeds
file size");
}
+ const size_t available = file_reader->size() - bloom_offset;
+ const size_t declared_available =
+ column_meta_data.__isset.bloom_filter_length
+ ?
std::min<size_t>(column_meta_data.bloom_filter_length, available)
+ : available;
+ const size_t header_read_size =
+ std::min<size_t>(declared_available,
BLOOM_FILTER_MAX_HEADER_LENGTH);
size_t bytes_read = 0;
- std::vector<uint8_t> header_buffer(size);
+ std::vector<uint8_t> header_buffer(header_read_size);
RETURN_IF_ERROR(file_reader->read_at(column_meta_data.bloom_filter_offset,
- Slice(header_buffer.data(),
size), &bytes_read,
- io_ctx));
+ Slice(header_buffer.data(),
header_buffer.size()),
+ &bytes_read, io_ctx));
tparquet::BloomFilterHeader t_bloom_filter_header;
uint32_t t_bloom_filter_header_size =
static_cast<uint32_t>(bytes_read);
- RETURN_IF_ERROR(deserialize_thrift_msg(header_buffer.data(),
&t_bloom_filter_header_size,
- true, &t_bloom_filter_header));
+ if (!deserialize_thrift_msg(header_buffer.data(),
&t_bloom_filter_header_size, true,
+ &t_bloom_filter_header)
+ .ok()) {
+ return Status::Corruption("Malformed Parquet bloom filter header");
+ }
// TODO the bloom filter could be encrypted, too, so need to double
check that this is NOT the case
if (!t_bloom_filter_header.algorithm.__isset.BLOCK ||
!t_bloom_filter_header.compression.__isset.UNCOMPRESSED ||
- !t_bloom_filter_header.hash.__isset.XXHASH) {
+ !t_bloom_filter_header.hash.__isset.XXHASH ||
t_bloom_filter_header.numBytes <= 0) {
return Status::NotSupported("Can not use this parquet bloom
filter.");
}
- ans_stat->bloom_filter =
std::make_unique<ParquetBlockSplitBloomFilter>();
+ const int64_t payload_size = t_bloom_filter_header.numBytes;
+ if (payload_size < segment_v2::BloomFilter::MINIMUM_BYTES ||
+ payload_size > segment_v2::BloomFilter::MAXIMUM_BYTES ||
payload_size % 32 != 0) {
+ return Status::Corruption("Invalid Parquet bloom filter payload
size {}", payload_size);
+ }
+ const uint64_t total_size =
+ static_cast<uint64_t>(t_bloom_filter_header_size) +
payload_size;
+ if (total_size > available) {
+ return Status::Corruption("Parquet bloom filter range exceeds file
size");
+ }
+ if (column_meta_data.__isset.bloom_filter_length &&
+ (static_cast<uint64_t>(column_meta_data.bloom_filter_length) <
total_size ||
+ static_cast<uint64_t>(column_meta_data.bloom_filter_length) >
available)) {
+ return Status::Corruption("Invalid Parquet bloom filter declared
length");
+ }
- std::vector<uint8_t> data_buffer(t_bloom_filter_header.numBytes);
+ // Validate the full split-block layout before allocating or adding
metadata-controlled
+ // offsets; the Bloom filter implementation assumes complete 32-byte
blocks.
+ std::vector<uint8_t> data_buffer(static_cast<size_t>(payload_size));
Review Comment:
[P1] Keep maximum-size Bloom filters inside the task memory budget
A valid `numBytes == MAXIMUM_BYTES` creates a 128 MiB `std::vector` here,
then `ParquetBlockSplitBloomFilter::init()` deep-copies it into another `new
char[128 MiB]`. Neither allocation uses Doris's allocator or `try_reserve()`,
and `_process_column_stat_filter()` can retain one 128 MiB filter per distinct
predicate column; several parallel scanners can therefore exceed the
query/workload-group limit and OOM the BE even though every individual filter
passes this new cap. Please reserve and allocator-track both temporary and
retained bytes (preferably transfer one tracked buffer instead of copying it),
and bound or avoid aggregate retention of very large filters. The format_v2
path has the same allocation pattern.
##########
be/src/util/thrift_util.h:
##########
@@ -134,15 +137,20 @@ Status deserialize_thrift_msg(const uint8_t* buf,
uint32_t* len, bool compact,
// transport. TMemoryBuffer is not const-safe, although we use it in
// a const-safe way, so we have to explicitly cast away the const.
auto conf = std::make_shared<apache::thrift::TConfiguration>();
- // On Thrift 0.14.0+, need use TConfiguration to raise the max message
size.
- // max message size is 100MB default, so make it unlimited.
- conf->setMaxMessageSize(std::numeric_limits<int>::max());
+ const int32_t size_limit =
+ *len == 0 ? 1
+ : static_cast<int32_t>(std::min<uint32_t>(
+ *len,
static_cast<uint32_t>(std::numeric_limits<int32_t>::max())));
Review Comment:
[P1] Bound decoded container allocation, not only element count
`container_limit = size_limit` still lets a modest input allocate far more
than its wire size. Thrift 0.16 generated list readers call `resize(count)`
immediately after `readListBegin()`, while compact and binary preflight treat
`T_STRUCT` as zero minimum wire bytes. Under the existing 100 MiB v2 footer
cap, metadata can declare nearly 100 million `SchemaElement`s (`count <= len`),
pass this guard, and resize the decoded vector to multiple GiB on the current
ABI before reading or validating the first element; padding makes the transport
window large enough, and failure afterward is too late. Please enforce a
decoded or reservation-aware container budget before generated resize, and add
compact and binary regressions where the count is within the input length but
decoded storage exceeds the allowed budget.
##########
be/src/format/parquet/parquet_predicate.h:
##########
@@ -441,44 +441,76 @@ class ParquetPredicate {
static Status read_bloom_filter(const tparquet::ColumnMetaData&
column_meta_data,
io::FileReaderSPtr file_reader,
io::IOContext* io_ctx,
ColumnStat* ans_stat) {
- size_t size;
if (!column_meta_data.__isset.bloom_filter_offset) {
return Status::NotSupported("Can not use this parquet bloom
filter.");
}
+ if (column_meta_data.bloom_filter_offset < 0 ||
+ (column_meta_data.__isset.bloom_filter_length &&
+ column_meta_data.bloom_filter_length <= 0)) {
+ return Status::Corruption("Invalid Parquet bloom filter offset or
declared length");
+ }
- if (column_meta_data.__isset.bloom_filter_length &&
- column_meta_data.bloom_filter_length > 0) {
- size = column_meta_data.bloom_filter_length;
- } else {
- size = BLOOM_FILTER_MAX_HEADER_LENGTH;
+ const uint64_t bloom_offset =
static_cast<uint64_t>(column_meta_data.bloom_filter_offset);
+ if (bloom_offset >= file_reader->size()) {
+ return Status::Corruption("Parquet bloom filter offset exceeds
file size");
}
+ const size_t available = file_reader->size() - bloom_offset;
+ const size_t declared_available =
+ column_meta_data.__isset.bloom_filter_length
+ ?
std::min<size_t>(column_meta_data.bloom_filter_length, available)
+ : available;
+ const size_t header_read_size =
+ std::min<size_t>(declared_available,
BLOOM_FILTER_MAX_HEADER_LENGTH);
size_t bytes_read = 0;
- std::vector<uint8_t> header_buffer(size);
+ std::vector<uint8_t> header_buffer(header_read_size);
RETURN_IF_ERROR(file_reader->read_at(column_meta_data.bloom_filter_offset,
- Slice(header_buffer.data(),
size), &bytes_read,
- io_ctx));
+ Slice(header_buffer.data(),
header_buffer.size()),
+ &bytes_read, io_ctx));
tparquet::BloomFilterHeader t_bloom_filter_header;
uint32_t t_bloom_filter_header_size =
static_cast<uint32_t>(bytes_read);
- RETURN_IF_ERROR(deserialize_thrift_msg(header_buffer.data(),
&t_bloom_filter_header_size,
- true, &t_bloom_filter_header));
+ if (!deserialize_thrift_msg(header_buffer.data(),
&t_bloom_filter_header_size, true,
+ &t_bloom_filter_header)
+ .ok()) {
+ return Status::Corruption("Malformed Parquet bloom filter header");
+ }
// TODO the bloom filter could be encrypted, too, so need to double
check that this is NOT the case
if (!t_bloom_filter_header.algorithm.__isset.BLOCK ||
!t_bloom_filter_header.compression.__isset.UNCOMPRESSED ||
- !t_bloom_filter_header.hash.__isset.XXHASH) {
+ !t_bloom_filter_header.hash.__isset.XXHASH ||
t_bloom_filter_header.numBytes <= 0) {
return Status::NotSupported("Can not use this parquet bloom
filter.");
}
- ans_stat->bloom_filter =
std::make_unique<ParquetBlockSplitBloomFilter>();
+ const int64_t payload_size = t_bloom_filter_header.numBytes;
+ if (payload_size < segment_v2::BloomFilter::MINIMUM_BYTES ||
+ payload_size > segment_v2::BloomFilter::MAXIMUM_BYTES ||
payload_size % 32 != 0) {
+ return Status::Corruption("Invalid Parquet bloom filter payload
size {}", payload_size);
+ }
+ const uint64_t total_size =
+ static_cast<uint64_t>(t_bloom_filter_header_size) +
payload_size;
+ if (total_size > available) {
+ return Status::Corruption("Parquet bloom filter range exceeds file
size");
+ }
+ if (column_meta_data.__isset.bloom_filter_length &&
+ (static_cast<uint64_t>(column_meta_data.bloom_filter_length) <
total_size ||
Review Comment:
[P1] Require the declared Bloom length to match the header
`bloom_filter_length` is the exact size of the serialized header plus
bitset, not just an upper bound. For example, metadata can declare `header_size
+ 64` while the header says `numBytes = 32`; this condition accepts it, reads
only the first block, and initializes a one-block filter. If the payload was
built as two blocks and a present value landed in block 1, `test_hash()`
recomputes its bucket for one block, sees the empty first block, and can
falsely prune the row group. Please reject any present declared length that is
not exactly `header_size + numBytes`, apply the same fix to
`validate_native_bloom_filter_layout()`, and cover the contradictory layout end
to end.
##########
be/test/format/parquet/parquet_thrift_test.cpp:
##########
@@ -71,6 +72,19 @@ class ParquetThriftReaderTest : public testing::Test {
void TearDown() override { TimezoneUtils::clear_timezone_caches(); }
};
Review Comment:
[P2] Cover the other newly bounded Thrift paths
This only exercises a compact-protocol container length, but the production
change separately constructs the binary protocol and enables both
`string_limit` and `container_limit` in both branches. A regression that drops
or misorders the binary limit, or either string limit, would therefore leave
this test green even though network-facing binary callers or Parquet string
fields could again allocate from a hostile length. Please add focused malformed
string/container cases for `compact=true` and `compact=false`, together with a
valid control, so every newly enabled allocation guard is covered.
##########
be/test/format/parquet/parquet_statistics_test.cpp:
##########
@@ -17,16 +17,111 @@
#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) :
_data(std::move(data)) {}
+
+ Status close() override {
+ _closed = true;
+ return Status::OK();
+ }
+
+ const io::Path& path() const override { return _path; }
+ size_t size() const override { return _data.size(); }
+ bool closed() const override { return _closed; }
+ int64_t mtime() const override { return 0; }
+
+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);
+ return Status::OK();
+ }
+
+private:
+ std::vector<uint8_t> _data;
+ io::Path _path = "parquet_bloom_filter_test";
+ bool _closed = false;
+};
+
+Status read_test_bloom_filter(int32_t header_payload_size, size_t
actual_payload_size,
+ int32_t declared_length_adjustment = 0) {
+ 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);
+ auto reader =
std::make_shared<BloomFilterFileReader>(std::move(file_bytes));
+ ParquetPredicate::ColumnStat stat;
+ return ParquetPredicate::read_bloom_filter(metadata, reader, nullptr,
&stat);
+}
+
+} // namespace
+
class ParquetStatisticsTest : public testing::Test {
public:
ParquetStatisticsTest() = default;
};
+TEST_F(ParquetStatisticsTest, reject_truncated_bloom_filter_payload) {
Review Comment:
[P2] Make this test reach the short-read check
This fake reports `_data.size()` as the file size, so with a 64-byte header
claim and only 32 payload bytes the production code returns from `total_size >
available` before it calls the payload `read_at()`. The test would still pass
if the new `bytes_read != data_buffer.size()` guard were removed. Please let
the fake advertise the full logical range while forcing a successful short
payload read, assert that no Bloom filter is installed, and keep the current
range-truncation scenario as a separate case.
--
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]
