github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3713544025
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java:
##########
@@ -481,6 +473,57 @@ private static Plan normalizePlanWithoutLock(LogicalPlan
plan, TableIf table,
return plan.withChildren(new
LogicalInlineTable(optimizedRowConstructors.build()));
}
+ private static List<Column> connectorWriteSchema(TableIf table, boolean
full) {
+ ConnectContext context = ConnectContext.get();
+ if (context != null && context.getStatementContext() != null) {
+ Optional<List<Column>> pinned =
+
context.getStatementContext().getConnectorWriteSchema(table.getId());
+ if (pinned.isPresent()) {
+ return pinned.get();
+ }
+ }
+ return table.getBaseSchema(full);
+ }
+
+ static void pinConnectorWriteSchema(StatementContext statementContext,
TableIf targetTableIf,
+ LogicalPlan logicalQuery, Optional<String> branchName) {
+ if (!(targetTableIf instanceof PluginDrivenExternalTable)
+ || !(logicalQuery instanceof UnboundConnectorTableSink)
+ || ((UnboundConnectorTableSink<?>) logicalQuery).isRewrite()
+ ||
statementContext.getConnectorWriteSchema(targetTableIf.getId()).isPresent()) {
Review Comment:
[P2] Repin the writer schema on an internal INSERT retry.
`InsertIntoTableCommand.initPlan()` can `continue` after its post-plan
table/schema check detects a concurrent change, but the first attempt has
already filled both connector caches. On attempt two this new `isPresent()` arm
returns without resolving the updated writer schema, and neither command-local
retry edge calls `resetConnectorStatementScope()`, so DEFAULT/omitted-column
normalization and sink planning keep using attempt one's schema until the retry
budget is exhausted or the later fence rejects it. Clear the connector
scope/pin before each internal retry and add a test that evolves an Iceberg
write default between planning and the post-plan schema check.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -1728,57 +1758,443 @@ private static List<String>
requestedLowerNames(List<ConnectorColumnHandle> colu
return names;
}
+ @VisibleForTesting
+ static boolean hasApplicableEqualityDeletes(TableScan scan) {
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null
+ || "0".equals(snapshot.summary().get(TOTAL_EQUALITY_DELETES)))
{
+ return false;
+ }
+ // planFiles binds delete files to the exact filtered data-file tasks
after partition and sequence
+ // pruning. A snapshot summary of zero returns above without planning;
a positive or missing summary
+ // needs this exact proof. Iterate whole-file tasks lazily and stop at
the first equality delete: this
+ // keeps memory O(1), does not create or retain byte-split tasks, and
avoids snapshot-wide delete
+ // counters forcing new-BE-only semantics when no dispatched task can
consume an equality delete.
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ for (DeleteFile delete : task.deletes()) {
+ if (delete.content() == FileContent.EQUALITY_DELETES) {
+ return true;
+ }
+ }
+ }
+ } catch (IOException e) {
+ throw new DorisConnectorException(
+ "Failed to inspect applicable Iceberg equality deletes: "
+ e.getMessage(), e);
+ }
+ return false;
+ }
+
/**
- * 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);
+ @VisibleForTesting
+ static List<NestedField> schemaForPotentialEqualityDeletes(
+ Table table, TableScan scan, Schema scanSchema) {
+ 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;
+ Set<Integer> missing = new HashSet<>();
+ for (int index = 0; index <= lastRelevantIndex; index++) {
+ Schema schema = metadataSchemas.get(index);
+ 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());
+ Map<Integer, Schema> schemasById = table.schemas();
+ Snapshot snapshot = scan.snapshot();
+ while (snapshot != null && !missing.isEmpty()) {
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null) {
+ Schema historicalSchema = schemasById.get(schemaId);
+ if (historicalSchema == null) {
+ throw new IllegalStateException(
+ "Iceberg snapshot schema " + schemaId + " is
absent from table metadata");
+ }
+ addHistoricalEqualityFields(fields, missing, historicalSchema);
+ }
+ if (missing.isEmpty()) {
+ break;
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null : table.snapshot(parentId);
+ }
+ for (int index = lastRelevantIndex; index >= 0 && !missing.isEmpty();
index--) {
+ addHistoricalEqualityFields(fields, missing,
metadataSchemas.get(index));
+ }
+ if (!missing.isEmpty()) {
+ throw new IllegalStateException(
+ "Iceberg historical primitive fields are absent from
schema history: " + missing);
+ }
+ return fields;
+ }
+
+ 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 hasApplicableEqualityDeletes,
+ Optional<Map<Integer, List<String>>> nameMapping) {
+ if (hasApplicableEqualityDeletes) {
+ 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;
+ }
+ return selectedHistoryRequiresMissingRequiredFieldRejection(
+ table, scanSchema, projectedFieldIds, scan.snapshot());
+ }
+
+ @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) {
+ @VisibleForTesting
+ static boolean selectedHistoryRequiresMissingRequiredFieldRejection(
+ Table table, Schema scanSchema, Set<Integer> projectedFieldIds,
+ Snapshot selectedSnapshot) {
+ Map<Integer, Schema> schemasById = table.schemas();
+ Set<Integer> relevantSchemaIds =
schemaIdsRequiringMissingRequiredFieldRejection(
+ scanSchema, projectedFieldIds, schemasById.values());
+ if (relevantSchemaIds.isEmpty()) {
return false;
}
- String equalityDeletes =
snapshot.summary().get(TOTAL_EQUALITY_DELETES);
- // Absent (compaction/replace snapshots omit the counter) -> unknown
-> assume present (safe superset).
- return equalityDeletes == null || !equalityDeletes.equals("0");
+ Deque<Snapshot> snapshots = new ArrayDeque<>();
+ if (selectedSnapshot != null) {
+ snapshots.add(selectedSnapshot);
+ }
+ Set<Long> visitedSnapshotIds = new HashSet<>();
+ while (!snapshots.isEmpty()) {
+ Snapshot snapshot = snapshots.removeFirst();
+ if (!visitedSnapshotIds.add(snapshot.snapshotId())) {
+ continue;
+ }
+ Integer schemaId = snapshot.schemaId();
Review Comment:
[P1] Gate legacy snapshots whose schema ID is unavailable.
`schemaIdsRequiringMissingRequiredFieldRejection` has already proved that some
historical schema would make a projected required field missing or optional,
but this loop simply skips a reachable snapshot when `Snapshot.schemaId()` is
null. Iceberg V1 snapshot metadata may omit `schema-id`, so after an
incompatible required-field evolution without an initial default this can
return false and omit `REQUIRED_CURRENT_BACKEND_SEMANTICS`; a rolling-upgrade
scan can then return legacy NULLs on an old BE while a new BE rejects the same
file. Treat an unavailable schema ID conservatively here (unless another
metadata source proves the schema), and cover legacy/V1 snapshot metadata with
this evolution.
--
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]
