hubgeter commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3664848088
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -530,30 +587,536 @@ 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.
+ */
+ @VisibleForTesting
+ 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();
+ snapshot = parentId == null ? null :
icebergTable.snapshot(parentId);
Review Comment:
Fixed in 6c245633220. The selected snapshot lineage walk now returns an
incomplete-history marker when a non-null parent cannot be resolved, and the
required-field compatibility check conservatively requires current scan
semantics in that case. Added
testRequiredFieldFenceRejectsTruncatedSnapshotLineage; the focused FE set
passed 86/86.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java:
##########
@@ -145,6 +148,9 @@ public void beginInsert(ExternalTable dorisTable,
Optional<InsertCommandContext>
+ " is a tag, not a branch. Tags
cannot be targets for producing snapshots");
}
}
+ if (writeSchemaContext.isPresent()) {
+ writeSchemaContext.get().validateCurrentSchema(table);
Review Comment:
Fixed in 6c245633220. Static partition overwrite now requires the fresh
current spec to exactly match the pinned spec, builds the replacement filter
from pinned writer metadata, and asserts that every supplied static key
produced a predicate. Added a concurrent retained-P1/current-P2 test that fails
before opening the transaction; the static overwrite regression passed.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergDmlCommandUtils.java:
##########
@@ -47,6 +55,49 @@ static void checkMergeMode(IcebergExternalTable table) {
TableProperties.MERGE_MODE_DEFAULT);
}
+ static Optional<IcebergWriteSchemaContext> installWriteSchemaContext(
+ ConnectContext context, IcebergWriteSchemaContext
writeSchemaContext) {
+ Optional<IcebergWriteSchemaContext> previous =
+ context.getStatementContext().getIcebergWriteSchemaContext();
+
context.getStatementContext().setIcebergWriteSchemaContext(Optional.of(writeSchemaContext));
+ return previous;
+ }
+
+ static void restoreWriteSchemaContext(
+ ConnectContext context, Optional<IcebergWriteSchemaContext>
previous) {
+ context.getStatementContext().setIcebergWriteSchemaContext(previous);
+ }
+
+ static Expression resolveDefaultReferences(
+ Expression expression, IcebergWriteSchemaContext
writeSchemaContext) {
+ return expression.rewriteDownShortCircuit(candidate -> {
+ if (!(candidate instanceof Default)) {
+ return candidate;
+ }
+ Expression reference = candidate.child(0);
+ Column column;
+ if (reference instanceof UnboundSlot) {
+ String columnName = ((UnboundSlot) reference).getNameParts()
Review Comment:
Fixed in 6c245633220. DEFAULT resolution now validates the complete unbound
qualifier against the target name or alias and, for bound slots, verifies
original-table ownership before using the pinned target field. Added valid
target-qualified and rejected source or unknown-qualified UPDATE and MERGE
cases; the focused FE set passed 86/86.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java:
##########
@@ -438,7 +516,8 @@ private static Plan normalizePlanWithoutLock(LogicalPlan
plan, TableIf table,
+ "' in table '" + table.getName() + "' is
not allowed.");
}
if (values.get(i) instanceof DefaultValueSlot) {
- NamedExpression defaultExpression =
generateDefaultExpression(sameNameColumn);
+ NamedExpression defaultExpression =
generateDefaultExpression(
+ sameNameColumn, icebergWriteSchemaContext);
Review Comment:
Fixed in 6c245633220. Inline Iceberg VALUES now rewrites
Default(UnboundSlot) from the pinned writer schema before empty-scope analysis
while preserving the surrounding named expression and target cast. Added
reordered multi-row VALUES coverage plus an unknown-field rejection case; the
focused FE set passed 86/86 and test_iceberg_initial_defaults passed.
--
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]
