Gabriel39 commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3841146632
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -309,6 +322,302 @@ void
checkVariantBackendCompatibilityForCurrentScan(Iterable<Backend> backends)
checkVariantBackendCompatibility(projectsVariant, backends);
}
+ private boolean requiresIcebergScanSemanticsV2() throws UserException {
+ if (isSystemTable) {
+ // position_deletes already has its stricter dedicated
mixed-version gate.
+ return false;
+ }
+ TableScan scan = createTableScan();
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ if (hasApplicableEqualityDeletes(scan)) {
+ return true;
+ }
+ Schema scanSchema = scan.schema();
+ Set<Integer> projectedFieldIds = projectedFieldIds(scanSchema);
+ Set<Integer> topLevelIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelIds.add(field.fieldId());
+ }
+ Map<Integer, NestedField> fieldsById =
TypeUtil.indexById(scanSchema.asStruct());
+ for (Integer fieldId : projectedFieldIds) {
+ NestedField field = fieldsById.get(fieldId);
+ if (field.initialDefault() != null
+ && (!topLevelIds.contains(field.fieldId()) ||
field.type().isNestedType())) {
+ return true;
+ }
+ }
+ if (hasProjectedNameAliasCollision(scanSchema, projectedFieldIds,
extractNameMapping())) {
+ return true;
+ }
+ return schemaHistoryRequiresMissingRequiredFieldRejection(
+ scanSchema, projectedFieldIds, reachableSchemas(icebergTable,
snapshot));
+ }
+
+ private boolean hasApplicableEqualityDeletes(TableScan scan) throws
UserException {
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ String equalityDeleteCount =
snapshot.summary().get("total-equality-deletes");
+ if (equalityDeleteCount != null) {
+ try {
+ // A positive snapshot total proves V2 semantics are required
without opening every
+ // delete manifest; only old summaries that omit the counter
need the fallback.
+ return Long.parseLong(equalityDeleteCount) > 0;
+ } catch (NumberFormatException ignored) {
+ // Fall through for non-standard summaries instead of
weakening compatibility.
+ }
+ }
+ // Inspect only delete manifests, not data tasks: equality-delete
semantics are snapshot-wide
+ // compatibility state even when the current predicate happens to
prune their partitions.
+ for (ManifestFile manifest :
snapshot.deleteManifests(icebergTable.io())) {
+ if (!manifest.hasAddedFiles() && !manifest.hasExistingFiles()) {
+ continue;
+ }
+ try (ManifestReader<DeleteFile> deletes =
ManifestFiles.readDeleteManifest(
+ manifest, icebergTable.io(), icebergTable.specs())) {
+ for (DeleteFile delete : deletes) {
+ if (delete.content() == FileContent.EQUALITY_DELETES) {
+ return true;
+ }
+ }
+ } catch (IOException e) {
+ throw new UserException(
+ "Failed to inspect Iceberg delete manifest " +
manifest.path(), e);
+ }
+ }
+ return false;
+ }
+
+ private static boolean hasSmoothUpgradeSource(Iterable<Backend> backends) {
+ for (Backend backend : backends) {
+ if (backend.isSmoothUpgradeSrc()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private Set<Integer> projectedFieldIds(Schema scanSchema) {
+ return projectedFieldIds(scanSchema, desc.getSlots());
+ }
+
+ @VisibleForTesting
+ static Set<Integer> projectedFieldIds(Schema scanSchema,
Iterable<SlotDescriptor> slots) {
+ Set<Integer> projected = new HashSet<>();
+ for (SlotDescriptor slot : slots) {
+ int fieldId = slot.getColumn().getUniqueId();
+ // Stable Iceberg IDs prevent a dropped-and-readded name from
selecting the wrong history.
+ NestedField field = fieldId >= 0 ? scanSchema.findField(fieldId)
+ :
scanSchema.caseInsensitiveFindField(slot.getColumn().getName());
+ if (field != null) {
+ collectProjectedFieldIds(field, slot.getType(), projected);
+ }
+ }
+ return projected;
+ }
+
+ private static void collectProjectedFieldIds(
+ NestedField field, org.apache.doris.catalog.Type projectedType,
+ Set<Integer> projected) {
+ projected.add(field.fieldId());
+ if (projectedType instanceof StructType &&
field.type().isStructType()) {
+ for (StructField projectedChild : ((StructType)
projectedType).getFields()) {
+ NestedField icebergChild =
field.type().asStructType().fields().stream()
+ .filter(child ->
child.name().equalsIgnoreCase(projectedChild.getName()))
+ .findFirst().orElse(null);
+ if (icebergChild != null) {
+ collectProjectedFieldIds(icebergChild,
projectedChild.getType(), projected);
+ }
+ }
+ } else if (projectedType instanceof ArrayType &&
field.type().isListType()) {
+ collectProjectedFieldIds(field.type().asListType().fields().get(0),
+ ((ArrayType) projectedType).getItemType(), projected);
+ } else if (projectedType instanceof MapType &&
field.type().isMapType()) {
+ collectProjectedFieldIds(field.type().asMapType().fields().get(0),
+ ((MapType) projectedType).getKeyType(), projected);
+ collectProjectedFieldIds(field.type().asMapType().fields().get(1),
+ ((MapType) projectedType).getValueType(), projected);
+ }
+ }
+
+ @VisibleForTesting
+ static Iterable<Schema> reachableSchemas(Table table, Snapshot
selectedSnapshot) {
+ Map<Integer, Schema> schemas = table.schemas();
+ List<Schema> reachable = new ArrayList<>();
+ Set<Long> visitedSnapshots = new HashSet<>();
+ Set<Integer> visitedSchemaIds = new HashSet<>();
+ Snapshot snapshot = selectedSnapshot;
+ while (snapshot != null &&
visitedSnapshots.add(snapshot.snapshotId())) {
+ Schema schema = schemas.get(snapshot.schemaId());
+ if (schema != null && visitedSchemaIds.add(snapshot.schemaId())) {
+ reachable.add(schema);
+ }
+ Map<String, String> summary = snapshot.summary();
+ String sourceSnapshotId = summary == null
+ ? null :
summary.get(SnapshotSummary.SOURCE_SNAPSHOT_ID_PROP);
+ if (sourceSnapshotId != null) {
+ Snapshot sourceSnapshot;
+ try {
+ sourceSnapshot =
table.snapshot(Long.parseLong(sourceSnapshotId));
+ } catch (NumberFormatException e) {
+ sourceSnapshot = null;
+ }
+ if (sourceSnapshot == null ||
!schemas.containsKey(sourceSnapshot.schemaId())) {
+ // A cherry-picked snapshot can contribute live files
outside the selected
+ // parent chain; unverifiable provenance must
conservatively gate all schemas.
+ for (Schema historicalSchema : schemas.values()) {
+ if (visitedSchemaIds.add(historicalSchema.schemaId()))
{
+ reachable.add(historicalSchema);
+ }
+ }
+ } else if (visitedSchemaIds.add(sourceSnapshot.schemaId())) {
+ reachable.add(schemas.get(sourceSnapshot.schemaId()));
+ }
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null : table.snapshot(parentId);
+ if (parentId != null && snapshot == null) {
+ // Expiration may remove the parent metadata while descendants
still inherit its
+ // files, so an incomplete lineage must gate against every
surviving schema.
+ for (Schema historicalSchema : schemas.values()) {
+ if (visitedSchemaIds.add(historicalSchema.schemaId())) {
+ reachable.add(historicalSchema);
+ }
+ }
+ }
+ }
+ // Only ancestors of the selected ref can have produced files visible
to this scan.
+ return reachable;
+ }
+
+ @VisibleForTesting
+ static void checkIcebergScanSemanticsV2Compatibility(
+ boolean requiresV2, Iterable<Backend> backends) throws
UserException {
+ if (!requiresV2) {
+ return;
+ }
+ for (Backend backend : backends) {
+ if (backend.isSmoothUpgradeSrc()) {
+ // A V1 BE accepts the Thrift field but does not enforce
nested defaults/requiredness.
+ throw new UserException("Current Iceberg scan semantics are
unavailable while backend "
+ + backend.getId() + " is a smooth upgrade source");
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static boolean schemaHistoryRequiresMissingRequiredFieldRejection(
+ Schema scanSchema, Iterable<Schema> historicalSchemas) {
+ return schemaHistoryRequiresMissingRequiredFieldRejection(
+ scanSchema,
TypeUtil.indexById(scanSchema.asStruct()).keySet(), historicalSchemas);
+ }
+
+ private static boolean schemaHistoryRequiresMissingRequiredFieldRejection(
+ Schema scanSchema, Set<Integer> projectedFieldIds,
Iterable<Schema> historicalSchemas) {
+ Map<Integer, NestedField> currentFields =
TypeUtil.indexById(scanSchema.asStruct());
+ Map<Integer, Integer> parentById =
TypeUtil.indexParents(scanSchema.asStruct());
+ Set<Integer> collectionWrapperIds = new HashSet<>();
+ collectCollectionWrapperFieldIds(scanSchema.asStruct(),
collectionWrapperIds);
+ for (Schema historicalSchema : historicalSchemas) {
+ Map<Integer, NestedField> historicalFields =
TypeUtil.indexById(historicalSchema.asStruct());
+ for (Integer fieldId : projectedFieldIds) {
+ NestedField field = currentFields.get(fieldId);
+ if (field == null ||
collectionWrapperIds.contains(field.fieldId())
Review Comment:
Thanks. This issue is part of the V1/V2 compatibility and File Scanner read
path. Per the current PR scope, compatibility fixes and additional File Scanner
V1 issues are intentionally not being addressed here, so no code change is
planned for this thread.
--
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]
