This is an automated email from the ASF dual-hosted git repository.
morningman 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 ee47a803766 [fix](be) Keep projected structs readable with nested
Iceberg equality keys in V1 Parquet (#69547)
ee47a803766 is described below
commit ee47a80376647f8efe4fea8ef0dcbf3481b9296d
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sun Oct 25 11:00:25 2026 +0000
[fix](be) Keep projected structs readable with nested Iceberg equality keys
in V1 Parquet (#69547)
### What problem does this PR solve?
Issue Number: None
Related PR: #66538
Problem Summary:
**In short.** This ports a fix from branch-4.1 #66538 (Iceberg V3
default values) to the V1 Parquet reader on master. When an Iceberg
equality delete is keyed on a struct field such as `payload.existing`,
and the query also projects `payload` (any `SELECT *`), or the delete
has several keys in the same struct, master reads only one of the
columns that map to the struct, and returns an empty struct or no rows.
**Background.** The V1 Parquet reader reads a nested equality key
through its top-level struct. A hidden carrier column named
`__equality_delete_column__<id>_<leaf>` maps to the physical root
(`payload`), the whole struct is read into it, and the key's leaf is
extracted after the read. `ParquetReader::_init_read_columns` builds a
map from each file column to one table column.
**The problem, and what it cost.** When two table columns name the same
file column, the later one replaces the earlier one in that map, and
only one of them gets a column reader. Two cases do this: a projected
`payload` next to the carrier of `payload.existing`, and the carriers of
two keys under one root, such as `payload.a` and `payload.b`. Reproduced
with BE unit tests on a file holding (1,{10}), (2,{20}), (3,{30}) and an
equality delete on `payload.existing = 20`:
| Query projects | Master today | After this PR |
|---|---|---|
| `id, payload` | `payload` is empty, and the block has fewer `payload`
rows than `id` rows (a DCHECK in `vparquet_group_reader.cpp` under ASAN)
| (1,{10}), (3,{30}) |
| `payload, id` | no rows | (1,{10}), (3,{30}) |
| `id`, with keys `payload.existing` and `payload.k` | the carrier of
`payload.existing` is never read | rows 1 and 3 |
ORC reads of a projected root struct are not affected: the ORC cases
below pass without this change.
**How this PR fixes it.** It follows branch-4.1 / branch-4.2 (the
Parquet carrier setup in `be/src/format/table/iceberg_reader.cpp` and
`ParquetReader::set_duplicate_file_column_aliases`):
- When the query projects the key's root struct, the file has the key's
full path and the projected struct holds the key with the key's type,
the key is taken from the projected column. Nothing extra is read. A key
is taken from the projected root only when the projected column's schema
mapping reads the same file field as the key's physical path at every
level; a plan without the Iceberg scan semantics marker can map a struct
level by name while the key is found by field ID, and such a key reads
its own carrier.
- Otherwise the first key under a physical root reads the root, and
every later key under that root is extracted from that carrier's column.
- A carrier that reads nothing gets a placeholder of the batch row
count, so the reader filters it with the other columns, and the leaf
replaces it after the read. All leaves are extracted before any carrier
column is replaced, because one carrier can be the source of other keys.
- `ParquetReader` reads a file column once for each table column that
names it, and prefetches it once. This covers a key that the projected
struct lacks, for example a key dropped from the table, whose carrier
reads the same root as the projected `payload`.
Two differences from branch-4.x, both on the safe side: a key whose path
is incomplete in the file (an older file written before the key existed)
keeps reading the physical root instead of relying on the projected
struct for the missing child, and a key whose type differs from the
projected type (a key written before a type promotion) also keeps
reading the physical root. Branch-4.x casts that case; master does not
cast equality keys yet, which is a separate fix.
Equality-delete keys that the V1 reader does not read are not
synthesized columns: a key extracted from a projected root or from
another key's carrier, and a key missing from an old data file, are
appended in on_after_read_block once the batch has its final row count.
This also fixes the missing-key path from #65502 on Parquet and ORC,
whose synthesized column was dropped by the lazy read's column trimming.
**Results.** All 58 tests of `IcebergReaderTest` pass. Without this
change, each of the five Parquet cases below aborts on the DCHECK
`block->rows() == col.column->size()` in `vparquet_group_reader.cpp`,
because one of the columns that share the root is never read:
- Four cases from #66538 for a projected root struct, Parquet and ORC,
with `payload` listed before and after `id`. Without this change the
Parquet cases leave `payload` unread in both orders; the ORC cases
already pass.
- `v1_parquet_equality_keys_share_unprojected_root`: two keys under an
unprojected root share one carrier. Without this change the carrier of
`payload.existing` is never read.
- `v1_parquet_equality_keys_come_from_projected_root`: two keys are
taken from the projected root. Without this change `payload` is never
read.
- `v1_parquet_equality_key_outside_projected_struct_rereads_root`: the
projected struct lacks the key, so the carrier reads the same file
column again. Without this change `payload` is never read.
-
`v1_parquet_legacy_plan_key_rereads_root_when_projected_child_matches_by_name`
(follow-up): a plan without the Iceberg scan semantics marker maps
`payload{existing without an ID, k#3}` by name, so the projected
`payload` reads the renamed key as NULL while the key is found by field
ID; the key must read its own carrier. Without the binding check id 2
survives the delete. Under a plan with the marker the key still comes
from the projected `payload`.
Eight more IcebergReaderTest cases cover lazy materialization under a
filter on id (nested keys on Parquet, missing keys on Parquet and ORC);
without the fix they abort at vparquet_group_reader.cpp:829 or fail with
"invalid input position". IcebergReaderTest has 59 cases, all passing
after the follow-up commit; the Iceberg reader suites (366 cases) passed
on 5acba3e2c97, before it.
Co-authored-by: daidai <[email protected]>
---
be/src/format/parquet/vparquet_reader.cpp | 26 +-
be/src/format/table/iceberg_reader.cpp | 253 ++++++-
be/src/format/table/iceberg_reader_mixin.h | 75 ++-
.../format/table/iceberg/iceberg_reader_test.cpp | 746 +++++++++++++++++++++
4 files changed, 1050 insertions(+), 50 deletions(-)
diff --git a/be/src/format/parquet/vparquet_reader.cpp
b/be/src/format/parquet/vparquet_reader.cpp
index 79ea2bb19f6..15f5f93c2ac 100644
--- a/be/src/format/parquet/vparquet_reader.cpp
+++ b/be/src/format/parquet/vparquet_reader.cpp
@@ -574,8 +574,10 @@ void ParquetReader::_init_read_columns(const
std::vector<std::string>& column_na
// Build file_col_name → table_col_name map, skipping missing columns.
// Must iterate file schema in physical order so that
_generate_random_access_ranges
// sees monotonically increasing chunk offsets.
+ // Several table columns may read the same file column, e.g. a projected
Iceberg struct and the
+ // hidden carrier of an equality key nested in it. Each of them gets its
own column reader.
auto schema_desc = _file_metadata->schema();
- std::map<std::string, std::string> required_file_columns;
+ std::multimap<std::string, std::string> required_file_columns;
for (const auto& col_name : column_names) {
if (_fill_missing_cols.contains(col_name)) {
continue;
@@ -584,14 +586,15 @@ void ParquetReader::_init_read_columns(const
std::vector<std::string>& column_na
if (_table_info_node_ptr &&
_table_info_node_ptr->children_column_exists(col_name)) {
file_col =
_table_info_node_ptr->children_file_column_name(col_name);
}
- required_file_columns[file_col] = col_name;
+ required_file_columns.emplace(std::move(file_col), col_name);
}
for (int i = 0; i < schema_desc.size(); ++i) {
const auto& name = schema_desc.get_column(i)->name;
- if (required_file_columns.contains(name)) {
+ const auto [begin, end] = required_file_columns.equal_range(name);
+ for (auto required = begin; required != end; ++required) {
_read_file_columns.emplace_back(name);
- _read_table_columns.emplace_back(required_file_columns[name]);
- _read_table_columns_set.insert(required_file_columns[name]);
+ _read_table_columns.emplace_back(required->second);
+ _read_table_columns_set.insert(required->second);
}
}
}
@@ -960,7 +963,13 @@ Status ParquetReader::_next_row_group_reader() {
return size;
};
int64_t group_size = 0; // only calculate the needed columns
+ const std::string* previous_read_col = nullptr;
for (auto& read_col : _read_file_columns) {
+ // A file column read under several table names is listed once per
name, consecutively.
+ if (previous_read_col != nullptr && *previous_read_col ==
read_col) {
+ continue;
+ }
+ previous_read_col = &read_col;
const FieldSchema* field =
_file_metadata->schema().get_column(read_col);
group_size += column_compressed_size(field);
}
@@ -1073,7 +1082,14 @@ std::vector<io::PrefetchRange>
ParquetReader::_generate_random_access_ranges(
}
};
const tparquet::RowGroup& row_group =
_t_metadata->row_groups[group.row_group_id];
+ const std::string* previous_read_col = nullptr;
for (const auto& read_col : _read_file_columns) {
+ // Chunk offsets must increase, so a file column read under several
table names is
+ // prefetched once.
+ if (previous_read_col != nullptr && *previous_read_col == read_col) {
+ continue;
+ }
+ previous_read_col = &read_col;
const FieldSchema* field =
_file_metadata->schema().get_column(read_col);
scalar_range(field, row_group);
}
diff --git a/be/src/format/table/iceberg_reader.cpp
b/be/src/format/table/iceberg_reader.cpp
index 2acfcdf7aac..d49d690a01e 100644
--- a/be/src/format/table/iceberg_reader.cpp
+++ b/be/src/format/table/iceberg_reader.cpp
@@ -29,6 +29,7 @@
#include <cstring>
#include <functional>
#include <memory>
+#include <optional>
#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/consts.h"
@@ -42,9 +43,11 @@
#include "core/column/column_vector.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_struct.h"
#include "core/data_type/define_primitive_type.h"
#include "core/data_type/primitive_type.h"
#include "core/string_ref.h"
+#include "core/typeid_cast.h"
#include "exprs/aggregate/aggregate_function.h"
#include "format/format_common.h"
#include "format/generic_reader.h"
@@ -372,6 +375,131 @@ bool find_orc_equality_field_prefix_by_name_path(
return true;
}
+struct ProjectedIcebergStructChild {
+ size_t index;
+ const schema::external::TField* field;
+};
+
+std::optional<ProjectedIcebergStructChild> find_projected_iceberg_struct_child(
+ const schema::external::TField& parent, const
schema::external::TField& target,
+ const DataTypeStruct& projected_type) {
+ DORIS_CHECK(parent.__isset.nestedField);
+ DORIS_CHECK(parent.nestedField.__isset.struct_field);
+ DORIS_CHECK(parent.nestedField.struct_field.__isset.fields);
+ DORIS_CHECK(target.__isset.name);
+ DORIS_CHECK(target.__isset.id);
+
+ // FE keeps current children first and appends historical equality fields
to the schema carrier
+ // by field ID. A dropped and re-added field may therefore appear twice
with the same name,
+ // while the query's DataTypeStruct contains only the current occurrence.
+ size_t schema_name_ordinal = 0;
+ const schema::external::TField* schema_child = nullptr;
+ for (const auto& child_ptr : parent.nestedField.struct_field.fields) {
+ DORIS_CHECK(child_ptr.__isset.field_ptr && child_ptr.field_ptr !=
nullptr);
+ const auto& child = child_ptr.field_ptr;
+ DORIS_CHECK(child->__isset.name);
+ DORIS_CHECK(child->__isset.id);
+ if (child->name != target.name) {
+ continue;
+ }
+ if (child->id == target.id) {
+ schema_child = child.get();
+ break;
+ }
+ ++schema_name_ordinal;
+ }
+ if (schema_child == nullptr) {
+ return std::nullopt;
+ }
+
+ size_t projected_name_ordinal = 0;
+ for (size_t index = 0; index < projected_type.get_elements().size();
++index) {
+ if (projected_type.get_element_name(index) != target.name) {
+ continue;
+ }
+ if (projected_name_ordinal == schema_name_ordinal) {
+ return ProjectedIcebergStructChild {.index = index, .field =
schema_child};
+ }
+ ++projected_name_ordinal;
+ }
+ return std::nullopt;
+}
+
+// Resolves a nested equality key inside the projected root column. That
column has the table's
+// projected struct type, not the file's physical struct, so the child indexes
come from the
+// projected type. `found` stays false when the projected type does not hold
the key, e.g. when the
+// key was dropped from the table or the struct was pruned.
+Status find_projected_equality_key_path(
+ const schema::external::TField& projected_root,
+ const std::vector<const schema::external::TField*>& key_path, const
DataTypePtr& root_type,
+ std::vector<size_t>* child_indexes, DataTypePtr* leaf_type, bool*
found) {
+ DORIS_CHECK(!key_path.empty());
+ DORIS_CHECK(child_indexes != nullptr);
+ DORIS_CHECK(leaf_type != nullptr);
+ DORIS_CHECK(found != nullptr);
+ child_indexes->clear();
+ *found = false;
+ const auto* projected_parent = &projected_root;
+ DataTypePtr current_type = root_type;
+ for (size_t path_index = 1; path_index < key_path.size(); ++path_index) {
+ const auto* child = key_path[path_index];
+ DORIS_CHECK(child != nullptr);
+ if (!projected_parent->__isset.nestedField ||
+ !projected_parent->nestedField.__isset.struct_field ||
+ !projected_parent->nestedField.struct_field.__isset.fields) {
+ return Status::NotSupported(
+ "Iceberg equality-delete field {} has a non-struct
projected-schema parent",
+ child->name);
+ }
+ const auto* struct_type =
+ typeid_cast<const
DataTypeStruct*>(remove_nullable(current_type).get());
+ if (struct_type == nullptr) {
+ return Status::OK();
+ }
+ const auto projected_child =
+ find_projected_iceberg_struct_child(*projected_parent, *child,
*struct_type);
+ if (!projected_child.has_value()) {
+ return Status::OK();
+ }
+ child_indexes->push_back(projected_child->index);
+ current_type = struct_type->get_element(projected_child->index);
+ projected_parent = projected_child->field;
+ }
+ *leaf_type = current_type;
+ *found = true;
+ return Status::OK();
+}
+
+// A projected column is read through its own schema mapping, as the struct
reader resolves each
+// child (`children_file_column_name`), while a hidden key is located in the
file by field ID or by
+// its name path. The two can bind different file fields: a plan without the
Iceberg scan semantics
+// marker maps a struct level by name when not all of its file children carry
field IDs. The key
+// can therefore come from the projected column only when that mapping reads
the same file field at
+// every level of the key's physical path.
+bool projected_root_reads_physical_path(const TableSchemaChangeHelper::Node&
root_node,
+ const std::string& root_name, const
DataTypePtr& root_type,
+ const std::vector<size_t>&
child_indexes,
+ const ParquetEqualityFieldPath&
file_path) {
+ if (child_indexes.size() + 1 != file_path.fields.size() ||
+ !root_node.children_column_exists(root_name) ||
+ root_node.children_file_column_name(root_name) !=
file_path.fields.front()->name) {
+ return false;
+ }
+ auto node = root_node.get_children_node(root_name);
+ DataTypePtr type = root_type;
+ for (size_t level = 0; level < child_indexes.size(); ++level) {
+ const auto& struct_type = assert_cast<const
DataTypeStruct&>(*remove_nullable(type));
+ const auto& child_name =
struct_type.get_element_name(child_indexes[level]);
+ if (!node->children_column_exists(child_name) ||
+ node->children_file_column_name(child_name) !=
file_path.fields[level + 1]->name) {
+ return false;
+ }
+ node = node->get_children_node(child_name);
+ type = struct_type.get_element(child_indexes[level]);
+ }
+ return true;
+}
+
} // namespace
const std::string IcebergOrcReader::ICEBERG_ORC_ATTRIBUTE =
kIcebergOrcAttribute;
@@ -570,6 +698,47 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
}
return nullptr;
};
+ std::unordered_map<int32_t, const SlotDescriptor*> projected_slots;
+ for (const auto* slot : ctx->tuple_descriptor->slots()) {
+ projected_slots.emplace(slot->col_unique_id(), slot);
+ }
+ // Finds a nested key inside its root struct when the query projects that
root, so the key is
+ // taken from the projected column instead of reading the root a second
time.
+ const auto find_projected_root = [&](int32_t field_id, const DataTypePtr&
key_type,
+ const ParquetEqualityFieldPath&
file_path,
+ std::string* root_name,
+ std::vector<size_t>* child_indexes)
-> Status {
+ const auto key_path = _find_schema_field_path(field_id);
+ if (key_path.empty()) {
+ return Status::OK();
+ }
+ DORIS_CHECK(key_path.front()->__isset.id);
+ const auto slot = projected_slots.find(key_path.front()->id);
+ if (slot == projected_slots.end()) {
+ return Status::OK();
+ }
+ std::vector<const schema::external::TField*> root_path;
+ if (!_find_schema_field_path_in_root(_current_schema_root(),
key_path.front()->id,
+ &root_path)) {
+ return Status::OK();
+ }
+ DORIS_CHECK(root_path.size() == 1);
+ DataTypePtr leaf_type;
+ bool found = false;
+ RETURN_IF_ERROR(find_projected_equality_key_path(*root_path.front(),
key_path,
+ slot->second->type(),
child_indexes,
+ &leaf_type, &found));
+ // The projected leaf has the table's current type. A key written
before a type promotion
+ // (INT promoted to BIGINT) keeps its own type, so it still reads the
physical root.
+ if (found &&
remove_nullable(leaf_type)->equals(*remove_nullable(key_type)) &&
+ projected_root_reads_physical_path(*ctx->table_info_node,
slot->second->col_name(),
+ slot->second->type(),
*child_indexes, file_path)) {
+ *root_name = slot->second->col_name();
+ }
+ return Status::OK();
+ };
+ // Physical root struct -> the carrier that reads it for every hidden key
under that root.
+ std::unordered_map<std::string, std::string> physical_root_carriers;
// Rebuild _expand_col_names with proper file-column-based names
std::vector<std::string> new_expand_col_names;
@@ -629,19 +798,69 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
if (file_column == nullptr) {
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.
+ // The old data file predates this equality key. Its logical
initial default is appended
+ // after the read; 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);
- if (!complete_file_path) {
+ // A hidden nested key is read through its containing top-level
struct. V1 column IDs are
+ // pre-order ranges, so include the complete subtree before extracting
the primitive leaf.
+ // A projected root that serves as the key's source then reads the
key's leaf as well.
+ for (uint64_t column_id = file_column->get_column_id();
+ column_id <= file_column->get_max_column_id(); ++column_id) {
+ ctx->column_ids.insert(column_id);
+ }
+
+ if (!complete_file_path || !file_path.child_indexes.empty()) {
+ // Parquet reads each table column through its own column reader.
Without the two
+ // branches below, a projected root and a carrier, or two carriers
of keys under one
+ // root, all named the same file column, and only one of them was
read. Example:
+ // `SELECT id, payload` with an equality delete on
payload.existing returned an empty
+ // payload, or no rows at all when payload was listed first.
+ std::string projected_root_name;
+ std::vector<size_t> projected_child_indexes;
+ if (complete_file_path) {
+ RETURN_IF_ERROR(find_projected_root(field_id,
_expand_columns[i].type, file_path,
+ &projected_root_name,
+ &projected_child_indexes));
+ }
+ if (!projected_root_name.empty()) {
+ // The query projects payload and the file has
payload.existing: extract the key
+ // from the projected payload column.
+ _nested_equality_delete_columns.push_back({
+ .field_id = field_id,
+ .block_name = table_col_name,
+ .leaf_type = _expand_columns[i].type,
+ .child_indexes = std::move(projected_child_indexes),
+ .missing_value = nullptr,
+ .source_block_name = std::move(projected_root_name),
+ });
+ continue;
+ }
ColumnPtr missing_value;
- RETURN_IF_ERROR(_create_missing_equality_delete_value(
- field_id, _expand_columns[i].type,
file_path.fields.size(), &missing_value));
+ if (!complete_file_path) {
+ RETURN_IF_ERROR(_create_missing_equality_delete_value(
+ field_id, _expand_columns[i].type,
file_path.fields.size(),
+ &missing_value));
+ }
+ const auto root_carrier =
physical_root_carriers.find(file_column->name);
+ if (root_carrier != physical_root_carriers.end()) {
+ // Keys payload.a and payload.b: the carrier of payload.a
reads payload, and
+ // payload.b is extracted from that carrier's column.
+ _nested_equality_delete_columns.push_back({
+ .field_id = field_id,
+ .block_name = table_col_name,
+ .leaf_type = _expand_columns[i].type,
+ .child_indexes = file_path.child_indexes,
+ .missing_value = std::move(missing_value),
+ .source_block_name = root_carrier->second,
+ });
+ continue;
+ }
+ physical_root_carriers.emplace(file_column->name, table_col_name);
_nested_equality_delete_columns.push_back({
.field_id = field_id,
.block_name = table_col_name,
@@ -651,23 +870,6 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
});
_expand_columns[i].type = make_nullable(file_column->data_type);
_expand_columns[i].column =
_expand_columns[i].type->create_column();
- } else if (!file_path.child_indexes.empty()) {
- _nested_equality_delete_columns.push_back({
- .field_id = field_id,
- .block_name = table_col_name,
- .leaf_type = _expand_columns[i].type,
- .child_indexes = file_path.child_indexes,
- .missing_value = nullptr,
- });
- _expand_columns[i].type = make_nullable(file_column->data_type);
- _expand_columns[i].column =
_expand_columns[i].type->create_column();
- }
-
- // A hidden nested key is read through its containing top-level
struct. V1 column IDs are
- // pre-order ranges, so include the complete subtree before extracting
the primitive leaf.
- for (uint64_t column_id = file_column->get_column_id();
- column_id <= file_column->get_max_column_id(); ++column_id) {
- ctx->column_ids.insert(column_id);
}
// Register in table_info_node: table_col_name → file_col_name
@@ -1016,9 +1218,8 @@ Status
IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
if (file_column == nullptr) {
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
+ // The old data file predates this equality key. Its logical
initial default is appended
+ // after the read. 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;
diff --git a/be/src/format/table/iceberg_reader_mixin.h
b/be/src/format/table/iceberg_reader_mixin.h
index a9d0925f10e..bf47010289e 100644
--- a/be/src/format/table/iceberg_reader_mixin.h
+++ b/be/src/format/table/iceberg_reader_mixin.h
@@ -294,6 +294,7 @@ protected:
// Called after reading a block: apply equality delete filter + shrink
block
Status on_after_read_block(Block* block, size_t* read_rows) override {
+ RETURN_IF_ERROR(_materialize_missing_equality_delete_columns(block,
block->rows()));
RETURN_IF_ERROR(_materialize_nested_equality_delete_columns(block));
if (!_equality_delete_impls.empty()) {
std::unique_ptr<IColumn::Filter> filter =
@@ -336,6 +337,7 @@ protected:
ColumnPtr* const value) const;
Status _register_missing_equality_delete_column(int32_t field_id, const
std::string& name,
const DataTypePtr&
delete_key_type);
+ bool _is_derived_equality_delete_column(const std::string& name) const;
Status _materialize_missing_equality_delete_column(Block* block, const
std::string& name,
const ColumnPtr& value,
size_t rows);
Status _materialize_missing_equality_delete_columns(Block* block, size_t
rows);
@@ -363,6 +365,9 @@ protected:
DataTypePtr leaf_type;
std::vector<size_t> child_indexes;
ColumnPtr missing_value;
+ // Block column holding the key's root struct. Empty when the key's
own column reads the
+ // root; otherwise the projected root, or the carrier that reads a
shared physical root.
+ std::string source_block_name {};
};
struct EqualityDeleteReadSpec {
NestedEqualityDeleteColumn nested_field;
@@ -919,9 +924,10 @@ Status
IcebergReaderMixin<BaseReader>::_expand_block_if_need(Block* block) {
auto block_names = block->get_names();
names.insert(block_names.begin(), block_names.end());
for (auto& col : _expand_columns) {
- if (_missing_equality_delete_values.contains(col.name)) {
- // Missing equality keys are logical columns, not physical file
columns. Add them only
- // after the base reader has established the batch row count.
+ if (_missing_equality_delete_values.contains(col.name) ||
+ _is_derived_equality_delete_column(col.name)) {
+ // The reader has no column for a key that is missing from the
file or extracted from
+ // another column; on_after_read_block appends it at the batch's
final row count.
continue;
}
if (names.contains(col.name)) {
@@ -1132,15 +1138,40 @@ Status
IcebergReaderMixin<BaseReader>::_extract_nested_equality_delete_column(
template <typename BaseReader>
Status
IcebergReaderMixin<BaseReader>::_materialize_nested_equality_delete_columns(Block*
block) {
DORIS_CHECK(block != nullptr);
+ // A carrier may be the source of other keys under the same physical root,
so extract every
+ // leaf before replacing any carrier column with its own leaf.
+ std::vector<ColumnPtr> leaves;
+ leaves.reserve(_nested_equality_delete_columns.size());
for (const auto& nested_field : _nested_equality_delete_columns) {
- const auto position =
this->col_name_to_block_idx_ref()->find(nested_field.block_name);
- DORIS_CHECK(position != this->col_name_to_block_idx_ref()->end());
- DORIS_CHECK(position->second < block->columns());
- auto& column = block->get_by_position(position->second);
+ const auto& source_name = nested_field.source_block_name.empty()
+ ? nested_field.block_name
+ : nested_field.source_block_name;
+ const auto source =
this->col_name_to_block_idx_ref()->find(source_name);
+ DORIS_CHECK(source != this->col_name_to_block_idx_ref()->end());
+ DORIS_CHECK(source->second < block->columns());
ColumnPtr leaf;
- RETURN_IF_ERROR(_extract_nested_equality_delete_column(column.column,
nested_field, &leaf));
- column.column = std::move(leaf);
- column.type = make_nullable(nested_field.leaf_type);
+ RETURN_IF_ERROR(_extract_nested_equality_delete_column(
+ block->get_by_position(source->second).column, nested_field,
&leaf));
+ leaves.push_back(std::move(leaf));
+ }
+ for (size_t index = 0; index < leaves.size(); ++index) {
+ const auto& nested_field = _nested_equality_delete_columns[index];
+ const auto type = make_nullable(nested_field.leaf_type);
+ auto& positions = *this->col_name_to_block_idx_ref();
+ if (!nested_field.source_block_name.empty()) {
+ // A key extracted from a projected root or another key's carrier
is not read, so the
+ // block gets its column only now.
+ DORIS_CHECK(!positions.contains(nested_field.block_name));
+ positions[nested_field.block_name] = block->columns();
+ block->insert({std::move(leaves[index]), type,
nested_field.block_name});
+ continue;
+ }
+ const auto target = positions.find(nested_field.block_name);
+ DORIS_CHECK(target != positions.end());
+ DORIS_CHECK(target->second < block->columns());
+ auto& column = block->get_by_position(target->second);
+ column.column = std::move(leaves[index]);
+ column.type = type;
}
return Status::OK();
}
@@ -1214,25 +1245,31 @@ Status
IcebergReaderMixin<BaseReader>::_register_missing_equality_delete_column(
ColumnPtr default_column;
RETURN_IF_ERROR(
_create_missing_equality_delete_value(field_id, delete_key_type,
0, &default_column));
+ // The key is appended after the read (on_after_read_block), not by a
synthesized-column
+ // handler: the reader filters only the columns its block had before the
read, and Parquet's
+ // lazy read also looks up a tuple slot for every synthesized column,
which this hidden key
+ // does not have.
const bool inserted =
_missing_equality_delete_values.emplace(name,
std::move(default_column)).second;
DORIS_CHECK(inserted);
- this->register_synthesized_column_handler(
- name, [this, name](Block* block, size_t rows) -> Status {
- DORIS_CHECK(_missing_equality_delete_values.contains(name));
- return _materialize_missing_equality_delete_column(
- block, name, _missing_equality_delete_values.at(name),
rows);
- });
return Status::OK();
}
+template <typename BaseReader>
+bool IcebergReaderMixin<BaseReader>::_is_derived_equality_delete_column(
+ const std::string& name) const {
+ return std::ranges::any_of(_nested_equality_delete_columns, [&](const
auto& nested_field) {
+ return nested_field.block_name == name &&
!nested_field.source_block_name.empty();
+ });
+}
+
template <typename BaseReader>
Status
IcebergReaderMixin<BaseReader>::_materialize_missing_equality_delete_column(
Block* block, const std::string& name, const ColumnPtr& value, size_t
rows) {
if (!this->col_name_to_block_idx_ref()->contains(name)) {
- // ORC must not register a key that is absent from the file as a
physical child. In that
- // case the reader block has no slot for the synthesized key, so
append one here before
- // equality-delete filtering. MultiEqualityDelete requires a full,
batch-sized column.
+ // A key that is absent from the file is not a physical child of the
reader, so the block
+ // has no column for it: append one before equality-delete filtering.
MultiEqualityDelete
+ // requires a full, batch-sized column.
const auto expand_col = std::ranges::find_if(
_expand_columns,
[&](const ColumnWithTypeAndName& col) { return col.name ==
name; });
diff --git a/be/test/format/table/iceberg/iceberg_reader_test.cpp
b/be/test/format/table/iceberg/iceberg_reader_test.cpp
index 528ed13f0f4..08767f5e022 100644
--- a/be/test/format/table/iceberg/iceberg_reader_test.cpp
+++ b/be/test/format/table/iceberg/iceberg_reader_test.cpp
@@ -27,6 +27,7 @@
#include <gtest/gtest.h>
#include <parquet/arrow/writer.h>
+#include <algorithm>
#include <array>
#include <bit>
#include <cmath>
@@ -57,6 +58,10 @@
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
#include "core/data_type/data_type_varbinary.h"
+#include "exprs/vectorized_fn_call.h"
+#include "exprs/vexpr_context.h"
+#include "exprs/vliteral.h"
+#include "exprs/vslot_ref.h"
#include "format/column_descriptor.h"
#include "format/format_common.h"
#include "format/orc/orc_memory_stream_test.h"
@@ -314,6 +319,59 @@ void write_iceberg_id_and_nested_struct_parquet_file(const
std::string& file_pat
properties.build()));
}
+// `payload STRUCT<existing INT, k INT>` (field 1, children 2 and 3), preceded
by `id INT` (field 0)
+// when `ids` is given.
+void write_iceberg_two_child_struct_parquet_file(const std::string& file_path,
+ const std::vector<int32_t>*
ids,
+ const std::vector<int32_t>&
existing_values,
+ const std::vector<int32_t>&
k_values,
+ bool existing_has_field_id =
true) {
+ DORIS_CHECK(existing_values.size() == k_values.size());
+ const auto with_field_id = [](const std::shared_ptr<arrow::Field>& field,
int32_t field_id) {
+ return field->WithMetadata(
+ arrow::key_value_metadata({"PARQUET:field_id"},
{std::to_string(field_id)}));
+ };
+ auto existing_field = arrow::field("existing", arrow::int32(), false);
+ auto struct_type = arrow::struct_(
+ {existing_has_field_id ? with_field_id(existing_field, 2) :
existing_field,
+ with_field_id(arrow::field("k", arrow::int32(), false), 3)});
+ std::vector<std::shared_ptr<arrow::ArrayBuilder>> field_builders;
+ field_builders.emplace_back(std::make_shared<arrow::Int32Builder>());
+ field_builders.emplace_back(std::make_shared<arrow::Int32Builder>());
+ arrow::StructBuilder builder(struct_type, arrow::default_memory_pool(),
+ std::move(field_builders));
+ auto* existing_builder =
assert_cast<arrow::Int32Builder*>(builder.field_builder(0));
+ auto* k_builder =
assert_cast<arrow::Int32Builder*>(builder.field_builder(1));
+ for (size_t row = 0; row < existing_values.size(); ++row) {
+ DORIS_CHECK(builder.Append().ok());
+ DORIS_CHECK(existing_builder->Append(existing_values[row]).ok());
+ DORIS_CHECK(k_builder->Append(k_values[row]).ok());
+ }
+ std::shared_ptr<arrow::Array> payload;
+ DORIS_CHECK(builder.Finish(&payload).ok());
+
+ std::vector<std::shared_ptr<arrow::Field>> fields;
+ std::vector<std::shared_ptr<arrow::Array>> arrays;
+ if (ids != nullptr) {
+ DORIS_CHECK(ids->size() == existing_values.size());
+ fields.push_back(with_field_id(arrow::field("id", arrow::int32(),
false), 0));
+ arrays.push_back(build_iceberg_int32_array(*ids));
+ }
+ fields.push_back(with_field_id(arrow::field("payload", struct_type, true),
1));
+ arrays.push_back(std::move(payload));
+ auto table = arrow::Table::Make(arrow::schema(fields), arrays);
+
+ auto output = arrow::io::FileOutputStream::Open(file_path);
+ DORIS_CHECK(output.ok());
+ ::parquet::WriterProperties::Builder properties;
+ properties.version(::parquet::ParquetVersion::PARQUET_2_6);
+ properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ properties.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), *output,
+
static_cast<int64_t>(existing_values.size()),
+ properties.build()));
+}
+
void write_iceberg_idless_nested_struct_parquet_file(
const std::string& file_path, const std::vector<int32_t>& ids,
const std::vector<int32_t>& values, const std::vector<bool>&
parent_nulls,
@@ -736,6 +794,81 @@ Status create_single_struct_tuple_descriptor(ObjectPool*
object_pool,
return Status::OK();
}
+// Slots `id INT` (field 0) and, unless `payload_children` is empty, `payload
STRUCT<...>`
+// (field 1) with INT children named by `payload_children`, in either order.
+Status create_id_and_payload_tuple_descriptor(ObjectPool* object_pool, bool
payload_first,
+ const std::vector<std::string>&
payload_children,
+ DescriptorTbl** descriptor_table,
+ const TupleDescriptor**
tuple_descriptor) {
+ TDescriptorTable thrift_table;
+ TTableDescriptor table_descriptor;
+ table_descriptor.__set_id(0);
+ table_descriptor.__set_tableType(TTableType::OLAP_TABLE);
+ table_descriptor.__set_numCols(payload_children.empty() ? 1 : 2);
+ table_descriptor.__set_numClusteringCols(0);
+ thrift_table.__set_tableDescriptors({table_descriptor});
+
+ TTypeNode int_node;
+ int_node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType int_scalar;
+ int_scalar.__set_type(TPrimitiveType::INT);
+ int_node.__set_scalar_type(int_scalar);
+ TTypeDesc id_type;
+ id_type.__set_types({int_node});
+
+ std::vector<TStructField> payload_fields;
+ std::vector<TTypeNode> payload_nodes(1);
+ for (const auto& child_name : payload_children) {
+ TStructField field;
+ field.__set_name(child_name);
+ field.__set_contains_null(true);
+ payload_fields.push_back(std::move(field));
+ payload_nodes.push_back(int_node);
+ }
+ payload_nodes[0].__set_type(TTypeNodeType::STRUCT);
+ payload_nodes[0].__set_struct_fields(payload_fields);
+ TTypeDesc payload_type;
+ payload_type.__set_types(payload_nodes);
+
+ std::vector<TSlotDescriptor> slots;
+ auto add_slot = [&slots](const std::string& name, int32_t field_id, const
TTypeDesc& type) {
+ const auto slot_index = static_cast<int32_t>(slots.size());
+ TSlotDescriptor slot;
+ slot.__set_id(slot_index);
+ slot.__set_parent(0);
+ slot.__set_col_unique_id(field_id);
+ slot.__set_slotType(type);
+ slot.__set_columnPos(slot_index);
+ slot.__set_byteOffset(0);
+ slot.__set_nullIndicatorByte(0);
+ slot.__set_nullIndicatorBit(-1);
+ slot.__set_colName(name);
+ slot.__set_slotIdx(slot_index);
+ slot.__set_isMaterialized(true);
+ slots.emplace_back(std::move(slot));
+ };
+ if (payload_children.empty()) {
+ add_slot("id", 0, id_type);
+ } else if (payload_first) {
+ add_slot("payload", 1, payload_type);
+ add_slot("id", 0, id_type);
+ } else {
+ add_slot("id", 0, id_type);
+ add_slot("payload", 1, payload_type);
+ }
+ thrift_table.__set_slotDescriptors(std::move(slots));
+
+ TTupleDescriptor thrift_tuple;
+ thrift_tuple.__set_id(0);
+ thrift_tuple.__set_byteSize(32);
+ thrift_tuple.__set_numNullBytes(0);
+ thrift_tuple.__set_tableId(0);
+ thrift_table.__set_tupleDescriptors({thrift_tuple});
+ RETURN_IF_ERROR(DescriptorTbl::create(object_pool, thrift_table,
descriptor_table));
+ *tuple_descriptor = (*descriptor_table)->get_tuple_descriptor(0);
+ return Status::OK();
+}
+
schema::external::TSchema make_nested_default_schema() {
TColumnType int_type;
int_type.__set_type(TPrimitiveType::INT);
@@ -3120,6 +3253,619 @@ TEST_F(IcebergReaderTest,
nested_equality_delete_filters_current_and_dropped_fie
}
}
+// A nested equality key is read through its top-level struct. When the query
also projects that
+// struct (any SELECT *), the projected column and the hidden key carrier name
the same physical
+// root, and both must still be filled. Example: data (1,{10}), (2,{20}),
(3,{30}) with an equality
+// delete on payload.existing = 20 must return (1,{10}) and (3,{30}).
+//
NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
+// Scan conjunct `id != 1` over the id slot of `tuple_descriptor`. With
another column to read, it
+// makes the V1 readers materialize lazily and filter part of the batch before
the lazy columns.
+struct IdFilter {
+ VExprContextSPtrs conjuncts;
+ std::unordered_map<int, VExprContextSPtrs> slot_id_to_conjuncts;
+};
+
+Status create_id_not_one_filter(const DescriptorTbl* descriptor_table,
+ const TupleDescriptor* tuple_descriptor,
+ RuntimeState* runtime_state, IdFilter* filter)
{
+ runtime_state->set_desc_tbl(descriptor_table);
+ const auto& slots = tuple_descriptor->slots();
+ const auto id_slot = std::ranges::find_if(
+ slots, [](const SlotDescriptor* slot) { return slot->col_name() ==
"id"; });
+ DORIS_CHECK(id_slot != slots.end());
+
+ TFunctionName function_name;
+ function_name.__set_db_name("");
+ function_name.__set_function_name("ne");
+ TFunction function;
+ function.__set_name(function_name);
+ function.__set_binary_type(TFunctionBinaryType::BUILTIN);
+ function.__set_arg_types({create_type_desc(TYPE_INT),
create_type_desc(TYPE_INT)});
+ function.__set_ret_type(create_type_desc(TYPE_BOOLEAN));
+ function.__set_has_var_args(false);
+ TExprNode predicate_node;
+ predicate_node.__set_type(create_type_desc(TYPE_BOOLEAN));
+ predicate_node.__set_node_type(TExprNodeType::BINARY_PRED);
+ predicate_node.__set_opcode(TExprOpcode::NE);
+ predicate_node.__set_fn(function);
+ predicate_node.__set_num_children(2);
+ predicate_node.__set_is_nullable(true);
+ VExprSPtr root = VectorizedFnCall::create_shared(predicate_node);
+ root->add_child(VSlotRef::create_shared(*id_slot));
+ TExprNode literal_node;
+ literal_node.__set_node_type(TExprNodeType::INT_LITERAL);
+ literal_node.__set_type(create_type_desc(TYPE_INT));
+ TIntLiteral int_literal;
+ int_literal.__set_value(1);
+ literal_node.__set_int_literal(int_literal);
+ literal_node.__set_is_nullable(false);
+ root->add_child(VLiteral::create_shared(literal_node));
+
+ auto context = VExprContext::create_shared(root);
+ RowDescriptor
row_descriptor(const_cast<TupleDescriptor*>(tuple_descriptor));
+ RETURN_IF_ERROR(context->prepare(runtime_state, row_descriptor));
+ RETURN_IF_ERROR(context->open(runtime_state));
+ filter->conjuncts = {context};
+ filter->slot_id_to_conjuncts = {{(*id_slot)->id(), {context}}};
+ return Status::OK();
+}
+
+void
check_nested_equality_key_keeps_projected_root_struct(TFileFormatType::type
file_format,
+ bool payload_first,
+ const
cctz::time_zone* timezone,
+ FileMetaCache*
meta_cache,
+ bool
filter_out_first_id = false) {
+ {
+ const bool is_parquet = file_format == TFileFormatType::FORMAT_PARQUET;
+ const std::string format_name = is_parquet ? "parquet" : "orc";
+ const std::string order_name = payload_first ? "payload_first" :
"id_first";
+ const std::string filter_name = filter_out_first_id ? "_filtered" : "";
+ SCOPED_TRACE(format_name + " " + order_name + filter_name);
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ ("doris_v1_nested_equality_projected_root_" +
format_name + "_" +
+ order_name + filter_name);
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto data_file = (test_dir / ("data." + format_name)).string();
+ const auto delete_file = (test_dir / ("equality-delete." +
format_name)).string();
+ if (is_parquet) {
+ write_iceberg_id_and_nested_struct_parquet_file(data_file, {1, 2,
3}, {10, 20, 30},
+ {false, false,
false});
+ write_iceberg_nested_struct_parquet_file(delete_file, {20},
{false});
+ } else {
+ write_iceberg_id_and_nested_struct_orc_file(data_file, {1, 2, 3},
{10, 20, 30},
+ {false, false, false});
+ write_iceberg_nested_struct_orc_file(delete_file, {20}, {false});
+ }
+
+ schema::external::TStructField current_root;
+ current_root.__set_fields(
+ {make_external_int_field("id", 0, std::nullopt),
+ make_external_struct_field(
+ "payload", 1, {make_external_int_field("existing", 2,
std::nullopt)})});
+ schema::external::TSchema current_schema;
+ current_schema.__set_schema_id(100);
+ current_schema.__set_root_field(std::move(current_root));
+
+ TFileScanRangeParams scan_params;
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_file_type(TFileType::FILE_LOCAL);
+ scan_params.__set_format_type(file_format);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({current_schema});
+
+ TIcebergDeleteFileDesc delete_descriptor;
+ delete_descriptor.__set_content(2);
+ delete_descriptor.__set_path(delete_file);
+ delete_descriptor.__set_field_ids({2});
+ delete_descriptor.__set_file_format(file_format);
+ TIcebergFileDesc iceberg_descriptor;
+ iceberg_descriptor.__set_format_version(2);
+ iceberg_descriptor.__set_original_file_path(data_file);
+ iceberg_descriptor.__set_delete_files({delete_descriptor});
+ TTableFormatFileDesc table_format_descriptor;
+
table_format_descriptor.__set_iceberg_params(std::move(iceberg_descriptor));
+
+ TFileRangeDesc scan_range;
+ scan_range.__set_fs_name("");
+ scan_range.__set_path(data_file);
+ scan_range.__set_start_offset(0);
+
scan_range.__set_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+
scan_range.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+
scan_range.__set_table_format_params(std::move(table_format_descriptor));
+
+ ObjectPool object_pool;
+ DescriptorTbl* descriptor_table = nullptr;
+ const TupleDescriptor* tuple_descriptor = nullptr;
+ ASSERT_TRUE(create_id_and_payload_tuple_descriptor(&object_pool,
payload_first,
+ {"existing"},
&descriptor_table,
+ &tuple_descriptor)
+ .ok());
+ ASSERT_NE(tuple_descriptor, nullptr);
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
+ runtime_state.set_timezone("UTC");
+ IdFilter filter;
+ if (filter_out_first_id) {
+ ASSERT_TRUE(create_id_not_one_filter(descriptor_table,
tuple_descriptor, &runtime_state,
+ &filter)
+ .ok());
+ }
+ io::IOContext io_ctx;
+ ShardedKVCache kv_cache(8);
+ const uint32_t id_position = payload_first ? 1 : 0;
+ const uint32_t payload_position = payload_first ? 0 : 1;
+ std::vector<ColumnDescriptor> column_descriptors(2);
+ column_descriptors[id_position].name = "id";
+ column_descriptors[payload_position].name = "payload";
+ std::unordered_map<std::string, uint32_t> block_positions = {{"id",
id_position},
+
{"payload", payload_position}};
+ const auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+ const auto payload_type =
make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {make_nullable(std::make_shared<DataTypeInt32>())},
+ Strings {"existing"}));
+ Block block;
+ if (payload_first) {
+ block.insert({payload_type->create_column(), payload_type,
"payload"});
+ block.insert({int_type->create_column(), int_type, "id"});
+ } else {
+ block.insert({int_type->create_column(), int_type, "id"});
+ block.insert({payload_type->create_column(), payload_type,
"payload"});
+ }
+ size_t read_rows = 0;
+ bool eof = false;
+
+ Status status;
+ if (is_parquet) {
+ IcebergParquetReader reader(&kv_cache, &profile, scan_params,
scan_range, 1024,
+ timezone, &io_ctx, &runtime_state,
meta_cache);
+ io::FileReaderSPtr file_reader;
+ ASSERT_TRUE(io::global_local_filesystem()->open_file(data_file,
&file_reader).ok());
+ reader.set_file_reader(file_reader);
+ ParquetInitContext context;
+ context.column_descs = &column_descriptors;
+ context.col_name_to_block_idx = &block_positions;
+ context.tuple_descriptor = tuple_descriptor;
+ context.params = &scan_params;
+ context.range = &scan_range;
+ context.conjuncts = &filter.conjuncts;
+ context.slot_id_to_filter_conjuncts = &filter.slot_id_to_conjuncts;
+ status = reader.init_reader(&context);
+ ASSERT_TRUE(status.ok()) << status;
+ status = reader.get_next_block(&block, &read_rows, &eof);
+ } else {
+ IcebergOrcReader reader(&kv_cache, &profile, &runtime_state,
scan_params, scan_range,
+ 1024, "UTC", &io_ctx, meta_cache);
+ OrcInitContext context;
+ context.column_descs = &column_descriptors;
+ context.col_name_to_block_idx = &block_positions;
+ context.tuple_descriptor = tuple_descriptor;
+ context.params = &scan_params;
+ context.range = &scan_range;
+ context.conjuncts = &filter.conjuncts;
+ context.slot_id_to_filter_conjuncts = &filter.slot_id_to_conjuncts;
+ status = reader.init_reader(&context);
+ ASSERT_TRUE(status.ok()) << status;
+ status = reader.get_next_block(&block, &read_rows, &eof);
+ }
+
+ // The equality delete removes id 2; the filter, when set, removes id
1.
+ const std::vector<std::string> expected_ids = filter_out_first_id
+ ?
std::vector<std::string> {"3"}
+ :
std::vector<std::string> {"1", "3"};
+ const std::vector<std::string> expected_payloads =
+ filter_out_first_id
+ ? std::vector<std::string> {"{\"existing\":30}"}
+ : std::vector<std::string> {"{\"existing\":10}",
"{\"existing\":30}"};
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(read_rows, expected_ids.size());
+ ASSERT_EQ(block.rows(), expected_ids.size());
+ const auto& id_column = *block.get_by_position(id_position).column;
+ const auto& payload_column =
*block.get_by_position(payload_position).column;
+ ASSERT_EQ(id_column.size(), expected_ids.size());
+ ASSERT_EQ(payload_column.size(), expected_ids.size());
+ for (size_t row = 0; row < expected_ids.size(); ++row) {
+ EXPECT_EQ(int_type->to_string(id_column, row), expected_ids[row]);
+ EXPECT_EQ(payload_type->to_string(payload_column, row),
expected_payloads[row]);
+ }
+ std::filesystem::remove_all(test_dir);
+ }
+}
+
+TEST_F(IcebergReaderTest,
v1_parquet_nested_equality_key_keeps_projected_root_struct) {
+
check_nested_equality_key_keeps_projected_root_struct(TFileFormatType::FORMAT_PARQUET,
false,
+ &timezone_obj,
cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_parquet_nested_equality_key_keeps_projected_root_struct_listed_first) {
+
check_nested_equality_key_keeps_projected_root_struct(TFileFormatType::FORMAT_PARQUET,
true,
+ &timezone_obj,
cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_orc_nested_equality_key_keeps_projected_root_struct) {
+
check_nested_equality_key_keeps_projected_root_struct(TFileFormatType::FORMAT_ORC,
false,
+ &timezone_obj,
cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_orc_nested_equality_key_keeps_projected_root_struct_listed_first) {
+
check_nested_equality_key_keeps_projected_root_struct(TFileFormatType::FORMAT_ORC,
true,
+ &timezone_obj,
cache.get());
+}
+
+// A filter on id makes the reader read payload lazily; the key taken from the
projected payload
+// must follow the rows that survive the filter.
+TEST_F(IcebergReaderTest,
v1_parquet_nested_equality_key_from_projected_root_after_lazy_filter) {
+
check_nested_equality_key_keeps_projected_root_struct(TFileFormatType::FORMAT_PARQUET,
false,
+ &timezone_obj,
cache.get(), true);
+}
+
+TEST_F(IcebergReaderTest,
+
v1_parquet_nested_equality_key_from_projected_root_listed_first_after_lazy_filter)
{
+
check_nested_equality_key_keeps_projected_root_struct(TFileFormatType::FORMAT_PARQUET,
true,
+ &timezone_obj,
cache.get(), true);
+}
+
+// Data `id INT, payload STRUCT<existing INT, k INT>` holds (1,{10,100}),
(2,{20,200}) and
+// (3,{30,300}). An equality delete on `delete_field_ids` (2 =
payload.existing, 3 = payload.k)
+// with the row {20,200} removes id 2. `projected_children` lists the payload
children the query
+// projects; empty means the query does not project payload.
+//
NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
+void check_two_child_root_equality_delete(
+ const std::string& case_name, const std::vector<std::string>&
projected_children,
+ const std::vector<int32_t>& delete_field_ids,
+ const std::vector<std::string>& expected_payloads, const
cctz::time_zone* timezone,
+ FileMetaCache* meta_cache, bool filter_out_first_id = false, bool
legacy_plan = false,
+ const std::string& current_k_name = "k", bool existing_has_field_id =
true) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
("doris_v1_two_child_root_" + case_name);
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto data_file = (test_dir / "data.parquet").string();
+ const auto delete_file = (test_dir / "equality-delete.parquet").string();
+ const std::vector<int32_t> ids = {1, 2, 3};
+ write_iceberg_two_child_struct_parquet_file(data_file, &ids, {10, 20, 30},
{100, 200, 300},
+ existing_has_field_id);
+ write_iceberg_two_child_struct_parquet_file(delete_file, nullptr, {20},
{200});
+
+ // The schema keeps payload.k even when the projected payload type lacks
it, as FE's scan
+ // dictionary keeps a dropped equality key.
+ schema::external::TStructField current_root;
+ current_root.__set_fields(
+ {make_external_int_field("id", 0, std::nullopt),
+ make_external_struct_field(
+ "payload", 1,
+ {make_external_int_field("existing", 2, std::nullopt),
+ make_external_int_field(current_k_name, 3,
std::nullopt)})});
+ schema::external::TSchema current_schema;
+ current_schema.__set_schema_id(100);
+ current_schema.__set_root_field(std::move(current_root));
+
+ TFileScanRangeParams scan_params;
+ if (!legacy_plan) {
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ }
+ scan_params.__set_file_type(TFileType::FILE_LOCAL);
+ scan_params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({current_schema});
+
+ TIcebergDeleteFileDesc delete_descriptor;
+ delete_descriptor.__set_content(2);
+ delete_descriptor.__set_path(delete_file);
+ delete_descriptor.__set_field_ids(delete_field_ids);
+ delete_descriptor.__set_file_format(TFileFormatType::FORMAT_PARQUET);
+ TIcebergFileDesc iceberg_descriptor;
+ iceberg_descriptor.__set_format_version(2);
+ iceberg_descriptor.__set_original_file_path(data_file);
+ iceberg_descriptor.__set_delete_files({delete_descriptor});
+ TTableFormatFileDesc table_format_descriptor;
+
table_format_descriptor.__set_iceberg_params(std::move(iceberg_descriptor));
+
+ TFileRangeDesc scan_range;
+ scan_range.__set_fs_name("");
+ scan_range.__set_path(data_file);
+ scan_range.__set_start_offset(0);
+
scan_range.__set_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+
scan_range.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+ scan_range.__set_table_format_params(std::move(table_format_descriptor));
+
+ ObjectPool object_pool;
+ DescriptorTbl* descriptor_table = nullptr;
+ const TupleDescriptor* tuple_descriptor = nullptr;
+ ASSERT_TRUE(create_id_and_payload_tuple_descriptor(&object_pool, false,
projected_children,
+ &descriptor_table,
&tuple_descriptor)
+ .ok());
+ ASSERT_NE(tuple_descriptor, nullptr);
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
+ runtime_state.set_timezone("UTC");
+ IdFilter filter;
+ if (filter_out_first_id) {
+ ASSERT_TRUE(create_id_not_one_filter(descriptor_table,
tuple_descriptor, &runtime_state,
+ &filter)
+ .ok());
+ }
+ io::IOContext io_ctx;
+ ShardedKVCache kv_cache(8);
+ const bool payload_projected = !projected_children.empty();
+ std::vector<ColumnDescriptor> column_descriptors(payload_projected ? 2 :
1);
+ column_descriptors[0].name = "id";
+ std::unordered_map<std::string, uint32_t> block_positions = {{"id", 0}};
+ const auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+ Block block;
+ block.insert({int_type->create_column(), int_type, "id"});
+ DataTypePtr payload_type;
+ if (payload_projected) {
+ column_descriptors[1].name = "payload";
+ block_positions.emplace("payload", 1);
+ payload_type = make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes(projected_children.size(), int_type),
projected_children));
+ block.insert({payload_type->create_column(), payload_type, "payload"});
+ }
+
+ IcebergParquetReader reader(&kv_cache, &profile, scan_params, scan_range,
1024, timezone,
+ &io_ctx, &runtime_state, meta_cache);
+ io::FileReaderSPtr file_reader;
+ ASSERT_TRUE(io::global_local_filesystem()->open_file(data_file,
&file_reader).ok());
+ reader.set_file_reader(file_reader);
+ ParquetInitContext context;
+ context.column_descs = &column_descriptors;
+ context.col_name_to_block_idx = &block_positions;
+ context.tuple_descriptor = tuple_descriptor;
+ context.params = &scan_params;
+ context.range = &scan_range;
+ context.conjuncts = &filter.conjuncts;
+ context.slot_id_to_filter_conjuncts = &filter.slot_id_to_conjuncts;
+ auto status = reader.init_reader(&context);
+ ASSERT_TRUE(status.ok()) << status;
+ size_t read_rows = 0;
+ bool eof = false;
+ status = reader.get_next_block(&block, &read_rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+
+ // The equality delete removes id 2; the filter, when set, removes id 1.
+ const std::vector<std::string> expected_ids = filter_out_first_id
+ ?
std::vector<std::string> {"3"}
+ :
std::vector<std::string> {"1", "3"};
+ ASSERT_EQ(read_rows, expected_ids.size());
+ ASSERT_EQ(block.rows(), expected_ids.size());
+ const auto& id_column = *block.get_by_position(0).column;
+ ASSERT_EQ(id_column.size(), expected_ids.size());
+ for (size_t row = 0; row < expected_ids.size(); ++row) {
+ EXPECT_EQ(int_type->to_string(id_column, row), expected_ids[row]);
+ }
+ if (payload_projected) {
+ const auto& payload_column = *block.get_by_position(1).column;
+ ASSERT_EQ(payload_column.size(), expected_ids.size());
+ ASSERT_EQ(expected_payloads.size(), expected_ids.size());
+ for (size_t row = 0; row < expected_ids.size(); ++row) {
+ EXPECT_EQ(payload_type->to_string(payload_column, row),
expected_payloads[row]);
+ }
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
+// Both keys sit under payload, which the query does not project: one carrier
reads payload and
+// the other key is extracted from it.
+TEST_F(IcebergReaderTest, v1_parquet_equality_keys_share_unprojected_root) {
+ check_two_child_root_equality_delete("shared_carrier", {}, {2, 3}, {},
&timezone_obj,
+ cache.get());
+}
+
+// Both keys are taken from the projected payload column.
+TEST_F(IcebergReaderTest, v1_parquet_equality_keys_come_from_projected_root) {
+ check_two_child_root_equality_delete(
+ "projected_root", {"existing", "k"}, {2, 3},
+ {"{\"existing\":10, \"k\":100}", "{\"existing\":30, \"k\":300}"},
&timezone_obj,
+ cache.get());
+}
+
+// The projected payload lacks k, so k's carrier reads the payload file column
a second time.
+TEST_F(IcebergReaderTest,
v1_parquet_equality_key_outside_projected_struct_rereads_root) {
+ check_two_child_root_equality_delete("reread_root", {"existing"}, {3},
+ {"{\"existing\":10}",
"{\"existing\":30}"}, &timezone_obj,
+ cache.get());
+}
+
+// The same two layouts under a filter on id, which makes the carrier and
payload lazy columns: the
+// keys extracted from them must follow the rows that survive the filter.
+TEST_F(IcebergReaderTest,
v1_parquet_equality_keys_share_unprojected_root_after_lazy_filter) {
+ check_two_child_root_equality_delete("shared_carrier_filtered", {}, {2,
3}, {}, &timezone_obj,
+ cache.get(), true);
+}
+
+TEST_F(IcebergReaderTest,
v1_parquet_equality_keys_come_from_projected_root_after_lazy_filter) {
+ check_two_child_root_equality_delete("projected_root_filtered",
{"existing", "k"}, {2, 3},
+ {"{\"existing\":30, \"k\":300}"},
&timezone_obj,
+ cache.get(), true);
+}
+
+// The data file's payload holds existing without a field ID and k#3, which
the table has renamed
+// to renamed_k. A plan without the Iceberg scan semantics marker maps that
payload level by name,
+// so the projected payload reads renamed_k as NULL, while the hidden key
finds k#3 by field ID: the
+// key must come from its own carrier, or the delete on renamed_k = 200 keeps
id 2. With the marker
+// both sides bind k#3 by field ID, and the key comes from the projected
payload.
+TEST_F(IcebergReaderTest,
+
v1_parquet_legacy_plan_key_rereads_root_when_projected_child_matches_by_name) {
+ check_two_child_root_equality_delete(
+ "legacy_plan_renamed_key", {"existing", "renamed_k"}, {3},
+ {"{\"existing\":10, \"renamed_k\":null}", "{\"existing\":30,
\"renamed_k\":null}"},
+ &timezone_obj, cache.get(), false, true, "renamed_k", false);
+ check_two_child_root_equality_delete(
+ "marked_plan_renamed_key", {"existing", "renamed_k"}, {3},
+ {"{\"existing\":null, \"renamed_k\":100}", "{\"existing\":null,
\"renamed_k\":300}"},
+ &timezone_obj, cache.get(), false, false, "renamed_k", false);
+}
+
+// Data `id INT, v INT, unused INT` holds (1,10,0), (2,20,0) and (3,30,0). The
table also has
+// added_column (field 3) with initial default 7, which the file predates. An
equality delete on
+// (id, v, added_column) with the row (2, 20, 7) removes id 2, so the key that
the reader has no
+// column for must still carry its default; the filter, when set, removes id 1.
+//
NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
+void check_missing_equality_key_delete(TFileFormatType::type file_format, bool
filter_out_first_id,
+ const cctz::time_zone* timezone,
FileMetaCache* meta_cache) {
+ const bool is_parquet = file_format == TFileFormatType::FORMAT_PARQUET;
+ const std::string format_name = is_parquet ? "parquet" : "orc";
+ const std::string filter_name = filter_out_first_id ? "_filtered" : "";
+ SCOPED_TRACE(format_name + filter_name);
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ ("doris_v1_missing_equality_key_" + format_name +
filter_name);
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto data_file = (test_dir / ("data." + format_name)).string();
+ const auto delete_file = (test_dir / ("equality-delete." +
format_name)).string();
+ if (is_parquet) {
+ write_iceberg_three_int_parquet_file(data_file, "id", 0, {1, 2, 3},
"v", 1, {10, 20, 30},
+ "unused", 2, {0, 0, 0});
+ write_iceberg_three_int_parquet_file(delete_file, "id", 0, {2}, "v",
1, {20},
+ "added_column", 3, {7});
+ } else {
+ write_iceberg_three_int_orc_file(data_file, "id", 0, {1, 2, 3}, "v",
1, {10, 20, 30},
+ "unused", 2, {0, 0, 0});
+ write_iceberg_three_int_orc_file(delete_file, "id", 0, {2}, "v", 1,
{20}, "added_column", 3,
+ {7});
+ }
+
+ schema::external::TStructField current_root;
+ current_root.__set_fields({make_external_int_field("id", 0, std::nullopt),
+ make_external_int_field("v", 1, std::nullopt),
+ make_external_int_field("unused", 2,
std::nullopt),
+ make_external_int_field("added_column", 3,
"7")});
+ schema::external::TSchema current_schema;
+ current_schema.__set_schema_id(100);
+ current_schema.__set_root_field(std::move(current_root));
+
+ TFileScanRangeParams scan_params;
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_file_type(TFileType::FILE_LOCAL);
+ scan_params.__set_format_type(file_format);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({current_schema});
+
+ TIcebergDeleteFileDesc delete_descriptor;
+ delete_descriptor.__set_content(2);
+ delete_descriptor.__set_path(delete_file);
+ delete_descriptor.__set_field_ids({0, 1, 3});
+ delete_descriptor.__set_file_format(file_format);
+ TIcebergFileDesc iceberg_descriptor;
+ iceberg_descriptor.__set_format_version(2);
+ iceberg_descriptor.__set_original_file_path(data_file);
+ iceberg_descriptor.__set_delete_files({delete_descriptor});
+ TTableFormatFileDesc table_format_descriptor;
+
table_format_descriptor.__set_iceberg_params(std::move(iceberg_descriptor));
+
+ TFileRangeDesc scan_range;
+ scan_range.__set_fs_name("");
+ scan_range.__set_path(data_file);
+ scan_range.__set_start_offset(0);
+
scan_range.__set_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+
scan_range.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+ scan_range.__set_table_format_params(std::move(table_format_descriptor));
+
+ ObjectPool object_pool;
+ DescriptorTbl* descriptor_table = nullptr;
+ const TupleDescriptor* tuple_descriptor = nullptr;
+ ASSERT_TRUE(create_two_int_tuple_descriptor(&object_pool, {{{"id", 0},
{"v", 1}}},
+ &descriptor_table,
&tuple_descriptor)
+ .ok());
+ ASSERT_NE(tuple_descriptor, nullptr);
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
+ runtime_state.set_timezone("UTC");
+ IdFilter filter;
+ if (filter_out_first_id) {
+ ASSERT_TRUE(create_id_not_one_filter(descriptor_table,
tuple_descriptor, &runtime_state,
+ &filter)
+ .ok());
+ }
+ io::IOContext io_ctx;
+ ShardedKVCache kv_cache(8);
+ std::vector<ColumnDescriptor> column_descriptors(2);
+ column_descriptors[0].name = "id";
+ column_descriptors[1].name = "v";
+ std::unordered_map<std::string, uint32_t> block_positions = {{"id", 0},
{"v", 1}};
+ const auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+ Block block;
+ block.insert({int_type->create_column(), int_type, "id"});
+ block.insert({int_type->create_column(), int_type, "v"});
+ size_t read_rows = 0;
+ bool eof = false;
+
+ Status status;
+ if (is_parquet) {
+ IcebergParquetReader reader(&kv_cache, &profile, scan_params,
scan_range, 1024, timezone,
+ &io_ctx, &runtime_state, meta_cache);
+ io::FileReaderSPtr file_reader;
+ ASSERT_TRUE(io::global_local_filesystem()->open_file(data_file,
&file_reader).ok());
+ reader.set_file_reader(file_reader);
+ ParquetInitContext context;
+ context.column_descs = &column_descriptors;
+ context.col_name_to_block_idx = &block_positions;
+ context.tuple_descriptor = tuple_descriptor;
+ context.params = &scan_params;
+ context.range = &scan_range;
+ context.conjuncts = &filter.conjuncts;
+ context.slot_id_to_filter_conjuncts = &filter.slot_id_to_conjuncts;
+ status = reader.init_reader(&context);
+ ASSERT_TRUE(status.ok()) << status;
+ status = reader.get_next_block(&block, &read_rows, &eof);
+ } else {
+ IcebergOrcReader reader(&kv_cache, &profile, &runtime_state,
scan_params, scan_range, 1024,
+ "UTC", &io_ctx, meta_cache);
+ OrcInitContext context;
+ context.column_descs = &column_descriptors;
+ context.col_name_to_block_idx = &block_positions;
+ context.tuple_descriptor = tuple_descriptor;
+ context.params = &scan_params;
+ context.range = &scan_range;
+ context.conjuncts = &filter.conjuncts;
+ context.slot_id_to_filter_conjuncts = &filter.slot_id_to_conjuncts;
+ status = reader.init_reader(&context);
+ ASSERT_TRUE(status.ok()) << status;
+ status = reader.get_next_block(&block, &read_rows, &eof);
+ }
+
+ ASSERT_TRUE(status.ok()) << status;
+ const std::vector<std::pair<std::string, std::string>> expected_rows =
+ filter_out_first_id
+ ? std::vector<std::pair<std::string, std::string>> {{"3",
"30"}}
+ : std::vector<std::pair<std::string, std::string>> {{"1",
"10"}, {"3", "30"}};
+ ASSERT_EQ(read_rows, expected_rows.size());
+ ASSERT_EQ(block.columns(), 2);
+ ASSERT_EQ(block.rows(), expected_rows.size());
+ for (size_t row = 0; row < expected_rows.size(); ++row) {
+ EXPECT_EQ(int_type->to_string(*block.get_by_position(0).column, row),
+ expected_rows[row].first);
+ EXPECT_EQ(int_type->to_string(*block.get_by_position(1).column, row),
+ expected_rows[row].second);
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
+TEST_F(IcebergReaderTest,
v1_parquet_missing_equality_key_matches_initial_default) {
+ check_missing_equality_key_delete(TFileFormatType::FORMAT_PARQUET, false,
&timezone_obj,
+ cache.get());
+}
+
+TEST_F(IcebergReaderTest,
+
v1_parquet_missing_equality_key_matches_initial_default_after_lazy_filter) {
+ check_missing_equality_key_delete(TFileFormatType::FORMAT_PARQUET, true,
&timezone_obj,
+ cache.get());
+}
+
+TEST_F(IcebergReaderTest, v1_orc_missing_equality_key_matches_initial_default)
{
+ check_missing_equality_key_delete(TFileFormatType::FORMAT_ORC, false,
&timezone_obj,
+ cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_orc_missing_equality_key_matches_initial_default_after_lazy_filter) {
+ check_missing_equality_key_delete(TFileFormatType::FORMAT_ORC, true,
&timezone_obj,
+ cache.get());
+}
+
// Keep the shared V1 Parquet/ORC reader setup together so both
rolling-upgrade scan-semantics
// revisions exercise the same ID-less nested-name resolution path.
//
NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
