Gabriel39 commented on code in PR #65851: URL: https://github.com/apache/doris/pull/65851#discussion_r3653955865
########## docs/iceberg-v3-default-values-agent-handoff.md: ########## @@ -0,0 +1,1066 @@ +# Iceberg V3 Default Values 全权处理交接 + +> 状态日期：2026-07-24 +> +> 当前 PR：[apache/doris#65851](https://github.com/apache/doris/pull/65851) +> +> 唯一工作目录：`/mnt/disk1/changyuwei/doris-worktrees/iceberg-v3-defaults` Review Comment: [P2] Remove contributor-local operational state from the repository This 1,066-line agent handoff is transient task state rather than durable Doris documentation. It publishes contributor-local absolute paths/usernames, disk usage, temp and log locations, local tool commands, and head/merge status that was already stale before this PR current head. Please remove this file from the PR; any durable design information belongs in the accompanying design document without environment-specific details. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java: ########## @@ -524,30 +578,504 @@ void enableCurrentIcebergScanSemantics() { params.setIcebergScanSemanticsVersion(ICEBERG_SCAN_SEMANTICS_VERSION); } + /** + * Build the schema metadata carrier used by both scanners and equality-delete readers. + * + * <p>Batch-mode delete files are planned asynchronously after scan parameters are sent to BE. + * The authenticated manifest preflight therefore supplies the live equality field IDs before + * the schema carrier is serialized. Only historical fields referenced by those delete files are + * added, so an unrelated dropped type cannot make an otherwise supported scan fail. + */ + @VisibleForTesting + List<NestedField> getSchemaFieldsForScan( + Schema scanSchema, Set<Integer> equalityDeleteFieldIds) throws UserException { + List<NestedField> fields = new ArrayList<>(scanSchema.columns()); + if (isSystemTable || equalityDeleteFieldIds.isEmpty()) { + return fields; + } + + Set<Integer> missingFieldIds = new HashSet<>(equalityDeleteFieldIds); + missingFieldIds.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet()); + if (missingFieldIds.isEmpty()) { + return fields; + } + + List<Schema> schemaHistory = getMetadataSchemaHistory(); + // Schema IDs may be reused when evolution returns to an earlier schema, while the metadata + // list may also contain schemas committed after a time-travel or branch target. Follow the + // actual scan snapshot's parent chain first so the field definition active on that lineage + // wins. Then use the complete metadata list as a fallback for schema-only changes and + // expired ancestors. A fallback definition may come from a later rename, so BE resolves an + // ID-less equality key through the target mapping first and the delete file's original key + // name second. Initial-default and field identity remain bound to the stable field ID. + Snapshot snapshot = createTableScan().snapshot(); + while (snapshot != null) { + Integer schemaId = snapshot.schemaId(); + if (schemaId != null) { + Schema historicalSchema = icebergTable.schemas().get(schemaId); + Preconditions.checkState(historicalSchema != null, + "Iceberg snapshot schema %s is absent from table metadata", schemaId); + addHistoricalEqualityFields(fields, missingFieldIds, historicalSchema); + } + Long parentId = snapshot.parentId(); + snapshot = parentId == null ? null : icebergTable.snapshot(parentId); + } + for (int index = schemaHistory.size() - 1; index >= 0; index--) { + addHistoricalEqualityFields(fields, missingFieldIds, schemaHistory.get(index)); + } + Preconditions.checkState(missingFieldIds.isEmpty(), + "Iceberg equality-delete fields are absent from schema history: %s", + missingFieldIds); + return fields; + } + + private List<Schema> getMetadataSchemaHistory() { + Preconditions.checkState(icebergTable instanceof HasTableOperations, + "Iceberg table does not expose metadata schema history: %s", icebergTable.name()); + return ((HasTableOperations) icebergTable).operations().current().schemas(); + } + + private static void addHistoricalEqualityFields(List<NestedField> fields, + Set<Integer> missingFieldIds, Schema historicalSchema) { + for (NestedField field : TypeUtil.indexById(historicalSchema.asStruct()).values()) { + if (missingFieldIds.remove(field.fieldId())) { + Preconditions.checkState(field.type().isPrimitiveType(), + "Iceberg equality-delete field %s must be primitive", field.fieldId()); + fields.add(field); + } + } + } + + @VisibleForTesting + static boolean requiresRecursiveInitialDefaultMaterialization( + Schema scanSchema, List<SlotDescriptor> projectedSlots) { + return requiresProjectedIcebergField(scanSchema, projectedSlots, + (field, isTopLevel) -> field.initialDefault() != null + && (!isTopLevel || field.type().isNestedType())); + } + + @VisibleForTesting + static boolean requiresMissingRequiredFieldRejection( + Schema scanSchema, List<SlotDescriptor> projectedSlots, + List<Schema> historicalSchemas) { + Map<Integer, NestedField> fieldById = TypeUtil.indexById(scanSchema.asStruct()); + Map<Integer, Integer> parentById = TypeUtil.indexParents(scanSchema.asStruct()); + Set<Integer> collectionWrapperFieldIds = new HashSet<>(); + collectCollectionWrapperFieldIds(scanSchema.asStruct(), collectionWrapperFieldIds); + Set<Integer> potentiallyMissingRequiredFieldIds = new HashSet<>(); + for (Schema historicalSchema : historicalSchemas) { + Map<Integer, NestedField> historicalFieldById = + TypeUtil.indexById(historicalSchema.asStruct()); + for (NestedField field : fieldById.values()) { + NestedField historicalField = historicalFieldById.get(field.fieldId()); + if (historicalField != null) { + if (!collectionWrapperFieldIds.contains(field.fieldId()) + && field.isRequired() && field.initialDefault() == null + && historicalField.isOptional()) { + potentiallyMissingRequiredFieldIds.add(field.fieldId()); + } + continue; + } + NestedField highestMissingField = field; + Integer parentId = parentById.get(field.fieldId()); + while (parentId != null && !historicalFieldById.containsKey(parentId)) { + highestMissingField = Preconditions.checkNotNull(fieldById.get(parentId), + "Iceberg parent field %s is absent from scan schema", parentId); + parentId = parentById.get(parentId); + } + // If the highest missing ancestor is optional, the old physical subtree is NULL + // and no required descendant is materialized. A non-null initial default is + // already covered by requiresRecursiveInitialDefaultMaterialization(). + if (!collectionWrapperFieldIds.contains(highestMissingField.fieldId()) + && highestMissingField.isRequired() + && highestMissingField.initialDefault() == null) { + potentiallyMissingRequiredFieldIds.add(highestMissingField.fieldId()); + } + } + } + return requiresProjectedIcebergField(scanSchema, projectedSlots, + (field, isTopLevel) -> potentiallyMissingRequiredFieldIds.contains( + field.fieldId())); + } + + private static void collectCollectionWrapperFieldIds( + Type type, Set<Integer> collectionWrapperFieldIds) { + switch (type.typeId()) { + case STRUCT: + for (NestedField field : type.asStructType().fields()) { + collectCollectionWrapperFieldIds(field.type(), collectionWrapperFieldIds); + } + break; + case LIST: + Types.ListType listType = (Types.ListType) type; + collectionWrapperFieldIds.add(listType.elementId()); + collectCollectionWrapperFieldIds( + listType.elementType(), collectionWrapperFieldIds); + break; + case MAP: + Types.MapType mapType = (Types.MapType) type; + collectionWrapperFieldIds.add(mapType.keyId()); + collectionWrapperFieldIds.add(mapType.valueId()); + collectCollectionWrapperFieldIds(mapType.keyType(), collectionWrapperFieldIds); + collectCollectionWrapperFieldIds(mapType.valueType(), collectionWrapperFieldIds); + break; + default: + break; + } + } + + private static boolean requiresProjectedIcebergField( + Schema scanSchema, List<SlotDescriptor> projectedSlots, + ProjectedFieldRequirement requirement) { + Map<Integer, NestedField> fieldById = TypeUtil.indexById(scanSchema.asStruct()); + Set<Integer> topLevelFieldIds = new HashSet<>(); + for (NestedField field : scanSchema.columns()) { + topLevelFieldIds.add(field.fieldId()); + } + for (SlotDescriptor slot : projectedSlots) { + Column column = slot.getColumn(); + List<ColumnAccessPath> accessPaths = slot.getAllAccessPaths(); + if (accessPaths != null && !accessPaths.isEmpty()) { + for (ColumnAccessPath accessPath : accessPaths) { + List<String> path = accessPath.getPath(); + Preconditions.checkState(!path.isEmpty(), + "Iceberg column access path must not be empty"); + Preconditions.checkState(matchesAccessPathComponent(column, path.get(0)), + "Iceberg access path root %s does not match column %s", path.get(0), + column.getName()); + if (requiresProjectedIcebergField( + column, path, 1, fieldById, + topLevelFieldIds.contains(column.getUniqueId()), requirement)) { + return true; + } + } + } else if (requiresProjectedIcebergField( + column, slot.getType(), fieldById, + topLevelFieldIds.contains(column.getUniqueId()), requirement)) { + return true; + } + } + return false; + } + + private static boolean requiresProjectedIcebergField( + Column column, org.apache.doris.catalog.Type projectedType, + Map<Integer, NestedField> fieldById, boolean isTopLevel, + ProjectedFieldRequirement requirement) { + if (requiresIcebergField(column, fieldById, isTopLevel, requirement)) { + return true; + } + if (column.getChildren() == null) { + return false; + } + if (projectedType.isStructType()) { + for (StructField projectedField : ((StructType) projectedType).getFields()) { + Column child = findChildByName(column, projectedField.getName()); + Preconditions.checkState(child != null, + "Projected Iceberg child %s is absent from column %s", + projectedField.getName(), column.getName()); + if (requiresProjectedIcebergField( + child, projectedField.getType(), fieldById, false, requirement)) { + return true; + } + } + } else if (projectedType.isArrayType()) { + Preconditions.checkState(column.getChildren().size() == 1, + "Iceberg array column %s must have one child", column.getName()); + if (requiresProjectedIcebergField( + column.getChildren().get(0), ((ArrayType) projectedType).getItemType(), + fieldById, false, requirement)) { + return true; + } + } else if (projectedType.isMapType()) { + Preconditions.checkState(column.getChildren().size() == 2, + "Iceberg map column %s must have two children", column.getName()); + MapType mapType = (MapType) projectedType; + if (requiresProjectedIcebergField( + column.getChildren().get(0), mapType.getKeyType(), fieldById, false, + requirement) + || requiresProjectedIcebergField( + column.getChildren().get(1), mapType.getValueType(), fieldById, false, + requirement)) { + return true; + } + } + return false; + } + + private static boolean requiresProjectedIcebergField( + Column column, List<String> path, int pathIndex, + Map<Integer, NestedField> fieldById, boolean isTopLevel, + ProjectedFieldRequirement requirement) { + if (requiresIcebergField(column, fieldById, isTopLevel, requirement)) { + return true; + } + if (pathIndex == path.size()) { + return requiresProjectedIcebergField(column, fieldById, requirement); + } + + String component = path.get(pathIndex); + if (AccessPathInfo.ACCESS_NULL.equals(component) + || AccessPathInfo.ACCESS_OFFSET.equals(component)) { + return false; + } + Preconditions.checkState(column.getChildren() != null, + "Iceberg access path continues below primitive column %s", column.getName()); + + if (AccessPathInfo.ACCESS_ALL.equals(component)) { + if (column.getType().isArrayType()) { + Preconditions.checkState(column.getChildren().size() == 1, + "Iceberg array column %s must have one child", column.getName()); + return requiresProjectedIcebergField( + column.getChildren().get(0), path, pathIndex + 1, fieldById, false, + requirement); + } + Preconditions.checkState(column.getType().isMapType(), + "Unexpected Iceberg access-all path below column %s", column.getName()); + Preconditions.checkState(column.getChildren().size() == 2, + "Iceberg map column %s must have two children", column.getName()); + Column key = column.getChildren().get(0); + // element_at(map, key) reads the complete key subtree, while any path after '*' + // describes only the selected value subtree. + if (requiresIcebergField(key, fieldById, false, requirement) + || requiresProjectedIcebergField(key, fieldById, requirement)) { + return true; + } + return requiresProjectedIcebergField( + column.getChildren().get(1), path, pathIndex + 1, fieldById, false, + requirement); + } + if (column.getType().isMapType()) { + Preconditions.checkState(column.getChildren().size() == 2, + "Iceberg map column %s must have two children", column.getName()); + int childIndex; + if (AccessPathInfo.ACCESS_MAP_KEYS.equals(component)) { + childIndex = 0; + } else { + Preconditions.checkState(AccessPathInfo.ACCESS_MAP_VALUES.equals(component), + "Unexpected Iceberg map access path component %s", component); + childIndex = 1; + } + return requiresProjectedIcebergField( + column.getChildren().get(childIndex), path, pathIndex + 1, fieldById, false, + requirement); + } + + Column child = findAccessPathChild(column, component); + Preconditions.checkState(child != null, + "Iceberg access path child %s is absent from column %s", component, + column.getName()); + return requiresProjectedIcebergField( + child, path, pathIndex + 1, fieldById, false, requirement); + } + + private static boolean requiresProjectedIcebergField( + Column column, Map<Integer, NestedField> fieldById, + ProjectedFieldRequirement requirement) { + if (column.getChildren() == null) { + return false; + } + for (Column child : column.getChildren()) { + if (requiresIcebergField(child, fieldById, false, requirement) + || requiresProjectedIcebergField(child, fieldById, requirement)) { + return true; + } + } + return false; + } + + private static boolean requiresIcebergField( + Column column, Map<Integer, NestedField> fieldById, boolean isTopLevel, + ProjectedFieldRequirement requirement) { + NestedField field = fieldById.get(column.getUniqueId()); + return field != null && requirement.requires(field, isTopLevel); + } + + private interface ProjectedFieldRequirement { + boolean requires(NestedField field, boolean isTopLevel); + } + + private static boolean matchesAccessPathComponent(Column column, String component) { + return Integer.toString(column.getUniqueId()).equals(component) + || column.getName().equalsIgnoreCase(component); + } + + private static Column findAccessPathChild(Column column, String component) { + for (Column child : column.getChildren()) { + if (matchesAccessPathComponent(child, component)) { + return child; + } + } + return null; + } + + private static Column findChildByName(Column column, String childName) { + for (Column child : column.getChildren()) { + if (child.getName().equalsIgnoreCase(childName)) { + return child; + } + } + return null; + } + + @VisibleForTesting + Set<Integer> getEqualityDeleteFieldIdsForScan() throws UserException { + Snapshot snapshot = createTableScan().snapshot(); + if (snapshot == null) { + return Collections.emptySet(); + } + try { + // deleteManifests() can lazily read the manifest list. Keep both that call and the + // manifest-entry cache load inside the catalog's authentication context. + return preExecutionAuthenticator.execute( + () -> loadEqualityDeleteFieldIds(snapshot)); + } catch (Exception e) { + Optional<NotSupportedException> opt = checkNotSupportedException(e); + if (opt.isPresent()) { + throw opt.get(); + } + throw new UserException(ExceptionUtils.getRootCauseMessage(e), e); + } + } + + @VisibleForTesting + Set<Integer> loadEqualityDeleteFieldIds(Snapshot snapshot) { + IcebergExternalMetaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr() + .iceberg(source.getCatalog().getId()); + Preconditions.checkState(source.getTargetTable() instanceof ExternalTable, + "Iceberg scan target table is not an external table"); + ExternalTable targetExternalTable = (ExternalTable) source.getTargetTable(); + Set<Integer> equalityDeleteFieldIds = new HashSet<>(); + for (ManifestFile manifest : snapshot.deleteManifests(icebergTable.io())) { + Preconditions.checkState(manifest.content() == ManifestContent.DELETES, + "Iceberg delete manifest %s has content %s", manifest.path(), + manifest.content()); + ManifestCacheValue value = IcebergManifestCacheLoader.loadDeleteFilesWithCache( + cache, targetExternalTable, manifest, icebergTable, + this::recordManifestCacheAccess); + equalityDeleteFieldIds.addAll( Review Comment: [P1] Scope equality fields to deletes applicable to this scan This loads every live entry from every delete manifest in the snapshot, but it never applies `createTableScan().filter()` or pairs deletes with the selected data tasks. Consequently `equalityDeleteFieldIds` includes keys from unrelated partitions and deletes that cannot apply by partition/sequence. For example, if an equality delete in partition `p=1` references a subsequently dropped `TIMESTAMP_NANO` key, a query filtered to `p=2` still carries that historical field and fails in `getScanColumns()` because Doris cannot map the type—even though neither the data nor the delete is part of the query. The same over-approximation unnecessarily rejects scans during a smooth upgrade. Please derive the IDs from the applicable deletes on the planned `FileScanTask`s (while preserving batch-mode planning), or otherwise prune entries with the same filter/applicability rules, and add a cross-partition regression. -- 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]
