Skip to content

[fix](be) Correct FileScannerV2 Iceberg delete handling #65502

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
Gabriel39 wants to merge 8 commits into
base: master
Choose a base branch
from
Open

[fix](be) Correct FileScannerV2 Iceberg delete handling #65502

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
0cbee24
[fix](be) Correct FileScannerV2 Iceberg delete handling
Gabriel39 Jul 12, 2026
74dd136
[fix](be) Harden Iceberg V2 delete handling
Gabriel39 Jul 12, 2026
1dc84ad
[fix](be) Harden Iceberg equality delete metadata
Gabriel39 Jul 12, 2026
05de225
[fix](regression) Fix session variable result extraction
Gabriel39 Jul 13, 2026
c570f2c
[fix](be) Handle Iceberg equality defaults in both scanners
Gabriel39 Jul 13, 2026
f1dfda5
[chore](be) Format Iceberg reader helper
Gabriel39 Jul 13, 2026
3de65f1
[fix](be) Handle Iceberg equality deletes across schema mappings
Gabriel39 Jul 13, 2026
aaf4bd7
[fix](be) Address Iceberg equality delete edge cases
Gabriel39 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
184 changes: 138 additions & 46 deletions be/src/format/table/iceberg_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,7 +221,8 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
_all_required_col_names = ctx->column_names;

// Create column IDs from field descriptor
auto column_id_result = _create_column_ids(field_desc, ctx->tuple_descriptor);
auto column_id_result =
_create_column_ids(field_desc, ctx->tuple_descriptor, ctx->table_info_node);
ctx->column_ids = std::move(column_id_result.column_ids);
ctx->filter_column_ids = std::move(column_id_result.filter_column_ids);

Expand All @@ -240,11 +241,11 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
// - Prefix with __equality_delete_column__ to avoid name conflicts
// - Correctly map table_col_name → file_col_name in table_info_node
const static std::string EQ_DELETE_PRE = "__equality_delete_column__";
std::unordered_map<int, std::string> field_id_to_file_col_name;
std::unordered_map<int, const FieldSchema*> field_id_to_file_column;
for (int i = 0; i < field_desc->size(); ++i) {
auto field_schema = field_desc->get_column(i);
const auto* field_schema = field_desc->get_column(i);
if (field_schema) {
field_id_to_file_col_name[field_schema->field_id] = field_schema->name;
field_id_to_file_column[field_schema->field_id] = field_schema;
}
}

Expand All @@ -261,12 +262,30 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
}
}

std::string file_col_name = old_name;
auto it = field_id_to_file_col_name.find(field_id);
if (it != field_id_to_file_col_name.end()) {
file_col_name = it->second;
const FieldSchema* file_column = nullptr;
auto id_it = field_id_to_file_column.find(field_id);

@github-actions github-actions Bot Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This still bypasses name-mapping mode for partial-ID files. The normal Parquet and ORC *_field_id_with_name_mapping() helpers switch to name/name_mapping matching once any top-level file field lacks an Iceberg id, but both hidden equality-key loops first accept any physical column whose id equals the delete key id and only consult table_info_node if that lookup misses. For a migrated file where the real key is an id-less alias and another column still carries the stale key id, the equality delete will compare against the unrelated column and miss or misapply deletes. Please prefer the already-resolved table_info_node mapping in name-mapping mode, or skip this id fast path whenever the file is not all-id-mapped, and cover both V1 Parquet and V1 ORC partial-ID/name-mapping files.

if (id_it != field_id_to_file_column.end()) {
file_column = id_it->second;
} else if (const auto struct_node =
std::dynamic_pointer_cast<TableSchemaChangeHelper::StructNode>(
ctx->table_info_node);
struct_node != nullptr && struct_node->get_children().contains(old_name) &&
struct_node->children_column_exists(old_name)) {
// Iceberg files written without field ids must use schema.name-mapping.default. The
// full-schema table_info_node already resolved the current key name (old_name) to its
// physical alias; reuse that result before deciding that the key is absent.
const auto& mapped_name = struct_node->children_file_column_name(old_name);
for (int j = 0; j < field_desc->size(); ++j) {
const auto* candidate = field_desc->get_column(j);
if (candidate != nullptr && candidate->name == mapped_name) {
file_column = candidate;
break;
}
}
DORIS_CHECK(file_column != nullptr);
}

const std::string file_col_name = file_column == nullptr ? old_name : file_column->name;
std::string table_col_name = EQ_DELETE_PRE + file_col_name;

// Update _id_to_block_column_name
Expand All @@ -279,18 +298,21 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
_expand_columns[i].name = table_col_name;
}

if (file_column == nullptr) {
DORIS_CHECK(i < _expand_columns.size());
RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id, table_col_name,
_expand_columns[i].type));
// The old data file predates this equality key. Keep it in the expand block so the
// synthesized-column hook can materialize its logical initial default before reader
// filtering, but do not advertise it to Parquet as a physical child.
new_expand_col_names.push_back(table_col_name);
continue;
}

