github-actions[bot] commented on code in PR #66538:
URL: https://github.com/apache/doris/pull/66538#discussion_r3868468906
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -183,13 +771,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));
+ RETURN_IF_ERROR(_materialize_nested_equality_delete_columns(block));
+ RETURN_IF_ERROR(_validate_required_table_columns(block));
Review Comment:
[P2] Validate required fields after equality deletes
This validates the block before the equality-delete filter below removes
logically invisible rows. If an older file has NULL in a field that is now
required, but an equality delete on another key removes that row, forced V1
returns `Required Iceberg field ... contains NULL`. Position deletes already
run in the physical reader, and V2 likewise applies equality deletes before
required-field validation, so the result currently depends on delete type and
scanner. Apply the V1 equality filter first, then validate only its survivors.
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -487,134 +1680,158 @@ 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 table_path = _find_schema_field_path(field_id);
+ bool uses_projected_root = false;
+ if (!table_path.empty() && table_path.front()->__isset.id &&
+ _id_to_block_column_name.contains(table_path.front()->id)) {
+ RETURN_IF_ERROR(_get_projected_schema_equality_delete_path(
+ field_id, &source_child_indexes, &source_leaf_type,
&uses_projected_root));
+ }
+ if (uses_projected_root) {
+ source_block_name =
_id_to_block_column_name.at(table_path.front()->id);
+ } else {
+ const std::string root_name = to_lower(file_column->name);
+ const auto root_source = physical_root_sources.find(root_name);
+ if (root_source == physical_root_sources.end()) {
+ source_block_name = block_name;
+ physical_root_sources.emplace(root_name, source_block_name);
+ reads_physical_root = true;
+ _expand_columns[index].type =
make_nullable(file_column->data_type);
+ _expand_columns[index].column =
_expand_columns[index].type->create_column();
+ table_info_node_ptr->add_children(
+ block_name, file_column->name,
+ TableSchemaChangeHelper::ConstNode::get_instance());
+ } else {
+ source_block_name = root_source->second;
+ }
+ source_child_indexes = file_path.child_indexes;
+ if (complete_file_path) {
+ source_leaf_type =
make_nullable(file_path.fields.back()->data_type);
+ } else {
+ source_leaf_type = target_leaf_type;
+ RETURN_IF_ERROR(_create_missing_equality_delete_value(
+ field_id, target_leaf_type, file_path.fields.size(),
&missing_value));
}
}
+ if (!reads_physical_root) {
+ _physical_missing_equality_delete_columns.insert(block_name);
+ }
+ _nested_equality_delete_columns.push_back({
+ .field_id = field_id,
+ .block_name = block_name,
+ .source_block_name = source_block_name,
+ .source_leaf_type = source_leaf_type,
+ .leaf_type = target_leaf_type,
+ .child_indexes = std::move(source_child_indexes),
+ .missing_value = std::move(missing_value),
+ .cast_context = nullptr,
+ });
+ RETURN_IF_ERROR(
+
_prepare_nested_equality_delete_column(&_nested_equality_delete_columns.back()));
+ for (uint64_t column_id = file_column->get_column_id();
+ column_id <= file_column->get_max_column_id(); ++column_id) {
+ column_ids.insert(column_id);
+ }
+ _all_required_col_names.push_back(block_name);
}
+ _expand_col_names = std::move(new_expand_col_names);
return parquet_reader->init_reader(
_all_required_col_names, _col_name_to_block_idx, conjuncts,
slot_id_to_predicates,
Review Comment:
[P2] Defer predicates until requiredness validation
V1 forwards these conjuncts into Parquet/ORC, so a predicate such as `WHERE
strengthened_field > 0` can discard a historical explicit NULL before
`_validate_required_table_columns` runs at line 777. That bypasses the new
optional-to-required validation and makes forced V1 accept data that V2
rejects; V2 explicitly marks `reject_null_value` mappings `FINALIZE_ONLY` for
this reason. Keep predicates that touch such mappings above the physical reader
until requiredness has been validated, for both Parquet and ORC.
--
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]
