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 b75649609ef [fix](be) Fix deletion vector cache identity (#65281)
b75649609ef is described below
commit b75649609ef90e10d45619ca939d1aaefbaa41b5
Author: Gabriel <[email protected]>
AuthorDate: Wed Jul 8 09:50:03 2026 +0800
[fix](be) Fix deletion vector cache identity (#65281)
---
.../table/iceberg_delete_file_reader_helper.cpp | 15 +-
.../table/iceberg_delete_file_reader_helper.h | 6 +
be/src/format/table/iceberg_reader_mixin.h | 86 +---
be/src/format/table/paimon_reader.cpp | 264 +++++------
be/src/format/table/paimon_reader.h | 9 +
be/src/format_v2/table/iceberg_reader.cpp | 36 +-
be/src/format_v2/table/iceberg_reader.h | 2 -
be/src/format_v2/table/paimon_reader.cpp | 11 +-
be/src/format_v2/table/paimon_reader.h | 4 +
be/src/format_v2/table_reader.cpp | 56 +--
.../iceberg_delete_file_reader_helper_test.cpp | 48 ++
be/test/format/table/paimon_cpp_reader_test.cpp | 106 +++++
be/test/format_v2/table/iceberg_reader_test.cpp | 51 ++-
be/test/format_v2/table/paimon_reader_test.cpp | 102 ++++-
.../docker-compose/iceberg/iceberg.yaml.tpl | 17 +
.../create_preinstalled_scripts/iceberg/run28.sql | 50 ++
.../iceberg/test_iceberg_deletion_vector.out | 119 +++++
.../paimon/test_paimon_deletion_vector.out | 24 +
.../iceberg/test_iceberg_deletion_vector.groovy | 510 +++++++++++++++++++++
.../paimon/test_paimon_deletion_vector.groovy | 14 +-
20 files changed, 1230 insertions(+), 300 deletions(-)
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 c0fe46e1d35..5845e440d95 100644
--- a/be/src/format/table/iceberg_delete_file_reader_helper.cpp
+++ b/be/src/format/table/iceberg_delete_file_reader_helper.cpp
@@ -17,6 +17,7 @@
#include "format/table/iceberg_delete_file_reader_helper.h"
+#include <fmt/format.h>
#include <gen_cpp/parquet_types.h>
#include <parallel_hashmap/phmap.h>
@@ -216,6 +217,13 @@ bool is_iceberg_deletion_vector(const
TIcebergDeleteFileDesc& delete_file) {
return delete_file.__isset.content && delete_file.content == 3;
}
+std::string build_iceberg_deletion_vector_cache_key(const std::string&
data_file_path,
+ const
TIcebergDeleteFileDesc& delete_file) {
+ return fmt::format("delete_dv_{}:{}{}:{}#{}#{}", data_file_path.size(),
data_file_path,
+ delete_file.path.size(), delete_file.path,
delete_file.content_offset,
+ delete_file.content_size_in_bytes);
+}
+
Status read_iceberg_position_delete_file(const TIcebergDeleteFileDesc&
delete_file,
const IcebergDeleteFileReaderOptions&
options,
IcebergPositionDeleteVisitor*
visitor) {
@@ -308,7 +316,12 @@ Status read_iceberg_deletion_vector(const
TIcebergDeleteFileDesc& delete_file,
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)}));
- return decode_deletion_vector_buffer(buf.data(), delete_range.size,
rows_to_delete);
+ return decode_iceberg_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) {
+ return decode_deletion_vector_buffer(buf, buffer_size, rows_to_delete);
}
} // namespace doris
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 a6771212851..95109384076 100644
--- a/be/src/format/table/iceberg_delete_file_reader_helper.h
+++ b/be/src/format/table/iceberg_delete_file_reader_helper.h
@@ -67,6 +67,12 @@ TFileRangeDesc build_iceberg_delete_file_range(const
std::string& path);
bool is_iceberg_deletion_vector(const TIcebergDeleteFileDesc& delete_file);
+std::string build_iceberg_deletion_vector_cache_key(const std::string&
data_file_path,
+ const
TIcebergDeleteFileDesc& delete_file);
+
+Status decode_iceberg_deletion_vector_buffer(const char* buf, size_t
buffer_size,
+ roaring::Roaring64Map*
rows_to_delete);
+
Status read_iceberg_position_delete_file(const TIcebergDeleteFileDesc&
delete_file,
const IcebergDeleteFileReaderOptions&
options,
IcebergPositionDeleteVisitor*
visitor);
diff --git a/be/src/format/table/iceberg_reader_mixin.h
b/be/src/format/table/iceberg_reader_mixin.h
index c88703325a8..f27fa83467c 100644
--- a/be/src/format/table/iceberg_reader_mixin.h
+++ b/be/src/format/table/iceberg_reader_mixin.h
@@ -33,8 +33,8 @@
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "format/generic_reader.h"
-#include "format/table/deletion_vector_reader.h"
#include "format/table/equality_delete.h"
+#include "format/table/iceberg_delete_file_reader_helper.h"
#include "format/table/table_schema_change_helper.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
@@ -805,67 +805,31 @@ 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>(data_file_path, [&]() -> DeleteRows* {
- auto* delete_rows = new DeleteRows;
-
- TFileRangeDesc delete_range;
- delete_range.__set_fs_name(this->get_scan_range().fs_name);
- delete_range.path = delete_file_desc.path;
- delete_range.start_offset = delete_file_desc.content_offset;
- delete_range.size = delete_file_desc.content_size_in_bytes;
- delete_range.file_size = -1;
-
- DeletionVectorReader dv_reader(this->get_state(), this->get_profile(),
- this->get_scan_params(), delete_range,
this->get_io_ctx());
- create_status = dv_reader.open();
- if (!create_status.ok()) [[unlikely]] {
- return nullptr;
- }
-
- size_t buffer_size = delete_range.size;
- std::vector<char> buf(buffer_size);
- if (buffer_size < 12) [[unlikely]] {
- create_status = Status::DataQualityError("Deletion vector file
size too small: {}",
- buffer_size);
- return nullptr;
- }
-
- create_status = dv_reader.read_at(delete_range.start_offset,
{buf.data(), buffer_size});
- if (!create_status) [[unlikely]] {
- return nullptr;
- }
-
- auto total_length = BigEndian::Load32(buf.data());
- if (total_length + 8 != buffer_size) [[unlikely]] {
- create_status = Status::DataQualityError(
- "Deletion vector length mismatch, expected: {}, actual:
{}", total_length + 8,
- buffer_size);
- return nullptr;
- }
-
- constexpr static char MAGIC_NUMBER[] = {'\xD1', '\xD3', '\x39',
'\x64'};
- if (memcmp(buf.data() + sizeof(total_length), MAGIC_NUMBER, 4))
[[unlikely]] {
- create_status = Status::DataQualityError("Deletion vector magic
number mismatch");
- return nullptr;
- }
-
- roaring::Roaring64Map bitmap;
- SCOPED_TIMER(_iceberg_profile.parse_delete_file_time);
- try {
- bitmap = roaring::Roaring64Map::readSafe(buf.data() + 8,
buffer_size - 12);
- } catch (const std::runtime_error& e) {
- create_status = Status::DataQualityError("Decode roaring bitmap
failed, {}", e.what());
- return nullptr;
- }
+ _iceberg_delete_rows = _kv_cache->template get<DeleteRows>(
+ build_iceberg_deletion_vector_cache_key(data_file_path,
delete_file_desc),
+ [&]() -> DeleteRows* {
+ auto* delete_rows = new DeleteRows;
+
+ roaring::Roaring64Map bitmap;
+ 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);
+ if (!create_status.ok()) [[unlikely]] {
+ return nullptr;
+ }
- 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;
- });
+ 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;
+ });
RETURN_IF_ERROR(create_status);
if (!_iceberg_delete_rows->empty()) [[likely]] {
diff --git a/be/src/format/table/paimon_reader.cpp
b/be/src/format/table/paimon_reader.cpp
index 59a448def18..98f85658e32 100644
--- a/be/src/format/table/paimon_reader.cpp
+++ b/be/src/format/table/paimon_reader.cpp
@@ -17,14 +17,64 @@
#include "format/table/paimon_reader.h"
+#include <fmt/format.h>
+
+#include <cstring>
#include <vector>
#include "common/status.h"
+#include "exec/common/endian.h"
#include "format/table/deletion_vector_reader.h"
#include "runtime/runtime_state.h"
namespace doris {
+namespace {
+
+constexpr static char PAIMON_BITMAP_MAGIC[] = {'\x5E', '\x43', '\xF2', '\xD0'};
+
+} // namespace
+
+std::string build_paimon_deletion_vector_cache_key(const
TPaimonDeletionFileDesc& deletion_file) {
+ return fmt::format("paimon_dv_{}#{}#{}", deletion_file.path,
deletion_file.offset,
+ deletion_file.length);
+}
+
+Status decode_paimon_deletion_vector_buffer(const char* buf, size_t
buffer_size,
+ std::vector<int64_t>* delete_rows)
{
+ if (buffer_size < 8) [[unlikely]] {
+ return Status::DataQualityError("Deletion vector file size too small:
{}", buffer_size);
+ }
+
+ const uint32_t actual_length = BigEndian::Load32(buf);
+ if (actual_length + 4 != buffer_size) [[unlikely]] {
+ return Status::RuntimeError(
+ "DeletionVector deserialize error: length not match, "
+ "actual length: {}, expect length: {}",
+ actual_length, buffer_size - 4);
+ }
+
+ if (memcmp(buf + sizeof(actual_length), PAIMON_BITMAP_MAGIC, 4) != 0)
[[unlikely]] {
+ return Status::RuntimeError("DeletionVector deserialize error: invalid
magic number {}",
+ BigEndian::Load32(buf +
sizeof(actual_length)));
+ }
+
+ roaring::Roaring roaring_bitmap;
+ try {
+ roaring_bitmap = roaring::Roaring::readSafe(buf + 8, buffer_size - 8);
+ } catch (const std::runtime_error& e) {
+ return Status::RuntimeError(
+ "DeletionVector deserialize error: failed to deserialize
roaring bitmap, {}",
+ e.what());
+ }
+
+ delete_rows->reserve(roaring_bitmap.cardinality());
+ for (auto it = roaring_bitmap.begin(); it != roaring_bitmap.end(); it++) {
+ delete_rows->push_back(*it);
+ }
+ return Status::OK();
+}
+
// ============================================================================
// PaimonOrcReader
// ============================================================================
@@ -80,80 +130,43 @@ Status PaimonOrcReader::_init_deletion_vector() {
Status create_status = Status::OK();
- std::string key;
- key.resize(deletion_file.path.size() + sizeof(deletion_file.offset));
- memcpy(key.data(), deletion_file.path.data(), deletion_file.path.size());
- memcpy(key.data() + deletion_file.path.size(), &deletion_file.offset,
- sizeof(deletion_file.offset));
-
SCOPED_TIMER(_paimon_profile.delete_files_read_time);
using DeleteRows = std::vector<int64_t>;
- _delete_rows = _kv_cache->get<DeleteRows>(key, [&]() -> DeleteRows* {
- auto* delete_rows = new DeleteRows;
-
- TFileRangeDesc delete_range;
- delete_range.__set_fs_name(get_scan_range().fs_name);
- delete_range.path = deletion_file.path;
- delete_range.start_offset = deletion_file.offset;
- delete_range.size = deletion_file.length + 4;
- delete_range.file_size = -1;
-
- DeletionVectorReader dv_reader(get_state(), get_profile(),
get_scan_params(), delete_range,
- get_io_ctx());
- create_status = dv_reader.open();
- if (!create_status.ok()) [[unlikely]] {
- return nullptr;
- }
-
- size_t bytes_read = deletion_file.length + 4;
- std::vector<char> buffer(bytes_read);
- create_status = dv_reader.read_at(deletion_file.offset,
{buffer.data(), bytes_read});
- if (!create_status.ok()) [[unlikely]] {
- return nullptr;
- }
-
- const char* buf = buffer.data();
- uint32_t actual_length;
- std::memcpy(reinterpret_cast<char*>(&actual_length), buf, 4);
- std::reverse(reinterpret_cast<char*>(&actual_length),
- reinterpret_cast<char*>(&actual_length) + 4);
- buf += 4;
- if (actual_length != bytes_read - 4) [[unlikely]] {
- create_status = Status::RuntimeError(
- "DeletionVector deserialize error: length not match, "
- "actual length: {}, expect length: {}",
- actual_length, bytes_read - 4);
- return nullptr;
- }
- uint32_t magic_number;
- std::memcpy(reinterpret_cast<char*>(&magic_number), buf, 4);
- std::reverse(reinterpret_cast<char*>(&magic_number),
- reinterpret_cast<char*>(&magic_number) + 4);
- buf += 4;
- const static uint32_t MAGIC_NUMBER = 1581511376;
- if (magic_number != MAGIC_NUMBER) [[unlikely]] {
- create_status = Status::RuntimeError(
- "DeletionVector deserialize error: invalid magic number
{}", magic_number);
- return nullptr;
- }
-
- roaring::Roaring roaring_bitmap;
- SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
- try {
- roaring_bitmap = roaring::Roaring::readSafe(buf, bytes_read - 4);
- } catch (const std::runtime_error& e) {
- create_status = Status::RuntimeError(
- "DeletionVector deserialize error: failed to deserialize
roaring bitmap, {}",
- e.what());
- return nullptr;
- }
- delete_rows->reserve(roaring_bitmap.cardinality());
- for (auto it = roaring_bitmap.begin(); it != roaring_bitmap.end();
it++) {
- delete_rows->push_back(*it);
- }
- COUNTER_UPDATE(_paimon_profile.num_delete_rows, delete_rows->size());
- return delete_rows;
- });
+ _delete_rows = _kv_cache->get<DeleteRows>(
+ build_paimon_deletion_vector_cache_key(deletion_file), [&]() ->
DeleteRows* {
+ auto* delete_rows = new DeleteRows;
+
+ TFileRangeDesc delete_range;
+ delete_range.__set_fs_name(get_scan_range().fs_name);
+ delete_range.path = deletion_file.path;
+ delete_range.start_offset = deletion_file.offset;
+ delete_range.size = deletion_file.length + 4;
+ delete_range.file_size = -1;
+
+ DeletionVectorReader dv_reader(get_state(), get_profile(),
get_scan_params(),
+ delete_range, get_io_ctx());
+ create_status = dv_reader.open();
+ if (!create_status.ok()) [[unlikely]] {
+ return nullptr;
+ }
+
+ size_t bytes_read = deletion_file.length + 4;
+ std::vector<char> buffer(bytes_read);
+ create_status =
+ dv_reader.read_at(deletion_file.offset,
{buffer.data(), bytes_read});
+ 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);
+ if (!create_status.ok()) [[unlikely]] {
+ return nullptr;
+ }
+ COUNTER_UPDATE(_paimon_profile.num_delete_rows,
delete_rows->size());
+ return delete_rows;
+ });
RETURN_IF_ERROR(create_status);
if (!_delete_rows->empty()) [[likely]] {
set_position_delete_rowids(_delete_rows);
@@ -216,80 +229,43 @@ Status PaimonParquetReader::_init_deletion_vector() {
Status create_status = Status::OK();
- std::string key;
- key.resize(deletion_file.path.size() + sizeof(deletion_file.offset));
- memcpy(key.data(), deletion_file.path.data(), deletion_file.path.size());
- memcpy(key.data() + deletion_file.path.size(), &deletion_file.offset,
- sizeof(deletion_file.offset));
-
SCOPED_TIMER(_paimon_profile.delete_files_read_time);
using DeleteRows = std::vector<int64_t>;
- _delete_rows = _kv_cache->get<DeleteRows>(key, [&]() -> DeleteRows* {
- auto* delete_rows = new DeleteRows;
-
- TFileRangeDesc delete_range;
- delete_range.__set_fs_name(get_scan_range().fs_name);
- delete_range.path = deletion_file.path;
- delete_range.start_offset = deletion_file.offset;
- delete_range.size = deletion_file.length + 4;
- delete_range.file_size = -1;
-
- DeletionVectorReader dv_reader(get_state(), get_profile(),
get_scan_params(), delete_range,
- get_io_ctx());
- create_status = dv_reader.open();
- if (!create_status.ok()) [[unlikely]] {
- return nullptr;
- }
-
- size_t bytes_read = deletion_file.length + 4;
- std::vector<char> buffer(bytes_read);
- create_status = dv_reader.read_at(deletion_file.offset,
{buffer.data(), bytes_read});
- if (!create_status.ok()) [[unlikely]] {
- return nullptr;
- }
-
- const char* buf = buffer.data();
- uint32_t actual_length;
- std::memcpy(reinterpret_cast<char*>(&actual_length), buf, 4);
- std::reverse(reinterpret_cast<char*>(&actual_length),
- reinterpret_cast<char*>(&actual_length) + 4);
- buf += 4;
- if (actual_length != bytes_read - 4) [[unlikely]] {
- create_status = Status::RuntimeError(
- "DeletionVector deserialize error: length not match, "
- "actual length: {}, expect length: {}",
- actual_length, bytes_read - 4);
- return nullptr;
- }
- uint32_t magic_number;
- std::memcpy(reinterpret_cast<char*>(&magic_number), buf, 4);
- std::reverse(reinterpret_cast<char*>(&magic_number),
- reinterpret_cast<char*>(&magic_number) + 4);
- buf += 4;
- const static uint32_t MAGIC_NUMBER = 1581511376;
- if (magic_number != MAGIC_NUMBER) [[unlikely]] {
- create_status = Status::RuntimeError(
- "DeletionVector deserialize error: invalid magic number
{}", magic_number);
- return nullptr;
- }
-
- roaring::Roaring roaring_bitmap;
- SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
- try {
- roaring_bitmap = roaring::Roaring::readSafe(buf, bytes_read - 4);
- } catch (const std::runtime_error& e) {
- create_status = Status::RuntimeError(
- "DeletionVector deserialize error: failed to deserialize
roaring bitmap, {}",
- e.what());
- return nullptr;
- }
- delete_rows->reserve(roaring_bitmap.cardinality());
- for (auto it = roaring_bitmap.begin(); it != roaring_bitmap.end();
it++) {
- delete_rows->push_back(*it);
- }
- COUNTER_UPDATE(_paimon_profile.num_delete_rows, delete_rows->size());
- return delete_rows;
- });
+ _delete_rows = _kv_cache->get<DeleteRows>(
+ build_paimon_deletion_vector_cache_key(deletion_file), [&]() ->
DeleteRows* {
+ auto* delete_rows = new DeleteRows;
+
+ TFileRangeDesc delete_range;
+ delete_range.__set_fs_name(get_scan_range().fs_name);
+ delete_range.path = deletion_file.path;
+ delete_range.start_offset = deletion_file.offset;
+ delete_range.size = deletion_file.length + 4;
+ delete_range.file_size = -1;
+
+ DeletionVectorReader dv_reader(get_state(), get_profile(),
get_scan_params(),
+ delete_range, get_io_ctx());
+ create_status = dv_reader.open();
+ if (!create_status.ok()) [[unlikely]] {
+ return nullptr;
+ }
+
+ size_t bytes_read = deletion_file.length + 4;
+ std::vector<char> buffer(bytes_read);
+ create_status =
+ dv_reader.read_at(deletion_file.offset,
{buffer.data(), bytes_read});
+ 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);
+ if (!create_status.ok()) [[unlikely]] {
+ return nullptr;
+ }
+ COUNTER_UPDATE(_paimon_profile.num_delete_rows,
delete_rows->size());
+ return delete_rows;
+ });
RETURN_IF_ERROR(create_status);
if (!_delete_rows->empty()) [[likely]] {
ParquetReader::set_delete_rows(_delete_rows);
diff --git a/be/src/format/table/paimon_reader.h
b/be/src/format/table/paimon_reader.h
index 42a5d211a9b..10da0b784e2 100644
--- a/be/src/format/table/paimon_reader.h
+++ b/be/src/format/table/paimon_reader.h
@@ -17,10 +17,14 @@
#pragma once
+#include <gen_cpp/PlanNodes_types.h>
+
#include <memory>
+#include <string>
#include <utility>
#include <vector>
+#include "common/status.h"
#include "format/orc/vorc_reader.h"
#include "format/parquet/vparquet_reader.h"
#include "format/table/table_schema_change_helper.h"
@@ -28,6 +32,11 @@
namespace doris {
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);
+
// PaimonOrcReader: directly inherits OrcReader (no composition wrapping).
// Schema mapping in on_before_init_reader, deletion vector reading in
on_after_init_reader.
class PaimonOrcReader final : public OrcReader, public TableSchemaChangeHelper
{
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index ccc100f0504..eb1172ed358 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -18,7 +18,6 @@
#include "format_v2/table/iceberg_reader.h"
#include <algorithm>
-#include <cstring>
#include <memory>
#include <sstream>
#include <utility>
@@ -323,7 +322,10 @@ Status
IcebergTableReader::_parse_deletion_vector_file(const TTableFormatFileDes
return Status::InternalError("Deletion vector is missing content
offset or length");
}
- desc->key = _iceberg_delete_vector_cache_key(*deletion_vector);
+ const std::string data_file_path =
iceberg_params.__isset.original_file_path
+ ?
iceberg_params.original_file_path
+ : _data_file_path();
+ desc->key = build_iceberg_deletion_vector_cache_key(data_file_path,
*deletion_vector);
desc->path = deletion_vector->path;
desc->start_offset = deletion_vector->content_offset;
desc->size = deletion_vector->content_size_in_bytes;
@@ -358,16 +360,17 @@ Status IcebergTableReader::_init_delete_predicates(const
TTableFormatFileDesc& t
equality_delete_files.push_back(delete_file);
}
}
- // `_delete_rows != nullptr` means DeleteVector is parsed
+ // `_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;
+ position_delete_files.clear();
}
- // Combine position delete rows from both deletion vector and position
delete files, and
- // initialize equality delete predicates. Position delete files contain
row positions of
- // deleted rows, which can be directly added to `_delete_rows`. Equality
delete files contain
- // values of deleted rows, which require reading the files and building
predicates for later
- // filtering.
+ // Initialize position and equality delete predicates. Position delete
files contain row
+ // positions of deleted rows, which can be directly added to
`_delete_rows`. Equality delete
+ // files contain values of deleted rows, which require reading the files
and building
+ // predicates for later filtering.
if (!position_delete_files.empty()) {
RETURN_IF_ERROR(_init_position_delete_rows(position_delete_files));
}
@@ -379,23 +382,6 @@ Status IcebergTableReader::_init_delete_predicates(const
TTableFormatFileDesc& t
return Status::OK();
}
-std::string IcebergTableReader::_iceberg_delete_vector_cache_key(
- const TIcebergDeleteFileDesc& delete_file) {
- const std::string key_prefix = "iceberg_dv:";
- std::string key;
- key.resize(key_prefix.size() + delete_file.path.size() +
sizeof(delete_file.content_offset) +
- sizeof(delete_file.content_size_in_bytes));
- char* data = key.data();
- memcpy(data, key_prefix.data(), key_prefix.size());
- data += key_prefix.size();
- memcpy(data, delete_file.path.data(), delete_file.path.size());
- data += delete_file.path.size();
- memcpy(data, &delete_file.content_offset,
sizeof(delete_file.content_offset));
- data += sizeof(delete_file.content_offset);
- memcpy(data, &delete_file.content_size_in_bytes,
sizeof(delete_file.content_size_in_bytes));
- return key;
-}
-
std::shared_ptr<io::FileSystemProperties>
IcebergTableReader::_delete_file_system_properties(
const TFileScanRangeParams& scan_params) {
auto system_properties = std::make_shared<io::FileSystemProperties>();
diff --git a/be/src/format_v2/table/iceberg_reader.h
b/be/src/format_v2/table/iceberg_reader.h
index 1a2811ef968..15e03342d4b 100644
--- a/be/src/format_v2/table/iceberg_reader.h
+++ b/be/src/format_v2/table/iceberg_reader.h
@@ -119,8 +119,6 @@ private:
format::DeleteRows* _rows = nullptr;
};
- static std::string _iceberg_delete_vector_cache_key(const
TIcebergDeleteFileDesc& delete_file);
-
static std::shared_ptr<io::FileSystemProperties>
_delete_file_system_properties(
const TFileScanRangeParams& scan_params);
diff --git a/be/src/format_v2/table/paimon_reader.cpp
b/be/src/format_v2/table/paimon_reader.cpp
index 5b79b743fd3..c1a0f441950 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -19,12 +19,12 @@
#include <glog/logging.h>
-#include <cstring>
#include <string>
#include <utility>
#include "exprs/vexpr_context.h"
#include "format/table/deletion_vector_reader.h"
+#include "format/table/paimon_reader.h"
#include "format_v2/column_mapper.h"
#include "format_v2/jni/paimon_jni_reader.h"
#include "format_v2/table/schema_history_util.h"
@@ -66,14 +66,7 @@ Status PaimonReader::_parse_deletion_vector_file(const
TTableFormatFileDesc& t_d
}
const auto& deletion_file = table_desc.deletion_file;
- const std::string key_prefix = "paimon_dv:";
- desc->key.resize(key_prefix.size() + deletion_file.path.size() +
sizeof(deletion_file.offset));
- char* key_data = desc->key.data();
- memcpy(key_data, key_prefix.data(), key_prefix.size());
- key_data += key_prefix.size();
- memcpy(key_data, deletion_file.path.data(), deletion_file.path.size());
- key_data += deletion_file.path.size();
- memcpy(key_data, &deletion_file.offset, sizeof(deletion_file.offset));
+ desc->key = build_paimon_deletion_vector_cache_key(deletion_file);
desc->path = deletion_file.path;
desc->start_offset = deletion_file.offset;
desc->size = deletion_file.length + 4;
diff --git a/be/src/format_v2/table/paimon_reader.h
b/be/src/format_v2/table/paimon_reader.h
index 200c4e885b5..a7aa227ddb4 100644
--- a/be/src/format_v2/table/paimon_reader.h
+++ b/be/src/format_v2/table/paimon_reader.h
@@ -36,6 +36,10 @@ public:
Status TEST_annotate_file_schema(std::vector<format::ColumnDefinition>*
file_schema) {
return annotate_file_schema(file_schema);
}
+ Status TEST_parse_deletion_vector_file(const TTableFormatFileDesc& t_desc,
DeleteFileDesc* desc,
+ bool* has_delete_file) {
+ return _parse_deletion_vector_file(t_desc, desc, has_delete_file);
+ }
#endif
protected:
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 11317c5fd7b..e1bd6b04ad2 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -22,11 +22,9 @@
#include <gen_cpp/Types_types.h>
#include <algorithm>
-#include <cstring>
#include <ranges>
#include <set>
#include <sstream>
-#include <stdexcept>
#include <utility>
#include <vector>
@@ -36,17 +34,17 @@
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_struct.h"
-#include "exec/common/endian.h"
#include "exprs/vexpr_context.h"
#include "exprs/vslot_ref.h"
#include "format/table/deletion_vector_reader.h"
+#include "format/table/iceberg_delete_file_reader_helper.h"
+#include "format/table/paimon_reader.h"
#include "format_v2/column_mapper.h"
#include "format_v2/delimited_text/csv_reader.h"
#include "format_v2/delimited_text/text_reader.h"
#include "format_v2/json/json_reader.h"
#include "format_v2/native/native_reader.h"
#include "format_v2/parquet/parquet_reader.h"
-#include "roaring/roaring64map.hh"
#include "storage/segment/condition_cache.h"
#include "util/string_util.h"
@@ -362,59 +360,13 @@ Status parse_deletion_vector(const char* buf, size_t
buffer_size, DeleteFileDesc
DORIS_CHECK(format == DeleteFileDesc::Format::PAIMON ||
format == DeleteFileDesc::Format::ICEBERG);
- const size_t checksum_size = format == DeleteFileDesc::Format::ICEBERG ? 4
: 0;
- if (buffer_size < 8 + checksum_size) [[unlikely]] {
- return Status::DataQualityError("Deletion vector file size too small:
{}", buffer_size);
- }
-
- auto total_length = BigEndian::Load32(buf);
- if (total_length + 4 + checksum_size != buffer_size) [[unlikely]] {
- return Status::DataQualityError("Deletion vector length mismatch,
expected: {}, actual: {}",
- total_length + 4 + checksum_size,
buffer_size);
- }
-
- const char* bitmap_buf = buf + 8;
- const size_t bitmap_size = buffer_size - 8 - checksum_size;
if (format == DeleteFileDesc::Format::PAIMON) {
- // Paimon BitmapDeletionVector stores:
- // [4-byte big-endian length][4-byte magic 0x5E43F2D0][32-bit
roaring bitmap]
- // The length covers magic + bitmap, and does not include the leading
length field.
- constexpr static char PAIMON_BITMAP_MAGIC[] = {'\x5E', '\x43', '\xF2',
'\xD0'};
- if (memcmp(buf + sizeof(total_length), PAIMON_BITMAP_MAGIC, 4) != 0)
[[unlikely]] {
- return Status::DataQualityError(
- "Paimon deletion vector magic number mismatch, expected:
{}, actual: {}",
- BigEndian::Load32(PAIMON_BITMAP_MAGIC),
- BigEndian::Load32(buf + sizeof(total_length)));
- }
-
- roaring::Roaring bitmap;
- try {
- bitmap = roaring::Roaring::readSafe(bitmap_buf, bitmap_size);
- } catch (const std::runtime_error& e) {
- return Status::DataQualityError("Decode roaring bitmap failed,
{}", e.what());
- }
-
- delete_rows->reserve(bitmap.cardinality());
- for (auto it = bitmap.begin(); it != bitmap.end(); it++) {
- delete_rows->push_back(*it);
- }
+ RETURN_IF_ERROR(decode_paimon_deletion_vector_buffer(buf, buffer_size,
delete_rows));
return Status::OK();
}
- constexpr static char ICEBERG_DV_MAGIC[] = {'\xD1', '\xD3', '\x39',
'\x64'};
- if (memcmp(buf + sizeof(total_length), ICEBERG_DV_MAGIC, 4) != 0)
[[unlikely]] {
- return Status::DataQualityError(
- "Iceberg deletion vector magic number mismatch, expected: {},
actual: {}",
- BigEndian::Load32(ICEBERG_DV_MAGIC), BigEndian::Load32(buf +
sizeof(total_length)));
- }
-
roaring::Roaring64Map bitmap;
- try {
- bitmap = roaring::Roaring64Map::readSafe(bitmap_buf, bitmap_size);
- } catch (const std::runtime_error& e) {
- return Status::DataQualityError("Decode roaring bitmap failed, {}",
e.what());
- }
-
+ 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));
diff --git
a/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
b/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
index 658d3f70d0a..6ce562d6de0 100644
--- a/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
+++ b/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
@@ -71,6 +71,54 @@ TEST(IcebergDeleteFileReaderHelperTest,
IsNotDeletionVectorWhenContentMissing) {
EXPECT_FALSE(is_iceberg_deletion_vector(delete_file));
}
+TEST(IcebergDeleteFileReaderHelperTest,
DeletionVectorCacheKeyIncludesLocationAndRange) {
+ // Scenario: one Puffin file can hold several DV blobs. The cache key must
isolate both the
+ // data file and the blob range so a scanner never reuses another file's
DV.
+ TIcebergDeleteFileDesc first_delete_file;
+ first_delete_file.__set_path("s3://bucket/shared/delete.puffin");
+ first_delete_file.__set_content_offset(128);
+ first_delete_file.__set_content_size_in_bytes(64);
+
+ TIcebergDeleteFileDesc different_offset = first_delete_file;
+ different_offset.__set_content_offset(256);
+
+ TIcebergDeleteFileDesc different_length = first_delete_file;
+ different_length.__set_content_size_in_bytes(96);
+
+ TIcebergDeleteFileDesc different_delete_file = first_delete_file;
+ different_delete_file.__set_path("s3://bucket/shared/other-delete.puffin");
+
+ const std::string data_file_path = "s3://bucket/table/data-00001.parquet";
+ const auto first_key =
+ build_iceberg_deletion_vector_cache_key(data_file_path,
first_delete_file);
+
+ EXPECT_NE(first_key,
build_iceberg_deletion_vector_cache_key(data_file_path, different_offset));
+ EXPECT_NE(first_key,
build_iceberg_deletion_vector_cache_key(data_file_path, different_length));
+ EXPECT_NE(first_key,
+ build_iceberg_deletion_vector_cache_key(data_file_path,
different_delete_file));
+ EXPECT_NE(first_key,
+ build_iceberg_deletion_vector_cache_key(
+ "s3://bucket/table/snapshot-branch/data-00001.parquet",
first_delete_file));
+}
+
+TEST(IcebergDeleteFileReaderHelperTest,
DeletionVectorCacheKeyEscapesPathBoundaries) {
+ TIcebergDeleteFileDesc first_delete_file;
+ first_delete_file.__set_path("middle#right#tail.puffin");
+ first_delete_file.__set_content_offset(1);
+ first_delete_file.__set_content_size_in_bytes(2);
+
+ TIcebergDeleteFileDesc second_delete_file = first_delete_file;
+ second_delete_file.__set_path("right#tail.puffin");
+
+ const std::string first_data_file_path = "s3://bucket/table/data#left";
+ const std::string second_data_file_path =
"s3://bucket/table/data#left#middle";
+ ASSERT_EQ(first_data_file_path + "#" + first_delete_file.path,
+ second_data_file_path + "#" + second_delete_file.path);
+
+ EXPECT_NE(build_iceberg_deletion_vector_cache_key(first_data_file_path,
first_delete_file),
+ build_iceberg_deletion_vector_cache_key(second_data_file_path,
second_delete_file));
+}
+
TEST(IcebergDeleteFileReaderHelperTest,
ReadMixedEncodingParquetPositionDeleteFile) {
RuntimeProfile profile("test_profile");
RuntimeState runtime_state((TQueryOptions()), TQueryGlobals());
diff --git a/be/test/format/table/paimon_cpp_reader_test.cpp
b/be/test/format/table/paimon_cpp_reader_test.cpp
index b352696927e..ecec461d6b1 100644
--- a/be/test/format/table/paimon_cpp_reader_test.cpp
+++ b/be/test/format/table/paimon_cpp_reader_test.cpp
@@ -19,15 +19,39 @@
#include <gtest/gtest.h>
+#include <cstring>
#include <string>
#include <vector>
#include "core/block/block.h"
+#include "exec/common/endian.h"
+#include "format/table/paimon_reader.h"
+#include "roaring/roaring.hh"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
namespace doris {
+namespace {
+
+std::vector<char> build_paimon_deletion_vector_buffer(const
std::vector<uint32_t>& positions) {
+ roaring::Roaring rows;
+ for (const auto position : positions) {
+ rows.add(position);
+ }
+
+ const size_t bitmap_size = rows.getSizeInBytes();
+ const uint32_t total_length = static_cast<uint32_t>(4 + bitmap_size);
+ std::vector<char> buffer(4 + total_length);
+ BigEndian::Store32(buffer.data(), total_length);
+ constexpr char PAIMON_BITMAP_MAGIC[] = {'\x5E', '\x43', '\xF2', '\xD0'};
+ memcpy(buffer.data() + 4, PAIMON_BITMAP_MAGIC, 4);
+ rows.write(buffer.data() + 8);
+ return buffer;
+}
+
+} // namespace
+
class PaimonCppReaderTest : public testing::Test {
protected:
void SetUp() override {
@@ -92,4 +116,86 @@ TEST_F(PaimonCppReaderTest,
InitReaderFailsWithoutPaimonSplit) {
EXPECT_NE(status.to_string().find("missing paimon_split"),
std::string::npos);
}
+TEST(PaimonDeletionVectorTest, DecodeValidBuffer) {
+ // Scenario: a valid Paimon DV buffer should decode to sorted row
positions that readers pass
+ // into the common position-delete filter.
+ const auto buffer = build_paimon_deletion_vector_buffer({0, 3, 5});
+ std::vector<int64_t> delete_rows;
+ const auto status =
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(delete_rows, std::vector<int64_t>({0, 3, 5}));
+}
+
+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;
+ const auto status =
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("file size too small"),
std::string::npos);
+}
+
+TEST(PaimonDeletionVectorTest, RejectLengthMismatch) {
+ // Scenario: the big-endian length prefix protects against using a
truncated or over-read DV
+ // 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;
+ const auto status =
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("length not match"), std::string::npos);
+}
+
+TEST(PaimonDeletionVectorTest, RejectMagicMismatch) {
+ // Scenario: Paimon DV buffers have a fixed magic header, so a cache entry
or offset pointing
+ // to unrelated bytes must be rejected.
+ auto buffer = build_paimon_deletion_vector_buffer({1});
+ buffer[4] = '\0';
+ std::vector<int64_t> delete_rows;
+ const auto status =
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("invalid magic number"),
std::string::npos);
+}
+
+TEST(PaimonDeletionVectorTest, RejectCorruptRoaringBitmap) {
+ // Scenario: a buffer with a valid header but incomplete Roaring payload
should surface as a
+ // data-quality error instead of silently producing partial delete rows.
+ 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;
+ const auto status =
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows);
+
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("failed to deserialize roaring bitmap"),
std::string::npos);
+}
+
+TEST(PaimonDeletionVectorTest, CacheKeyIncludesOffsetAndLength) {
+ // Scenario: different Paimon tables or splits may reference the same DV
file with different
+ // ranges; cache keys must include both offset and length to avoid sharing
the wrong bitmap.
+ TPaimonDeletionFileDesc first_deletion_file;
+ first_deletion_file.__set_path("s3://bucket/table/deletion.dv");
+ first_deletion_file.__set_offset(128);
+ first_deletion_file.__set_length(64);
+
+ TPaimonDeletionFileDesc different_offset = first_deletion_file;
+ different_offset.__set_offset(256);
+
+ TPaimonDeletionFileDesc different_length = first_deletion_file;
+ different_length.__set_length(96);
+
+ const auto first_key =
build_paimon_deletion_vector_cache_key(first_deletion_file);
+ EXPECT_NE(first_key,
build_paimon_deletion_vector_cache_key(different_offset));
+ EXPECT_NE(first_key,
build_paimon_deletion_vector_cache_key(different_length));
+}
+
} // namespace doris
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 442ea660f35..4d57fcd950b 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -1179,6 +1179,7 @@ TEST(IcebergV2ReaderTest,
IcebergDeletionVectorUsesTableReaderDeleteFileInterfac
TTableFormatFileDesc table_format_desc;
TIcebergFileDesc iceberg_desc;
iceberg_desc.__set_format_version(2);
+ iceberg_desc.__set_original_file_path("data-a.parquet");
iceberg_desc.__set_delete_files({make_iceberg_deletion_vector("dv.bin", 8,
128)});
table_format_desc.__set_iceberg_params(iceberg_desc);
@@ -1195,6 +1196,52 @@ TEST(IcebergV2ReaderTest,
IcebergDeletionVectorUsesTableReaderDeleteFileInterfac
EXPECT_EQ(desc.format, DeleteFileDesc::Format::ICEBERG);
}
+TEST(IcebergV2ReaderTest,
IcebergDeletionVectorCacheKeyIncludesDataFileAndRange) {
+ // Scenario: Iceberg Puffin deletion vectors are scoped to the data file.
The same Puffin blob
+ // referenced by different data files, offsets or lengths must not share a
DeleteFileDesc key.
+ const auto shared_dv =
+ make_iceberg_deletion_vector("s3://bucket/table/shared-dv.puffin",
8, 128);
+ IcebergTableReaderDeleteFileTestHelper reader;
+
+ DeleteFileDesc first_desc;
+ bool has_delete_file = false;
+ ASSERT_TRUE(reader.parse_deletion_vector_file(
+
make_iceberg_table_format_desc("s3://bucket/table/data-a.parquet",
+ {shared_dv}),
+ &first_desc, &has_delete_file)
+ .ok());
+ EXPECT_TRUE(has_delete_file);
+
+ DeleteFileDesc different_data_file_desc;
+ ASSERT_TRUE(reader.parse_deletion_vector_file(
+
make_iceberg_table_format_desc("s3://bucket/table/data-b.parquet",
+ {shared_dv}),
+ &different_data_file_desc, &has_delete_file)
+ .ok());
+
+ DeleteFileDesc different_offset_desc;
+ ASSERT_TRUE(reader.parse_deletion_vector_file(
+ make_iceberg_table_format_desc(
+ "s3://bucket/table/data-a.parquet",
+ {make_iceberg_deletion_vector(
+
"s3://bucket/table/shared-dv.puffin", 16, 128)}),
+ &different_offset_desc, &has_delete_file)
+ .ok());
+
+ DeleteFileDesc different_length_desc;
+ ASSERT_TRUE(reader.parse_deletion_vector_file(
+ make_iceberg_table_format_desc(
+ "s3://bucket/table/data-a.parquet",
+ {make_iceberg_deletion_vector(
+
"s3://bucket/table/shared-dv.puffin", 8, 256)}),
+ &different_length_desc, &has_delete_file)
+ .ok());
+
+ EXPECT_NE(first_desc.key, different_data_file_desc.key);
+ EXPECT_NE(first_desc.key, different_offset_desc.key);
+ EXPECT_NE(first_desc.key, different_length_desc.key);
+}
+
TEST(IcebergV2ReaderTest, IcebergDeletionVectorRejectsMultipleDeleteFiles) {
TTableFormatFileDesc table_format_desc;
TIcebergFileDesc iceberg_desc;
@@ -1741,7 +1788,7 @@ TEST(IcebergV2ReaderTest,
IcebergTableReaderAppliesPositionDeleteFile) {
std::filesystem::remove_all(test_dir);
}
-TEST(IcebergV2ReaderTest,
IcebergTableReaderMergesDeletionVectorAndPositionDeleteFiles) {
+TEST(IcebergV2ReaderTest,
IcebergTableReaderIgnoresPositionDeleteFilesWhenDeletionVectorPresent) {
const auto test_dir =
std::filesystem::temp_directory_path() /
"doris_iceberg_delete_files_merge_test";
std::filesystem::remove_all(test_dir);
@@ -1784,7 +1831,7 @@ TEST(IcebergV2ReaderTest,
IcebergTableReaderMergesDeletionVectorAndPositionDelet
make_iceberg_position_delete_file(position_delete_path)}));
ASSERT_TRUE(reader.prepare_split(split_options).ok());
- EXPECT_EQ(read_iceberg_ids(&reader, projected_columns),
std::vector<int32_t>({2, 3, 5}));
+ EXPECT_EQ(read_iceberg_ids(&reader, projected_columns),
std::vector<int32_t>({2, 3, 4, 5}));
ASSERT_TRUE(reader.close().ok());
std::filesystem::remove_all(test_dir);
diff --git a/be/test/format_v2/table/paimon_reader_test.cpp
b/be/test/format_v2/table/paimon_reader_test.cpp
index 2e07706a71a..caa4de6ce44 100644
--- a/be/test/format_v2/table/paimon_reader_test.cpp
+++ b/be/test/format_v2/table/paimon_reader_test.cpp
@@ -44,6 +44,8 @@
#include "core/field.h"
#include "exec/common/endian.h"
#include "format/format_common.h"
+#include "format/table/deletion_vector_reader.h"
+#include "format/table/paimon_reader.h"
#include "format_v2/column_data.h"
#include "gen_cpp/ExternalTableSchema_types.h"
#include "gen_cpp/PlanNodes_types.h"
@@ -193,8 +195,8 @@ void write_int_pair_parquet_file(const std::string&
file_path, const std::vector
builder.build()));
}
-int64_t write_paimon_deletion_vector_file(const std::string& file_path,
- const std::vector<uint32_t>&
deleted_positions) {
+std::vector<char> build_paimon_deletion_vector_buffer(
+ const std::vector<uint32_t>& deleted_positions) {
roaring::Roaring rows;
for (const auto position : deleted_positions) {
rows.add(position);
@@ -207,13 +209,18 @@ int64_t write_paimon_deletion_vector_file(const
std::string& file_path,
constexpr char PAIMON_BITMAP_MAGIC[] = {'\x5E', '\x43', '\xF2', '\xD0'};
memcpy(blob.data() + 4, PAIMON_BITMAP_MAGIC, 4);
rows.write(blob.data() + 8);
+ return blob;
+}
+int64_t write_paimon_deletion_vector_file(const std::string& file_path,
+ const std::vector<uint32_t>&
deleted_positions) {
+ const auto blob = build_paimon_deletion_vector_buffer(deleted_positions);
std::ofstream output(file_path, std::ios::binary);
EXPECT_TRUE(output.is_open());
output.write(blob.data(), static_cast<std::streamsize>(blob.size()));
EXPECT_TRUE(output.good());
// Paimon DeletionFile.length is magic + bitmap length, excluding the
leading length field.
- return static_cast<int64_t>(total_length);
+ return static_cast<int64_t>(blob.size() - 4);
}
TFileScanRangeParams make_local_parquet_scan_params() {
@@ -382,6 +389,95 @@ TEST(PaimonReaderTest,
FallsBackToByNameWhenSplitHistorySchemaIsMissing) {
EXPECT_TRUE(file_schema[0].name_mapping.empty());
}
+TEST(PaimonReaderTest, DeletionVectorCacheKeyIncludesOffsetAndLength) {
+ // Scenario: format_v2 converts Paimon split metadata into a generic
DeleteFileDesc. The
+ // generated key must preserve offset and length so shared DV files do not
collide.
+ TTableFormatFileDesc table_format_params;
+ table_format_params.__isset.paimon_params = true;
+ TPaimonDeletionFileDesc deletion_file;
+ deletion_file.__set_path("s3://bucket/table/deletion.dv");
+ deletion_file.__set_offset(128);
+ deletion_file.__set_length(64);
+ table_format_params.paimon_params.__set_deletion_file(deletion_file);
+
+ paimon::PaimonReader reader;
+ DeleteFileDesc first_desc;
+ bool has_delete_file = false;
+ ASSERT_TRUE(reader.TEST_parse_deletion_vector_file(table_format_params,
&first_desc,
+ &has_delete_file)
+ .ok());
+ EXPECT_TRUE(has_delete_file);
+
+ table_format_params.paimon_params.deletion_file.__set_offset(256);
+ DeleteFileDesc different_offset_desc;
+ ASSERT_TRUE(reader.TEST_parse_deletion_vector_file(table_format_params,
&different_offset_desc,
+ &has_delete_file)
+ .ok());
+
+ table_format_params.paimon_params.deletion_file.__set_offset(128);
+ table_format_params.paimon_params.deletion_file.__set_length(96);
+ DeleteFileDesc different_length_desc;
+ ASSERT_TRUE(reader.TEST_parse_deletion_vector_file(table_format_params,
&different_length_desc,
+ &has_delete_file)
+ .ok());
+
+ EXPECT_NE(first_desc.key, different_offset_desc.key);
+ EXPECT_NE(first_desc.key, different_length_desc.key);
+}
+
+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;
+
+ ASSERT_TRUE(
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
+ EXPECT_EQ(delete_rows, DeleteRows({0, 3, 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;
+
+ EXPECT_FALSE(
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
+}
+
+TEST(PaimonReaderTest, DecodeDeletionVectorBufferRejectsLengthMismatch) {
+ // Scenario: a cached or remote Paimon DV range with a mismatched leading
length must not be
+ // 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;
+
+ EXPECT_FALSE(
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).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;
+
+ EXPECT_FALSE(
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
+}
+
+TEST(PaimonReaderTest, DecodeDeletionVectorBufferRejectsCorruptRoaringBitmap) {
+ // Scenario: a valid Paimon DV header with a corrupt roaring body should
return a data quality
+ // error instead of producing partial delete rows.
+ auto buffer = build_paimon_deletion_vector_buffer({1, 2});
+ buffer.resize(8);
+ BigEndian::Store32(buffer.data(), 4);
+ DeleteRows delete_rows;
+
+ EXPECT_FALSE(
+ decode_paimon_deletion_vector_buffer(buffer.data(), buffer.size(),
&delete_rows).ok());
+}
+
// Scenario: PaimonReader must clear the previous split schema id before
reading a new split. A
// schema-evolved split must not force the following split without schema id
to keep BY_FIELD_ID.
TEST(PaimonReaderTest, ResetsSplitSchemaIdBeforePreparingNextSplit) {
diff --git a/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
b/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
index 9b1704a7891..c51bbcc9f82 100644
--- a/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
+++ b/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
@@ -101,6 +101,23 @@ services:
- /usr/lib/iceberg-rest/iceberg-rest-adapter.jar:/opt/jdbc/postgresql.jar
- org.apache.iceberg.rest.RESTCatalogServer
+ trino:
+ image: trinodb/trino:482
+ container_name: doris--iceberg-trino
+ depends_on:
+ rest:
+ condition: service_started
+ mc:
+ condition: service_completed_successfully
+ user: root
+ networks:
+ - doris--iceberg
+ healthcheck:
+ test: ["CMD-SHELL", "test -d /etc/trino/catalog && command -v trino
>/dev/null"]
+ interval: 5s
+ timeout: 60s
+ retries: 120
+
minio:
image: minio/minio:RELEASE.2025-01-20T14-49-07Z
container_name: doris--iceberg-minio
diff --git
a/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run28.sql
b/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run28.sql
index adbfafcf976..31d75a8d50b 100644
---
a/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run28.sql
+++
b/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run28.sql
@@ -34,6 +34,56 @@ where batch = 2 and id >= 14;
delete from dv_test
where id % 2 = 1;
+drop table if exists dv_test_orc;
+create table dv_test_orc (
+ id int,
+ batch int,
+ data string
+)
+using iceberg
+tblproperties (
+ 'format-version' = '3',
+ 'write.format.default' = 'orc',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read'
+);
+
+insert into dv_test_orc values
+ (1, 1, 'a'), (2, 1, 'b'), (3, 1, 'c'), (4, 1, 'd'),
+ (5, 1, 'e'), (6, 1, 'f'), (7, 1, 'g'), (8, 1, 'h');
+
+delete from dv_test_orc
+where batch = 1 and id in (3, 4, 5);
+
+insert into dv_test_orc values
+ (9, 2, 'i'), (10, 2, 'j'), (11, 2, 'k'), (12, 2, 'l'),
+ (13, 2, 'm'), (14, 2, 'n'), (15, 2, 'o'), (16, 2, 'p');
+
+delete from dv_test_orc
+where batch = 2 and id >= 14;
+
+delete from dv_test_orc
+where id % 2 = 1;
+
+drop table if exists dv_test_no_delete;
+create table dv_test_no_delete (
+ id int,
+ batch int,
+ data string
+)
+using iceberg
+tblproperties (
+ 'format-version' = '3',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read'
+);
+
+insert into dv_test_no_delete values
+ (1, 1, 'a'), (2, 1, 'b'), (3, 1, 'c'), (4, 1, 'd'),
+ (5, 1, 'e'), (6, 1, 'f'), (7, 1, 'g'), (8, 1, 'h');
+
drop table if exists dv_test_v2;
create table dv_test_v2 (
id int,
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_deletion_vector.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_deletion_vector.out
index 41819b90bd7..6d3531f345e 100644
---
a/regression-test/data/external_table_p0/iceberg/test_iceberg_deletion_vector.out
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_deletion_vector.out
@@ -11,40 +11,97 @@
6 1 f
8 1 h
+-- !q3 --
+2 1 b
+6 1 f
+8 1 h
+10 2 j
+12 2 l
+
+-- !no_delete --
+1 1 a
+2 1 b
+3 1 c
+4 1 d
+5 1 e
+6 1 f
+7 1 g
+8 1 h
+
+-- !equality_only --
+1 a
+2 b
+
+-- !position_only_count --
+5632
+
+-- !position_and_dv_count --
+5120
+
+-- !equality_and_dv --
+2 1 b
+4 1 d
+6 1 f
+8 1 h
+12 2 l
+
+-- !split_cache_single_file_count --
+2048 4192256
+
-- !q4 --
2 1 b
-- !q5 --
2 1 b
+-- !q5_orc --
+2 1 b
+
-- !q6 --
-- !q7 --
+-- !q7_orc --
+
-- !q8 --
2 1 b
-- !q9 --
2 1 b
+-- !q9_orc --
+2 1 b
+
-- !q10 --
5
-- !q11 --
3
+-- !q11_orc --
+5
+
+-- !no_delete_count --
+8
+
-- !q12 --
5
-- !q13 --
3
+-- !q13_orc --
+5
+
-- !q14 --
5
-- !q15 --
3
+-- !q15_orc --
+5
+
-- !q16 --
33334
@@ -63,3 +120,65 @@
-- !q21 --
33334
+-- !q22 --
+12 2 l
+10 2 j
+8 1 h
+
+-- !q23 --
+8 1 h
+6 1 f
+2 1 b
+
+-- !q24 --
+1 3 16
+2 2 22
+
+-- !q25 --
+1 3 16
+
+-- !q26 --
+1 3 16
+2 2 22
+
+-- !q27 --
+2 b
+6 f
+
+-- !q28 --
+2 b
+6 f
+
+-- !q29 --
+2 b
+6 f
+
+-- !q30 --
+100
+
+-- !q31 --
+100
+
+-- !delete_type_matrix --
+dv_equality 0 PARQUET 12 12
+dv_equality 1 PUFFIN 6 6
+dv_equality 2 PARQUET 1 1
+dv_only 0 PARQUET 10 10
+dv_only 1 PUFFIN 5 5
+dv_position 0 PARQUET 1 7680
+dv_position 1 PUFFIN 1 2560
+equality_only 0 PARQUET 2 2
+equality_only 2 PARQUET 2 2
+no_delete 0 PARQUET 8 8
+position_only 0 PARQUET 1 7680
+position_only 1 PARQUET 4 2048
+
+-- !position_delete_ignored_with_dv --
+
+-- !equality_delete_active_with_dv --
+2 1 b
+4 1 d
+6 1 f
+8 1 h
+12 2 l
+
diff --git
a/regression-test/data/external_table_p0/paimon/test_paimon_deletion_vector.out
b/regression-test/data/external_table_p0/paimon/test_paimon_deletion_vector.out
index f0b1e92a088..d8b9e547101 100644
---
a/regression-test/data/external_table_p0/paimon/test_paimon_deletion_vector.out
+++
b/regression-test/data/external_table_p0/paimon/test_paimon_deletion_vector.out
@@ -35,6 +35,18 @@
-- !9 --
7
+-- !10 --
+3 3_1
+
+-- !11 --
+3 3_1
+
+-- !12 --
+4
+
+-- !13 --
+4
+
-- !1 --
3
@@ -71,3 +83,15 @@
-- !9 --
7
+-- !10 --
+3 3_1
+
+-- !11 --
+3 3_1
+
+-- !12 --
+4
+
+-- !13 --
+4
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_deletion_vector.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_deletion_vector.groovy
index cdd1764c794..63fcf1cfda0 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_deletion_vector.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_deletion_vector.groovy
@@ -42,30 +42,329 @@ suite("test_iceberg_deletion_vector", "p0,external") {
sql """switch ${catalog_name};"""
sql """ use format_v3;"""
+ sql """ set file_split_size=1;"""
+ try {
+ def executeCommandWithStatus = { String cmd, int timeoutSeconds = 300,
Boolean logFailure = true,
+ Boolean logCommand = true ->
+ StringBuilder stdout = new StringBuilder()
+ StringBuilder stderr = new StringBuilder()
+ try {
+ if (logCommand) {
+ logger.info("execute ${cmd}")
+ }
+ def proc = new ProcessBuilder("/bin/bash", "-c", cmd).start()
+ proc.consumeProcessOutput(stdout, stderr)
+ proc.waitForOrKill(timeoutSeconds * 1000)
+ int exitcode = proc.exitValue()
+ String output = stdout.toString()
+ String error = stderr.toString()
+ if (exitcode != 0 && logFailure) {
+ logger.info("exit code: ${exitcode}, stdout\n:
${output}\nstderr\n: ${error}")
+ }
+ return [exitCode: exitcode, stdout: output, stderr: error]
+ } catch (IOException e) {
+ assertTrue(false, "Execute failed: ${cmd}, err: ${e.message}")
+ }
+ }
+
+ def executeCommand = { String cmd, Boolean mustSuc, int timeoutSeconds =
300 ->
+ def result = executeCommandWithStatus(cmd, timeoutSeconds)
+ if (mustSuc && result.exitCode != 0) {
+ assertTrue(false,
+ "Execute failed:
${cmd}\nstdout:\n${result.stdout}\nstderr:\n${result.stderr}")
+ }
+ return result.stdout
+ }
+
+ String dockerCommand =
context.config.otherConfigs.get("externalDockerCommand") ?: "docker"
+ def listDockerContainers = {
+ String containers =
+ executeCommand("${dockerCommand} ps --format
'{{.ID}}\t{{.Names}}\t{{.Image}}'", true, 30) ?:
+ ""
+ return containers.readLines().collect { it.trim() }.findAll {
!it.isEmpty() }
+ }
+
+ def findRequiredDockerContainer = { String role, String configKey, String
probeCommand ->
+ String configuredContainer = context.config.otherConfigs.get(configKey)
+ if (configuredContainer != null && !configuredContainer.isEmpty()) {
+ def probe = executeCommandWithStatus(
+ "${dockerCommand} exec ${configuredContainer} bash -lc
'${probeCommand}'",
+ 30)
+ assertEquals(0, probe.exitCode,
+ "${role} container configured by
${configKey}=${configuredContainer} is not usable")
+ return configuredContainer
+ }
+
+ def matchedContainers = []
+ listDockerContainers().each { String containerLine ->
+ def fields = containerLine.split(/\t/, 3)
+ assertTrue(fields.length >= 2, "Unexpected docker ps output:
${containerLine}")
+ String containerId = fields[0].trim()
+ String containerName = fields[1].trim()
+ String containerImage = fields.length >= 3 ? fields[2].trim() : ""
+ def probe = executeCommandWithStatus(
+ "${dockerCommand} exec ${containerId} bash -lc
'${probeCommand}'",
+ 30,
+ false,
+ false)
+ if (probe.exitCode == 0) {
+ matchedContainers.add(
+ [id: containerId, name: containerName, image:
containerImage])
+ }
+ }
+
+ assertFalse(matchedContainers.isEmpty(),
+ "No usable ${role} container found. Set ${configKey} to the
exact container name " +
+ "or start the required external environment.")
+ assertEquals(1, matchedContainers.size(),
+ "Multiple usable ${role} containers found:
${matchedContainers}. " +
+ "Set ${configKey} to the exact container name.")
+ logger.info("use ${role} container ${matchedContainers[0].name} " +
+ "(${matchedContainers[0].image})")
+ return matchedContainers[0].id
+ }
+
+ String sparkContainerName = findRequiredDockerContainer(
+ "Spark Iceberg", "icebergSparkContainer",
+ "command -v spark-sql >/dev/null && test -f /mnt/SUCCESS && " +
+ "spark-sql -e \"SHOW NAMESPACES IN demo\" >/dev/null")
+
+ def encodeBase64 = { String text ->
+ return text.getBytes("UTF-8").encodeBase64().toString()
+ }
+
+ def runInSparkContainer = { String command, int timeout = 300 ->
+ executeCommand("${dockerCommand} exec ${sparkContainerName} bash -lc
'${command}'", true,
+ timeout)
+ }
+
+ def runSparkSql = { String sqlText, int timeout = 600 ->
+ String encodedSql = encodeBase64(sqlText)
+ runInSparkContainer(
+ "echo ${encodedSql} | base64 -d
>/tmp/test_iceberg_deletion_vector.sql && " +
+ "spark-sql -f /tmp/test_iceberg_deletion_vector.sql",
+ timeout
+ )
+ }
+
+ String setupSql = """
+use format_v3;
+
+-- Delete-type matrix fixtures: reuse existing v2 equality/position-delete
tables and create
+-- v3 MOR tables that Spark writes with Iceberg deletion vectors.
+drop table if exists dv_delete_matrix_equality_only;
+CALL demo.system.register_table(
+ table => 'format_v3.dv_delete_matrix_equality_only',
+ metadata_file =>
's3a://warehouse/wh/test_db/customer_flink_one/metadata/00002-c38103a7-237b-4350-aa98-05fe2ad14d16.metadata.json'
+);
+
+drop table if exists dv_delete_matrix_position_only;
+CALL demo.system.register_table(
+ table => 'format_v3.dv_delete_matrix_position_only',
+ metadata_file =>
's3a://warehouse/wh/test_db/iceberg_position_parquet/metadata/00005-6ff36319-7a9a-4357-a9fc-965d5f280c9b.metadata.json'
+);
+
+drop table if exists dv_delete_matrix_position_and_dv;
+CALL demo.system.register_table(
+ table => 'format_v3.dv_delete_matrix_position_and_dv',
+ metadata_file =>
's3a://warehouse/wh/test_db/iceberg_position_parquet/metadata/00005-6ff36319-7a9a-4357-a9fc-965d5f280c9b.metadata.json'
+);
+alter table dv_delete_matrix_position_and_dv set tblproperties (
+ 'format-version' = '3',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read'
+);
+delete from dv_delete_matrix_position_and_dv where id = 2;
+
+drop table if exists dv_delete_matrix_equality_and_dv;
+create table dv_delete_matrix_equality_and_dv (
+ id int,
+ batch int,
+ data string
+)
+using iceberg
+tblproperties (
+ 'format-version' = '3',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read'
+);
+insert into dv_delete_matrix_equality_and_dv values
+ (1, 1, 'a'), (2, 1, 'b'), (3, 1, 'c'), (4, 1, 'd'),
+ (5, 1, 'e'), (6, 1, 'f'), (7, 1, 'g'), (8, 1, 'h'),
+ (9, 2, 'i'), (10, 2, 'j'), (11, 2, 'k'), (12, 2, 'l');
+delete from dv_delete_matrix_equality_and_dv where id % 2 = 1;
+
+-- Split/cache fixture: one sufficiently large data file plus one DV. Doris
reads it with a tiny
+-- file_split_size so the same data file is split across scanners and the
shared scan-node cache
+-- must prevent the DV from being parsed once per split.
+drop table if exists dv_split_cache_single_file;
+create table dv_split_cache_single_file (
+ id bigint,
+ grp int,
+ payload string
+)
+using iceberg
+tblproperties (
+ 'format-version' = '3',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read',
+ 'write.parquet.row-group-size-bytes' = '1024'
+);
+set spark.sql.shuffle.partitions = 1;
+insert into dv_split_cache_single_file
+select /*+ COALESCE(1) */
+ id,
+ cast(id % 8 as int) as grp,
+ repeat(concat('payload_', cast(id as string)), 20) as payload
+ from range(0, 4096);
+delete from dv_split_cache_single_file where id % 2 = 1;
+reset spark.sql.shuffle.partitions;
+"""
+ runSparkSql(setupSql)
+
+ String javaSource = '''
+import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+
+public class AppendEqualityDelete {
+ public static void main(String[] args) throws Exception {
+ Catalog catalog = IcebergRestCatalog.load();
+ Table table = catalog.loadTable(TableIdentifier.of(args[0], args[1]));
+ Schema eqSchema = table.schema().select(args[2]);
+ int fieldId = table.schema().findField(args[2]).fieldId();
+ String location = table.location() + "/data/equality-delete-" +
args[2] + "-"
+ + args[3] + "-" + System.currentTimeMillis() + ".parquet";
+ OutputFile outputFile = table.io().newOutputFile(location);
+ EqualityDeleteWriter<Record> writer = Parquet.writeDeletes(outputFile)
+ .forTable(table)
+ .rowSchema(eqSchema)
+ .withSpec(table.spec())
+ .createWriterFunc(GenericParquetWriter::create)
+ .equalityFieldIds(fieldId)
+ .overwrite()
+ .buildEqualityWriter();
+ GenericRecord record = GenericRecord.create(eqSchema);
+ record.setField(args[2], Integer.valueOf(args[3]));
+ writer.write(record);
+ writer.close();
+ DeleteFile deleteFile = writer.toDeleteFile();
+ table.newRowDelta().addDeletes(deleteFile).commit();
+ }
+}
+class ReadIcebergRows {
+ public static void main(String[] args) throws Exception {
+ Catalog catalog = IcebergRestCatalog.load();
+ Table table = catalog.loadTable(TableIdentifier.of(args[0], args[1]));
+ String[] columns = Arrays.copyOfRange(args, 2, args.length);
+ List<String> rows = new ArrayList<>();
+ try (CloseableIterable<Record> records =
IcebergGenerics.read(table).select(columns).build()) {
+ for (Record record : records) {
+ List<String> values = new ArrayList<>();
+ for (String column : columns) {
+ values.add(String.valueOf(record.getField(column)));
+ }
+ rows.add(String.join("\\t", values));
+ }
+ }
+ Collections.sort(rows, Comparator.comparingInt(row ->
Integer.parseInt(row.split("\\t")[0])));
+ for (String row : rows) {
+ System.out.println(row);
+ }
+ }
+}
+
+class IcebergRestCatalog {
+ static Catalog load() {
+ Map<String, String> props = new HashMap<>();
+ props.put("type", "rest");
+ props.put("uri", "http://rest:8181";);
+ props.put("warehouse", "s3://warehouse/wh/");
+ props.put("io-impl", "org.apache.iceberg.aws.s3.S3FileIO");
+ props.put("s3.endpoint", "http://minio:9000";);
+ props.put("s3.path-style-access", "true");
+ props.put("s3.region", "us-east-1");
+ return CatalogUtil.buildIcebergCatalog("demo", props, null);
+ }
+}
+'''
+ String encodedJavaSource = encodeBase64(javaSource)
+ runInSparkContainer(
+ "echo ${encodedJavaSource} | base64 -d
>/tmp/AppendEqualityDelete.java && " +
+ "javac -cp \"/opt/spark/jars/*\"
/tmp/AppendEqualityDelete.java && " +
+ "java -cp \"/tmp:/opt/spark/jars/*\" AppendEqualityDelete
" +
+ "format_v3 dv_delete_matrix_equality_and_dv id 10",
+ 300
+ )
+
+
+ // Baseline DV correctness across Parquet, ORC, no-delete, and v2-to-v3
upgrade fixtures.
qt_q1 """ SELECT * FROM dv_test ORDER BY id; """
qt_q2 """ SELECT * FROM dv_test_v2 ORDER BY id; """
+ qt_q3 """ SELECT * FROM dv_test_orc ORDER BY id; """
+ qt_no_delete """ SELECT * FROM dv_test_no_delete ORDER BY id; """
+
+ // Delete-type matrix checks cover equality-only, position-only, DV-only,
DV+position,
+ // and DV+equality combinations.
+ qt_equality_only """ SELECT * FROM dv_delete_matrix_equality_only ORDER BY
id; """
+ qt_position_only_count """ SELECT count(*) FROM
dv_delete_matrix_position_only; """
+ qt_position_and_dv_count """ SELECT count(*) FROM
dv_delete_matrix_position_and_dv; """
+ qt_equality_and_dv """ SELECT * FROM dv_delete_matrix_equality_and_dv
ORDER BY id; """
+ qt_split_cache_single_file_count """
+ SELECT count(*), sum(id)
+ FROM dv_split_cache_single_file;
+ """
qt_q4 """ SELECT * FROM dv_test where id = 2 ORDER BY id ; """
qt_q5 """ SELECT * FROM dv_test_v2 where id = 2 ORDER BY id ; """
+ qt_q5_orc """ SELECT * FROM dv_test_orc where id = 2 ORDER BY id; """
qt_q6 """ SELECT * FROM dv_test where id %2 =1 ORDER BY id; """
qt_q7 """ SELECT * FROM dv_test_v2 where id %2 =1 ORDER BY id; """
+ qt_q7_orc """ SELECT * FROM dv_test_orc where id %2 =1 ORDER BY id; """
qt_q8 """ SELECT * FROM dv_test where data < 'f' ORDER BY id; """
qt_q9 """ SELECT * FROM dv_test_v2 where data < 'f' ORDER BY id; """
+ qt_q9_orc """ SELECT * FROM dv_test_orc where data < 'f' ORDER BY id; """
qt_q10 """ SELECT count(*) FROM dv_test ; """
qt_q11 """ SELECT count(*) FROM dv_test_v2 ; """
+ qt_q11_orc """ SELECT count(*) FROM dv_test_orc; """
+ qt_no_delete_count """ SELECT count(*) FROM dv_test_no_delete; """
qt_q12 """ SELECT count(id) FROM dv_test ; """
qt_q13 """ SELECT count(id) FROM dv_test_v2 ; """
+ qt_q13_orc """ SELECT count(id) FROM dv_test_orc; """
qt_q14 """ SELECT count(batch) FROM dv_test ; """
qt_q15 """ SELECT count(batch) FROM dv_test_v2 ; """
+ qt_q15_orc """ SELECT count(batch) FROM dv_test_orc; """
qt_q16 """ SELECT count(*) FROM dv_test_1w ; """
@@ -74,9 +373,220 @@ suite("test_iceberg_deletion_vector", "p0,external") {
qt_q19 """ SELECT count(value) FROM dv_test_1w where id%2 = 1; """
qt_q20 """ SELECT count(id) FROM dv_test_1w where id%3 = 1; """
qt_q21 """ SELECT count(ts) FROM dv_test_1w where id%3 != 1; """
+ qt_q22 """ SELECT * FROM dv_test ORDER BY data DESC LIMIT 3; """
+ qt_q23 """ SELECT * FROM dv_test_v2 ORDER BY data DESC LIMIT 3; """
+ qt_q24 """ SELECT batch, count(*), sum(id) FROM dv_test GROUP BY batch
ORDER BY batch; """
+ qt_q25 """ SELECT batch, count(*), sum(id) FROM dv_test_v2 GROUP BY batch
ORDER BY batch; """
+ qt_q26 """ SELECT batch, count(*), sum(id) FROM dv_test_orc GROUP BY batch
ORDER BY batch; """
+ qt_q27 """ SELECT id, data FROM dv_test WHERE id BETWEEN 2 AND 10 ORDER BY
id LIMIT 2; """
+ qt_q28 """ SELECT id, data FROM dv_test_v2 WHERE id BETWEEN 2 AND 10 ORDER
BY id LIMIT 2; """
+ qt_q29 """ SELECT id, data FROM dv_test_orc WHERE id BETWEEN 2 AND 10
ORDER BY id LIMIT 2; """
+ qt_q30 """ SELECT count(*) FROM (SELECT * FROM dv_test_1w WHERE id % 7 = 0
ORDER BY id LIMIT 100) t; """
+ qt_q31 """ SELECT count(*) FROM (SELECT * FROM dv_test_1w WHERE id % 11 =
0 ORDER BY id LIMIT 100) t; """
+
+ // Metadata matrix verifies the physical delete-file content types that
back the logical
+ // scenario checks above.
+ qt_delete_type_matrix """
+ SELECT 'dv_equality' scenario, content, file_format, count(*) files,
sum(record_count) records
+ FROM `dv_delete_matrix_equality_and_dv\$files`
+ GROUP BY content, file_format
+ UNION ALL
+ SELECT 'dv_only' scenario, content, file_format, count(*) files,
sum(record_count) records
+ FROM `dv_test\$files`
+ GROUP BY content, file_format
+ UNION ALL
+ SELECT 'dv_position' scenario, content, file_format, count(*) files,
sum(record_count) records
+ FROM `dv_delete_matrix_position_and_dv\$files`
+ GROUP BY content, file_format
+ UNION ALL
+ SELECT 'equality_only' scenario, content, file_format, count(*) files,
sum(record_count) records
+ FROM `dv_delete_matrix_equality_only\$files`
+ GROUP BY content, file_format
+ UNION ALL
+ SELECT 'no_delete' scenario, content, file_format, count(*) files,
sum(record_count) records
+ FROM `dv_test_no_delete\$files`
+ GROUP BY content, file_format
+ UNION ALL
+ SELECT 'position_only' scenario, content, file_format, count(*) files,
sum(record_count) records
+ FROM `dv_delete_matrix_position_only\$files`
+ GROUP BY content, file_format
+ ORDER BY scenario, content, file_format;
+ """
+ // Iceberg v3 semantics: when a data file has a DV, old position deletes
for that file must be
+ // ignored, while equality deletes still filter matching rows.
+ qt_position_delete_ignored_with_dv """
+ SELECT id, name
+ FROM dv_delete_matrix_position_and_dv
+ WHERE id IN (1, 2, 5, 10, 15)
+ ORDER BY id, name
+ LIMIT 20;
+ """
+ qt_equality_delete_active_with_dv """
+ SELECT id, batch, data
+ FROM dv_delete_matrix_equality_and_dv
+ ORDER BY id;
+ """
+
+ def normalizeExternalRows = { String output ->
+ return output.readLines()
+ .collect { it.trim() }
+ .findAll { it ==~ /^-?[0-9].*/ && it.contains("\t") }
+ .join("\n")
+ }
+ String expectedRows = ["2\t1\tb", "4\t1\td", "6\t1\tf", "8\t1\th",
"12\t2\tl"].join("\n")
+ String sparkRows = normalizeExternalRows(runSparkSql(
+ "use format_v3; select id, batch, data from
dv_delete_matrix_equality_and_dv order by id;",
+ 300
+ ))
+ assertEquals(expectedRows, sparkRows)
+ String javaRows = normalizeExternalRows(runInSparkContainer(
+ "java -cp \"/tmp:/opt/spark/jars/*\" ReadIcebergRows " +
+ "format_v3 dv_delete_matrix_equality_and_dv id batch data",
+ 300
+ ))
+ assertEquals(expectedRows, javaRows)
+
+ String trinoContainerName = findRequiredDockerContainer(
+ "Trino", "icebergTrinoContainer",
+ "test -d /etc/trino/catalog && command -v trino >/dev/null")
+ String trinoExternalEnvIp = externalEnvIp
+ if (trinoExternalEnvIp == "127.0.0.1" ||
trinoExternalEnvIp.equalsIgnoreCase("localhost")) {
+ trinoExternalEnvIp = executeCommand("hostname -I | cut -d' ' -f1",
true, 30)?.trim()
+ }
+ String trinoCatalogProps = """
+connector.name=iceberg
+iceberg.catalog.type=rest
+iceberg.rest-catalog.uri=http://${trinoExternalEnvIp}:${rest_port}
+fs.native-s3.enabled=true
+s3.endpoint=http://${trinoExternalEnvIp}:${minio_port}
+s3.aws-access-key=admin
+s3.aws-secret-key=password
+s3.region=us-east-1
+s3.path-style-access=true
+"""
+ String encodedTrinoCatalogProps = encodeBase64(trinoCatalogProps)
+ executeCommand(
+ "${dockerCommand} exec ${trinoContainerName} bash -lc " +
+ "'echo ${encodedTrinoCatalogProps} | " +
+ "base64 -d >/etc/trino/catalog/iceberg.properties'",
+ true,
+ 30
+ )
+ executeCommand("${dockerCommand} restart ${trinoContainerName}", true, 60)
+ String trinoRows = ""
+ for (int i = 0; i < 12; i++) {
+ Thread.sleep(5000)
+ trinoRows = normalizeExternalRows(executeCommand(
+ "${dockerCommand} exec ${trinoContainerName} trino
--output-format TSV " +
+ "--catalog iceberg --schema format_v3 --execute " +
+ "\"SELECT id, batch, data " +
+ "FROM dv_delete_matrix_equality_and_dv ORDER BY id\"",
+ false,
+ 120
+ ))
+ if (!trinoRows.isEmpty()) {
+ break
+ }
+ }
+ assertEquals(expectedRows, trinoRows)
+ def profileCounterValues = { String profileText, String counterName ->
+ def values = []
+ def matcher = profileText =~ ("(?m)^\\s*(?:-\\s*)?" +
+ java.util.regex.Pattern.quote(counterName) + ":\\s+([^\\n]+)")
+ while (matcher.find()) {
+ String valueText = matcher.group(1).toString()
+ def exact = valueText =~ /\(([0-9,]+)\)/
+ String rawValue
+ if (exact.find()) {
+ rawValue = exact.group(1)
+ } else {
+ def number = valueText =~ /([0-9,]+)/
+ rawValue = number.find() ? number.group(1) : null
+ }
+ if (rawValue != null) {
+ values.add(Long.parseLong(rawValue.replace(",", "")))
+ }
+ }
+ return values
+ }
+
+ def profileInfoValueCount = { String profileText, String infoName ->
+ def matcher = profileText =~ (java.util.regex.Pattern.quote(infoName)
+ ":\\s*\\[([^\\]]*)\\]")
+ if (!matcher.find()) {
+ return 0
+ }
+ return matcher.group(1).split(",").collect { it.trim() }.findAll {
!it.isEmpty() }.size()
+ }
+
+ // Split/cache scenario for 3.4: this query scans one data file with
file_split_size=1. The
+ // result is captured by qt_split_cache_single_file_count; the profile
assertion checks the DV
+ // rows were materialized once for the shared scan-node cache, not once
per split.
+ String splitCacheProfileTag = "iceberg_dv_split_cache_" +
UUID.randomUUID().toString()
+ profile(splitCacheProfileTag) {
+ run {
+ sql """
+ /* ${splitCacheProfileTag} */
+ SELECT count(*), sum(id)
+ FROM dv_split_cache_single_file;
+ """
+ // The detailed scanner counters are populated asynchronously
after the query returns.
+ Thread.sleep(5000)
+ }
+ check { profileString, exception ->
+ if (exception != null) {
+ throw exception
+ }
+ String mergedProfile = profileString
+ if (profileString.contains("MergedProfile:")) {
+ mergedProfile =
profileString.substring(profileString.indexOf("MergedProfile:"))
+ int mergedProfileEnd = mergedProfile.length()
+ ["DetailProfile(", "Execution Profile:", "Appendix:"].each {
String sectionName ->
+ int sectionIndex = mergedProfile.indexOf(sectionName)
+ if (sectionIndex > 0) {
+ mergedProfileEnd = Math.min(mergedProfileEnd,
sectionIndex)
+ }
+ }
+ mergedProfile = mergedProfile.substring(0, mergedProfileEnd)
+ }
+ def numDeleteRowsValues = profileCounterValues(mergedProfile,
"NumDeleteRows")
+ numDeleteRowsValues = numDeleteRowsValues.findAll { it > 0L }
+ int scannerCount = profileInfoValueCount(profileString,
"PerScannerRowsRead")
+ assertFalse(numDeleteRowsValues.isEmpty(),
+ "Expected NumDeleteRows counter for split DV scan,
profile: ${profileString}")
+ assertEquals(2048L, numDeleteRowsValues.sum(0L),
+ "Expected DV rows to be materialized once across split
scanners, " +
+ "NumDeleteRows counters: ${numDeleteRowsValues},
profile: ${profileString}")
+ assertTrue(scannerCount > 1,
+ "Expected multiple scanner entries for split DV scan,
profile: ${profileString}")
+ }
+ }
+
+ // Cross-reader comparison protects against Doris accepting a
DV/equality-delete result that
+ // diverges from Spark, Trino, or Iceberg's Java reader.
+ explain {
+ sql("SELECT count(*) FROM dv_test;")
+ verbose(true)
+ contains "deleteFileNum"
+ }
+
+ explain {
+ sql("SELECT count(*) FROM dv_test_orc;")
+ verbose(true)
+ contains "deleteFileNum"
+ }
+
+ explain {
+ sql("SELECT count(*) FROM dv_test_v2;")
+ verbose(true)
+ contains "deleteFileNum"
+ }
+
+ } finally {
+ sql """ unset variable file_split_size;"""
+ }
}
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_deletion_vector.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_deletion_vector.groovy
index 044166c3c5d..172df875927 100644
---
a/regression-test/suites/external_table_p0/paimon/test_paimon_deletion_vector.groovy
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_deletion_vector.groovy
@@ -38,6 +38,9 @@ suite("test_paimon_deletion_vector", "p0,external") {
def test_cases = { String force ->
sql """ set force_jni_scanner=${force} """
+ // Force tiny splits so both native and JNI paths read Paimon DV
tables through multiple
+ // scanner ranges while still returning the same filtered rows.
+ sql """ set file_split_size=1 """
qt_1 """select count(*) from deletion_vector_orc;"""
qt_2 """select count(*) from deletion_vector_parquet;"""
qt_3 """select count(*) from deletion_vector_orc where id > 2;"""
@@ -47,10 +50,16 @@ suite("test_paimon_deletion_vector", "p0,external") {
qt_7 """select * from deletion_vector_table_1_0 order by id;"""
qt_8 """select count(*) from deletion_vector_table_1_0;"""
qt_9 """select count(*) from deletion_vector_table_1_0 where id >
2;"""
+ qt_10 """select * from deletion_vector_orc where id > 2 order by
id limit 1;"""
+ qt_11 """select * from deletion_vector_parquet where id > 2 order
by id limit 1;"""
+ qt_12 """select count(*) from (select * from deletion_vector_orc
where id > 2 union all select * from deletion_vector_orc where id > 2) t;"""
+ qt_13 """select count(*) from (select * from
deletion_vector_parquet where id > 2 union all select * from
deletion_vector_parquet where id > 2) t;"""
}
def test_table_count_push_down = { String force ->
sql """ set force_jni_scanner=${force} """
+ // DV tables cannot use table-level count pushdown because deleted
rows must be applied;
+ // the no-DV v1.0 table keeps the positive pushdown baseline.
explain {
sql("select count(*) from deletion_vector_orc;")
contains "pushdown agg=COUNT (-1)"
@@ -68,6 +77,8 @@ suite("test_paimon_deletion_vector", "p0,external") {
def test_not_table_count_push_down = { String force ->
sql """ set enable_count_push_down_for_external_table=false; """
sql """ set force_jni_scanner=${force} """
+ // With count pushdown disabled, all DV and non-DV tables must
report NONE regardless
+ // of the selected scanner path.
explain {
sql("select count(*) from deletion_vector_orc;")
contains "pushdown agg=NONE"
@@ -90,7 +101,8 @@ suite("test_paimon_deletion_vector", "p0,external") {
test_not_table_count_push_down("true")
} finally {
sql """ set enable_count_push_down_for_external_table=true; """
+ sql """unset variable file_split_size"""
sql """set force_jni_scanner=false"""
}
-}
\ No newline at end of file
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
