Skip to content

[feature](iceberg) Support Iceberg position deletes system table #65135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hubgeter wants to merge 16 commits into
base: master
Choose a base branch
from
Open

[feature](iceberg) Support Iceberg position deletes system table #65135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
ba7ff24
[feature](iceberg) Support Iceberg position deletes system table
hubgeter Jul 1, 2026
2c42b3e
[fix](iceberg) Preserve typed partition values for position deletes
hubgeter Jul 2, 2026
2bfd638
[fix](iceberg) Address position deletes system table review comments
hubgeter Jul 3, 2026
80e4825
[fix](iceberg) Support position deletes with file scanner v2
hubgeter Jul 3, 2026
9585dca
[fix](iceberg) Decouple position delete v2 reader from v1
hubgeter Jul 4, 2026
acf68f1
[fix](iceberg) Remove unused JNI scanner changes
hubgeter Jul 6, 2026
44e041c
[fix](iceberg) Move position delete reader selection into format init
hubgeter Jul 6, 2026
72c7ed8
Revert "[fix](iceberg) Move position delete reader selection into for…
hubgeter Jul 6, 2026
6122b73
[fix](iceberg) Stream position delete system table reads
hubgeter Jul 6, 2026
5967b06
[fix](iceberg) Read position delete rows by field id
hubgeter Jul 6, 2026
c841108
[fix](iceberg) Fix position_deletes reader build and refine delete fi…
hubgeter Jul 7, 2026
a997282
[test](be) Cover Iceberg position delete scanner v2 fallback
hubgeter Jul 8, 2026
2120186
fix build
hubgeter Jul 9, 2026
2e6bc0a
[fix](iceberg) Match evolved partition fields by ID
hubgeter Jul 12, 2026
3f9bc30
[fix](iceberg) Resolve scanner v2 conflict for ORC position deletes
hubgeter Jul 12, 2026
40b8d7a
[fix](iceberg) Guard position deletes during smooth upgrade
hubgeter Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions be/src/exec/scan/file_scanner.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@
#include "format/table/hive_reader.h"
#include "format/table/hudi_jni_reader.h"
#include "format/table/hudi_reader.h"
#include "format/table/iceberg_position_delete_sys_table_reader.h"
#include "format/table/iceberg_reader.h"
#include "format/table/iceberg_sys_table_jni_reader.h"
#include "format/table/jdbc_jni_reader.h"
Expand Down Expand Up @@ -98,6 +99,20 @@ class ShardedKVCache;
namespace doris {
using namespace ErrorCode;

namespace {
constexpr int kIcebergPositionDeleteContent = 1;
constexpr int kIcebergDeletionVectorContent = 3;

bool is_iceberg_position_deletes_sys_table(const TFileRangeDesc& range) {
return range.__isset.table_format_params &&
range.table_format_params.table_format_type == "iceberg" &&
range.table_format_params.__isset.iceberg_params &&
range.table_format_params.iceberg_params.__isset.content &&
(range.table_format_params.iceberg_params.content == kIcebergPositionDeleteContent ||
range.table_format_params.iceberg_params.content == kIcebergDeletionVectorContent);
}
} // namespace

const std::string FileScanner::FileReadBytesProfile = "FileReadBytes";
const std::string FileScanner::FileReadTimeProfile = "FileReadTime";

Expand Down Expand Up @@ -1038,6 +1053,7 @@ Status FileScanner::_get_next_reader() {
// create reader for specific format
Status init_status = Status::OK();
TFileFormatType::type format_type = _get_current_format_type();
const bool is_position_deletes_sys_table = is_iceberg_position_deletes_sys_table(range);
// for compatibility, this logic is deprecated in 3.1
if (format_type == TFileFormatType::FORMAT_JNI && range.__isset.table_format_params) {
if (range.table_format_params.table_format_type == "paimon" &&
Expand Down Expand Up @@ -1163,6 +1179,16 @@ Status FileScanner::_get_next_reader() {
auto file_meta_cache_ptr = _should_enable_file_meta_cache()
? ExecEnv::GetInstance()->file_meta_cache()
: nullptr;
if (is_position_deletes_sys_table) {
ReaderInitContext ctx;
_fill_base_init_context(&ctx);
auto reader = IcebergPositionDeleteSysTableReader::create_unique(
_file_slot_descs, _state, _profile, range, _params, file_meta_cache_ptr);
init_status = static_cast<GenericReader*>(reader.get())->init_reader(&ctx);
_cur_reader = std::move(reader);
need_to_get_parsed_schema = false;
break;
}
if (push_down_predicates) {
RETURN_IF_ERROR(_process_late_arrival_conjuncts());
}
Expand All @@ -1175,6 +1201,16 @@ Status FileScanner::_get_next_reader() {
auto file_meta_cache_ptr = _should_enable_file_meta_cache()
? ExecEnv::GetInstance()->file_meta_cache()
: nullptr;
if (is_position_deletes_sys_table) {
ReaderInitContext ctx;
_fill_base_init_context(&ctx);
auto reader = IcebergPositionDeleteSysTableReader::create_unique(
_file_slot_descs, _state, _profile, range, _params, file_meta_cache_ptr);
init_status = static_cast<GenericReader*>(reader.get())->init_reader(&ctx);
_cur_reader = std::move(reader);
need_to_get_parsed_schema = false;
break;
}
if (push_down_predicates) {
RETURN_IF_ERROR(_process_late_arrival_conjuncts());
}
Expand Down
17 changes: 16 additions & 1 deletion be/src/exec/scan/file_scanner_v2.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
#include "format_v2/jni/trino_connector_jni_reader.h"
#include "format_v2/table/hive_reader.h"
#include "format_v2/table/hudi_reader.h"
#include "format_v2/table/iceberg_position_delete_sys_table_reader.h"
#include "format_v2/table/iceberg_reader.h"
#include "format_v2/table/paimon_reader.h"
#include "format_v2/table/remote_doris_reader.h"
Expand All @@ -70,6 +71,9 @@
namespace doris {
namespace {

constexpr int kIcebergPositionDeleteContent = 1;
constexpr int kIcebergDeletionVectorContent = 3;

std::string table_format_name(const TFileRangeDesc& range) {
return range.__isset.table_format_params ? range.table_format_params.table_format_type
: "NotSet";
Expand Down Expand Up @@ -110,6 +114,15 @@ bool is_supported_jni_table_format(const TFileRangeDesc& range) {
table_format == "max_compute" || table_format == "trino_connector";
}

bool is_iceberg_position_deletes_sys_table(const TFileRangeDesc& range) {
return range.__isset.table_format_params &&
range.table_format_params.table_format_type == "iceberg" &&
range.table_format_params.__isset.iceberg_params &&
range.table_format_params.iceberg_params.__isset.content &&
(range.table_format_params.iceberg_params.content == kIcebergPositionDeleteContent ||
range.table_format_params.iceberg_params.content == kIcebergDeletionVectorContent);
}

bool is_csv_format(TFileFormatType::type format_type) {
switch (format_type) {
case TFileFormatType::FORMAT_CSV_PLAIN:
Expand Down Expand Up @@ -462,7 +475,9 @@ Status FileScannerV2::_create_table_reader_for_format(
} else if (table_format == "hive") {
*reader = format::hive::HiveReader::create_unique();
} else if (table_format == "iceberg") {
if (get_range_format_type(*_params, range) == TFileFormatType::FORMAT_JNI) {
if (is_iceberg_position_deletes_sys_table(range)) {
*reader = std::make_unique<format::iceberg::IcebergPositionDeleteSysTableV2Reader>();
} else if (get_range_format_type(*_params, range) == TFileFormatType::FORMAT_JNI) {
*reader = std::make_unique<format::iceberg::IcebergSysTableJniReader>();
} else {
*reader = std::make_unique<format::iceberg::IcebergTableReader>();
Expand Down
Loading
Loading