github-actions[bot] commented on code in PR #66538:
URL: https://github.com/apache/doris/pull/66538#discussion_r3861936118
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -183,13 +720,20 @@ Status IcebergTableReader::get_next_block_inner(Block*
block, size_t* read_rows,
RETURN_IF_ERROR(_expand_block_if_need(block));
RETURN_IF_ERROR(_file_format_reader->get_next_block(block, read_rows,
eof));
+ RETURN_IF_ERROR(_materialize_missing_table_columns(block));
+ RETURN_IF_ERROR(_materialize_missing_equality_delete_columns(block,
*read_rows));
Review Comment:
[P1] Size missing equality carriers from the filtered block
Parquet's non-lazy `can_filter_all` path clears every projected column and
returns before resetting `read_rows`, so this can repeat a missing equality key
to the physical batch size while `Block::rows()` is zero. A single
BINARY/FIXED/UUID key then makes `SimpleEqualityDelete` write matches past its
zero-length filter; a multi-column equality delete similarly hashes the
nonempty carrier into a zero-length row hash buffer. Both happen before the
later zero-row cleanup. Materialize these carriers with `block->rows()`, as
`_materialize_missing_table_columns()` already does, or reset `read_rows` on
every filtered return.
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -487,134 +1612,155 @@ Status IcebergParquetReader::init_reader(
parquet_reader->set_row_lineage_columns(_row_lineage_columns);
}
- auto column_id_result = _create_column_ids(_data_file_field_desc,
tuple_descriptor);
- auto& column_ids = column_id_result.column_ids;
- const auto& filter_column_ids = column_id_result.filter_column_ids;
-
- RETURN_IF_ERROR(init_row_filters());
_all_required_col_names = file_col_names;
+ for (const auto* slot : tuple_descriptor->slots()) {
+ _id_to_block_column_name.emplace(slot->col_unique_id(),
slot->col_name());
+ _required_column_types.emplace(slot->col_name(), slot->type());
+ }
+ RETURN_IF_ERROR(init_row_filters());
if (!_params.__isset.history_schema_info ||
_params.history_schema_info.empty()) [[unlikely]] {
RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_name(
tuple_descriptor, *_data_file_field_desc,
table_info_node_ptr));
} else {
- std::set<std::string> read_col_name_set(file_col_names.begin(),
file_col_names.end());
+
RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id_with_name_mapping(
+ _params.history_schema_info.front().root_field,
*_data_file_field_desc,
+ table_info_node_ptr,
supports_iceberg_scan_semantics_v2(&_params)));
+ }
+ RETURN_IF_ERROR(_validate_projected_missing_required_fields());
- bool exist_field_id = true;
- for (int idx = 0; idx < _data_file_field_desc->size(); idx++) {
- if (_data_file_field_desc->get_column(idx)->field_id == -1) {
- // the data file may be from hive table migrated to iceberg,
field id is missing
- exist_field_id = false;
- break;
- }
- }
- const auto& table_schema =
_params.history_schema_info.front().root_field;
-
- table_info_node_ptr =
std::make_shared<TableSchemaChangeHelper::StructNode>();
- if (exist_field_id) {
- // id -> table column name. columns that need read data file.
- std::unordered_map<int, std::shared_ptr<schema::external::TField>>
id_to_table_field;
- for (const auto& table_field : table_schema.fields) {
- auto field = table_field.field_ptr;
- DCHECK(field->__isset.name);
- if (!read_col_name_set.contains(field->name)) {
- continue;
- }
- id_to_table_field.emplace(field->id, field);
- }
+ auto column_id_result =
+ _create_column_ids(_data_file_field_desc, tuple_descriptor,
table_info_node_ptr);
+ auto& column_ids = column_id_result.column_ids;
+ const auto& filter_column_ids = column_id_result.filter_column_ids;
- for (int idx = 0; idx < _data_file_field_desc->size(); idx++) {
- const auto& data_file_field =
_data_file_field_desc->get_column(idx);
- auto data_file_column_id =
_data_file_field_desc->get_column(idx)->field_id;
-
- if (id_to_table_field.contains(data_file_column_id)) {
- const auto& table_field =
id_to_table_field[data_file_column_id];
-
- std::shared_ptr<TableSchemaChangeHelper::Node> field_node
= nullptr;
- RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id(
- *table_field, *data_file_field, exist_field_id,
field_node));
- table_info_node_ptr->add_children(table_field->name,
data_file_field->name,
- field_node);
-
- _id_to_block_column_name.emplace(data_file_column_id,
table_field->name);
- id_to_table_field.erase(data_file_column_id);
- } else if
(_equality_delete_col_ids.contains(data_file_column_id)) {
- // Columns that need to be read for equality delete.
- const static std::string EQ_DELETE_PRE =
"__equality_delete_column__";
-
- // Construct table column names that avoid duplication
with current table schema.
- // As the columns currently being read may have been
deleted in the latest
- // table structure or have undergone a series of schema
changes...
- std::string table_column_name = EQ_DELETE_PRE +
data_file_field->name;
- table_info_node_ptr->add_children(
- table_column_name, data_file_field->name,
-
std::make_shared<TableSchemaChangeHelper::ConstNode>());
-
- _id_to_block_column_name.emplace(data_file_column_id,
table_column_name);
- _expand_col_names.emplace_back(table_column_name);
- auto expand_data_type =
make_nullable(data_file_field->data_type);
- _expand_columns.emplace_back(
- ColumnWithTypeAndName
{expand_data_type->create_column(),
- expand_data_type,
table_column_name});
-
- _all_required_col_names.emplace_back(table_column_name);
- column_ids.insert(data_file_field->get_column_id());
+ bool all_file_columns_have_field_ids = true;
+ bool any_file_column_has_field_id = false;
+ for (int index = 0; index < _data_file_field_desc->size(); ++index) {
+ const auto* field = _data_file_field_desc->get_column(index);
+ if (field == nullptr) {
+ continue;
+ }
+ if (field->field_id < 0) {
+ all_file_columns_have_field_ids = false;
+ }
+ if (parquet_subtree_has_iceberg_id(*field)) {
+ any_file_column_has_field_id = true;
+ }
+ }
+ const bool use_field_ids = supports_iceberg_scan_semantics_v2(&_params)
+ ? any_file_column_has_field_id
+ : all_file_columns_have_field_ids;
+ std::unordered_map<std::string, std::string> physical_root_sources;
+ std::vector<std::string> new_expand_col_names;
+ DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size());
+ DORIS_CHECK(_expand_col_names.size() == _expand_columns.size());
+ for (size_t index = 0; index < _expand_col_names.size(); ++index) {
+ const std::string old_name = _expand_col_names[index];
+ const int32_t field_id = _expand_col_field_ids[index];
+ const FieldSchema* file_column = nullptr;
+ ParquetEqualityFieldPath file_path;
+ bool complete_file_path = false;
+ if (use_field_ids) {
+ complete_file_path =
find_parquet_equality_field_path_by_id(_data_file_field_desc,
+
field_id, &file_path);
+ if (!complete_file_path &&
supports_iceberg_scan_semantics_v2(&_params)) {
+ const auto table_path = _find_schema_field_path(field_id);
+ if (!table_path.empty()) {
+ complete_file_path =
find_parquet_equality_field_prefix_by_id_path(
+ _data_file_field_desc, table_path, &file_path);
}
}
- for (const auto& [id, table_field] : id_to_table_field) {
- table_info_node_ptr->add_not_exist_children(table_field->name);
+ if (!file_path.fields.empty()) {
+ file_column = file_path.fields.front();
}
} else {
- if (!_equality_delete_col_ids.empty()) [[unlikely]] {
- return Status::InternalError(
- "Can not read missing field id data file when have
equality delete");
- }
- std::map<std::string, size_t> file_column_idx_map;
- for (size_t idx = 0; idx < _data_file_field_desc->size(); idx++) {
-
file_column_idx_map.emplace(_data_file_field_desc->get_column(idx)->name, idx);
+ const auto table_path = _find_schema_field_path(field_id);
+ if (!table_path.empty()) {
+ complete_file_path =
find_parquet_equality_field_prefix_by_name_path(
+ _data_file_field_desc, table_path, old_name,
&file_path);
+ if (!file_path.fields.empty()) {
+ file_column = file_path.fields.front();
+ }
}
+ }
- for (const auto& table_field : table_schema.fields) {
- DCHECK(table_field.__isset.field_ptr);
- DCHECK(table_field.field_ptr->__isset.name);
- const auto& table_column_name = table_field.field_ptr->name;
- if (!read_col_name_set.contains(table_column_name)) {
- continue;
- }
- if (!table_field.field_ptr->__isset.name_mapping ||
- table_field.field_ptr->name_mapping.size() == 0) {
- return Status::DataQualityError(
- "name_mapping must be set when read missing field
id data file.");
- }
- bool have_mapping = false;
- for (const auto& mapped_name :
table_field.field_ptr->name_mapping) {
- if (file_column_idx_map.contains(mapped_name)) {
- std::shared_ptr<TableSchemaChangeHelper::Node>
field_node = nullptr;
- const auto& file_field =
_data_file_field_desc->get_column(
- file_column_idx_map.at(mapped_name));
-
RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id(
- *table_field.field_ptr, *file_field,
exist_field_id, field_node));
- table_info_node_ptr->add_children(table_column_name,
file_field->name,
- field_node);
- have_mapping = true;
- break;
- }
- }
- if (!have_mapping) {
-
table_info_node_ptr->add_not_exist_children(table_column_name);
- }
+ const std::string block_name = _expand_columns[index].name;
+ const DataTypePtr target_leaf_type = _expand_columns[index].type;
+ new_expand_col_names.push_back(block_name);
+ if (file_column == nullptr) {
+ RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id,
block_name,
+
target_leaf_type));
+ continue;
+ }
+
+ std::string source_block_name;
+ std::vector<size_t> source_child_indexes;
+ DataTypePtr source_leaf_type;
+ ColumnPtr missing_value;
+ bool reads_physical_root = false;
+ const auto current_path = _find_schema_field_path(field_id);
+ if (!current_path.empty() && current_path.front()->__isset.id &&
+ _id_to_block_column_name.contains(current_path.front()->id)) {
Review Comment:
[P1] Keep a physical carrier when the projected root lacks the delete field
ID
When a nested equality key is absent by ID from the current schema but its
ancestor struct is still projected, FE merges the historical child into the
schema carrier, so this path is nonempty and the root ID is present here.
`_get_current_schema_equality_delete_path()` then walks the current
`DataTypeStruct` by child name: it either fails initialization when that name
is absent, or silently selects a newly added same-name field with a different
ID and probes deletes against the wrong value. Both Parquet and ORC take this
branch. Reuse the projected root only when it contains the complete key path by
field identity; otherwise retain a shared raw physical-root carrier. This is
distinct from the existing shared-root thread because there are no competing
aliases here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
