github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3664639780
##########
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:
[P1] Resolve DEFAULT(column) before VALUES analysis
The new pinned handling here recognizes only bare `DefaultValueSlot`.
`VALUES (1, DEFAULT(score))` instead contains `Default(UnboundSlot(score))`,
follows the normal `addColumnValue()` path, and is analyzed against
`buildExprAnalyzer()`'s empty scope before `RewriteDefaultExpression` can use
the pinned Iceberg context. Thus the advertised INSERT `DEFAULT(column)` path
works in the added `INSERT ... SELECT ... FROM t` test but still fails for
inline VALUES. Please resolve `Default` against the pinned target schema before
empty-scope analysis, preserving unknown-field and cross-field casts, and add
reordered/multi-row VALUES coverage.
##########
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:
[P0] Reject spec changes before a static partition overwrite
This preflight accepts the pinned spec whenever it is still present in
`table.specs()`, even if another spec has become current. That is safe for
appends, but not for static overwrite: `commitStaticPartitionOverwrite()` later
builds its deletion filter from `transaction.table().spec()`. If planning
pinned identity spec P1 with `{p=7}`, then a concurrent evolution makes P2 on
`q` current while retaining P1, validation passes; P2 consumes none of the
saved `p` key, `buildPartitionFilter()` returns `alwaysTrue`, and
`overwriteByRowFilter(alwaysTrue)` replaces the whole table. Please require the
fresh current spec to equal the pinned spec for static overwrite, or build the
filter from the pinned spec, and fail unless every non-empty static key
produced a predicate. Add a concurrent spec-replacement test.
##########
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:
[P1] Fence scans when the selected snapshot ancestry is truncated
If snapshot B still inherits a file written under schema A, but A has been
expired, B keeps a non-null `parentId` while `table.snapshot(parentId)` returns
null. This loop then returns only B's schema. After an incompatible evolution
makes a projected field required without an initial default,
`requiresMissingRequiredFieldRejection()` therefore skips the smooth-upgrade
fence: a current BE rejects the old file as missing the required field, while a
source BE at the old semantics materializes NULL. This is distinct from the
earlier over-fencing thread for later/off-lineage schemas; here the selected
lineage is incomplete. Please conservatively require current semantics when a
non-null parent cannot be resolved (or derive file-era evidence from retained
manifests), and add the expired-parent case.
##########
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:
[P1] Preserve qualifier ownership when resolving DEFAULT
`DEFAULT(s.score)` reaches this helper as `UnboundSlot(["s", "score"])`, but
this keeps only `score` and resolves it against the target schema before normal
binding can validate `s`. In MERGE, `DEFAULT(source.score)` (and even
`DEFAULT(no_such_alias.score)`) therefore silently writes the target field's
pinned default whenever target `score` exists. The ordinary
`RewriteDefaultExpression` path correctly checks `SlotReference.originalTable`
against the target. Please bind the child normally or validate the complete
qualifier against the allowed target names, and add target-, source-, and
unknown-qualified UPDATE/MERGE cases.
--
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]
