This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 e05c331da1c [improvement](be) Retain deletion vectors as Roaring
bitmaps (#65451)
e05c331da1c is described below
commit e05c331da1ce4b4041c62bafd9b1576c36af5fb8
Author: Gabriel <[email protected]>
AuthorDate: Fri Jul 10 23:30:15 2026 +0800
[improvement](be) Retain deletion vectors as Roaring bitmaps (#65451)
---
be/src/format/format_common.h | 13 ++-
be/src/format/orc/vorc_reader.cpp | 13 ++-
be/src/format/orc/vorc_reader.h | 7 ++
be/src/format/parquet/vparquet_reader.cpp | 27 +++++
be/src/format/parquet/vparquet_reader.h | 12 ++-
be/src/format/table/deletion_vector.h | 29 +++++
be/src/format/table/deletion_vector_reader.cpp | 45 +++++++-
be/src/format/table/deletion_vector_reader.h | 25 ++++-
.../table/iceberg_delete_file_reader_helper.cpp | 12 ++-
.../table/iceberg_delete_file_reader_helper.h | 8 +-
be/src/format/table/iceberg_reader.h | 6 ++
be/src/format/table/iceberg_reader_mixin.h | 59 +++++++---
be/src/format/table/paimon_reader.cpp | 93 +++++++++++-----
be/src/format/table/paimon_reader.h | 17 ++-
be/src/format_v2/expr/delete_predicate.cpp | 35 +++++-
be/src/format_v2/expr/delete_predicate.h | 5 +-
be/src/format_v2/table/iceberg_reader.cpp | 9 +-
be/src/format_v2/table_reader.cpp | 120 +++++++++++++--------
be/src/format_v2/table_reader.h | 36 +++++--
be/test/format/orc/orc_reader_test.cpp | 72 +++++++++++++
be/test/format/table/paimon_cpp_reader_test.cpp | 54 +++++++---
be/test/format_v2/expr/delete_predicate_test.cpp | 24 +++++
be/test/format_v2/table/paimon_reader_test.cpp | 31 +++---
23 files changed, 602 insertions(+), 150 deletions(-)
diff --git a/be/src/format/format_common.h b/be/src/format/format_common.h
index 70fb701d515..2d65ed721ef 100644
--- a/be/src/format/format_common.h
+++ b/be/src/format/format_common.h
@@ -89,12 +89,18 @@ public:
}
template <class T>
- T* get(const KType& key, const std::function<T*()> create_func) {
+ T* get(const KType& key, const std::function<T*()> create_func, bool*
cache_hit = nullptr) {
std::lock_guard<std::mutex> lock(_lock);
auto it = _storage.find(key);
if (it != _storage.end()) {
+ if (cache_hit != nullptr) {
+ *cache_hit = true;
+ }
return reinterpret_cast<T*>(it->second);
} else {
+ if (cache_hit != nullptr) {
+ *cache_hit = false;
+ }
T* rawPtr = create_func();
if (rawPtr != nullptr) {
_delete_fn[key] = [](void* obj) { delete
reinterpret_cast<T*>(obj); };
@@ -129,8 +135,9 @@ public:
}
template <class T>
- T* get(const std::string& key, const std::function<T*()> create_func) {
- return _shards[_get_idx(key)]->get(key, create_func);
+ T* get(const std::string& key, const std::function<T*()> create_func,
+ bool* cache_hit = nullptr) {
+ return _shards[_get_idx(key)]->get(key, create_func, cache_hit);
}
private:
diff --git a/be/src/format/orc/vorc_reader.cpp
b/be/src/format/orc/vorc_reader.cpp
index 0b8b55cd2ba..28ef930057f 100644
--- a/be/src/format/orc/vorc_reader.cpp
+++ b/be/src/format/orc/vorc_reader.cpp
@@ -2706,7 +2706,8 @@ Status OrcReader::_get_next_block_impl(Block* block,
size_t* read_rows, bool* eo
_execute_filter_position_delete_rowids(*_delete_rows_filter_ptr, start_row);
RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(
block, columns_to_filter,
(*_delete_rows_filter_ptr)));
- } else if (_position_delete_ordered_rowids != nullptr) {
+ } else if (_position_delete_ordered_rowids != nullptr ||
+ (_deletion_vector != nullptr &&
!_deletion_vector->isEmpty())) {
std::unique_ptr<IColumn::Filter> filter(new
IColumn::Filter(block->rows(), 1));
_execute_filter_position_delete_rowids(*filter, start_row);
RETURN_IF_CATCH_EXCEPTION(
@@ -3492,6 +3493,16 @@ void
ORCFileInputStream::_build_large_ranges_input_stripe_streams(
}
void OrcReader::_execute_filter_position_delete_rowids(IColumn::Filter&
filter, int64_t start_row) {
+ if (_deletion_vector != nullptr) {
+ auto it = _deletion_vector->begin();
+ it.move(cast_set<uint64_t>(start_row));
+ const auto end = _deletion_vector->end();
+ const auto end_row = cast_set<uint64_t>(start_row +
_batch->numElements);
+ while (it != end && *it < end_row) {
+ filter[cast_set<size_t>(*it - cast_set<uint64_t>(start_row))] = 0;
+ ++it;
+ }
+ }
if (_position_delete_ordered_rowids == nullptr) {
return;
}
diff --git a/be/src/format/orc/vorc_reader.h b/be/src/format/orc/vorc_reader.h
index 9f13726fd51..6d64f604cde 100644
--- a/be/src/format/orc/vorc_reader.h
+++ b/be/src/format/orc/vorc_reader.h
@@ -53,6 +53,7 @@
#include "orc/Type.hh"
#include "orc/Vector.hh"
#include "orc/sargs/Literal.hh"
+#include "roaring/roaring64map.hh"
#include "runtime/runtime_profile.h"
namespace doris {
@@ -197,6 +198,10 @@ public:
_position_delete_ordered_rowids = delete_rows;
}
+ void set_deletion_vector(const roaring::Roaring64Map* deletion_vector) {
+ _deletion_vector = deletion_vector;
+ }
+
void set_delete_rows(const AcidRowIDSet* delete_rows) { _delete_rows =
delete_rows; }
Status filter(orc::ColumnVectorBatch& data, uint16_t* sel, uint16_t size,
void* arg);
@@ -260,6 +265,7 @@ public:
bool has_delete_operations() const override {
return (_position_delete_ordered_rowids != nullptr &&
!_position_delete_ordered_rowids->empty()) ||
+ (_deletion_vector != nullptr && !_deletion_vector->isEmpty()) ||
(_delete_rows != nullptr && !_delete_rows->empty());
}
@@ -831,6 +837,7 @@ private:
//support iceberg position delete .
const std::vector<int64_t>* _position_delete_ordered_rowids = nullptr;
+ const roaring::Roaring64Map* _deletion_vector = nullptr;
std::unordered_map<const VSlotRef*, orc::PredicateDataType>
_vslot_ref_to_orc_predicate_data_type;
std::unordered_map<const VLiteral*, orc::Literal> _vliteral_to_orc_literal;
diff --git a/be/src/format/parquet/vparquet_reader.cpp
b/be/src/format/parquet/vparquet_reader.cpp
index 11758c76410..26caf7faac6 100644
--- a/be/src/format/parquet/vparquet_reader.cpp
+++ b/be/src/format/parquet/vparquet_reader.cpp
@@ -862,6 +862,33 @@ Status ParquetReader::_do_get_next_block(Block* block,
size_t* read_rows, bool*
RowGroupReader::PositionDeleteContext ParquetReader::_get_position_delete_ctx(
const tparquet::RowGroup& row_group, const
RowGroupReader::RowGroupIndex& row_group_index) {
+ if (_deletion_vector != nullptr) {
+ _row_group_deletion_vector_rows.clear();
+ auto it = _deletion_vector->begin();
+ it.move(cast_set<uint64_t>(row_group_index.first_row));
+ const auto end = _deletion_vector->end();
+ while (it != end && *it <
cast_set<uint64_t>(row_group_index.last_row)) {
+ _row_group_deletion_vector_rows.push_back(cast_set<int64_t>(*it));
+ ++it;
+ }
+ // Iceberg may plan a DV together with older position delete files.
Preserve both sources
+ // while keeping only the current row group's temporary expansion.
+ if (_delete_rows != nullptr) {
+ const auto position_begin =
std::lower_bound(_delete_rows->begin(), _delete_rows->end(),
+
row_group_index.first_row);
+ const auto position_end =
+ std::lower_bound(position_begin, _delete_rows->end(),
row_group_index.last_row);
+
_row_group_deletion_vector_rows.insert(_row_group_deletion_vector_rows.end(),
+ position_begin,
position_end);
+ std::ranges::sort(_row_group_deletion_vector_rows);
+ auto unique_end =
std::ranges::unique(_row_group_deletion_vector_rows).begin();
+ _row_group_deletion_vector_rows.erase(unique_end,
+
_row_group_deletion_vector_rows.end());
+ }
+ return RowGroupReader::PositionDeleteContext(
+ _row_group_deletion_vector_rows, row_group.num_rows,
row_group_index.first_row, 0,
+ cast_set<int64_t>(_row_group_deletion_vector_rows.size()));
+ }
if (_delete_rows == nullptr) {
return RowGroupReader::PositionDeleteContext(row_group.num_rows,
row_group_index.first_row);
}
diff --git a/be/src/format/parquet/vparquet_reader.h
b/be/src/format/parquet/vparquet_reader.h
index 4ce686803c7..9046641c6c0 100644
--- a/be/src/format/parquet/vparquet_reader.h
+++ b/be/src/format/parquet/vparquet_reader.h
@@ -40,6 +40,7 @@
#include "io/fs/file_meta_cache.h"
#include "io/fs/file_reader.h"
#include "io/fs/file_reader_writer_fwd.h"
+#include "roaring/roaring64map.hh"
#include "runtime/runtime_profile.h"
#include "storage/olap_scan_common.h"
#include "util/obj_lru_cache.h"
@@ -160,6 +161,12 @@ public:
// set the delete rows in current parquet file
void set_delete_rows(const std::vector<int64_t>* delete_rows) {
_delete_rows = delete_rows; }
+ // DVs stay compressed in the query cache. Only row ids belonging to the
row group being opened
+ // are materialized for the legacy RowGroupReader position-delete
interface.
+ void set_deletion_vector(const roaring::Roaring64Map* deletion_vector) {
+ _deletion_vector = deletion_vector;
+ }
+
int64_t size() const { return _file_reader->size(); }
Status _get_columns_impl(std::unordered_map<std::string, DataTypePtr>*
name_to_type) override;
@@ -212,7 +219,8 @@ public:
int64_t get_total_rows() const override;
bool has_delete_operations() const override {
- return _delete_rows != nullptr && !_delete_rows->empty();
+ return (_delete_rows != nullptr && !_delete_rows->empty()) ||
+ (_deletion_vector != nullptr && !_deletion_vector->isEmpty());
}
/// Disable row-group range filtering (needed when reading delete files
@@ -408,6 +416,8 @@ private:
// Deleted rows will be marked by Iceberg/Paimon. So we should filter
deleted rows when reading it.
const std::vector<int64_t>* _delete_rows = nullptr;
int64_t _delete_rows_index = 0;
+ const roaring::Roaring64Map* _deletion_vector = nullptr;
+ std::vector<int64_t> _row_group_deletion_vector_rows;
// parquet file reader object
RuntimeProfile* _profile = nullptr;
diff --git a/be/src/format/table/deletion_vector.h
b/be/src/format/table/deletion_vector.h
new file mode 100644
index 00000000000..2c89771b2e6
--- /dev/null
+++ b/be/src/format/table/deletion_vector.h
@@ -0,0 +1,29 @@
+// 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 "roaring/roaring64map.hh"
+
+namespace doris {
+
+// A deletion vector is already a bitmap on the wire. Keep decoded DVs
compressed in the
+// query-local cache instead of expanding every set bit into an int64_t.
Position delete files use
+// a different representation because their input is a stream of (file_path,
row_position) rows.
+using DeletionVector = roaring::Roaring64Map;
+
+} // namespace doris
diff --git a/be/src/format/table/deletion_vector_reader.cpp
b/be/src/format/table/deletion_vector_reader.cpp
index d7e33c923d9..65ce2831a09 100644
--- a/be/src/format/table/deletion_vector_reader.cpp
+++ b/be/src/format/table/deletion_vector_reader.cpp
@@ -22,6 +22,39 @@
#include "util/block_compression.h"
namespace doris {
+DeletionVectorReader::~DeletionVectorReader() {
+ // The file reader may retain the child IOContext. Destroy it before
merging and before the
+ // child statistics storage goes away.
+ _file_reader.reset();
+ _merge_io_statistics();
+}
+
+void DeletionVectorReader::_init_io_context() {
+ if (_parent_io_ctx == nullptr) {
+ return;
+ }
+ _reader_io_ctx = *_parent_io_ctx;
+ _reader_io_ctx.file_cache_stats = &_file_cache_stats;
+ _reader_io_ctx.file_reader_stats = &_file_reader_stats;
+ _io_ctx = &_reader_io_ctx;
+}
+
+void DeletionVectorReader::_merge_io_statistics() {
+ if (_statistics_merged || _parent_io_ctx == nullptr) {
+ return;
+ }
+ if (_parent_io_ctx->file_cache_stats != nullptr) {
+ _parent_io_ctx->file_cache_stats->merge_from(_file_cache_stats);
+ }
+ if (_parent_io_ctx->file_reader_stats != nullptr) {
+ _parent_io_ctx->file_reader_stats->read_calls +=
_file_reader_stats.read_calls;
+ _parent_io_ctx->file_reader_stats->read_bytes +=
_file_reader_stats.read_bytes;
+ _parent_io_ctx->file_reader_stats->read_time_ns +=
_file_reader_stats.read_time_ns;
+ _parent_io_ctx->file_reader_stats->read_rows +=
_file_reader_stats.read_rows;
+ }
+ _statistics_merged = true;
+}
+
Status DeletionVectorReader::open() {
if (_is_opened) [[unlikely]] {
return Status::OK();
@@ -37,9 +70,12 @@ Status DeletionVectorReader::open() {
}
Status DeletionVectorReader::read_at(size_t offset, Slice result) {
- if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) {
+ if (UNLIKELY(_parent_io_ctx && _parent_io_ctx->should_stop)) {
return Status::EndOfFile("stop read.");
}
+ if (_io_ctx != nullptr) {
+ _io_ctx->should_stop = _parent_io_ctx->should_stop;
+ }
size_t bytes_read = 0;
RETURN_IF_ERROR(_file_reader->read_at(offset, result, &bytes_read,
_io_ctx));
if (bytes_read != result.size) [[unlikely]] {
@@ -50,11 +86,14 @@ Status DeletionVectorReader::read_at(size_t offset, Slice
result) {
}
Status DeletionVectorReader::_create_file_reader() {
- if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) {
+ if (UNLIKELY(_parent_io_ctx && _parent_io_ctx->should_stop)) {
return Status::EndOfFile("stop read.");
}
_file_description.mtime = _desc.modification_time;
+ // Keep DV blobs on the normal remote-file path. When file cache is
enabled this creates a
+ // CachedRemoteFileReader, so a Puffin/delete-file range is persisted in
the disk File Cache;
+ // the query-local decoded cache above it only owns the Roaring bitmap.
io::FileReaderOptions reader_options =
FileFactory::get_reader_options(_state->query_options(),
_file_description);
_file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
@@ -79,4 +118,4 @@ void DeletionVectorReader::_init_system_properties() {
}
}
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/src/format/table/deletion_vector_reader.h
b/be/src/format/table/deletion_vector_reader.h
index 968344a8496..55f0099b6fe 100644
--- a/be/src/format/table/deletion_vector_reader.h
+++ b/be/src/format/table/deletion_vector_reader.h
@@ -24,10 +24,11 @@
#include "common/status.h"
#include "format/generic_reader.h"
+#include "format/table/deletion_vector.h"
#include "io/file_factory.h"
#include "io/fs/buffered_reader.h"
#include "io/fs/file_reader.h"
-#include "roaring/roaring64map.hh"
+#include "io/io_common.h"
#include "util/profile_collector.h"
#include "util/slice.h"
@@ -59,7 +60,7 @@ public:
DeletionVectorReader(RuntimeState* state, RuntimeProfile* profile,
const TFileScanRangeParams& params, const
TFileRangeDesc& range,
io::IOContext* io_ctx)
- : _state(state), _profile(profile), _params(params),
_io_ctx(io_ctx) {
+ : _state(state), _profile(profile), _params(params),
_parent_io_ctx(io_ctx) {
_desc = DeleteFileDesc {
.key = "",
.path = range.path,
@@ -68,20 +69,31 @@ public:
.size = range.size,
.file_size = range.__isset.file_size ? range.file_size : -1,
.modification_time = range.__isset.modification_time ?
range.modification_time : 0};
+ _init_io_context();
}
DeletionVectorReader(RuntimeState* state, RuntimeProfile* profile,
const TFileScanRangeParams& params, const
DeleteFileDesc& desc,
io::IOContext* io_ctx)
- : _state(state), _profile(profile), _params(params),
_io_ctx(io_ctx) {
+ : _state(state), _profile(profile), _params(params),
_parent_io_ctx(io_ctx) {
_desc = desc;
+ _init_io_context();
}
- ~DeletionVectorReader() = default;
+ ~DeletionVectorReader();
+ DeletionVectorReader(const DeletionVectorReader&) = delete;
+ DeletionVectorReader& operator=(const DeletionVectorReader&) = delete;
Status open();
Status read_at(size_t offset, Slice result);
+ // These are deliberately exposed separately from the decoded-DV cache
counters. Local I/O is
+ // a hit in the disk-backed File Cache; remote I/O is a File Cache miss. A
decoded-DV cache hit
+ // does not create a DeletionVectorReader and therefore changes neither
value.
+ const io::FileCacheStatistics& file_cache_statistics() const { return
_file_cache_stats; }
+
private:
void _init_system_properties();
void _init_file_description();
+ void _init_io_context();
+ void _merge_io_statistics();
Status _create_file_reader();
private:
@@ -89,7 +101,12 @@ private:
RuntimeProfile* _profile = nullptr;
DeleteFileDesc _desc;
const TFileScanRangeParams& _params;
+ io::IOContext* _parent_io_ctx = nullptr;
+ io::IOContext _reader_io_ctx;
io::IOContext* _io_ctx = nullptr;
+ io::FileCacheStatistics _file_cache_stats;
+ io::FileReaderStats _file_reader_stats;
+ bool _statistics_merged = false;
io::FileSystemProperties _system_properties;
io::FileDescription _file_description;
diff --git a/be/src/format/table/iceberg_delete_file_reader_helper.cpp
b/be/src/format/table/iceberg_delete_file_reader_helper.cpp
index 50738587d8b..ae5723c5d90 100644
--- a/be/src/format/table/iceberg_delete_file_reader_helper.cpp
+++ b/be/src/format/table/iceberg_delete_file_reader_helper.cpp
@@ -312,7 +312,7 @@ Status read_iceberg_position_delete_file(const
TIcebergDeleteFileDesc& delete_fi
Status read_iceberg_deletion_vector(const TIcebergDeleteFileDesc& delete_file,
const IcebergDeleteFileReaderOptions&
options,
- roaring::Roaring64Map* rows_to_delete) {
+ DeletionVector* rows_to_delete) {
if (options.state == nullptr || options.profile == nullptr ||
options.scan_params == nullptr ||
options.io_ctx == nullptr || rows_to_delete == nullptr) {
return Status::InvalidArgument("invalid deletion vector reader
options");
@@ -337,13 +337,17 @@ Status read_iceberg_deletion_vector(const
TIcebergDeleteFileDesc& delete_file,
RETURN_IF_ERROR(dv_reader.open());
std::vector<char> buf(delete_range.size);
- RETURN_IF_ERROR(dv_reader.read_at(delete_range.start_offset,
- {buf.data(),
cast_set<size_t>(delete_range.size)}));
+ const auto read_status = dv_reader.read_at(delete_range.start_offset,
+ {buf.data(),
cast_set<size_t>(delete_range.size)});
+ if (options.deletion_vector_file_cache_stats != nullptr) {
+
options.deletion_vector_file_cache_stats->merge_from(dv_reader.file_cache_statistics());
+ }
+ RETURN_IF_ERROR(read_status);
return decode_deletion_vector_buffer(buf.data(), delete_range.size,
rows_to_delete);
}
Status decode_iceberg_deletion_vector_buffer(const char* buf, size_t
buffer_size,
- roaring::Roaring64Map*
rows_to_delete) {
+ DeletionVector* rows_to_delete) {
return decode_deletion_vector_buffer(buf, buffer_size, rows_to_delete);
}
diff --git a/be/src/format/table/iceberg_delete_file_reader_helper.h
b/be/src/format/table/iceberg_delete_file_reader_helper.h
index 95109384076..0b4f601c0e8 100644
--- a/be/src/format/table/iceberg_delete_file_reader_helper.h
+++ b/be/src/format/table/iceberg_delete_file_reader_helper.h
@@ -26,6 +26,7 @@
#include <vector>
#include "common/status.h"
+#include "format/table/deletion_vector.h"
#include "io/io_common.h"
#include "roaring/roaring64map.hh"
@@ -50,6 +51,9 @@ struct IcebergDeleteFileReaderOptions {
io::IOContext* io_ctx = nullptr;
FileMetaCache* meta_cache = nullptr;
const std::string* fs_name = nullptr;
+ // Optional per-DV attribution. The reader still merges these values into
io_ctx so the
+ // scanner-wide File Cache profile remains complete.
+ io::FileCacheStatistics* deletion_vector_file_cache_stats = nullptr;
size_t batch_size = 102400;
};
@@ -71,7 +75,7 @@ std::string build_iceberg_deletion_vector_cache_key(const
std::string& data_file
const
TIcebergDeleteFileDesc& delete_file);
Status decode_iceberg_deletion_vector_buffer(const char* buf, size_t
buffer_size,
- roaring::Roaring64Map*
rows_to_delete);
+ DeletionVector* rows_to_delete);
Status read_iceberg_position_delete_file(const TIcebergDeleteFileDesc&
delete_file,
const IcebergDeleteFileReaderOptions&
options,
@@ -79,6 +83,6 @@ Status read_iceberg_position_delete_file(const
TIcebergDeleteFileDesc& delete_fi
Status read_iceberg_deletion_vector(const TIcebergDeleteFileDesc& delete_file,
const IcebergDeleteFileReaderOptions&
options,
- roaring::Roaring64Map* rows_to_delete);
+ DeletionVector* rows_to_delete);
} // namespace doris
diff --git a/be/src/format/table/iceberg_reader.h
b/be/src/format/table/iceberg_reader.h
index d4156964efe..205e46a3a1b 100644
--- a/be/src/format/table/iceberg_reader.h
+++ b/be/src/format/table/iceberg_reader.h
@@ -89,6 +89,10 @@ public:
ParquetReader::set_delete_rows(_iceberg_delete_rows);
}
+ void set_deletion_vector() final {
+ ParquetReader::set_deletion_vector(_iceberg_deletion_vector);
+ }
+
protected:
// Parquet-specific schema matching via on_before_init_reader hook
Status on_before_init_reader(ReaderInitContext* ctx) override;
@@ -133,6 +137,8 @@ public:
this->set_position_delete_rowids(_iceberg_delete_rows);
}
+ void set_deletion_vector() final {
OrcReader::set_deletion_vector(_iceberg_deletion_vector); }
+
protected:
// ORC-specific schema matching via on_before_init_reader hook
Status on_before_init_reader(ReaderInitContext* ctx) override;
diff --git a/be/src/format/table/iceberg_reader_mixin.h
b/be/src/format/table/iceberg_reader_mixin.h
index c0b2969678a..2bc7be1741f 100644
--- a/be/src/format/table/iceberg_reader_mixin.h
+++ b/be/src/format/table/iceberg_reader_mixin.h
@@ -81,6 +81,21 @@ public:
ADD_CHILD_TIMER(this->get_profile(), "DeleteRowsSortTime",
iceberg_profile);
_iceberg_profile.parse_delete_file_time =
ADD_CHILD_TIMER(this->get_profile(), "ParseDeleteFileTime",
iceberg_profile);
+ _iceberg_profile.decoded_cache_hit_count =
+ ADD_CHILD_COUNTER(this->get_profile(),
"DeletionVectorDecodedCacheHitCount",
+ TUnit::UNIT, iceberg_profile);
+ _iceberg_profile.decoded_cache_miss_count =
+ ADD_CHILD_COUNTER(this->get_profile(),
"DeletionVectorDecodedCacheMissCount",
+ TUnit::UNIT, iceberg_profile);
+ _iceberg_profile.file_cache_hit_count =
+ ADD_CHILD_COUNTER(this->get_profile(),
"DeletionVectorFileCacheHitCount",
+ TUnit::UNIT, iceberg_profile);
+ _iceberg_profile.file_cache_miss_count =
+ ADD_CHILD_COUNTER(this->get_profile(),
"DeletionVectorFileCacheMissCount",
+ TUnit::UNIT, iceberg_profile);
+ _iceberg_profile.file_cache_peer_read_count =
+ ADD_CHILD_COUNTER(this->get_profile(),
"DeletionVectorFileCachePeerReadCount",
+ TUnit::UNIT, iceberg_profile);
}
~IcebergReaderMixin() override = default;
@@ -96,6 +111,7 @@ public:
enum Fileformat { NONE, PARQUET, ORC, AVRO };
virtual void set_delete_rows() = 0;
+ virtual void set_deletion_vector() = 0;
// Table-level COUNT(*) is handled by CountReader (created by FileScanner
after
// init_reader). If _do_get_next_block is called, COUNT must have been
resolved.
@@ -318,6 +334,11 @@ protected:
RuntimeProfile::Counter* delete_files_read_time;
RuntimeProfile::Counter* delete_rows_sort_time;
RuntimeProfile::Counter* parse_delete_file_time;
+ RuntimeProfile::Counter* decoded_cache_hit_count;
+ RuntimeProfile::Counter* decoded_cache_miss_count;
+ RuntimeProfile::Counter* file_cache_hit_count;
+ RuntimeProfile::Counter* file_cache_miss_count;
+ RuntimeProfile::Counter* file_cache_peer_read_count;
};
bool _need_row_id_column = false;
@@ -328,6 +349,7 @@ protected:
ShardedKVCache* _kv_cache;
IcebergProfile _iceberg_profile;
const std::vector<int64_t>* _iceberg_delete_rows = nullptr;
+ const DeletionVector* _iceberg_deletion_vector = nullptr;
std::vector<std::string> _expand_col_names;
std::vector<ColumnWithTypeAndName> _expand_columns;
std::vector<std::string> _all_required_col_names;
@@ -834,35 +856,44 @@ Status
IcebergReaderMixin<BaseReader>::read_deletion_vector(
const std::string& data_file_path, const TIcebergDeleteFileDesc&
delete_file_desc) {
Status create_status = Status::OK();
SCOPED_TIMER(_iceberg_profile.delete_files_read_time);
- _iceberg_delete_rows = _kv_cache->template get<DeleteRows>(
+ bool decoded_cache_hit = false;
+ _iceberg_deletion_vector = _kv_cache->template get<DeletionVector>(
build_iceberg_deletion_vector_cache_key(data_file_path,
delete_file_desc),
- [&]() -> DeleteRows* {
- auto delete_rows = std::make_unique<DeleteRows>();
+ [&]() -> DeletionVector* {
+ auto deletion_vector = std::make_unique<DeletionVector>();
- roaring::Roaring64Map bitmap;
+ io::FileCacheStatistics file_cache_stats;
IcebergDeleteFileReaderOptions options;
options.state = this->get_state();
options.profile = this->get_profile();
options.scan_params = &this->get_scan_params();
options.io_ctx = this->get_io_ctx();
options.fs_name = &this->get_scan_range().fs_name;
- create_status = read_iceberg_deletion_vector(delete_file_desc,
options, &bitmap);
+ options.deletion_vector_file_cache_stats = &file_cache_stats;
+ create_status = read_iceberg_deletion_vector(delete_file_desc,
options,
+
deletion_vector.get());
+ COUNTER_UPDATE(_iceberg_profile.file_cache_hit_count,
+ file_cache_stats.num_local_io_total);
+ COUNTER_UPDATE(_iceberg_profile.file_cache_miss_count,
+ file_cache_stats.num_remote_io_total);
+ COUNTER_UPDATE(_iceberg_profile.file_cache_peer_read_count,
+ file_cache_stats.num_peer_io_total);
if (!create_status.ok()) [[unlikely]] {
return nullptr;
}
SCOPED_TIMER(_iceberg_profile.parse_delete_file_time);
- delete_rows->reserve(bitmap.cardinality());
- for (auto it = bitmap.begin(); it != bitmap.end(); it++) {
- delete_rows->push_back(*it);
- }
- COUNTER_UPDATE(_iceberg_profile.num_delete_rows,
delete_rows->size());
- return delete_rows.release();
- });
+ COUNTER_UPDATE(_iceberg_profile.num_delete_rows,
deletion_vector->cardinality());
+ return deletion_vector.release();
+ },
+ &decoded_cache_hit);
RETURN_IF_ERROR(create_status);
- if (!_iceberg_delete_rows->empty()) [[likely]] {
- set_delete_rows();
+ COUNTER_UPDATE(decoded_cache_hit ? _iceberg_profile.decoded_cache_hit_count
+ :
_iceberg_profile.decoded_cache_miss_count,
+ 1);
+ if (!_iceberg_deletion_vector->isEmpty()) [[likely]] {
+ set_deletion_vector();
}
return Status::OK();
}
diff --git a/be/src/format/table/paimon_reader.cpp
b/be/src/format/table/paimon_reader.cpp
index 51a80f3a967..4953078c32d 100644
--- a/be/src/format/table/paimon_reader.cpp
+++ b/be/src/format/table/paimon_reader.cpp
@@ -21,6 +21,7 @@
#include <cstring>
#include <memory>
+#include <utility>
#include <vector>
#include "common/status.h"
@@ -42,7 +43,10 @@ std::string build_paimon_deletion_vector_cache_key(const
TPaimonDeletionFileDesc
}
Status decode_paimon_deletion_vector_buffer(const char* buf, size_t
buffer_size,
- std::vector<int64_t>* delete_rows)
{
+ DeletionVector* deletion_vector) {
+ if (deletion_vector == nullptr) {
+ return Status::InvalidArgument("deletion_vector must not be null");
+ }
if (buffer_size < 8) [[unlikely]] {
return Status::DataQualityError("Deletion vector file size too small:
{}", buffer_size);
}
@@ -69,13 +73,38 @@ Status decode_paimon_deletion_vector_buffer(const char*
buf, size_t buffer_size,
e.what());
}
- delete_rows->reserve(roaring_bitmap.cardinality());
- for (auto it = roaring_bitmap.begin(); it != roaring_bitmap.end(); it++) {
- delete_rows->push_back(*it);
- }
+ *deletion_vector |= DeletionVector(std::move(roaring_bitmap));
return Status::OK();
}
+namespace {
+
+template <typename Profile>
+void init_deletion_vector_cache_profile(RuntimeProfile* profile, const char*
parent,
+ Profile* counters) {
+ counters->decoded_cache_hit_count =
+ ADD_CHILD_COUNTER(profile, "DeletionVectorDecodedCacheHitCount",
TUnit::UNIT, parent);
+ counters->decoded_cache_miss_count =
+ ADD_CHILD_COUNTER(profile, "DeletionVectorDecodedCacheMissCount",
TUnit::UNIT, parent);
+ counters->file_cache_hit_count =
+ ADD_CHILD_COUNTER(profile, "DeletionVectorFileCacheHitCount",
TUnit::UNIT, parent);
+ counters->file_cache_miss_count =
+ ADD_CHILD_COUNTER(profile, "DeletionVectorFileCacheMissCount",
TUnit::UNIT, parent);
+ counters->file_cache_peer_read_count =
+ ADD_CHILD_COUNTER(profile, "DeletionVectorFileCachePeerReadCount",
TUnit::UNIT, parent);
+}
+
+template <typename Profile>
+void update_deletion_vector_file_cache_profile(const DeletionVectorReader&
reader,
+ Profile* counters) {
+ const auto& stats = reader.file_cache_statistics();
+ COUNTER_UPDATE(counters->file_cache_hit_count, stats.num_local_io_total);
+ COUNTER_UPDATE(counters->file_cache_miss_count, stats.num_remote_io_total);
+ COUNTER_UPDATE(counters->file_cache_peer_read_count,
stats.num_peer_io_total);
+}
+
+} // namespace
+
// ============================================================================
// PaimonOrcReader
// ============================================================================
@@ -88,6 +117,7 @@ void PaimonOrcReader::_init_paimon_profile() {
ADD_CHILD_TIMER(get_profile(), "DeleteFileReadTime",
paimon_profile);
_paimon_profile.parse_deletion_vector_time =
ADD_CHILD_TIMER(get_profile(), "ParseDeletionVectorTime",
paimon_profile);
+ init_deletion_vector_cache_profile(get_profile(), paimon_profile,
&_paimon_profile);
}
Status PaimonOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
@@ -132,10 +162,11 @@ Status PaimonOrcReader::_init_deletion_vector() {
Status create_status = Status::OK();
SCOPED_TIMER(_paimon_profile.delete_files_read_time);
- using DeleteRows = std::vector<int64_t>;
- _delete_rows = _kv_cache->get<DeleteRows>(
- build_paimon_deletion_vector_cache_key(deletion_file), [&]() ->
DeleteRows* {
- auto delete_rows = std::make_unique<DeleteRows>();
+ bool decoded_cache_hit = false;
+ _deletion_vector = _kv_cache->get<DeletionVector>(
+ build_paimon_deletion_vector_cache_key(deletion_file),
+ [&]() -> DeletionVector* {
+ auto deletion_vector = std::make_unique<DeletionVector>();
TFileRangeDesc delete_range;
delete_range.__set_fs_name(get_scan_range().fs_name);
@@ -155,22 +186,27 @@ Status PaimonOrcReader::_init_deletion_vector() {
std::vector<char> buffer(bytes_read);
create_status =
dv_reader.read_at(deletion_file.offset,
{buffer.data(), bytes_read});
+ update_deletion_vector_file_cache_profile(dv_reader,
&_paimon_profile);
if (!create_status.ok()) [[unlikely]] {
return nullptr;
}
SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
create_status =
decode_paimon_deletion_vector_buffer(buffer.data(), bytes_read,
-
delete_rows.get());
+
deletion_vector.get());
if (!create_status.ok()) [[unlikely]] {
return nullptr;
}
- COUNTER_UPDATE(_paimon_profile.num_delete_rows,
delete_rows->size());
- return delete_rows.release();
- });
+ COUNTER_UPDATE(_paimon_profile.num_delete_rows,
deletion_vector->cardinality());
+ return deletion_vector.release();
+ },
+ &decoded_cache_hit);
RETURN_IF_ERROR(create_status);
- if (!_delete_rows->empty()) [[likely]] {
- set_position_delete_rowids(_delete_rows);
+ COUNTER_UPDATE(decoded_cache_hit ? _paimon_profile.decoded_cache_hit_count
+ :
_paimon_profile.decoded_cache_miss_count,
+ 1);
+ if (!_deletion_vector->isEmpty()) [[likely]] {
+ set_deletion_vector(_deletion_vector);
}
return Status::OK();
}
@@ -187,6 +223,7 @@ void PaimonParquetReader::_init_paimon_profile() {
ADD_CHILD_TIMER(get_profile(), "DeleteFileReadTime",
paimon_profile);
_paimon_profile.parse_deletion_vector_time =
ADD_CHILD_TIMER(get_profile(), "ParseDeletionVectorTime",
paimon_profile);
+ init_deletion_vector_cache_profile(get_profile(), paimon_profile,
&_paimon_profile);
}
Status PaimonParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
@@ -231,10 +268,11 @@ Status PaimonParquetReader::_init_deletion_vector() {
Status create_status = Status::OK();
SCOPED_TIMER(_paimon_profile.delete_files_read_time);
- using DeleteRows = std::vector<int64_t>;
- _delete_rows = _kv_cache->get<DeleteRows>(
- build_paimon_deletion_vector_cache_key(deletion_file), [&]() ->
DeleteRows* {
- auto delete_rows = std::make_unique<DeleteRows>();
+ bool decoded_cache_hit = false;
+ _deletion_vector = _kv_cache->get<DeletionVector>(
+ build_paimon_deletion_vector_cache_key(deletion_file),
+ [&]() -> DeletionVector* {
+ auto deletion_vector = std::make_unique<DeletionVector>();
TFileRangeDesc delete_range;
delete_range.__set_fs_name(get_scan_range().fs_name);
@@ -254,22 +292,27 @@ Status PaimonParquetReader::_init_deletion_vector() {
std::vector<char> buffer(bytes_read);
create_status =
dv_reader.read_at(deletion_file.offset,
{buffer.data(), bytes_read});
+ update_deletion_vector_file_cache_profile(dv_reader,
&_paimon_profile);
if (!create_status.ok()) [[unlikely]] {
return nullptr;
}
SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
create_status =
decode_paimon_deletion_vector_buffer(buffer.data(), bytes_read,
-
delete_rows.get());
+
deletion_vector.get());
if (!create_status.ok()) [[unlikely]] {
return nullptr;
}
- COUNTER_UPDATE(_paimon_profile.num_delete_rows,
delete_rows->size());
- return delete_rows.release();
- });
+ COUNTER_UPDATE(_paimon_profile.num_delete_rows,
deletion_vector->cardinality());
+ return deletion_vector.release();
+ },
+ &decoded_cache_hit);
RETURN_IF_ERROR(create_status);
- if (!_delete_rows->empty()) [[likely]] {
- ParquetReader::set_delete_rows(_delete_rows);
+ COUNTER_UPDATE(decoded_cache_hit ? _paimon_profile.decoded_cache_hit_count
+ :
_paimon_profile.decoded_cache_miss_count,
+ 1);
+ if (!_deletion_vector->isEmpty()) [[likely]] {
+ ParquetReader::set_deletion_vector(_deletion_vector);
}
return Status::OK();
}
diff --git a/be/src/format/table/paimon_reader.h
b/be/src/format/table/paimon_reader.h
index 616c58275b3..0f11825ce89 100644
--- a/be/src/format/table/paimon_reader.h
+++ b/be/src/format/table/paimon_reader.h
@@ -27,6 +27,7 @@
#include "common/status.h"
#include "format/orc/vorc_reader.h"
#include "format/parquet/vparquet_reader.h"
+#include "format/table/deletion_vector.h"
#include "format/table/table_schema_change_helper.h"
namespace doris {
@@ -35,7 +36,7 @@ class ShardedKVCache;
std::string build_paimon_deletion_vector_cache_key(const
TPaimonDeletionFileDesc& deletion_file);
Status decode_paimon_deletion_vector_buffer(const char* buf, size_t
buffer_size,
- std::vector<int64_t>* delete_rows);
+ DeletionVector* deletion_vector);
// PaimonOrcReader: directly inherits OrcReader (no composition wrapping).
// Schema mapping in on_before_init_reader, deletion vector reading in
on_after_init_reader.
@@ -79,9 +80,14 @@ private:
RuntimeProfile::Counter* num_delete_rows = nullptr;
RuntimeProfile::Counter* delete_files_read_time = nullptr;
RuntimeProfile::Counter* parse_deletion_vector_time = nullptr;
+ RuntimeProfile::Counter* decoded_cache_hit_count = nullptr;
+ RuntimeProfile::Counter* decoded_cache_miss_count = nullptr;
+ RuntimeProfile::Counter* file_cache_hit_count = nullptr;
+ RuntimeProfile::Counter* file_cache_miss_count = nullptr;
+ RuntimeProfile::Counter* file_cache_peer_read_count = nullptr;
};
- const std::vector<int64_t>* _delete_rows = nullptr;
+ const DeletionVector* _deletion_vector = nullptr;
ShardedKVCache* _kv_cache;
PaimonProfile _paimon_profile;
};
@@ -126,9 +132,14 @@ private:
RuntimeProfile::Counter* num_delete_rows = nullptr;
RuntimeProfile::Counter* delete_files_read_time = nullptr;
RuntimeProfile::Counter* parse_deletion_vector_time = nullptr;
+ RuntimeProfile::Counter* decoded_cache_hit_count = nullptr;
+ RuntimeProfile::Counter* decoded_cache_miss_count = nullptr;
+ RuntimeProfile::Counter* file_cache_hit_count = nullptr;
+ RuntimeProfile::Counter* file_cache_miss_count = nullptr;
+ RuntimeProfile::Counter* file_cache_peer_read_count = nullptr;
};
- const std::vector<int64_t>* _delete_rows = nullptr;
+ const DeletionVector* _deletion_vector = nullptr;
ShardedKVCache* _kv_cache;
PaimonProfile _paimon_profile;
};
diff --git a/be/src/format_v2/expr/delete_predicate.cpp
b/be/src/format_v2/expr/delete_predicate.cpp
index 9ab1090247c..772305290de 100644
--- a/be/src/format_v2/expr/delete_predicate.cpp
+++ b/be/src/format_v2/expr/delete_predicate.cpp
@@ -25,6 +25,7 @@
#include <cstddef>
#include <ostream>
+#include "common/cast_set.h"
#include "common/status.h"
#include "core/block/block.h"
#include "core/block/column_numbers.h"
@@ -34,7 +35,14 @@
namespace doris::format {
DeletePredicate::DeletePredicate(const std::vector<int64_t>& deleted_rows)
- : VExpr(), _deleted_rows(deleted_rows) {
+ : VExpr(), _deleted_rows(&deleted_rows) {
+ _node_type = TExprNodeType::PREDICATE;
+ _opcode = TExprOpcode::DELETE;
+ _data_type = std::make_shared<DataTypeBool>();
+}
+
+DeletePredicate::DeletePredicate(const roaring::Roaring64Map& deletion_vector)
+ : VExpr(), _deletion_vector(&deletion_vector) {
_node_type = TExprNodeType::PREDICATE;
_opcode = TExprOpcode::DELETE;
_data_type = std::make_shared<DataTypeBool>();
@@ -84,7 +92,8 @@ Status DeletePredicate::execute(VExprContext* context, Block*
block, int* result
assert_cast<const
ColumnInt64&>(*block->get_by_position(slot).column).get_data();
const auto count = row_ids.size();
auto res_col = ColumnBool::create(count, 0);
- if (_deleted_rows.empty()) {
+ if ((_deleted_rows == nullptr || _deleted_rows->empty()) &&
+ (_deletion_vector == nullptr || _deletion_vector->isEmpty())) {
block->insert({std::move(res_col), std::make_shared<DataTypeBool>(),
expr_name()});
*result_column_id = static_cast<int>(block->get_columns().size() - 1);
return Status::OK();
@@ -94,8 +103,26 @@ Status DeletePredicate::execute(VExprContext* context,
Block* block, int* result
*result_column_id = static_cast<int>(block->get_columns().size() - 1);
return Status::OK();
}
- const int64_t* delete_rows = _deleted_rows.data();
- const int64_t* delete_rows_end = delete_rows + _deleted_rows.size();
+ if (_deletion_vector != nullptr) {
+ auto it = _deletion_vector->begin();
+ it.move(cast_set<uint64_t>(row_ids[0]));
+ const auto end = _deletion_vector->end();
+ const auto last_row_id = cast_set<uint64_t>(row_ids[count - 1]);
+ while (it != end && *it <= last_row_id) {
+ const auto row = cast_set<int64_t>(*it);
+ if (const auto row_it = std::ranges::lower_bound(row_ids, row);
+ row_it != row_ids.end() && *row_it == row) {
+ res_col->get_data()[row_it - row_ids.begin()] = true;
+ }
+ ++it;
+ }
+ block->insert({std::move(res_col), std::make_shared<DataTypeBool>(),
expr_name()});
+ *result_column_id = static_cast<int>(block->get_columns().size() - 1);
+ return Status::OK();
+ }
+
+ const int64_t* delete_rows = _deleted_rows->data();
+ const int64_t* delete_rows_end = delete_rows + _deleted_rows->size();
const int64_t* start_pos = std::lower_bound(delete_rows, delete_rows_end,
row_ids[0]);
int64_t start_index = start_pos - delete_rows;
const int64_t* end_pos = std::upper_bound(start_pos, delete_rows_end,
row_ids[count - 1]);
diff --git a/be/src/format_v2/expr/delete_predicate.h
b/be/src/format_v2/expr/delete_predicate.h
index dce2de3edf2..85bde274d51 100644
--- a/be/src/format_v2/expr/delete_predicate.h
+++ b/be/src/format_v2/expr/delete_predicate.h
@@ -23,6 +23,7 @@
#include "common/status.h"
#include "exprs/function_context.h"
#include "exprs/vexpr.h"
+#include "roaring/roaring64map.hh"
namespace doris {
class RowDescriptor;
@@ -39,6 +40,7 @@ class DeletePredicate final : public VExpr {
public:
DeletePredicate(const std::vector<int64_t>& deleted_rows);
+ DeletePredicate(const roaring::Roaring64Map& deletion_vector);
~DeletePredicate() override = default;
Status execute(VExprContext* context, Block* block, int* result_column_id)
const override;
Status execute_column_impl(VExprContext* context, const Block* block,
const Selector* selector,
@@ -55,6 +57,7 @@ public:
private:
std::string _expr_name;
- const std::vector<int64_t>& _deleted_rows;
+ const std::vector<int64_t>* _deleted_rows = nullptr;
+ const roaring::Roaring64Map* _deletion_vector = nullptr;
};
} // namespace doris::format
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index 5c0693cbba6..1598fa783cd 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -374,11 +374,10 @@ Status IcebergTableReader::_init_delete_predicates(const
TTableFormatFileDesc& t
equality_delete_files.push_back(delete_file);
}
}
- // `_delete_rows != nullptr` means a deletion vector is parsed. Per
Iceberg scan planning,
- // position delete files apply only when there is no deletion vector for
the data file.
- if (_delete_rows != nullptr) {
- _position_delete_rows_storage = *_delete_rows;
- _delete_rows = &_position_delete_rows_storage;
+ // Per Iceberg scan planning, position delete files apply only when there
is no deletion vector
+ // for the data file. DVs and position deletes now intentionally use
different in-memory
+ // representations, so use the Roaring pointer as the DV sentinel.
+ if (_deletion_vector != nullptr) {
position_delete_files.clear();
}
// Initialize position and equality delete predicates. Position delete
files contain row
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 17938066420..11e2b30df6d 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -357,24 +357,18 @@ Status build_table_filters_from_conjunct(const
VExprContextSPtr& conjunct, Runti
}
Status parse_deletion_vector(const char* buf, size_t buffer_size,
DeleteFileDesc::Format format,
- DeleteRows* delete_rows) {
+ DeletionVector* deletion_vector) {
DORIS_CHECK(buf != nullptr);
- DORIS_CHECK(delete_rows != nullptr);
+ DORIS_CHECK(deletion_vector != nullptr);
DORIS_CHECK(format == DeleteFileDesc::Format::PAIMON ||
format == DeleteFileDesc::Format::ICEBERG);
if (format == DeleteFileDesc::Format::PAIMON) {
- RETURN_IF_ERROR(decode_paimon_deletion_vector_buffer(buf, buffer_size,
delete_rows));
+ RETURN_IF_ERROR(decode_paimon_deletion_vector_buffer(buf, buffer_size,
deletion_vector));
return Status::OK();
}
- roaring::Roaring64Map bitmap;
- RETURN_IF_ERROR(decode_iceberg_deletion_vector_buffer(buf, buffer_size,
&bitmap));
- delete_rows->reserve(bitmap.cardinality());
- for (auto it = bitmap.begin(); it != bitmap.end(); it++) {
- delete_rows->push_back(cast_set<int64_t>(*it));
- }
- return Status::OK();
+ return decode_iceberg_deletion_vector_buffer(buf, buffer_size,
deletion_vector);
}
} // namespace
@@ -406,6 +400,9 @@ std::string TableReader::debug_string() const {
<< ", current_file=" << current_file_debug_string(_current_task)
<< ", has_delete_rows=" << (_delete_rows != nullptr)
<< ", delete_row_count=" << (_delete_rows == nullptr ? 0 :
_delete_rows->size())
+ << ", has_deletion_vector=" << (_deletion_vector != nullptr)
+ << ", deletion_vector_cardinality="
+ << (_deletion_vector == nullptr ? 0 : _deletion_vector->cardinality())
<< ", has_system_properties=" << (_system_properties != nullptr) << ",
system_type="
<< (_system_properties == nullptr ?
static_cast<int>(TFileType::FILE_LOCAL)
:
static_cast<int>(_system_properties->system_type))
@@ -490,6 +487,20 @@ 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.decoded_dv_cache_hit_count =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"DeletionVectorDecodedCacheHitCount",
+ TUnit::UNIT, table_profile, 1);
+ _profile.decoded_dv_cache_miss_count = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "DeletionVectorDecodedCacheMissCount",
TUnit::UNIT, table_profile,
+ 1);
+ _profile.dv_file_cache_hit_count = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "DeletionVectorFileCacheHitCount",
TUnit::UNIT, table_profile, 1);
+ _profile.dv_file_cache_miss_count =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"DeletionVectorFileCacheMissCount",
+ TUnit::UNIT, table_profile, 1);
+ _profile.dv_file_cache_peer_read_count = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "DeletionVectorFileCachePeerReadCount",
TUnit::UNIT,
+ table_profile, 1);
_profile.exec_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "GetBlockTime",
table_profile, 1);
_profile.prepare_split_timer =
@@ -546,7 +557,8 @@ bool TableReader::_should_enable_condition_cache(const
FileScanRequest& file_req
// Delete files/deletion vectors are table-format state. They may change
independently of the
// data file path/mtime/size used by the external cache key, so caching
their result can become
// stale. Keep delete filtering enabled, but do not read or write
condition cache.
- if (_delete_rows != nullptr || !file_request.delete_conjuncts.empty()) {
+ if (_delete_rows != nullptr || _deletion_vector != nullptr ||
+ !file_request.delete_conjuncts.empty()) {
return false;
}
// Runtime filters can arrive late and their payload is not guaranteed to
be represented by the
@@ -752,6 +764,7 @@ Status TableReader::prepare_split(const SplitReadOptions&
options) {
: std::nullopt;
_global_rowid_context = options.global_rowid_context;
_delete_rows = nullptr;
+ _deletion_vector = nullptr;
_aggregate_pushdown_tried = false;
_remaining_table_level_count = -1;
_current_reader_reached_eof = false;
@@ -850,41 +863,58 @@ Status TableReader::_parse_delete_predicates(const
SplitReadOptions& options) {
DORIS_CHECK(options.cache != nullptr);
Status create_status = Status::OK();
- _delete_rows = options.cache->get<DeleteRows>(desc.key, [&]() ->
DeleteRows* {
- auto delete_rows = std::make_unique<DeleteRows>();
-
- DeletionVectorReader dv_reader(_runtime_state, _scanner_profile,
*_scan_params, desc,
- _io_ctx.get());
- create_status = dv_reader.open();
- if (!create_status.ok()) [[unlikely]] {
- return nullptr;
- }
-
- size_t bytes_read = desc.size;
- std::vector<char> buffer(bytes_read);
- DBUG_EXECUTE_IF("TableReader.parse_deletion_vector.io_error", {
- create_status = Status::IOError("injected format v2 deletion
vector read failure");
- return nullptr;
- });
- DBUG_EXECUTE_IF("TableReader.parse_deletion_vector.should_stop", {
- create_status = Status::EndOfFile("stop read.");
- return nullptr;
- });
- create_status = dv_reader.read_at(desc.start_offset,
{buffer.data(), bytes_read});
- if (!create_status.ok()) [[unlikely]] {
- return nullptr;
- }
-
- const char* buf = buffer.data();
- SCOPED_TIMER(_profile.parse_delete_file_time);
- create_status = parse_deletion_vector(buf, bytes_read,
desc.format, delete_rows.get());
- if (!create_status.ok()) [[unlikely]] {
- return nullptr;
- }
- COUNTER_UPDATE(_profile.num_delete_rows, delete_rows->size());
- return delete_rows.release();
- });
+ bool decoded_cache_hit = false;
+ _deletion_vector = options.cache->get<DeletionVector>(
+ desc.key,
+ [&]() -> DeletionVector* {
+ auto deletion_vector = std::make_unique<DeletionVector>();
+
+ DeletionVectorReader dv_reader(_runtime_state,
_scanner_profile, *_scan_params,
+ desc, _io_ctx.get());
+ create_status = dv_reader.open();
+ if (!create_status.ok()) [[unlikely]] {
+ return nullptr;
+ }
+
+ size_t bytes_read = desc.size;
+ std::vector<char> buffer(bytes_read);
+
DBUG_EXECUTE_IF("TableReader.parse_deletion_vector.io_error", {
+ create_status =
+ Status::IOError("injected format v2 deletion
vector read failure");
+ return nullptr;
+ });
+
DBUG_EXECUTE_IF("TableReader.parse_deletion_vector.should_stop", {
+ create_status = Status::EndOfFile("stop read.");
+ return nullptr;
+ });
+ create_status =
+ dv_reader.read_at(desc.start_offset,
{buffer.data(), bytes_read});
+ const auto& file_cache_stats =
dv_reader.file_cache_statistics();
+ COUNTER_UPDATE(_profile.dv_file_cache_hit_count,
+ file_cache_stats.num_local_io_total);
+ COUNTER_UPDATE(_profile.dv_file_cache_miss_count,
+ file_cache_stats.num_remote_io_total);
+ COUNTER_UPDATE(_profile.dv_file_cache_peer_read_count,
+ file_cache_stats.num_peer_io_total);
+ if (!create_status.ok()) [[unlikely]] {
+ return nullptr;
+ }
+
+ const char* buf = buffer.data();
+ SCOPED_TIMER(_profile.parse_delete_file_time);
+ create_status = parse_deletion_vector(buf, bytes_read,
desc.format,
+
deletion_vector.get());
+ if (!create_status.ok()) [[unlikely]] {
+ return nullptr;
+ }
+ COUNTER_UPDATE(_profile.num_delete_rows,
deletion_vector->cardinality());
+ return deletion_vector.release();
+ },
+ &decoded_cache_hit);
RETURN_IF_ERROR(create_status);
+ COUNTER_UPDATE(decoded_cache_hit ? _profile.decoded_dv_cache_hit_count
+ :
_profile.decoded_dv_cache_miss_count,
+ 1);
}
return Status::OK();
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 2a8a95817e8..da2cb351ff6 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -54,6 +54,7 @@
#include "exprs/vexpr_context.h"
#include "exprs/vexpr_fwd.h"
#include "exprs/vslot_ref.h"
+#include "format/table/deletion_vector.h"
#include "format_v2/column_data.h"
#include "format_v2/column_mapper.h"
#include "format_v2/expr/cast.h"
@@ -101,6 +102,11 @@ struct ReadProfile {
RuntimeProfile::Counter* num_delete_files = nullptr;
RuntimeProfile::Counter* num_delete_rows = nullptr;
RuntimeProfile::Counter* parse_delete_file_time = 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;
+ RuntimeProfile::Counter* dv_file_cache_miss_count = nullptr;
+ RuntimeProfile::Counter* dv_file_cache_peer_read_count = nullptr;
RuntimeProfile::Counter* exec_timer = nullptr;
RuntimeProfile::Counter* prepare_split_timer = nullptr;
RuntimeProfile::Counter* finalize_timer = nullptr;
@@ -568,20 +574,28 @@ protected:
// Append DeletePredicate to file scan request if there are deletes. The
predicate will be evaluated in file reader level and filter out deleted rows
before returning data to table reader.
Status _append_delete_predicate(FileScanRequest* request) {
DORIS_CHECK(request != nullptr);
- if (_delete_rows == nullptr || _delete_rows->empty()) {
+ if ((_delete_rows == nullptr || _delete_rows->empty()) &&
+ (_deletion_vector == nullptr || _deletion_vector->isEmpty())) {
return Status::OK();
}
const auto row_position_column_id =
LocalColumnId(ROW_POSITION_COLUMN_ID);
_append_file_scan_column(request, row_position_column_id,
&request->predicate_columns);
- auto delete_predicate =
std::make_shared<DeletePredicate>(*_delete_rows);
const auto block_position =
request->local_positions.at(row_position_column_id);
- delete_predicate->add_child(VSlotRef::create_shared(
- cast_set<int>(block_position.value()),
cast_set<int>(block_position.value()), -1,
- std::make_shared<DataTypeInt64>(), ROW_POSITION_COLUMN_NAME));
-
- request->delete_conjuncts.push_back(
- VExprContext::create_shared(std::move(delete_predicate)));
+ auto append_predicate = [&](auto& deleted_rows) {
+ auto delete_predicate =
std::make_shared<DeletePredicate>(deleted_rows);
+ delete_predicate->add_child(VSlotRef::create_shared(
+ cast_set<int>(block_position.value()),
cast_set<int>(block_position.value()),
+ -1, std::make_shared<DataTypeInt64>(),
ROW_POSITION_COLUMN_NAME));
+ request->delete_conjuncts.push_back(
+ VExprContext::create_shared(std::move(delete_predicate)));
+ };
+ if (_delete_rows != nullptr && !_delete_rows->empty()) {
+ append_predicate(*_delete_rows);
+ }
+ if (_deletion_vector != nullptr && !_deletion_vector->isEmpty()) {
+ append_predicate(*_deletion_vector);
+ }
return Status::OK();
}
@@ -889,7 +903,8 @@ protected:
// Only support aggregate pushdown when there is no delete or filter,
so
// the reduced rows consumed by the upper aggregate remain
semantically equivalent to a
// normal scan.
- if (_delete_rows != nullptr && !_delete_rows->empty()) {
+ if ((_delete_rows != nullptr && !_delete_rows->empty()) ||
+ (_deletion_vector != nullptr && !_deletion_vector->isEmpty())) {
return false;
}
if (!_table_filters.empty()) {
@@ -1505,6 +1520,7 @@ protected:
ReadProfile _profile;
// Parsed from row-position based delete files, including position delete
and deletion vector.
DeleteRows* _delete_rows = nullptr;
+ DeletionVector* _deletion_vector = nullptr;
TFileScanRangeParams* _scan_params;
std::shared_ptr<io::IOContext> _io_ctx;
RuntimeState* _runtime_state;
@@ -1602,7 +1618,7 @@ private:
return Status::OK();
}
- // Parse row-position deletes from table format specific parameters, and
fill in _delete_rows.
+ // Parse a DV into its compressed bitmap. Position delete files continue
to use _delete_rows.
Status _parse_delete_predicates(const SplitReadOptions& options);
};
diff --git a/be/test/format/orc/orc_reader_test.cpp
b/be/test/format/orc/orc_reader_test.cpp
index 4c830ad643b..b678f16c014 100644
--- a/be/test/format/orc/orc_reader_test.cpp
+++ b/be/test/format/orc/orc_reader_test.cpp
@@ -23,6 +23,10 @@
#include <tuple>
#include <vector>
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
#include "core/data_type/define_primitive_type.h"
#include "exec/common/util.hpp"
#include "exprs/vexpr_context.h"
@@ -32,6 +36,7 @@
#include "io/fs/file_meta_cache.h"
#include "orc/sargs/SearchArgument.hh"
#include "runtime/exec_env.h"
+#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
#include "testutil/desc_tbl_builder.h"
namespace doris {
@@ -246,4 +251,71 @@ TEST_F(OrcReaderTest,
set_batch_size_without_row_reader_is_safe) {
EXPECT_EQ(reader->_batch, nullptr);
}
+TEST_F(OrcReaderTest, deletion_vector_filters_rows_without_query_conjuncts) {
+ auto read_order_keys = [&](const roaring::Roaring64Map* deletion_vector,
+ std::vector<int32_t>* order_keys) -> Status {
+ ObjectPool object_pool;
+ DescriptorTblBuilder builder(&object_pool);
+ builder.declare_tuple() << std::make_tuple(
+ DataTypeFactory::instance().create_data_type(TYPE_INT, false),
"o_orderkey");
+ DescriptorTbl* desc_tbl = builder.build();
+ auto* tuple_desc =
const_cast<TupleDescriptor*>(desc_tbl->get_tuple_descriptor(0));
+ RowDescriptor row_desc(tuple_desc);
+
+ TFileScanRangeParams params;
+ params.__set_file_type(TFileType::FILE_LOCAL);
+ params.__set_format_type(TFileFormatType::FORMAT_ORC);
+ TFileRangeDesc range;
+ range.__set_path("./be/test/exec/test_data/orc_scanner/orders.orc");
+ range.__set_start_offset(0);
+ range.__set_size(1293);
+
+ RuntimeState state;
+ state.set_desc_tbl(desc_tbl);
+ RuntimeProfile profile("deletion_vector_without_conjuncts");
+ io::IOContext io_ctx;
+ auto reader = std::make_unique<OrcReader>(&profile, &state, params,
range, 2, "UTC",
+ &io_ctx, &cache, true);
+ std::vector<std::string> column_names = {"o_orderkey"};
+ std::unordered_map<std::string, uint32_t> col_name_to_block_idx =
{{"o_orderkey", 0}};
+ OrcInitContext orc_ctx;
+ orc_ctx.column_names = column_names;
+ orc_ctx.col_name_to_block_idx = &col_name_to_block_idx;
+ orc_ctx.tuple_descriptor = tuple_desc;
+ orc_ctx.row_descriptor = &row_desc;
+ orc_ctx.params = ¶ms;
+ orc_ctx.range = ⦥
+ RETURN_IF_ERROR(reader->init_reader(&orc_ctx));
+ reader->set_deletion_vector(deletion_vector);
+
+ const auto data_type = tuple_desc->slots()[0]->type();
+ Block block;
+ block.insert({data_type->create_column(), data_type, "o_orderkey"});
+ bool eof = false;
+ while (!eof) {
+ block.clear_column_data();
+ size_t read_rows = 0;
+ RETURN_IF_ERROR(reader->get_next_block(&block, &read_rows, &eof));
+ const auto values_column =
remove_nullable(block.get_by_position(0).column);
+ const auto& values = assert_cast<const
ColumnInt32&>(*values_column).get_data();
+ order_keys->insert(order_keys->end(), values.begin(),
values.end());
+ }
+ return Status::OK();
+ };
+
+ std::vector<int32_t> all_order_keys;
+ ASSERT_TRUE(read_order_keys(nullptr, &all_order_keys).ok());
+ ASSERT_FALSE(all_order_keys.empty());
+
+ // No conjuncts are installed in OrcInitContext. A DV-only split must
still enter the
+ // position-delete filtering branch and remove the row at the requested
absolute position.
+ roaring::Roaring64Map deletion_vector {0};
+ std::vector<int32_t> filtered_order_keys;
+ ASSERT_TRUE(read_order_keys(&deletion_vector, &filtered_order_keys).ok());
+
+ auto expected = all_order_keys;
+ expected.erase(expected.begin());
+ EXPECT_EQ(filtered_order_keys, expected);
+}
+
} // namespace doris
diff --git a/be/test/format/table/paimon_cpp_reader_test.cpp
b/be/test/format/table/paimon_cpp_reader_test.cpp
index 2858956c8f0..e323f4f1af4 100644
--- a/be/test/format/table/paimon_cpp_reader_test.cpp
+++ b/be/test/format/table/paimon_cpp_reader_test.cpp
@@ -149,23 +149,27 @@ TEST_F(PaimonCppReaderTest,
InitReaderFailsWithoutPaimonSplit) {
}
TEST(PaimonDeletionVectorTest, DecodeValidBuffer) {
- // Scenario: a valid Paimon DV buffer should decode to sorted row
positions that readers pass
- // into the common position-delete filter.
+ // Scenario: a valid Paimon DV stays compressed after decoding instead of
becoming one int64_t
+ // per deleted row in the query cache.
const auto buffer = build_paimon_deletion_vector_buffer({0, 3, 5});
- std::vector<int64_t> delete_rows;
+ DeletionVector deletion_vector;
const auto status =
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector);
ASSERT_TRUE(status.ok()) << status;
- EXPECT_EQ(delete_rows, std::vector<int64_t>({0, 3, 5}));
+ EXPECT_EQ(deletion_vector.cardinality(), 3);
+ EXPECT_TRUE(deletion_vector.contains(uint64_t {0}));
+ EXPECT_TRUE(deletion_vector.contains(uint64_t {3}));
+ EXPECT_TRUE(deletion_vector.contains(uint64_t {5}));
+ EXPECT_FALSE(deletion_vector.contains(uint64_t {4}));
}
TEST(PaimonDeletionVectorTest, RejectShortBuffer) {
// Scenario: malformed DV content must fail before reading the length and
magic fields.
const std::vector<char> buffer(7, '\0');
- std::vector<int64_t> delete_rows;
+ DeletionVector deletion_vector;
const auto status =
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector);
ASSERT_FALSE(status.ok());
EXPECT_NE(status.to_string().find("file size too small"),
std::string::npos);
@@ -176,9 +180,9 @@ TEST(PaimonDeletionVectorTest, RejectLengthMismatch) {
// slice from a shared deletion-vector file.
auto buffer = build_paimon_deletion_vector_buffer({1});
BigEndian::Store32(buffer.data(), static_cast<uint32_t>(buffer.size()));
- std::vector<int64_t> delete_rows;
+ DeletionVector deletion_vector;
const auto status =
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector);
ASSERT_FALSE(status.ok());
EXPECT_NE(status.to_string().find("length not match"), std::string::npos);
@@ -189,9 +193,9 @@ TEST(PaimonDeletionVectorTest, RejectMagicMismatch) {
// to unrelated bytes must be rejected.
auto buffer = build_paimon_deletion_vector_buffer({1});
buffer[4] = '\0';
- std::vector<int64_t> delete_rows;
+ DeletionVector deletion_vector;
const auto status =
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector);
ASSERT_FALSE(status.ok());
EXPECT_NE(status.to_string().find("invalid magic number"),
std::string::npos);
@@ -203,9 +207,9 @@ TEST(PaimonDeletionVectorTest, RejectCorruptRoaringBitmap) {
auto buffer = build_paimon_deletion_vector_buffer({1, 2});
buffer.resize(10);
BigEndian::Store32(buffer.data(), static_cast<uint32_t>(buffer.size() -
4));
- std::vector<int64_t> delete_rows;
+ DeletionVector deletion_vector;
const auto status =
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector);
ASSERT_FALSE(status.ok());
EXPECT_NE(status.to_string().find("failed to deserialize roaring bitmap"),
std::string::npos);
@@ -230,6 +234,30 @@ TEST(PaimonDeletionVectorTest,
CacheKeyIncludesOffsetAndLength) {
EXPECT_NE(first_key,
build_paimon_deletion_vector_cache_key(different_length));
}
+TEST(PaimonDeletionVectorTest, DecodedCacheReportsHitSeparatelyFromFileCache) {
+ // The decoded cache lookup result is reported by ShardedKVCache itself.
The creator represents
+ // the lower File Cache/read/decode path and must only run for the miss.
+ ShardedKVCache cache(1);
+ int create_count = 0;
+ bool cache_hit = true;
+ auto create = [&]() {
+ ++create_count;
+ auto* deletion_vector = new DeletionVector();
+ deletion_vector->add(uint64_t {7});
+ return deletion_vector;
+ };
+
+ const auto* first = cache.get<DeletionVector>("dv", create, &cache_hit);
+ EXPECT_FALSE(cache_hit);
+ ASSERT_NE(first, nullptr);
+ EXPECT_TRUE(first->contains(uint64_t {7}));
+
+ const auto* second = cache.get<DeletionVector>("dv", create, &cache_hit);
+ EXPECT_TRUE(cache_hit);
+ EXPECT_EQ(first, second);
+ EXPECT_EQ(create_count, 1);
+}
+
TEST(PaimonDeletionVectorTest, V1ParquetReaderReadErrorReleasesCacheEntry) {
RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
RuntimeProfile profile("paimon_v1_parquet_dv_test");
diff --git a/be/test/format_v2/expr/delete_predicate_test.cpp
b/be/test/format_v2/expr/delete_predicate_test.cpp
index 264a9fdf9b1..2f8d4af0f5b 100644
--- a/be/test/format_v2/expr/delete_predicate_test.cpp
+++ b/be/test/format_v2/expr/delete_predicate_test.cpp
@@ -62,6 +62,17 @@ protected:
VExprContext context(delete_predicate);
return delete_predicate->execute(&context, block, result_column_id);
}
+
+ static Status execute_delete_predicate(const roaring::Roaring64Map&
deletion_vector,
+ Block* block, int*
result_column_id) {
+ auto delete_predicate =
std::make_shared<DeletePredicate>(deletion_vector);
+ delete_predicate->_open_finished = true;
+ delete_predicate->add_child(
+ std::make_shared<MockSlotRef>(0,
std::make_shared<DataTypeInt64>()));
+
+ VExprContext context(delete_predicate);
+ return delete_predicate->execute(&context, block, result_column_id);
+ }
};
TEST_F(DeletePredicateTest, MatchDeletedRowsInInputRange) {
@@ -77,6 +88,19 @@ TEST_F(DeletePredicateTest, MatchDeletedRowsInInputRange) {
std::vector<UInt8>({0, 1, 0, 0, 1, 0, 1, 1}));
}
+TEST_F(DeletePredicateTest, MatchCompressedDeletionVectorWithoutExpansion) {
+ // The cached DV remains a Roaring bitmap all the way into the file-level
predicate. Include
+ // values on both sides of the current batch to exercise the iterator
range seek.
+ const roaring::Roaring64Map deletion_vector {1, 4, 8, 12, 20};
+ auto block = make_block({0, 1, 2, 3, 4, 5, 8, 12});
+
+ int result_column_id = -1;
+ const auto status = execute_delete_predicate(deletion_vector, &block,
&result_column_id);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(result_column_data(block, result_column_id),
+ std::vector<UInt8>({0, 1, 0, 0, 1, 0, 1, 1}));
+}
+
TEST_F(DeletePredicateTest, EmptyDeletedRowsReturnAllFalse) {
const std::vector<int64_t> deleted_rows;
auto block = make_block({1, 2, 3});
diff --git a/be/test/format_v2/table/paimon_reader_test.cpp
b/be/test/format_v2/table/paimon_reader_test.cpp
index 57dec5e4b2d..f1c9196e733 100644
--- a/be/test/format_v2/table/paimon_reader_test.cpp
+++ b/be/test/format_v2/table/paimon_reader_test.cpp
@@ -465,20 +465,24 @@ TEST(PaimonReaderTest,
DecodeDeletionVectorBufferUsesSharedFormatHelper) {
// Scenario: format_v2 TableReader reads a raw Paimon BitmapDeletionVector
range and delegates
// the binary parsing to the same helper used by the format reader path.
const auto buffer = build_paimon_deletion_vector_buffer({0, 3, 5});
- DeleteRows delete_rows;
+ DeletionVector deletion_vector;
- ASSERT_TRUE(
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
- EXPECT_EQ(delete_rows, DeleteRows({0, 3, 5}));
+ ASSERT_TRUE(decode_paimon_deletion_vector_buffer(buffer.data(),
buffer.size(), &deletion_vector)
+ .ok());
+ EXPECT_EQ(deletion_vector.cardinality(), 3);
+ EXPECT_TRUE(deletion_vector.contains(uint64_t {0}));
+ EXPECT_TRUE(deletion_vector.contains(uint64_t {3}));
+ EXPECT_TRUE(deletion_vector.contains(uint64_t {5}));
}
TEST(PaimonReaderTest, DecodeDeletionVectorBufferRejectsShortBuffer) {
// Scenario: a truncated Paimon DV must fail before reading the magic or
roaring payload.
const std::vector<char> buffer = {'\0', '\0', '\0', '\4'};
- DeleteRows delete_rows;
+ DeletionVector deletion_vector;
EXPECT_FALSE(
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector)
+ .ok());
}
TEST(PaimonReaderTest, DecodeDeletionVectorBufferRejectsLengthMismatch) {
@@ -486,20 +490,22 @@ TEST(PaimonReaderTest,
DecodeDeletionVectorBufferRejectsLengthMismatch) {
// accepted as a valid bitmap.
auto buffer = build_paimon_deletion_vector_buffer({1, 2});
BigEndian::Store32(buffer.data(), static_cast<uint32_t>(buffer.size()));
- DeleteRows delete_rows;
+ DeletionVector deletion_vector;
EXPECT_FALSE(
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector)
+ .ok());
}
TEST(PaimonReaderTest, DecodeDeletionVectorBufferRejectsMagicMismatch) {
// Scenario: format_v2 must reject non-Paimon payloads even when the range
length is valid.
auto buffer = build_paimon_deletion_vector_buffer({1, 2});
buffer[4] = '\0';
- DeleteRows delete_rows;
+ DeletionVector deletion_vector;
EXPECT_FALSE(
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector)
+ .ok());
}
TEST(PaimonReaderTest, DecodeDeletionVectorBufferRejectsCorruptRoaringBitmap) {
@@ -508,10 +514,11 @@ TEST(PaimonReaderTest,
DecodeDeletionVectorBufferRejectsCorruptRoaringBitmap) {
auto buffer = build_paimon_deletion_vector_buffer({1, 2});
buffer.resize(8);
BigEndian::Store32(buffer.data(), 4);
- DeleteRows delete_rows;
+ DeletionVector deletion_vector;
EXPECT_FALSE(
- decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&deletion_vector)
+ .ok());
}
// Scenario: PaimonReader must clear the previous split schema id before
reading a new split. A
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