new_expand_col_names.push_back(table_col_name);

// Add column IDs
if (it != field_id_to_file_col_name.end()) {
for (int j = 0; j < field_desc->size(); ++j) {
auto field_schema = field_desc->get_column(j);
if (field_schema && field_schema->field_id == field_id) {
ctx->column_ids.insert(field_schema->get_column_id());
break;
}
}
}
ctx->column_ids.insert(file_column->get_column_id());

// Register in table_info_node: table_col_name → file_col_name
ctx->column_names.push_back(table_col_name);
Expand All @@ -308,8 +330,9 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
// ============================================================================
// IcebergParquetReader: _create_column_ids
// ============================================================================
ColumnIdResult IcebergParquetReader::_create_column_ids(const FieldDescriptor* field_desc,
const TupleDescriptor* tuple_descriptor) {
ColumnIdResult IcebergParquetReader::_create_column_ids(
const FieldDescriptor* field_desc, const TupleDescriptor* tuple_descriptor,
const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node) {
auto* mutable_field_desc = const_cast<FieldDescriptor*>(field_desc);
mutable_field_desc->assign_ids();

Expand All @@ -335,11 +358,32 @@ ColumnIdResult IcebergParquetReader::_create_column_ids(const FieldDescriptor* f
};

for (const auto* slot : tuple_descriptor->slots()) {
auto it = iceberg_id_to_field_schema_map.find(slot->col_unique_id());
if (it == iceberg_id_to_field_schema_map.end()) {
const FieldSchema* field_schema = nullptr;
if (table_info_node != nullptr) {
if (table_info_node->children_column_exists(slot->col_name())) {
// Use the physical child selected by the schema-mapping pass. This keeps partial-id
// files in BY_NAME mode from binding a projected column through an unrelated stale
// field id.
const auto& file_column_name =
table_info_node->children_file_column_name(slot->col_name());
for (int i = 0; i < field_desc->size(); ++i) {
const auto* candidate = field_desc->get_column(i);
if (candidate != nullptr && candidate->name == file_column_name) {
field_schema = candidate;
break;
}
}
DORIS_CHECK(field_schema != nullptr);
}
} else {
auto it = iceberg_id_to_field_schema_map.find(slot->col_unique_id());
if (it != iceberg_id_to_field_schema_map.end()) {
field_schema = it->second;
}
}
if (field_schema == nullptr) {
continue;
}
auto field_schema = it->second;

if ((slot->col_type() != TYPE_STRUCT && slot->col_type() != TYPE_ARRAY &&
slot->col_type() != TYPE_MAP)) {
Expand Down Expand Up @@ -503,7 +547,8 @@ Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
_all_required_col_names = ctx->column_names;

// Create column IDs from ORC type
auto column_id_result = _create_column_ids(orc_type_ptr, ctx->tuple_descriptor);
auto column_id_result =
_create_column_ids(orc_type_ptr, ctx->tuple_descriptor, ctx->table_info_node);
ctx->column_ids = std::move(column_id_result.column_ids);
ctx->filter_column_ids = std::move(column_id_result.filter_column_ids);

Expand All @@ -518,13 +563,12 @@ Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
// Add expand column IDs for equality delete and remap expand column names
// (matching master's behavior with __equality_delete_column__ prefix)
const static std::string EQ_DELETE_PRE = "__equality_delete_column__";
std::unordered_map<int, std::string> field_id_to_file_col_name;
std::unordered_map<int, const orc::Type*> field_id_to_file_column;
for (uint64_t i = 0; i < orc_type_ptr->getSubtypeCount(); ++i) {
std::string col_name = orc_type_ptr->getFieldName(i);
const orc::Type* sub_type = orc_type_ptr->getSubtype(i);
if (sub_type->hasAttributeKey(ICEBERG_ORC_ATTRIBUTE)) {
int fid = std::stoi(sub_type->getAttributeValue(ICEBERG_ORC_ATTRIBUTE));
field_id_to_file_col_name[fid] = col_name;
field_id_to_file_column[fid] = sub_type;
}
}

Expand All @@ -539,12 +583,34 @@ Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
}
}

std::string file_col_name = old_name;
auto it = field_id_to_file_col_name.find(field_id);
if (it != field_id_to_file_col_name.end()) {
file_col_name = it->second;
const orc::Type* file_column = nullptr;
auto id_it = field_id_to_file_column.find(field_id);
if (id_it != field_id_to_file_column.end()) {
file_column = id_it->second;
} else if (const auto struct_node =
std::dynamic_pointer_cast<TableSchemaChangeHelper::StructNode>(
ctx->table_info_node);
struct_node != nullptr && struct_node->get_children().contains(old_name) &&
struct_node->children_column_exists(old_name)) {
const auto& mapped_name = struct_node->children_file_column_name(old_name);
for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) {
if (orc_type_ptr->getFieldName(j) == mapped_name) {
file_column = orc_type_ptr->getSubtype(j);
break;
}
}
DORIS_CHECK(file_column != nullptr);
}

std::string file_col_name = old_name;
if (file_column != nullptr) {
for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) {
if (orc_type_ptr->getSubtype(j) == file_column) {
file_col_name = orc_type_ptr->getFieldName(j);
break;
}
}
}
std::string table_col_name = EQ_DELETE_PRE + file_col_name;

if (field_id >= 0) {
Expand All @@ -553,18 +619,21 @@ Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
if (i < _expand_columns.size()) {
_expand_columns[i].name = table_col_name;
}
if (file_column == nullptr) {
DORIS_CHECK(i < _expand_columns.size());
RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id, table_col_name,
_expand_columns[i].type));
// The old data file predates this equality key. Keep it in the expand block so the
// synthesized-column hook can materialize its logical initial default before ORC's
// block-size checks. Adding it to column_names/table_info_node would mark it as an
// existing ORC child and make OrcReader read a column that is not present in the file.
new_expand_col_names.push_back(table_col_name);
continue;
}
new_expand_col_names.push_back(table_col_name);

// Add column IDs
if (it != field_id_to_file_col_name.end()) {
for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) {
const orc::Type* sub_type = orc_type_ptr->getSubtype(j);
if (orc_type_ptr->getFieldName(j) == file_col_name) {
ctx->column_ids.insert(sub_type->getColumnId());
break;
}
}
}
ctx->column_ids.insert(file_column->getColumnId());

