This is an automated email from the ASF dual-hosted git repository.
yiguolei 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 53264e01c6a [fix](be) Reduce Iceberg equality delete index memory
(#67570)
53264e01c6a is described below
commit 53264e01c6ab7fcb10c53ad6c00c38c537a2b2ba
Author: Gabriel <[email protected]>
AuthorDate: Tue Sep 8 11:04:38 2026 +0800
[fix](be) Reduce Iceberg equality delete index memory (#67570)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
FileScannerV2 keeps a node-based hash-to-row map for every row in a
multi-column Iceberg equality delete file. It also rebuilds that map for
every data split even though the parsed delete file is shared. This
amplifies memory usage and repeats index construction.
This change stores hash candidates in a sorted contiguous index and
shares the immutable index through the split cache. Full-key comparison
remains in place to resolve hash collisions. The optimization is
intentionally scoped to FileScannerV2; the legacy format reader is
unchanged.
### Optimization approach
Before this change, FileScannerV2 cached the parsed equality delete
block, but every split-local `EqualityDeletePredicate` still recomputed
all delete-row hashes and inserted them into a `std::multimap<hash,
row_index>`. This caused repeated `O(N log N)` construction and one
tree-node allocation per delete row for every predicate.
The new flow is:
1. When an equality delete file is loaded for the first time, compute
one hash for each delete row.
2. Store `{hash, row_index}` entries in a contiguous vector and sort it
by hash.
3. Cache this immutable index together with the parsed delete block.
4. Pass a shared reference to every split-local predicate instead of
rebuilding the index.
5. Use `lower_bound` and `upper_bound` to find candidates with the same
hash, then compare the complete equality key to preserve correctness
under hash collisions.
This removes the per-row node and allocator overhead of `std::multimap`,
improves cache locality, and changes index construction from once per
split to once per cached delete file. Runtime profile counters expose
index cache hits, misses, and index memory usage for validation.
### Release note
None
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- Full BE build
- [ ] 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:
- [x] No.
- [ ] Yes.
- 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/format_v2/expr/equality_delete_hash_index.h | 61 ++++++++++++++++++++++
.../format_v2/expr/equality_delete_predicate.cpp | 24 ++++++---
be/src/format_v2/expr/equality_delete_predicate.h | 12 +++--
be/src/format_v2/table/iceberg_reader.cpp | 23 +++++---
be/src/format_v2/table/iceberg_reader.h | 2 +
be/src/format_v2/table_reader.cpp | 8 +++
be/src/format_v2/table_reader.h | 3 ++
be/test/format_v2/table/iceberg_reader_test.cpp | 13 +++++
8 files changed, 128 insertions(+), 18 deletions(-)
diff --git a/be/src/format_v2/expr/equality_delete_hash_index.h
b/be/src/format_v2/expr/equality_delete_hash_index.h
new file mode 100644
index 00000000000..36685112b6d
--- /dev/null
+++ b/be/src/format_v2/expr/equality_delete_hash_index.h
@@ -0,0 +1,61 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <utility>
+#include <vector>
+
+namespace doris {
+
+class EqualityDeleteHashIndex {
+public:
+ struct Entry {
+ uint64_t hash;
+ size_t row_index;
+ };
+
+ using const_iterator = std::vector<Entry>::const_iterator;
+
+ explicit EqualityDeleteHashIndex(std::vector<uint64_t> hashes) {
+ _entries.reserve(hashes.size());
+ for (size_t row = 0; row < hashes.size(); ++row) {
+ _entries.push_back({hashes[row], row});
+ }
+ // A node-based map amplifies memory for large delete files. Keep
candidates contiguous;
+ // the retained delete block still resolves hash collisions with
full-key comparisons.
+ std::ranges::sort(_entries, {}, &Entry::hash);
+ }
+
+ std::pair<const_iterator, const_iterator> equal_range(uint64_t hash) const
{
+ const auto first = std::ranges::lower_bound(_entries, hash, {},
&Entry::hash);
+ const auto last = std::ranges::upper_bound(_entries, hash, {},
&Entry::hash);
+ return {first, last};
+ }
+
+ bool empty() const { return _entries.empty(); }
+
+ size_t memory_usage() const { return _entries.size() * sizeof(Entry); }
+
+private:
+ std::vector<Entry> _entries;
+};
+
+} // namespace doris
diff --git a/be/src/format_v2/expr/equality_delete_predicate.cpp
b/be/src/format_v2/expr/equality_delete_predicate.cpp
index 1d111aa7a74..57d7337f837 100644
--- a/be/src/format_v2/expr/equality_delete_predicate.cpp
+++ b/be/src/format_v2/expr/equality_delete_predicate.cpp
@@ -87,19 +87,27 @@ void update_varbinary_hashes(const ColumnWithTypeAndName&
entry, uint64_t* hashe
} // namespace
-EqualityDeletePredicate::EqualityDeletePredicate(Block delete_block,
std::vector<int> field_ids)
- : VExpr(), _delete_block(std::move(delete_block)),
_field_ids(std::move(field_ids)) {
+EqualityDeletePredicate::EqualityDeletePredicate(
+ Block delete_block, std::vector<int> field_ids,
+ std::shared_ptr<const EqualityDeleteHashIndex> delete_hash_index)
+ : _delete_block(std::move(delete_block)),
+ _field_ids(std::move(field_ids)),
+ _delete_hash_index(std::move(delete_hash_index)) {
_node_type = TExprNodeType::PREDICATE;
_opcode = TExprOpcode::DELETE;
_data_type = std::make_shared<DataTypeBool>();
_expr_name = "EqualityDeletePredicate";
DCHECK_EQ(_delete_block.columns(), _field_ids.size());
- _delete_hashes = _build_hashes(_delete_block);
- for (size_t row = 0; row < _delete_hashes.size(); ++row) {
- _delete_hash_map.emplace(_delete_hashes[row], row);
+ if (_delete_hash_index == nullptr) {
+ _delete_hash_index = build_hash_index(_delete_block);
}
}
+std::shared_ptr<const EqualityDeleteHashIndex>
EqualityDeletePredicate::build_hash_index(
+ const Block& delete_block) {
+ return std::make_shared<const
EqualityDeleteHashIndex>(_build_hashes(delete_block));
+}
+
Status EqualityDeletePredicate::prepare(RuntimeState* state, const
RowDescriptor& desc,
VExprContext* context) {
RETURN_IF_ERROR_OR_PREPARED(VExpr::prepare(state, desc, context));
@@ -171,15 +179,15 @@ Status
EqualityDeletePredicate::execute_column_impl(VExprContext* context, const
ColumnPtr EqualityDeletePredicate::_evaluate_key_block(const Block&
data_key_block) const {
const auto rows = data_key_block.rows();
auto res_col = ColumnBool::create(rows, 0);
- if (_delete_hash_map.empty() || rows == 0) {
+ if (_delete_hash_index->empty() || rows == 0) {
return res_col;
}
auto data_hashes = _build_hashes(data_key_block);
auto& result_data = res_col->get_data();
for (size_t row = 0; row < rows; ++row) {
- const auto range = _delete_hash_map.equal_range(data_hashes[row]);
+ const auto range = _delete_hash_index->equal_range(data_hashes[row]);
for (auto it = range.first; it != range.second; ++it) {
- if (_equal(data_key_block, row, it->second)) {
+ if (_equal(data_key_block, row, it->row_index)) {
result_data[row] = true;
break;
}
diff --git a/be/src/format_v2/expr/equality_delete_predicate.h
b/be/src/format_v2/expr/equality_delete_predicate.h
index 0e6f127cee2..8854cac00ca 100644
--- a/be/src/format_v2/expr/equality_delete_predicate.h
+++ b/be/src/format_v2/expr/equality_delete_predicate.h
@@ -19,7 +19,6 @@
#include <cstddef>
#include <cstdint>
-#include <map>
#include <memory>
#include <string>
#include <vector>
@@ -28,6 +27,7 @@
#include "core/block/block.h"
#include "exprs/function_context.h"
#include "exprs/vexpr.h"
+#include "format_v2/expr/equality_delete_hash_index.h"
namespace doris {
class RowDescriptor;
@@ -41,7 +41,9 @@ class EqualityDeletePredicate final : public VExpr {
ENABLE_FACTORY_CREATOR(EqualityDeletePredicate);
public:
- EqualityDeletePredicate(Block delete_block, std::vector<int> field_ids);
+ EqualityDeletePredicate(
+ Block delete_block, std::vector<int> field_ids,
+ std::shared_ptr<const EqualityDeleteHashIndex> delete_hash_index =
nullptr);
~EqualityDeletePredicate() override = default;
Status execute(VExprContext* context, Block* block, int* result_column_id)
const override;
@@ -55,6 +57,9 @@ public:
uint64_t get_digest(uint64_t seed) const override { return 0; }
const std::string& expr_name() const override { return _expr_name; }
+ static std::shared_ptr<const EqualityDeleteHashIndex> build_hash_index(
+ const Block& delete_block);
+
private:
static std::vector<uint64_t> _build_hashes(const Block& block);
ColumnPtr _evaluate_key_block(const Block& data_key_block) const;
@@ -63,8 +68,7 @@ private:
std::string _expr_name;
Block _delete_block;
std::vector<int> _field_ids;
- std::vector<uint64_t> _delete_hashes;
- std::multimap<uint64_t, size_t> _delete_hash_map;
+ std::shared_ptr<const EqualityDeleteHashIndex> _delete_hash_index;
};
} // namespace doris::format
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index 61403ec0f36..3fba91a3a5b 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -1481,8 +1481,8 @@ Status
IcebergTableReader::_build_missing_equality_delete_key_expr(
Status
IcebergTableReader::_append_equality_delete_predicates(format::FileScanRequest*
request) {
DORIS_CHECK(request != nullptr);
for (const auto& filter : _equality_delete_filters) {
- auto delete_predicate =
- std::make_shared<EqualityDeletePredicate>(filter.delete_block,
filter.field_ids);
+ auto delete_predicate = std::make_shared<EqualityDeletePredicate>(
+ filter.delete_block, filter.field_ids, filter.hash_index);
DCHECK_EQ(filter.field_ids.size(), filter.key_types.size());
bool has_missing_key = false;
for (size_t idx = 0; idx < filter.field_ids.size(); ++idx) {
@@ -1792,6 +1792,7 @@ Status
IcebergTableReader::_load_equality_delete_file(const TIcebergDeleteFileDe
}
RETURN_IF_ERROR(reader->close());
result->delete_block = mutable_delete_block.to_block();
+ result->hash_index =
EqualityDeletePredicate::build_hash_index(result->delete_block);
return Status::OK();
}
@@ -1808,11 +1809,13 @@ Status
IcebergTableReader::_read_equality_delete_file(const TIcebergDeleteFileDe
cache_key << ':' << field_id;
}
Status read_status = Status::OK();
+ bool cache_hit = false;
// Include the ordered equality ids in the key because the same physical
delete file can be
- // projected with different key layouts. The cached block and its key
metadata are immutable
- // after construction and therefore safe to copy into each split-local
predicate.
+ // projected with different key layouts. The cached block, key metadata,
and contiguous index
+ // are immutable after construction and therefore safe to share across
split-local predicates.
auto* cached_filter = _split_cache->get<EqualityDeleteFilter>(
- cache_key.str(), [&]() -> EqualityDeleteFilter* {
+ cache_key.str(),
+ [&]() -> EqualityDeleteFilter* {
auto result = std::make_unique<EqualityDeleteFilter>();
read_status = _load_equality_delete_file(delete_file,
scan_params, delete_io_ctx,
result.get());
@@ -1820,9 +1823,17 @@ Status
IcebergTableReader::_read_equality_delete_file(const TIcebergDeleteFileDe
return nullptr;
}
return result.release();
- });
+ },
+ &cache_hit);
RETURN_IF_ERROR(read_status);
DORIS_CHECK(cached_filter != nullptr);
+ COUNTER_UPDATE(cache_hit ? _profile.equality_delete_index_cache_hit_count
+ : _profile.equality_delete_index_cache_miss_count,
+ 1);
+ if (!cache_hit) {
+ COUNTER_UPDATE(_profile.equality_delete_hash_index_memory,
+
static_cast<int64_t>(cached_filter->hash_index->memory_usage()));
+ }
_equality_delete_filters.push_back(*cached_filter);
return Status::OK();
}
diff --git a/be/src/format_v2/table/iceberg_reader.h
b/be/src/format_v2/table/iceberg_reader.h
index b9fc79f2416..cc4f6058aef 100644
--- a/be/src/format_v2/table/iceberg_reader.h
+++ b/be/src/format_v2/table/iceberg_reader.h
@@ -33,6 +33,7 @@
namespace doris {
class Block;
+class EqualityDeleteHashIndex;
struct DeleteFileDesc;
namespace io {
struct FileDescription;
@@ -201,6 +202,7 @@ private:
std::vector<std::string> field_names;
std::vector<DataTypePtr> key_types;
Block delete_block;
+ std::shared_ptr<const EqualityDeleteHashIndex> hash_index;
};
std::vector<EqualityDeleteFilter> _equality_delete_filters;
// Scanner-shared cache supplied in SplitReadOptions. Parsed delete files
outlive one data-file
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index cf56a41f420..7161cf9adb5 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -955,6 +955,14 @@ Status TableReader::init(TableReadOptions&& options) {
TUnit::UNIT,
table_profile, 1);
_profile.parse_delete_file_time = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "ParseDeleteFileTime", table_profile, 1);
+ _profile.equality_delete_index_cache_hit_count =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"EqualityDeleteIndexCacheHitCount",
+ TUnit::UNIT, table_profile, 1);
+ _profile.equality_delete_index_cache_miss_count =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"EqualityDeleteIndexCacheMissCount",
+ TUnit::UNIT, table_profile, 1);
+ _profile.equality_delete_hash_index_memory =
ADD_CHILD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "EqualityDeleteHashIndexMemory",
TUnit::BYTES, table_profile, 1);
_profile.decoded_dv_cache_hit_count =
ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"DeletionVectorDecodedCacheHitCount",
TUnit::UNIT, table_profile, 1);
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index c763a2301ba..ccee6f8e406 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -106,6 +106,9 @@ struct ReadProfile {
RuntimeProfile::Counter* num_delete_files = nullptr;
RuntimeProfile::Counter* num_delete_rows = nullptr;
RuntimeProfile::Counter* parse_delete_file_time = nullptr;
+ RuntimeProfile::Counter* equality_delete_index_cache_hit_count = nullptr;
+ RuntimeProfile::Counter* equality_delete_index_cache_miss_count = nullptr;
+ RuntimeProfile::Counter* equality_delete_hash_index_memory = nullptr;
RuntimeProfile::Counter* decoded_dv_cache_hit_count = nullptr;
RuntimeProfile::Counter* decoded_dv_cache_miss_count = nullptr;
RuntimeProfile::Counter* dv_file_cache_hit_count = nullptr;
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 8e2b7dc778b..65a905a9e18 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -4523,6 +4523,16 @@ TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteFileIsReusedAcrossSplits) {
first_file_path,
{make_iceberg_equality_delete_file(delete_file_path, {0})}));
ASSERT_TRUE(reader.prepare_split(first_split).ok());
EXPECT_EQ(read_iceberg_ids(&reader, projected_columns),
std::vector<int32_t>({1, 3}));
+ auto* cache_misses =
profile.get_counter("EqualityDeleteIndexCacheMissCount");
+ auto* cache_hits = profile.get_counter("EqualityDeleteIndexCacheHitCount");
+ auto* index_memory = profile.get_counter("EqualityDeleteHashIndexMemory");
+ ASSERT_NE(cache_misses, nullptr);
+ ASSERT_NE(cache_hits, nullptr);
+ ASSERT_NE(index_memory, nullptr);
+ EXPECT_EQ(cache_misses->value(), 1);
+ EXPECT_EQ(cache_hits->value(), 0);
+ EXPECT_GT(index_memory->value(), 0);
+ const auto first_index_memory = index_memory->value();
// Removing the source after the first split proves that the second split
consumes the parsed
// delete block from SplitReadOptions.cache instead of reopening the
delete file.
@@ -4533,6 +4543,9 @@ TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteFileIsReusedAcrossSplits) {
second_file_path,
{make_iceberg_equality_delete_file(delete_file_path, {0})}));
ASSERT_TRUE(reader.prepare_split(second_split).ok());
EXPECT_EQ(read_iceberg_ids(&reader, projected_columns),
std::vector<int32_t>({1, 3}));
+ EXPECT_EQ(cache_misses->value(), 1);
+ EXPECT_EQ(cache_hits->value(), 1);
+ EXPECT_EQ(index_memory->value(), first_index_memory);
ASSERT_TRUE(reader.close().ok());
std::filesystem::remove_all(test_dir);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
