github-actions[bot] commented on code in PR #65502:
URL: https://github.com/apache/doris/pull/65502#discussion_r3566355379
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -438,38 +440,99 @@ 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;
+ }
+
+ // Migrated Hive tables may have no field ids in their physical files.
Once the mapper has
+ // selected BY_NAME mode, bind the delete key by its delete-file name or
by the table column's
+ // historical name mapping, exactly as normal column materialization does
for a renamed field.
+ const auto table_field_it = std::ranges::find_if(
Review Comment:
This lookup makes equality-delete binding depend on the query projection.
`_projected_columns` comes from the scan's required slots, so a query that
reads only `data` or `count(*)` does not necessarily include the equality key's
table metadata. For an old field-id-less file with physical `legacy_id`, a
current table field `current_id` with `name_mapping = legacy_id`, and an
equality delete file for field id 0 named `current_id`, the test where
`current_id` is projected works, but `SELECT data ...` reaches this code with
no table field for id 0. The fallback then treats the key as a missing column
and uses a NULL literal, so a non-null delete key like `current_id = 2` will
not delete the row. Equality-delete key binding needs to use table
schema/history metadata independent of the selected columns, or add the delete
key as a hidden mapping before resolving the data file field.
##########
be/src/format_v2/expr/equality_delete_predicate.cpp:
##########
@@ -104,17 +104,55 @@ Status EqualityDeletePredicate::execute(VExprContext*
context, Block* block,
int slot = -1;
RETURN_IF_ERROR(child->execute(context, &eval_block, &slot));
const auto& key_column = eval_block.get_by_position(slot);
- data_key_block.insert({key_column.column, key_column.type,
key_column.name});
+
data_key_block.insert({key_column.column->convert_to_full_column_if_const(),
+ key_column.type, key_column.name});
}
+ auto result_column = _evaluate_key_block(data_key_block);
+ if (result_column->size() != block->rows()) {
Review Comment:
`block->rows()` is not a reliable batch size here. In lazy Parquet/ORC scans
the shared file block can contain an unread projected column at position 0
while the equality-delete key column has already been decoded later in the
block. `Block::rows()` then returns 0, so a normal present-key equality delete
produces a batch-sized result and hits this new
`DORIS_CHECK(result_column->size() == 1)`. For all-missing-key deletes, the
literal children are sized from the same 0-row block and a zero-length delete
filter is inserted before the reader checks it against the real batch size.
Please pass the reader-known row count into the predicate path, or derive the
expansion size from populated predicate/carrier columns rather than from the
first block column.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_equality_delete_with_schema_change.groovy:
##########
@@ -42,7 +42,11 @@ suite("test_iceberg_equality_delete_with_schema_change",
"p0,external,doris,exte
sql """switch ${catalog_name};"""
sql """ use multi_catalog;"""
-
+ def originalEnableFileScannerV2 = sql """show variables like
'enable_file_scanner_v2'"""[0][1]
+ // The existing output was generated by the legacy scanner and is kept as
the differential
+ // baseline. Run the complete schema-evolution matrix through V2 so
missing/renamed equality
+ // delete keys must produce byte-for-byte identical query results.
+ sql """set enable_file_scanner_v2=true"""
Review Comment:
Please put this session-variable override in a `try/finally`. If any
`order_qt` in the Parquet/ORC matrix fails, execution skips the restore at the
end of the suite and leaves `enable_file_scanner_v2` forced to true for
whatever runs next in this session. Nearby Iceberg suites use `try { ... }
finally { set enable_file_scanner_v2=${originalEnableFileScannerV2} }` for this
same cleanup pattern.
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -438,38 +440,99 @@ 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;
+ }
+
+ // Migrated Hive tables may have no field ids in their physical files.
Once the mapper has
+ // selected BY_NAME mode, bind the delete key by its delete-file name or
by the table column's
+ // historical name mapping, exactly as normal column materialization does
for a renamed field.
+ const auto table_field_it = 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;
+ });
+ field_it = std::ranges::find_if(
+ _data_reader.file_schema, [&](const format::ColumnDefinition&
file_field) {
+ if (file_field.name == filter.field_names[key_idx]) {
Review Comment:
This fallback does not actually match the BY_NAME rules used by normal
materialization. `ColumnMapper` compares names case-insensitively and also
checks string identifiers plus aliases on both the table and file fields, but
this code only does exact string comparisons and only checks
`file_field.name_mapping` against the current table name. For field-id-less
files, a physical column like `ID` normally maps to table column `id`, and
aliases/identifier names are also valid, but this path classifies those fields
as missing and substitutes a NULL literal. A non-null equality-delete key then
never matches, leaving deleted rows visible. Please reuse the mapper's BY_NAME
matching semantics here or factor the same matcher for equality-delete binding.
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -566,14 +629,50 @@ Status
IcebergTableReader::_read_position_delete_file(const TIcebergDeleteFileDe
Status IcebergTableReader::_init_position_delete_rows(
const std::vector<TIcebergDeleteFileDesc>& delete_files) {
+ DORIS_CHECK(_split_cache != nullptr);
TFileScanRangeParams delete_scan_params =
_scan_params == nullptr ? TFileScanRangeParams() : *_scan_params;
format::DeleteRows position_delete_rows;
IcebergDeleteFileIOContext delete_io_ctx(_runtime_state);
- PositionDeleteRowsCollector collector(_data_file_path(),
&position_delete_rows);
for (const auto& delete_file : delete_files) {
- RETURN_IF_ERROR(_read_position_delete_file(delete_file,
delete_scan_params, &delete_io_ctx,
- &collector));
+ Status read_status = Status::OK();
+ // A position delete file normally references many data files. Cache
the complete
+ // path-to-position map once; caching only the current data file would
still rescan the
+ // shared delete file for every subsequent split.
+ auto* rows_by_data_file =
+
_split_cache->get<PositionDeleteRowsCollector::PositionDeleteFile>(
+ "iceberg_v2_position_delete_" + delete_file.path,
Review Comment:
The cache key needs to include the filesystem identity used to open the
delete file. `_create_delete_file_reader()` resolves `delete_file.path` through
the current data split's `fs_name` when it is set, but the cached
position-delete map is keyed only by `delete_file.path` here, and the
equality-delete cache below uses only the path plus field ids. Since the
`ShardedKVCache` is shared across all splits for the scan node, two splits
whose delete descriptors both say e.g. `/warehouse/t/delete/00001.parquet` but
whose data ranges have different `fs_name` values can share this entry. The
second split would then reuse rows/equality keys parsed from the first
filesystem instead of opening the delete file in its own filesystem, which can
silently miss or apply the wrong deletes. Please include the resolved
filesystem identity (at least the current split `fs_name`, and keep the
equality field-id suffix) in both V2 position and equality delete-file cache
keys.
--
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]
