github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3709634890
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -371,6 +371,7 @@ private BuildInsertExecutorResult
initPlanOnce(ConnectContext ctx,
Optional<CascadesContext> analyzeContext = Optional.of(
CascadesContext.initContext(ctx.getStatementContext(),
originLogicalQuery, PhysicalProperties.ANY)
);
+ pinConnectorWriteSchema(ctx.getStatementContext(), targetTableIf);
Review Comment:
[P1] Pin the writer schema before overwrite normalization
This hook is too late for `INSERT OVERWRITE`: its wrapper has already called
`InsertUtils.normalizePlan()` on inline VALUES before constructing the inner
`InsertIntoTableCommand`. With no connector writer schema pinned at that first
pass, explicit `DEFAULT` is resolved against cached columns that lack Iceberg
write defaults, so an optional column with write default 42 becomes NULL and
the inner pass cannot recover it. The wrapper also passes an empty `branchName`
to the inner command, so this late pin uses the main/current schema even when
the overwrite context targets a branch. Please pin the target (including the
branch) before `InsertOverwriteTableCommand` first normalizes, reuse that pin
for the inner command/explain path, and add overwrite VALUES coverage for
DEFAULT and an old-schema branch.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -1728,57 +1758,437 @@ private static List<String>
requestedLowerNames(List<ConnectorColumnHandle> colu
return names;
}
+ private static boolean mayHaveEqualityDeletes(Snapshot snapshot) {
+ if (snapshot == null) {
+ return false;
+ }
+ return mayHaveEqualityDeletes(snapshot.summary());
+ }
+
+ @VisibleForTesting
+ static boolean mayHaveEqualityDeletes(Map<String, String> snapshotSummary)
{
+ String equalityDeletes = snapshotSummary.get(TOTAL_EQUALITY_DELETES);
+ // A missing counter is unknown (replace/cherry-pick snapshots can
omit it), so retain the bounded
+ // schema-history carrier. The exact task/delete binding is still
decided by Iceberg during split planning.
+ return equalityDeletes == null || !equalityDeletes.equals("0");
+ }
+
/**
- * Ensure the schema-evolution dict carries the table's equality-delete
KEY columns even when the query
- * does not project them (#65502). Equality-delete keys are hidden scan
dependencies: BE resolves a key
- * that is missing from an OLD data file by looking its field id up in
this dict to get the column type +
- * iceberg initial default; without the entry BE materializes the key as
NULL and mis-applies the delete.
- * The keys are the table's declared identifier fields (what
equality-delete writers key on) -> a few
- * columns, DCHECK-safe superset (BE looks up only its own scan slots; the
pin/top-N branches already ship
- * the full schema). If the table declares NO identifier yet the scan
carries equality deletes (whose
- * equality_ids we cannot cheaply enumerate here), fall back to the full
schema. Non-identifier /
- * append-only / position-delete-only tables are unaffected (the pruned
dict is returned verbatim).
+ * Build a schema carrier that can resolve any equality key reachable
before the selected schema without
+ * enumerating data files, manifests, or byte-split tasks. Its retained
state is bounded by table schema
+ * history rather than scan cardinality. At execution time BE looks fields
up by the exact IDs on each
+ * {@link FileScanTask#deletes()}; unrelated carrier fields never
participate in delete matching.
+ *
+ * <p>The selected snapshot lineage wins when a field was renamed. The
metadata schema list, in its actual
+ * chronology up to the selected schema (schema IDs are identifiers, not a
sequence), fills schema-only
+ * changes and expired ancestors. Current fields remain first, so a
dropped/re-added name still resolves the
+ * projected current field by name while a historical equality key
resolves by its stable field ID.</p>
*/
- private List<String> withEqualityDeleteKeyColumns(Table table,
List<String> requested) {
- if (requested.isEmpty()) {
- // An empty requested list already makes buildCurrentSchema fall
back to the FULL schema (every
- // top-level column) — a superset that covers every
equality-delete key — so there is nothing to
- // force-include. Returning early also preserves that all-columns
fallback (a non-empty identifier
- // set would otherwise prune it to identifier-only) and skips the
table.schema()/currentSnapshot()
- // probe when it cannot change the result.
- return requested;
- }
- Schema schema = table.schema();
- Set<Integer> identifierFieldIds = schema.identifierFieldIds();
- if (identifierFieldIds.isEmpty()) {
- return hasEqualityDeletes(table) ? Collections.emptyList() :
requested;
- }
- Set<String> present = new HashSet<>();
- for (String name : requested) {
- present.add(name.toLowerCase(Locale.ROOT));
- }
- List<String> result = new ArrayList<>(requested);
- for (int fieldId : identifierFieldIds) {
- Types.NestedField field = schema.findField(fieldId);
+ private static List<NestedField> schemaForPotentialEqualityDeletes(
+ Table table, TableScan scan, Schema scanSchema) {
+ List<Schema> history = potentialEqualityDeleteSchemaHistory(table,
scan, scanSchema);
+ Set<Integer> missing = new HashSet<>();
+ for (Schema schema : history) {
+ for (NestedField field :
TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isPrimitiveType()) {
+ missing.add(field.fieldId());
+ }
+ }
+ }
+ missing.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ if (missing.isEmpty()) {
+ return scanSchema.columns();
+ }
+
+ List<NestedField> fields = new ArrayList<>(scanSchema.columns());
+ for (Schema historicalSchema : history) {
+ addHistoricalEqualityFields(fields, missing, historicalSchema);
+ }
+ if (!missing.isEmpty()) {
+ throw new IllegalStateException(
+ "Iceberg historical primitive fields are absent from
schema history: " + missing);
+ }
+ return fields;
+ }
+
+ private static List<Schema> potentialEqualityDeleteSchemaHistory(
+ Table table, TableScan scan, Schema scanSchema) {
+ List<Schema> history = new ArrayList<>();
+ Set<Integer> seenSchemaIds = new HashSet<>();
+ addSchemaIfAbsent(history, seenSchemaIds, scanSchema);
+
+ Snapshot snapshot = scan.snapshot();
+ while (snapshot != null) {
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null) {
+ Schema historicalSchema = table.schemas().get(schemaId);
+ if (historicalSchema == null) {
+ throw new IllegalStateException(
+ "Iceberg snapshot schema " + schemaId + " is
absent from table metadata");
+ }
+ addSchemaIfAbsent(history, seenSchemaIds, historicalSchema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null : table.snapshot(parentId);
+ }
+
+ List<Schema> metadataSchemas = metadataSchemaHistory(table);
+ int selectedSchemaIndex = -1;
+ for (int index = 0; index < metadataSchemas.size(); index++) {
+ if (metadataSchemas.get(index).schemaId() ==
scanSchema.schemaId()) {
+ selectedSchemaIndex = index;
+ }
+ }
+ int lastRelevantIndex = selectedSchemaIndex >= 0
+ ? selectedSchemaIndex : metadataSchemas.size() - 1;
+ for (int index = lastRelevantIndex; index >= 0; index--) {
+ addSchemaIfAbsent(history, seenSchemaIds,
metadataSchemas.get(index));
+ }
+ return history;
+ }
+
+ private static void addSchemaIfAbsent(
+ List<Schema> schemas, Set<Integer> seenSchemaIds, Schema schema) {
+ if (seenSchemaIds.add(schema.schemaId())) {
+ schemas.add(schema);
+ }
+ }
+
+ private static List<Schema> metadataSchemaHistory(Table table) {
+ if (table instanceof HasTableOperations) {
+ return ((HasTableOperations)
table).operations().current().schemas();
+ }
+ return new ArrayList<>(table.schemas().values());
+ }
+
+ private static void addHistoricalEqualityFields(
+ List<NestedField> fields, Set<Integer> missingFieldIds, Schema
historicalSchema) {
+ Map<Integer, NestedField> historicalFields =
TypeUtil.indexById(historicalSchema.asStruct());
+ Set<Integer> selectedFieldIds = new HashSet<>();
+ for (Integer fieldId : missingFieldIds) {
+ NestedField field = historicalFields.get(fieldId);
+ if (field != null) {
+ if (!field.type().isPrimitiveType()) {
+ throw new IllegalStateException(
+ "Iceberg equality-delete field " + fieldId + "
must be primitive");
+ }
+ selectedFieldIds.add(fieldId);
+ }
+ }
+ if (selectedFieldIds.isEmpty()) {
+ return;
+ }
+ Schema selectedSchema = TypeUtil.select(historicalSchema,
selectedFieldIds);
+ mergeHistoricalEqualityFields(fields, selectedSchema.columns());
+ missingFieldIds.removeAll(selectedFieldIds);
+ }
+
+ private static void mergeHistoricalEqualityFields(
+ List<NestedField> fields, List<NestedField> historicalFields) {
+ for (NestedField historicalField : historicalFields) {
+ int currentIndex = -1;
+ for (int index = 0; index < fields.size(); index++) {
+ if (fields.get(index).fieldId() == historicalField.fieldId()) {
+ currentIndex = index;
+ break;
+ }
+ }
+ if (currentIndex < 0) {
+ fields.add(historicalField);
+ continue;
+ }
+ NestedField currentField = fields.get(currentIndex);
+ Type mergedType = mergeHistoricalEqualityType(currentField.type(),
historicalField.type());
+ if (mergedType != currentField.type()) {
+ fields.set(currentIndex,
+
Types.NestedField.from(currentField).ofType(mergedType).build());
+ }
+ }
+ }
+
+ private static Type mergeHistoricalEqualityType(Type currentType, Type
historicalType) {
+ if (currentType.typeId() != historicalType.typeId()) {
+ throw new IllegalStateException("Iceberg equality-delete ancestor
type changed from "
+ + historicalType + " to " + currentType);
+ }
+ switch (currentType.typeId()) {
+ case STRUCT:
+ List<NestedField> mergedFields =
+ new ArrayList<>(currentType.asStructType().fields());
+ mergeHistoricalEqualityFields(mergedFields,
historicalType.asStructType().fields());
+ return mergedFields.equals(currentType.asStructType().fields())
+ ? currentType : Types.StructType.of(mergedFields);
+ case LIST:
+ Types.ListType currentList = currentType.asListType();
+ Types.ListType historicalList = historicalType.asListType();
+ if (currentList.elementId() != historicalList.elementId()) {
+ throw new IllegalStateException(
+ "Iceberg equality-delete list element field ID
changed");
+ }
+ Type mergedElement = mergeHistoricalEqualityType(
+ currentList.elementType(),
historicalList.elementType());
+ if (mergedElement == currentList.elementType()) {
+ return currentType;
+ }
+ return currentList.isElementOptional()
+ ? Types.ListType.ofOptional(currentList.elementId(),
mergedElement)
+ : Types.ListType.ofRequired(currentList.elementId(),
mergedElement);
+ case MAP:
+ Types.MapType currentMap = currentType.asMapType();
+ Types.MapType historicalMap = historicalType.asMapType();
+ if (currentMap.keyId() != historicalMap.keyId()
+ || currentMap.valueId() != historicalMap.valueId()) {
+ throw new IllegalStateException(
+ "Iceberg equality-delete map field IDs changed");
+ }
+ Type mergedKey = mergeHistoricalEqualityType(
+ currentMap.keyType(), historicalMap.keyType());
+ Type mergedValue = mergeHistoricalEqualityType(
+ currentMap.valueType(), historicalMap.valueType());
+ if (mergedKey == currentMap.keyType() && mergedValue ==
currentMap.valueType()) {
+ return currentType;
+ }
+ return currentMap.isValueOptional()
+ ? Types.MapType.ofOptional(currentMap.keyId(),
currentMap.valueId(),
+ mergedKey, mergedValue)
+ : Types.MapType.ofRequired(currentMap.keyId(),
currentMap.valueId(),
+ mergedKey, mergedValue);
+ default:
+ if (!currentType.equals(historicalType)) {
+ throw new IllegalStateException("Iceberg equality-delete
field type changed from "
+ + historicalType + " to " + currentType);
+ }
+ return currentType;
+ }
+ }
+
+ private static boolean requiresCurrentScanSemantics(
+ Table table, TableScan scan, Schema scanSchema,
List<ConnectorColumnHandle> columns,
+ boolean mayHaveEqualityDeletes,
+ Optional<Map<Integer, List<String>>> nameMapping) {
+ if (mayHaveEqualityDeletes) {
Review Comment:
[P1] Base the upgrade fence on applicable deletes
`mayHaveEqualityDeletes` comes from the snapshot-wide
`total-equality-deletes`, so this makes every scan require current BE semantics
whenever any equality delete exists anywhere in the snapshot. Iceberg only
attaches deletes after partition/filter and delete/data-sequence applicability
are evaluated per selected `FileScanTask`; for example, `WHERE p = 2` can
select tasks with no deletes while an unrelated delete for `p = 1` makes this
branch reject the query whenever a smooth-upgrade source backend is present.
The legacy path already had to move this fence to exact task applicability.
Please derive the plugin gate from deletes attached to the dispatched tasks
(with an equivalent bounded/lazy proof for batch planning) and cover partition-
and sequence-pruned deletes during rolling upgrade.
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -511,35 +1163,97 @@ 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 {
+Status IcebergTableReader::_find_equality_delete_data_field(
+ const EqualityDeleteFilter& filter, size_t key_idx,
+ EqualityDeleteColumnPath* const data_path, bool* const complete_path)
const {
DORIS_CHECK(key_idx < filter.field_ids.size());
DORIS_CHECK(key_idx < filter.field_names.size());
- if (mapping_mode() != format::TableColumnMappingMode::BY_NAME) {
+ DORIS_CHECK(data_path != nullptr);
+ DORIS_CHECK(complete_path != nullptr);
+ data_path->clear();
+ *complete_path = false;
+
+ auto schema_path =
+
_find_table_column_identity_path_by_field_id(filter.field_ids[key_idx], true);
+ std::vector<const format::ColumnDefinition*> table_path;
+ if (schema_path.has_value()) {
+ for (const auto& field : *schema_path) {
+ table_path.push_back(&field);
+ }
+ } else {
+ static_cast<void>(find_equality_delete_column_path(_projected_columns,
+
filter.field_ids[key_idx], &table_path));
+ }
+ if (table_path.empty() && mapping_mode() !=
format::TableColumnMappingMode::BY_NAME) {
const int field_id = filter.field_ids[key_idx];
- const 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;
- });
- return field_it == _data_reader.file_schema.end() ? nullptr :
&*field_it;
+ *complete_path =
+ find_equality_delete_column_path(_data_reader.file_schema,
field_id, data_path);
+ return Status::OK();
}
// 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_equality_delete_table_field(filter, key_idx);
- return format::find_column_by_name(*table_field, _data_reader.file_schema);
+ // Reuse ColumnMapper's exact BY_NAME rules at every ancestor so a nested
key keeps its
+ // physical path, including historical aliases for ID-less files.
+ std::optional<format::ColumnDefinition> legacy_table_field;
+ if (table_path.empty() &&
!supports_iceberg_scan_semantics_v2(_scan_params)) {
+ legacy_table_field.emplace();
+ legacy_table_field->name = filter.field_names[key_idx];
+ legacy_table_field->type = filter.key_types[key_idx];
+ table_path.push_back(&*legacy_table_field);
+ }
+ if (table_path.empty()) {
+ return Status::InvalidArgument(
+ "Iceberg equality delete field id {} is absent from current
and historical table "
+ "schema metadata",
+ filter.field_ids[key_idx]);
+ }
+ const std::vector<format::ColumnDefinition>* candidates =
&_data_reader.file_schema;
+ for (size_t index = 0; index < table_path.size(); ++index) {
+ const auto* table_field = table_path[index];
+ DORIS_CHECK(table_field != nullptr);
+ const format::ColumnDefinition* data_field = nullptr;
+ if (mapping_mode() == format::TableColumnMappingMode::BY_NAME) {
Review Comment:
[P1] Resolve ID-bearing keys through ID-less wrappers
With scan semantics v2, a descendant Iceberg ID selects `BY_FIELD_ID`, and
the ordinary mapper explicitly accepts an ID-less Parquet struct wrapper when
an ID-bearing descendant identifies it. This loop instead requires every
physical ancestor to carry the transported ancestor ID, so an ID-less `payload`
fails before its ID-bearing `legacy_k` child is reached. The resolver then
reports an incomplete path and the caller hashes NULL/the initial default
instead of the stored key, allowing an applicable equality delete to be missed.
Please apply the mapper's ID-less-wrapper fallback (or resolve the leaf ID
recursively while retaining its ancestors) and add a V2 equality-delete case
for this file shape.
--
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]
