924060929 commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3711534163
##########
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) {
+ return true;
+ }
+ Set<Integer> projectedFieldIds = projectedFieldIds(scanSchema,
columns);
+ Set<Integer> topLevelFieldIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelFieldIds.add(field.fieldId());
+ }
+ Map<Integer, NestedField> fields =
TypeUtil.indexById(scanSchema.asStruct());
+ for (Integer fieldId : projectedFieldIds) {
+ NestedField field = fields.get(fieldId);
+ if (field != null && field.initialDefault() != null
+ && (!topLevelFieldIds.contains(fieldId) ||
field.type().isNestedType())) {
+ return true;
+ }
+ }
+ if (hasProjectedNameAliasCollision(scanSchema, projectedFieldIds,
nameMapping)) {
+ return true;
+ }
+ Optional<List<Schema>> history = requiredFieldSchemaHistory(table,
scanSchema, scan.snapshot());
+ return !history.isPresent()
+ || requiresMissingRequiredFieldRejection(scanSchema,
projectedFieldIds, history.get());
+ }
+
+ @VisibleForTesting
+ static Set<Integer> projectedFieldIds(
+ Schema scanSchema, List<ConnectorColumnHandle> columns) {
+ Set<Integer> projected = new HashSet<>();
+ if (columns == null || columns.isEmpty()) {
+
projected.addAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ return projected;
+ }
+ Map<Integer, NestedField> fieldsById =
TypeUtil.indexById(scanSchema.asStruct());
+ for (ConnectorColumnHandle column : columns) {
+ IcebergColumnHandle icebergColumn = (IcebergColumnHandle) column;
+ NestedField field =
scanSchema.findField(icebergColumn.getFieldId());
if (field == null) {
continue;
}
- String lower = field.name().toLowerCase(Locale.ROOT);
- if (present.add(lower)) {
- result.add(lower);
+ if (icebergColumn.hasProjectedFieldIds()) {
+ Set<Integer> scopedFieldIds =
icebergColumn.getProjectedFieldIds();
+ Set<Integer> selectableFieldIds = new HashSet<>();
+ selectableFieldIds.add(field.fieldId());
+ if (field.type().isNestedType()) {
+
selectableFieldIds.addAll(TypeUtil.getProjectedIds(field.type()));
+ }
+ for (Integer scopedFieldId : scopedFieldIds) {
+ NestedField scopedField = fieldsById.get(scopedFieldId);
+ if (scopedField == null ||
!selectableFieldIds.contains(scopedFieldId)) {
+ throw new IllegalStateException("Projected Iceberg
field ID " + scopedFieldId
+ + " does not belong to top-level field " +
field.fieldId());
+ }
+ projected.add(scopedFieldId);
+ if (scopedField.type().isNestedType()) {
+ Set<Integer> descendants =
TypeUtil.getProjectedIds(scopedField.type());
+ if (Collections.disjoint(scopedFieldIds, descendants))
{
+ // The access path terminates at this complex
field, so the entire subtree is
+ // projected. An ancestor with a selected
descendant must remain scoped instead.
+ projected.addAll(descendants);
+ }
+ }
+ }
+ continue;
+ }
+ projected.add(field.fieldId());
+ // Iceberg's type visitor returns null for a primitive root;
getProjectedIds(Type) then passes
+ // that null to ImmutableSet.copyOf. The top-level id is already
present, and only nested types
+ // have descendant ids to add.
+ if (field.type().isNestedType()) {
+ projected.addAll(TypeUtil.getProjectedIds(field.type()));
}
}
- return result;
+ return projected;
}
- private static boolean hasEqualityDeletes(Table table) {
- Snapshot snapshot = table.currentSnapshot();
- if (snapshot == null) {
+ private static Optional<List<Schema>> requiredFieldSchemaHistory(
+ Table table, Schema scanSchema, Snapshot selectedSnapshot) {
+ List<Schema> schemas = new ArrayList<>();
+ Set<Integer> schemaIds = new HashSet<>();
+ schemas.add(scanSchema);
+ schemaIds.add(scanSchema.schemaId());
+ Deque<Snapshot> snapshots = new ArrayDeque<>();
+ if (selectedSnapshot != null) {
+ snapshots.add(selectedSnapshot);
+ }
+ Set<Long> visitedSnapshotIds = new HashSet<>();
+ while (!snapshots.isEmpty()) {
Review Comment:
[P2] This only removes the single-schema fast case; the snapshot-count path
remains after a relevant historical schema exists. For example, make field id 1
optional in S1, require it without an initial default in S2, then append 100k
snapshots under S2. schemaIdsRequiringMissingRequiredFieldRejection returns S1,
so selectedHistoryRequiresMissingRequiredFieldRejection still follows every
parentId through all 100k S2 snapshots before it reaches S1. The equality
carrier has the same shape after a dropped field leaves missing non-empty. The
new test cannot detect either case because its one schema makes
relevantSchemaIds/missing empty before traversal starts. Please add a
two-schema counter test with a long same-schema tail and bound work by schema
transitions/relevant schema IDs rather than retained snapshot count.
--
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]
