hubgeter commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3666088045
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -530,30 +587,556 @@ 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();
+ }
+
+ /**
+ * Return only schemas that can describe files visible from the selected
target.
+ *
+ * <p>The query schema is included explicitly because a schema-only commit
does not create a
+ * snapshot. Other schemas are taken from the selected snapshot's parent
lineage, excluding
+ * later main-branch and unrelated branch schemas from the rolling-upgrade
fence. An empty
+ * optional means snapshot expiration truncated that lineage, so callers
must conservatively
+ * require current scan semantics.
+ */
+ @VisibleForTesting
+ Optional<List<Schema>> getRequiredFieldSchemaHistory(Schema scanSchema)
throws UserException {
+ List<Schema> schemas = new ArrayList<>();
+ Set<Integer> schemaIds = new HashSet<>();
+ schemas.add(scanSchema);
+ schemaIds.add(scanSchema.schemaId());
+
+ Snapshot snapshot = createTableScan().snapshot();
+ while (snapshot != null) {
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null && schemaIds.add(schemaId)) {
+ Schema lineageSchema = icebergTable.schemas().get(schemaId);
+ Preconditions.checkState(lineageSchema != null,
+ "Iceberg snapshot schema %s is absent from table
metadata", schemaId);
+ schemas.add(lineageSchema);
+ }
+ Long parentId = snapshot.parentId();
Review Comment:
Fixed in 990c536eeff. The required-field history now traverses both ordinary
parents and Iceberg `source-snapshot-id` provenance recursively, returns the
conservative unknown result when either lineage is truncated, and de-duplicates
the graph by snapshot ID. Added a non-ancestor cherry-pick source/ancestor
test; the expanded Iceberg FE suite passes 96/96 and the full FE build passes.
--
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]