ctx->column_names.push_back(table_col_name);
ctx->table_info_node->add_children(table_col_name, file_col_name,
Expand All @@ -578,12 +647,15 @@ Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
// ============================================================================
// IcebergOrcReader: _create_column_ids
// ============================================================================
ColumnIdResult IcebergOrcReader::_create_column_ids(const orc::Type* orc_type,
const TupleDescriptor* tuple_descriptor) {
ColumnIdResult IcebergOrcReader::_create_column_ids(
const orc::Type* orc_type, const TupleDescriptor* tuple_descriptor,
const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node) {
std::unordered_map<int, const orc::Type*> iceberg_id_to_orc_type_map;
for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) {
auto orc_sub_type = orc_type->getSubtype(i);
if (!orc_sub_type) continue;
const auto* orc_sub_type = orc_type->getSubtype(i);
if (!orc_sub_type) {
continue;
}
if (!orc_sub_type->hasAttributeKey(ICEBERG_ORC_ATTRIBUTE)) {
continue;
}
Expand All @@ -605,11 +677,31 @@ ColumnIdResult IcebergOrcReader::_create_column_ids(const orc::Type* orc_type,
};

for (const auto* slot : tuple_descriptor->slots()) {
auto it = iceberg_id_to_orc_type_map.find(slot->col_unique_id());
if (it == iceberg_id_to_orc_type_map.end()) {
const orc::Type* orc_field = nullptr;
if (table_info_node != nullptr) {
if (table_info_node->children_column_exists(slot->col_name())) {
// Select the physical child resolved by the shared schema-mapping pass. Hidden
// equality keys and projected columns must obey the same BY_NAME decision for
// partial-id ORC files.
const auto& file_column_name =
table_info_node->children_file_column_name(slot->col_name());
for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) {
if (orc_type->getFieldName(i) == file_column_name) {
orc_field = orc_type->getSubtype(i);
break;
}
}
DORIS_CHECK(orc_field != nullptr);
}
} else {
auto it = iceberg_id_to_orc_type_map.find(slot->col_unique_id());
if (it != iceberg_id_to_orc_type_map.end()) {
orc_field = it->second;
}
}
if (orc_field == nullptr) {
continue;
}
const orc::Type* orc_field = it->second;

if ((slot->col_type() != TYPE_STRUCT && slot->col_type() != TYPE_ARRAY &&
slot->col_type() != TYPE_MAP)) {
Expand Down
10 changes: 6 additions & 4 deletions be/src/format/table/iceberg_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,8 +105,9 @@ class IcebergParquetReader final : public IcebergReaderMixin<ParquetReader> {
this->get_state(), this->_meta_cache);
}

static ColumnIdResult _create_column_ids(const FieldDescriptor* field_desc,
const TupleDescriptor* tuple_descriptor);
static ColumnIdResult _create_column_ids(
const FieldDescriptor* field_desc, const TupleDescriptor* tuple_descriptor,
const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node = nullptr);

private:
Status _read_position_delete_file(const TFileRangeDesc* delete_range,
Expand Down Expand Up @@ -151,8 +152,9 @@ class IcebergOrcReader final : public IcebergReaderMixin<OrcReader> {
this->get_io_ctx(), this->_meta_cache);
}

static ColumnIdResult _create_column_ids(const orc::Type* orc_type,
const TupleDescriptor* tuple_descriptor);
static ColumnIdResult _create_column_ids(
const orc::Type* orc_type, const TupleDescriptor* tuple_descriptor,
const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node = nullptr);

static const std::string ICEBERG_ORC_ATTRIBUTE;

Expand Down
Loading
Loading