[fix](parquet) Bound Thrift and Bloom filter metadata allocations

be/src/format/parquet/parquet_block_split_bloom_filter.cpp

@@ -23,6 +23,22 @@
 
 namespace doris {
 
+ParquetBlockSplitBloomFilter::~ParquetBlockSplitBloomFilter() {
+    if (_data == nullptr) {
+        return;
+    }
+    if (_is_write) {
+        g_write_bloom_filter_total_bytes << -static_cast<int64_t>(_size);
+        g_write_bloom_filter_num << -1;
+    } else {
+        g_read_bloom_filter_total_bytes << -static_cast<int64_t>(_size);
+        g_read_bloom_filter_num << -1;
+    }
+    g_total_bloom_filter_total_bytes << -static_cast<int64_t>(_size);
+    // The derived owner uses Doris's tracked allocator, so the base must not delete this view.
+    _data = nullptr;
+}
+
 // for write
 Status ParquetBlockSplitBloomFilter::init(uint64_t filter_size,
                                           segment_v2::HashStrategyPB strategy) {
@@ -37,7 +53,8 @@ Status ParquetBlockSplitBloomFilter::init(uint64_t filter_size,
     }
     _num_bytes = filter_size;
     _size = _num_bytes;
-    _data = new char[_size];
+    _owned_data = make_unique_buffer<char>(_size);
+    _data = _owned_data.get();
     memset(_data, 0, _size);
     _has_null = nullptr;
     _is_write = true;
@@ -51,10 +68,20 @@ Status ParquetBlockSplitBloomFilter::init(uint64_t filter_size,
 // use deep copy to acquire the data
 Status ParquetBlockSplitBloomFilter::init(const char* buf, size_t size,
                                           segment_v2::HashStrategyPB strategy) {
-    if (size <= 1) {
+    if (buf == nullptr || size <= 1) {
         return Status::InvalidArgument("invalid size:{}", size);
     }
     DCHECK(size > 1);
+    RETURN_IF_ERROR(init_for_read(size, strategy));
+    memcpy(_data, buf, size);
+    return Status::OK();
+}
+
+Status ParquetBlockSplitBloomFilter::init_for_read(size_t size,
+                                                   segment_v2::HashStrategyPB strategy) {
+    if (size <= 1) {
+        return Status::InvalidArgument("invalid size:{}", size);
+    }
     if (strategy == XX_HASH_64) {
         _hash_func = [](const void* buf, const int64_t len, const uint64_t seed, void* out) {
             auto h =
@@ -64,12 +91,8 @@ Status ParquetBlockSplitBloomFilter::init(const char* buf, size_t size,
     } else {
         return Status::InvalidArgument("invalid strategy:{}", strategy);
     }
-    if (buf == nullptr) {
-        return Status::InvalidArgument("buf is nullptr");
-    }
-
-    _data = new char[size];
-    memcpy(_data, buf, size);
+    _owned_data = make_unique_buffer<char>(size);
+    _data = _owned_data.get();
     _size = size;
     _num_bytes = _size;
     _has_null = nullptr;

be/src/format/parquet/parquet_block_split_bloom_filter.h

@@ -19,6 +19,7 @@
 
 #include <stdint.h>
 
+#include "core/custom_allocator.h"
 #include "storage/index/bloom_filter/bloom_filter.h"
 
 namespace doris {
@@ -36,8 +37,11 @@ namespace doris {
 // https://parquet.apache.org/docs/file-format/bloomfilter/
 class ParquetBlockSplitBloomFilter : public segment_v2::BloomFilter {
 public:
+    ~ParquetBlockSplitBloomFilter() override;
     Status init(uint64_t filter_size, segment_v2::HashStrategyPB strategy) override;
     Status init(const char* buf, size_t size, segment_v2::HashStrategyPB strategy) override;
+    Status init_for_read(size_t size, segment_v2::HashStrategyPB strategy);
+    char* mutable_data() { return _data; }
     void add_bytes(const char* buf, size_t size) override;
     bool test_bytes(const char* buf, size_t size) const override;
     void set_has_null(bool has_null) override;
@@ -62,6 +66,8 @@ class ParquetBlockSplitBloomFilter : public segment_v2::BloomFilter {
     };
 
 private:
+    DorisUniqueBufferPtr<char> _owned_data;
+
     void _set_masks(uint32_t key, BlockMask& block_mask) const {
         for (int i = 0; i < BITS_SET_PER_BLOCK; ++i) {
             block_mask.item[i] = key * SALT[i];

be/src/format/parquet/parquet_predicate.h

@@ -441,45 +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");
+        }
+        const auto expected_declared_length = static_cast<int64_t>(total_size);
+        if (column_meta_data.__isset.bloom_filter_length &&
+            column_meta_data.bloom_filter_length != expected_declared_length) {
+            return Status::Corruption("Invalid Parquet bloom filter declared length");
+        }
 
-        std::vector<uint8_t> data_buffer(t_bloom_filter_header.numBytes);
+        auto bloom_filter = std::make_unique<ParquetBlockSplitBloomFilter>();
+        // Read directly into one tracked allocation so a valid maximum-size filter is admitted
+        // against the task budget without doubling its peak memory during initialization.
+        RETURN_IF_ERROR(bloom_filter->init_for_read(static_cast<size_t>(payload_size),
+                                                    segment_v2::HashStrategyPB::XX_HASH_64));
         RETURN_IF_ERROR(file_reader->read_at(
-                column_meta_data.bloom_filter_offset + t_bloom_filter_header_size,
-                Slice(data_buffer.data(), t_bloom_filter_header.numBytes), &bytes_read, io_ctx));
+                static_cast<size_t>(bloom_offset) + t_bloom_filter_header_size,
+                Slice(bloom_filter->mutable_data(), bloom_filter->size()), &bytes_read, io_ctx));
+        if (bytes_read != bloom_filter->size()) {
+            return Status::Corruption("Truncated Parquet bloom filter payload");
+        }
 
-        RETURN_IF_ERROR(ans_stat->bloom_filter->init(
-                reinterpret_cast<const char*>(data_buffer.data()), t_bloom_filter_header.numBytes,
-                segment_v2::HashStrategyPB::XX_HASH_64));
+        ans_stat->bloom_filter = std::move(bloom_filter);
 
         return Status::OK();
     }

be/src/format/parquet/vparquet_page_reader.cpp

@@ -30,6 +30,7 @@
 #include "io/fs/buffered_reader.h"
 #include "runtime/runtime_profile.h"
 #include "storage/cache/page_cache.h"
+#include "util/debug_points.h"
 #include "util/slice.h"
 #include "util/thrift_util.h"
 
@@ -42,6 +43,24 @@ struct IOContext;
 namespace doris {
 static constexpr size_t INIT_PAGE_HEADER_SIZE = 128;
 
+namespace {
+
+void inject_page_header_memory_failure(Status* status) {
+    DBUG_EXECUTE_IF("ParquetPageReader.parse_page_header.memory_failure", {
+        *status = Status::Error<ErrorCode::QUERY_MEMORY_EXCEEDED>(
+                "Injected page header memory admission failure");
+    });
+}
+
+bool is_memory_failure(const Status& status) {
+    return status.is<ErrorCode::QUERY_MEMORY_EXCEEDED>() ||
+           status.is<ErrorCode::WORKLOAD_GROUP_MEMORY_EXCEEDED>() ||
+           status.is<ErrorCode::PROCESS_MEMORY_EXCEEDED>() ||
+           status.is<ErrorCode::MEM_ALLOC_FAILED>() || status.is<ErrorCode::MEM_LIMIT_EXCEEDED>();
+}
+
+} // namespace
+
 void ParquetPageCacheKeyBuilder::init(const std::string& path, int64_t mtime) {
     _file_key_prefix = fmt::format("{}::{}", path, mtime);
 }
@@ -167,9 +186,15 @@ Status PageReader<IN_COLLECTION, OFFSET_INDEX>::parse_page_header() {
         SCOPED_RAW_TIMER(&_page_statistics.decode_header_time);
         auto st =
                 deserialize_thrift_msg(page_header_buf, &real_header_size, true, &_cur_page_header);
+        inject_page_header_memory_failure(&st);
         if (st.ok()) {
             break;
         }
+        if (is_memory_failure(st)) {
+            // A larger input window cannot recover a memory admission failure and would only turn
+            // the actionable status into a generic page-header I/O error after repeated reads.
+            return st;
+        }
         if (_offset + header_size >= _end_offset || real_header_size > MAX_PAGE_HEADER_SIZE) {
             return Status::IOError(
                     "Failed to deserialize parquet page header. offset: {}, "

be/src/format/parquet/vparquet_reader.cpp

@@ -1507,6 +1507,8 @@ Status ParquetReader::_process_column_stat_filter(
     // Cache bloom filters for each column to avoid reading the same bloom filter multiple times
     // when there are multiple predicates on the same column
     std::unordered_map<int, std::unique_ptr<ParquetBlockSplitBloomFilter>> bloom_filter_cache;
+    constexpr size_t MAX_CACHED_BLOOM_FILTER_BYTES = 16 * 1024 * 1024;
+    size_t cached_bloom_filter_bytes = 0;
 
     // Initialize output parameters
     *filtered_by_min_max = false;
@@ -1565,6 +1567,7 @@ Status ParquetReader::_process_column_stat_filter(
                     auto cache_iter = bloom_filter_cache.find(parquet_col_id);
                     if (cache_iter != bloom_filter_cache.end()) {
                         // Bloom filter already loaded for this column, reuse it
+                        cached_bloom_filter_bytes -= cache_iter->second->size();
                         stat->bloom_filter = std::move(cache_iter->second);
                         bloom_filter_cache.erase(cache_iter);
                         return stat->bloom_filter != nullptr;
@@ -1607,6 +1610,12 @@ Status ParquetReader::_process_column_stat_filter(
 
         // After evaluating, if the bloom filter was used, cache it for subsequent predicates
         if (stat.bloom_filter) {
+            // Large filters remain tracked but are not retained across predicates; this bounds the
+            // row-group cache independently of the number of predicate columns.
+            if (stat.bloom_filter->size() >
+                MAX_CACHED_BLOOM_FILTER_BYTES - cached_bloom_filter_bytes) {
+                continue;
+            }
             // Find the column id for caching
             for (auto* slot : _tuple_descriptor->slots()) {
                 if (_table_info_node_ptr->children_column_exists(slot->col_name())) {
@@ -1616,6 +1625,7 @@ Status ParquetReader::_process_column_stat_filter(
                             _file_metadata->schema().get_column(file_col_name);
                     int parquet_col_id = col_schema->physical_column_index;
                     if (stat.col_schema == col_schema) {
+                        cached_bloom_filter_bytes += stat.bloom_filter->size();
                         bloom_filter_cache[parquet_col_id] = std::move(stat.bloom_filter);
                         break;
                     }

be/src/format_v2/parquet/parquet_statistics.cpp

@@ -75,7 +75,7 @@ Status validate_native_bloom_filter_layout(int64_t offset, uint32_t header_size,
     }
     if (declared_length >= 0) {
         const uint64_t unsigned_declared_length = static_cast<uint64_t>(declared_length);
-        if (unsigned_declared_length < total_size ||
+        if (unsigned_declared_length != total_size ||
             unsigned_declared_length > file_size - unsigned_offset) {
             return Status::Corruption(
                     "Parquet Bloom filter requires {} bytes, metadata declares {}, file has {}",
@@ -190,15 +190,17 @@ Status read_native_bloom_filter(const tparquet::ColumnMetaData& metadata,
             metadata.__isset.bloom_filter_length ? metadata.bloom_filter_length : -1,
             file->size()));
 
-    std::vector<uint8_t> data(cast_set<size_t>(header.numBytes));
+    auto bloom_filter = std::make_unique<native::BlockSplitBloomFilter>();
+    // Read directly into one tracked allocation so maximum-size filters cannot bypass task
+    // admission or temporarily require a second payload-sized allocation.
+    RETURN_IF_ERROR(bloom_filter->init_for_read(cast_set<size_t>(header.numBytes),
+                                                segment_v2::HashStrategyPB::XX_HASH_64));
     RETURN_IF_ERROR(file->read_at(static_cast<size_t>(metadata.bloom_filter_offset) + header_size,
-                                  Slice(data.data(), data.size()), &bytes_read, io_ctx));
-    if (bytes_read != data.size()) {
+                                  Slice(bloom_filter->mutable_data(), bloom_filter->size()),
+                                  &bytes_read, io_ctx));
+    if (bytes_read != bloom_filter->size()) {
         return Status::Corruption("Truncated Parquet Bloom filter payload");
     }
-    auto bloom_filter = std::make_unique<native::BlockSplitBloomFilter>();
-    RETURN_IF_ERROR(bloom_filter->init(reinterpret_cast<const char*>(data.data()), data.size(),
-                                       segment_v2::HashStrategyPB::XX_HASH_64));
     *result = std::move(bloom_filter);
     return Status::OK();
 }

be/src/format_v2/parquet/reader/native/block_split_bloom_filter.cpp

@@ -37,11 +37,28 @@ Status set_hash_strategy(segment_v2::HashStrategyPB strategy,
 }
 } // namespace
 
+BlockSplitBloomFilter::~BlockSplitBloomFilter() {
+    if (_data == nullptr) {
+        return;
+    }
+    if (_is_write) {
+        segment_v2::g_write_bloom_filter_total_bytes << -static_cast<int64_t>(_size);
+        segment_v2::g_write_bloom_filter_num << -1;
+    } else {
+        segment_v2::g_read_bloom_filter_total_bytes << -static_cast<int64_t>(_size);
+        segment_v2::g_read_bloom_filter_num << -1;
+    }
+    segment_v2::g_total_bloom_filter_total_bytes << -static_cast<int64_t>(_size);
+    // The derived owner uses Doris's tracked allocator, so the base must not delete this view.
+    _data = nullptr;
+}
+
 Status BlockSplitBloomFilter::init(uint64_t filter_size, segment_v2::HashStrategyPB strategy) {
     RETURN_IF_ERROR(set_hash_strategy(strategy, &_hash_func));
     _num_bytes = filter_size;
     _size = _num_bytes;
-    _data = new char[_size];
+    _owned_data = make_unique_buffer<char>(_size);
+    _data = _owned_data.get();
     memset(_data, 0, _size);
     _has_null = nullptr;
     _is_write = true;
@@ -56,9 +73,18 @@ Status BlockSplitBloomFilter::init(const char* buf, size_t size,
     if (buf == nullptr || size <= 1) {
         return Status::InvalidArgument("Invalid Parquet Bloom filter buffer of size {}", size);
     }
-    RETURN_IF_ERROR(set_hash_strategy(strategy, &_hash_func));
-    _data = new char[size];
+    RETURN_IF_ERROR(init_for_read(size, strategy));
     memcpy(_data, buf, size);
+    return Status::OK();
+}
+
+Status BlockSplitBloomFilter::init_for_read(size_t size, segment_v2::HashStrategyPB strategy) {
+    if (size <= 1) {
+        return Status::InvalidArgument("Invalid Parquet Bloom filter buffer of size {}", size);
+    }
+    RETURN_IF_ERROR(set_hash_strategy(strategy, &_hash_func));
+    _owned_data = make_unique_buffer<char>(size);
+    _data = _owned_data.get();
     _size = size;
     _num_bytes = size;
     _has_null = nullptr;

be/src/format_v2/parquet/reader/native/block_split_bloom_filter.h

@@ -20,6 +20,7 @@
 #include <cstddef>
 #include <cstdint>
 
+#include "core/custom_allocator.h"
 #include "storage/index/bloom_filter/bloom_filter.h"
 
 namespace doris::format::parquet::native {
@@ -28,8 +29,11 @@ namespace doris::format::parquet::native {
 // legacy Parquet reader merely to evaluate footer Bloom filters.
 class BlockSplitBloomFilter final : public segment_v2::BloomFilter {
 public:
+    ~BlockSplitBloomFilter() override;
     Status init(uint64_t filter_size, segment_v2::HashStrategyPB strategy) override;
     Status init(const char* buf, size_t size, segment_v2::HashStrategyPB strategy) override;
+    Status init_for_read(size_t size, segment_v2::HashStrategyPB strategy);
+    char* mutable_data() { return _data; }
     void add_bytes(const char* buf, size_t size) override;
     bool test_bytes(const char* buf, size_t size) const override;
     void set_has_null(bool has_null) override;
@@ -38,6 +42,8 @@ class BlockSplitBloomFilter final : public segment_v2::BloomFilter {
     bool test_hash(uint64_t hash) const override;
 
 private:
+    DorisUniqueBufferPtr<char> _owned_data;
+
     static constexpr int BYTES_PER_BLOCK = 32;
     static constexpr int BITS_SET_PER_BLOCK = 8;
     static constexpr uint32_t SALT[BITS_SET_PER_BLOCK] = {0x47b6137bU, 0x44974d91U, 0x8824ad5bU,