github-actions[bot] commented on code in PR #65502:
URL: https://github.com/apache/doris/pull/65502#discussion_r3566483227
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -180,6 +181,7 @@ Status IcebergTableReader::prepare_split(const
format::SplitReadOptions& options
_delete_predicates_initialized = false;
_position_delete_rows_storage.clear();
_equality_delete_filters.clear();
+ _split_cache = options.cache;
Review Comment:
This now reads `options.cache` for every Iceberg split, but
`SplitReadOptions::cache` is still a raw pointer with no default initializer.
Several existing split helpers build `SplitReadOptions options;` and call
`prepare_split()` for no-delete Iceberg reads without setting `cache`, so this
assignment performs an indeterminate pointer read before
`_init_delete_predicates()` can decide whether delete-file caching is needed.
Please initialize the field to `nullptr` in `SplitReadOptions` or only copy it
after delete metadata proves a cache is required; the existing
`DORIS_CHECK(_split_cache != nullptr)` at the actual delete-file cache use
sites can still enforce the real cache requirement.
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -646,31 +777,49 @@ Status
IcebergTableReader::_read_equality_delete_file(const TIcebergDeleteFileDe
}
RETURN_IF_ERROR(reader->open(request));
- auto build_equality_delete_block =
- [](const std::vector<format::ColumnDefinition> fields) -> Block {
- Block block;
- for (const auto& field : fields) {
- block.insert({field.type->create_column(), field.type,
field.name});
- }
- return block;
- };
- Block delete_block = build_equality_delete_block(delete_fields);
- MutableBlock mutable_delete_block(std::move(delete_block));
+ MutableBlock mutable_delete_block(build_block(delete_fields));
bool eof = false;
while (!eof) {
- Block block = build_equality_delete_block(delete_fields);
+ Block block = build_block(delete_fields);
size_t read_rows = 0;
RETURN_IF_ERROR(reader->get_block(&block, &read_rows, &eof));
if (read_rows > 0) {
RETURN_IF_ERROR(mutable_delete_block.merge(block));
}
}
RETURN_IF_ERROR(reader->close());
- delete_block = mutable_delete_block.to_block();
- _equality_delete_filters.push_back(
- EqualityDeleteFilter {.field_ids = std::move(delete_field_ids),
- .key_types = std::move(delete_key_types),
- .delete_block = std::move(delete_block)});
+ result->delete_block = mutable_delete_block.to_block();
+ return Status::OK();
+}
+
+Status IcebergTableReader::_read_equality_delete_file(const
TIcebergDeleteFileDesc& delete_file,
+ const
TFileScanRangeParams& scan_params,
+
IcebergDeleteFileIOContext* delete_io_ctx) {
+ if (!delete_file.__isset.field_ids || delete_file.field_ids.empty()) {
+ return Status::InternalError("Iceberg equality delete file is missing
field ids");
+ }
+ std::ostringstream cache_key;
+ cache_key << _delete_file_cache_key("iceberg_v2_equality_delete_",
delete_file.path);
+ for (const auto field_id : delete_file.field_ids) {
+ cache_key << ':' << field_id;
Review Comment:
The equality-delete key still has a structural collision.
`_delete_file_cache_key()` length-encodes `fs_name`, but it appends raw `path`;
this code then appends each field id as `:<id>`. For the same filesystem, path
`s3://bucket/table/delete.parquet` with ids `[1, 2]` and path
`s3://bucket/table/delete.parquet:1` with ids `[2]` both produce the same
suffix `s3://bucket/table/delete.parquet:1:2`, so the shared split cache can
return the wrong parsed delete block. Please length-encode the path and the
field-id vector, or otherwise make the equality-delete cache key structurally
unambiguous end to end.
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -438,38 +440,115 @@ Status
IcebergTableReader::_append_row_position_output_column(format::FileScanRe
return Status::OK();
}
+const format::ColumnDefinition*
IcebergTableReader::_find_equality_delete_data_field(
+ const EqualityDeleteFilter& filter, size_t key_idx) const {
+ DORIS_CHECK(key_idx < filter.field_ids.size());
+ DORIS_CHECK(key_idx < filter.field_names.size());
+ const int field_id = filter.field_ids[key_idx];
+ auto field_it = std::ranges::find_if(_data_reader.file_schema,
+ [field_id](const
format::ColumnDefinition& field) {
+ return
field.has_identifier_field_id() &&
+
field.get_identifier_field_id() == field_id;
+ });
+ if (field_it != _data_reader.file_schema.end() ||
+ mapping_mode() != format::TableColumnMappingMode::BY_NAME) {
+ return field_it == _data_reader.file_schema.end() ? nullptr :
&*field_it;
+ }
+
+ // Equality keys are hidden scan dependencies and need not appear in the
query projection.
+ // Resolve their current name and aliases from the full table schema
supplied by FE, falling
+ // back to the delete-file name when history metadata is unavailable.
Reuse ColumnMapper's
+ // exact BY_NAME rules so case, string identifiers, and aliases on either
side stay consistent.
+ auto table_field = _find_current_table_column_by_field_id(field_id,
filter.key_types[key_idx]);
+ if (!table_field.has_value()) {
+ const auto projected_field = std::ranges::find_if(
+ _projected_columns, [field_id](const format::ColumnDefinition&
field) {
+ return field.has_identifier_field_id() &&
+ field.get_identifier_field_id() == field_id;
+ });
+ if (projected_field != _projected_columns.end()) {
+ // Older scan descriptors and focused unit tests may omit
history_schema_info. Keep the
+ // projected metadata as a compatibility fallback, but never
require projection when
+ // the complete current schema is available.
+ table_field = *projected_field;
+ }
+ }
+ if (!table_field.has_value()) {
+ table_field = format::ColumnDefinition {
+ .identifier = {},
+ .name = filter.field_names[key_idx],
+ .type = filter.key_types[key_idx],
+ };
+ }
+ return format::find_column_by_name(*table_field, _data_reader.file_schema);
+}
+
+std::string IcebergTableReader::_delete_file_cache_key(const char* prefix,
+ const std::string&
path) const {
+ DORIS_CHECK(prefix != nullptr);
+ std::string fs_name;
+ if (_current_task != nullptr && _current_task->data_file != nullptr) {
+ fs_name = _current_task->data_file->fs_name;
+ }
+ // Delete descriptors can reuse the same path text in different filesystem
namespaces. Encode
+ // the filesystem name with its length to avoid ambiguous concatenation
and cross-filesystem
+ // cache hits; scan-level credentials/properties are shared for all splits
in this cache.
+ std::ostringstream key;
+ key << prefix << fs_name.size() << ':' << fs_name << ':' << path;
+ return key.str();
+}
+
+void IcebergTableReader::_append_equality_delete_row_count_carrier(
+ format::FileScanRequest* request) {
+ DORIS_CHECK(request != nullptr);
+ // Columnar readers establish a filter batch's row count from predicate
columns. If all
+ // equality keys are missing, the predicate consists only of NULL literals
and the filter block
+ // would otherwise have zero rows. Read one physical column eagerly as a
row-count carrier;
+ // normal final materialization ignores this hidden dependency.
+ const auto carrier_it = std::ranges::find_if(
+ _data_reader.file_schema, [](const format::ColumnDefinition&
field) {
+ return field.column_type == format::ColumnType::DATA_COLUMN;
+ });
+ DORIS_CHECK(carrier_it != _data_reader.file_schema.end());
+ _append_file_scan_column(request,
format::LocalColumnId(carrier_it->file_local_id()),
+ &request->predicate_columns);
+}
+
Status
IcebergTableReader::_append_equality_delete_predicates(format::FileScanRequest*
request) {
DORIS_CHECK(request != nullptr);
for (const auto& filter : _equality_delete_filters) {
auto delete_predicate =
std::make_shared<EqualityDeletePredicate>(filter.delete_block,
filter.field_ids);
DCHECK_EQ(filter.field_ids.size(), filter.key_types.size());
+ bool has_missing_key = false;
for (size_t idx = 0; idx < filter.field_ids.size(); ++idx) {
- const int field_id = filter.field_ids[idx];
- auto field_it = std::ranges::find_if(
- _data_reader.file_schema, [field_id](const
format::ColumnDefinition& field) {
- return field.has_identifier_field_id() &&
- field.get_identifier_field_id() == field_id;
- });
- if (field_it == _data_reader.file_schema.end()) {
- return Status::InternalError(
- "Can not find equality delete column field id {} in
data file schema",
- field_id);
+ const auto* field = _find_equality_delete_data_field(filter, idx);
+ if (field == nullptr) {
+ // Iceberg schema evolution defines a field absent from an
older data file as NULL.
+ // EqualityDeletePredicate deliberately treats NULL == NULL as
a match, so a typed
+ // NULL literal preserves the spec semantics without inventing
a physical column.
+ delete_predicate->add_child(
+
VLiteral::create_shared(make_nullable(filter.key_types[idx]), Field()));
Review Comment:
Please do not hard-code every missing equality key to NULL. Iceberg schema
evolution can assign an added field a non-null initial default on table
versions that allow it, and Doris' Iceberg DDL passes column defaults into
`UpdateSchema.addColumn(...)`, but this branch has already lost the
table-field/default metadata and always builds `VLiteral(..., Field())`. For an
old data file missing field id 1 whose logical initial value is `7`, an
equality delete file containing key `7` should compare against `7`; this
predicate compares against NULL and keeps those rows. Please carry the equality
field's default/null semantics through the FE/BE schema/default path and fall
back to NULL only when the field's logical initial value is NULL.
--
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]
