github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3634913930
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -524,30 +567,354 @@ 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,
+ * so FE cannot first wait for their equality field IDs. Carry the latest
historical definition
+ * of every dropped primitive field; the field IDs referenced by any
planned equality delete are
+ * a subset of this bounded metadata. Iceberg field IDs are never reused.
+ */
+ @VisibleForTesting
+ List<NestedField> getSchemaFieldsForScan(
+ Schema scanSchema, boolean includeHistoricalEqualityFields) throws
UserException {
+ List<NestedField> fields = new ArrayList<>(scanSchema.columns());
+ if (isSystemTable || !includeHistoricalEqualityFields) {
+ return fields;
+ }
+
+ Set<Integer> carriedFieldIds = new HashSet<>(
+ TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ Preconditions.checkState(icebergTable instanceof HasTableOperations,
+ "Iceberg table does not expose metadata schema history: %s",
icebergTable.name());
+ List<Schema> schemaHistory = ((HasTableOperations) icebergTable)
+ .operations().current().schemas();
+ // 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);
+ addDroppedPrimitiveFields(fields, carriedFieldIds,
historicalSchema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null :
icebergTable.snapshot(parentId);
+ }
+ for (int index = schemaHistory.size() - 1; index >= 0; index--) {
+ addDroppedPrimitiveFields(fields, carriedFieldIds,
schemaHistory.get(index));
+ }
+ return fields;
+ }
+
+ private static void addDroppedPrimitiveFields(List<NestedField> fields,
+ Set<Integer> carriedFieldIds, Schema historicalSchema) {
+ for (NestedField field :
TypeUtil.indexById(historicalSchema.asStruct()).values()) {
+ if (field.type().isPrimitiveType() &&
carriedFieldIds.add(field.fieldId())) {
Review Comment:
[P1] Do not convert unrelated unsupported history fields
This appends every dropped primitive, and `getScanColumns()` later runs each
one through `IcebergUtils.parseField()`. In the pinned Iceberg 1.10.1 API
`TIMESTAMP_NANO` is a primitive, but `icebergPrimitiveTypeToDorisType()` has no
such case and throws. A table whose current schema and live equality key are
fully supported therefore becomes unqueryable merely because an unrelated
nano-timestamp was dropped earlier. Please carry only referenced live equality
IDs, or preserve/filter raw historical metadata so unsupported fields fail only
when they are actually required.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -524,30 +567,354 @@ 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,
+ * so FE cannot first wait for their equality field IDs. Carry the latest
historical definition
+ * of every dropped primitive field; the field IDs referenced by any
planned equality delete are
+ * a subset of this bounded metadata. Iceberg field IDs are never reused.
+ */
+ @VisibleForTesting
+ List<NestedField> getSchemaFieldsForScan(
+ Schema scanSchema, boolean includeHistoricalEqualityFields) throws
UserException {
+ List<NestedField> fields = new ArrayList<>(scanSchema.columns());
+ if (isSystemTable || !includeHistoricalEqualityFields) {
+ return fields;
+ }
+
+ Set<Integer> carriedFieldIds = new HashSet<>(
+ TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ Preconditions.checkState(icebergTable instanceof HasTableOperations,
+ "Iceberg table does not expose metadata schema history: %s",
icebergTable.name());
+ List<Schema> schemaHistory = ((HasTableOperations) icebergTable)
+ .operations().current().schemas();
+ // 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);
+ addDroppedPrimitiveFields(fields, carriedFieldIds,
historicalSchema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null :
icebergTable.snapshot(parentId);
+ }
+ for (int index = schemaHistory.size() - 1; index >= 0; index--) {
+ addDroppedPrimitiveFields(fields, carriedFieldIds,
schemaHistory.get(index));
+ }
+ return fields;
+ }
+
+ private static void addDroppedPrimitiveFields(List<NestedField> fields,
+ Set<Integer> carriedFieldIds, Schema historicalSchema) {
+ for (NestedField field :
TypeUtil.indexById(historicalSchema.asStruct()).values()) {
+ if (field.type().isPrimitiveType() &&
carriedFieldIds.add(field.fieldId())) {
+ fields.add(field);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static boolean requiresRecursiveInitialDefaultMaterialization(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots) {
+ Map<Integer, NestedField> fieldById =
TypeUtil.indexById(scanSchema.asStruct());
+ Set<Integer> topLevelFieldIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelFieldIds.add(field.fieldId());
+ }
+ for (SlotDescriptor slot : projectedSlots) {
+ Column column = slot.getColumn();
+ List<ColumnAccessPath> accessPaths = slot.getAllAccessPaths();
+ if (accessPaths != null && !accessPaths.isEmpty()) {
+ for (ColumnAccessPath accessPath : accessPaths) {
+ List<String> path = accessPath.getPath();
+ Preconditions.checkState(!path.isEmpty(),
+ "Iceberg column access path must not be empty");
+
Preconditions.checkState(matchesAccessPathComponent(column, path.get(0)),
+ "Iceberg access path root %s does not match column
%s", path.get(0),
+ column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column, path, 1, fieldById,
+ topLevelFieldIds.contains(column.getUniqueId()))) {
+ return true;
+ }
+ }
+ } else if (requiresRecursiveInitialDefaultMaterialization(
+ column, slot.getType(), fieldById,
+ topLevelFieldIds.contains(column.getUniqueId()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, org.apache.doris.catalog.Type projectedType,
+ Map<Integer, NestedField> fieldById, boolean isTopLevel) {
+ if (requiresInitialDefaultMaterialization(column, fieldById,
isTopLevel)) {
+ return true;
+ }
+ if (column.getChildren() == null) {
+ return false;
+ }
+ if (projectedType.isStructType()) {
+ for (StructField projectedField : ((StructType)
projectedType).getFields()) {
+ Column child = findChildByName(column,
projectedField.getName());
+ Preconditions.checkState(child != null,
+ "Projected Iceberg child %s is absent from column %s",
+ projectedField.getName(), column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ child, projectedField.getType(), fieldById, false)) {
+ return true;
+ }
+ }
+ } else if (projectedType.isArrayType()) {
+ Preconditions.checkState(column.getChildren().size() == 1,
+ "Iceberg array column %s must have one child",
column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), ((ArrayType)
projectedType).getItemType(),
+ fieldById, false)) {
+ return true;
+ }
+ } else if (projectedType.isMapType()) {
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ MapType mapType = (MapType) projectedType;
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), mapType.getKeyType(),
fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(1),
mapType.getValueType(), fieldById, false)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, List<String> path, int pathIndex,
+ Map<Integer, NestedField> fieldById, boolean isTopLevel) {
+ if (requiresInitialDefaultMaterialization(column, fieldById,
isTopLevel)) {
+ return true;
+ }
+ if (pathIndex == path.size()) {
+ return requiresRecursiveInitialDefaultMaterialization(column,
fieldById);
+ }
+
+ String component = path.get(pathIndex);
+ if (AccessPathInfo.ACCESS_NULL.equals(component)
+ || AccessPathInfo.ACCESS_OFFSET.equals(component)) {
+ return false;
+ }
+ Preconditions.checkState(column.getChildren() != null,
+ "Iceberg access path continues below primitive column %s",
column.getName());
+
+ if (AccessPathInfo.ACCESS_ALL.equals(component)) {
+ if (column.getType().isArrayType()) {
+ Preconditions.checkState(column.getChildren().size() == 1,
+ "Iceberg array column %s must have one child",
column.getName());
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), path, pathIndex + 1,
fieldById, false);
+ }
+ Preconditions.checkState(column.getType().isMapType(),
+ "Unexpected Iceberg access-all path below column %s",
column.getName());
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ Column key = column.getChildren().get(0);
+ // element_at(map, key) reads the complete key subtree, while any
path after '*'
+ // describes only the selected value subtree.
+ if (requiresInitialDefaultMaterialization(key, fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(key,
fieldById)) {
+ return true;
+ }
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(1), path, pathIndex + 1,
fieldById, false);
+ }
+ if (column.getType().isMapType()) {
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ int childIndex;
+ if (AccessPathInfo.ACCESS_MAP_KEYS.equals(component)) {
+ childIndex = 0;
+ } else {
+
Preconditions.checkState(AccessPathInfo.ACCESS_MAP_VALUES.equals(component),
+ "Unexpected Iceberg map access path component %s",
component);
+ childIndex = 1;
+ }
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(childIndex), path, pathIndex + 1,
fieldById, false);
+ }
+
+ Column child = findAccessPathChild(column, component);
+ Preconditions.checkState(child != null,
+ "Iceberg access path child %s is absent from column %s",
component,
+ column.getName());
+ return requiresRecursiveInitialDefaultMaterialization(
+ child, path, pathIndex + 1, fieldById, false);
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, Map<Integer, NestedField> fieldById) {
+ if (column.getChildren() == null) {
+ return false;
+ }
+ for (Column child : column.getChildren()) {
+ if (requiresInitialDefaultMaterialization(child, fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(child,
fieldById)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresInitialDefaultMaterialization(
+ Column column, Map<Integer, NestedField> fieldById, boolean
isTopLevel) {
+ NestedField field = fieldById.get(column.getUniqueId());
+ return field != null && field.initialDefault() != null
+ && (!isTopLevel || field.type().isNestedType());
+ }
+
+ private static boolean matchesAccessPathComponent(Column column, String
component) {
+ return Integer.toString(column.getUniqueId()).equals(component)
+ || column.getName().equalsIgnoreCase(component);
+ }
+
+ private static Column findAccessPathChild(Column column, String component)
{
+ for (Column child : column.getChildren()) {
+ if (matchesAccessPathComponent(child, component)) {
+ return child;
+ }
+ }
+ return null;
+ }
+
+ private static Column findChildByName(Column column, String childName) {
+ for (Column child : column.getChildren()) {
+ if (child.getName().equalsIgnoreCase(childName)) {
+ return child;
+ }
+ }
+ return null;
+ }
+
+ @VisibleForTesting
+ boolean mayRequireEqualityDeleteIdentitySemantics() throws UserException {
+ Snapshot snapshot = createTableScan().snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ Map<String, String> summary = snapshot.summary();
+ String equalityDeleteCount = summary == null
+ ? null : summary.get(IcebergUtils.TOTAL_EQUALITY_DELETES);
+ if (equalityDeleteCount != null && Long.parseLong(equalityDeleteCount)
> 0) {
+ return true;
+ }
+ // total-equality-deletes counts records, so a live zero-row
equality-delete file still
+ // reports 0 while carrying field IDs that may require historical
schema metadata. Missing
+ // counters have the same ambiguity. A non-empty delete-manifest list
may contain only
+ // position deletes, but conservatively using the current semantics is
safe and avoids
+ // reading manifest entries before asynchronous batch splits.
+ return mayRequireEqualityDeleteIdentitySemantics(
+ equalityDeleteCount,
!snapshot.deleteManifests(icebergTable.io()).isEmpty());
Review Comment:
[P1] Run this manifest read under the catalog authenticator
This helper is called from `createScanRangeLocations()` before
`getSplits()`, so `deleteManifests()` can load the manifest list through
`FileIO` outside `preExecutionAuthenticator`. On a Kerberized HDFS catalog an
uncached list is therefore opened without the existing Hadoop `doAs` boundary
whenever the summary counter is missing or zero, and planning can fail before
the authenticated split path is reached. Please execute this preflight under
the same authenticator (with consistent exception conversion), or avoid storage
I/O here.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -524,30 +567,354 @@ 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,
+ * so FE cannot first wait for their equality field IDs. Carry the latest
historical definition
+ * of every dropped primitive field; the field IDs referenced by any
planned equality delete are
+ * a subset of this bounded metadata. Iceberg field IDs are never reused.
+ */
+ @VisibleForTesting
+ List<NestedField> getSchemaFieldsForScan(
+ Schema scanSchema, boolean includeHistoricalEqualityFields) throws
UserException {
+ List<NestedField> fields = new ArrayList<>(scanSchema.columns());
+ if (isSystemTable || !includeHistoricalEqualityFields) {
+ return fields;
+ }
+
+ Set<Integer> carriedFieldIds = new HashSet<>(
+ TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ Preconditions.checkState(icebergTable instanceof HasTableOperations,
+ "Iceberg table does not expose metadata schema history: %s",
icebergTable.name());
+ List<Schema> schemaHistory = ((HasTableOperations) icebergTable)
+ .operations().current().schemas();
+ // 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);
+ addDroppedPrimitiveFields(fields, carriedFieldIds,
historicalSchema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null :
icebergTable.snapshot(parentId);
+ }
+ for (int index = schemaHistory.size() - 1; index >= 0; index--) {
+ addDroppedPrimitiveFields(fields, carriedFieldIds,
schemaHistory.get(index));
+ }
+ return fields;
+ }
+
+ private static void addDroppedPrimitiveFields(List<NestedField> fields,
+ Set<Integer> carriedFieldIds, Schema historicalSchema) {
+ for (NestedField field :
TypeUtil.indexById(historicalSchema.asStruct()).values()) {
+ if (field.type().isPrimitiveType() &&
carriedFieldIds.add(field.fieldId())) {
+ fields.add(field);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static boolean requiresRecursiveInitialDefaultMaterialization(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots) {
+ Map<Integer, NestedField> fieldById =
TypeUtil.indexById(scanSchema.asStruct());
+ Set<Integer> topLevelFieldIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelFieldIds.add(field.fieldId());
+ }
+ for (SlotDescriptor slot : projectedSlots) {
+ Column column = slot.getColumn();
+ List<ColumnAccessPath> accessPaths = slot.getAllAccessPaths();
+ if (accessPaths != null && !accessPaths.isEmpty()) {
+ for (ColumnAccessPath accessPath : accessPaths) {
+ List<String> path = accessPath.getPath();
+ Preconditions.checkState(!path.isEmpty(),
+ "Iceberg column access path must not be empty");
+
Preconditions.checkState(matchesAccessPathComponent(column, path.get(0)),
+ "Iceberg access path root %s does not match column
%s", path.get(0),
+ column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column, path, 1, fieldById,
+ topLevelFieldIds.contains(column.getUniqueId()))) {
+ return true;
+ }
+ }
+ } else if (requiresRecursiveInitialDefaultMaterialization(
+ column, slot.getType(), fieldById,
+ topLevelFieldIds.contains(column.getUniqueId()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, org.apache.doris.catalog.Type projectedType,
+ Map<Integer, NestedField> fieldById, boolean isTopLevel) {
+ if (requiresInitialDefaultMaterialization(column, fieldById,
isTopLevel)) {
+ return true;
+ }
+ if (column.getChildren() == null) {
+ return false;
+ }
+ if (projectedType.isStructType()) {
+ for (StructField projectedField : ((StructType)
projectedType).getFields()) {
+ Column child = findChildByName(column,
projectedField.getName());
+ Preconditions.checkState(child != null,
+ "Projected Iceberg child %s is absent from column %s",
+ projectedField.getName(), column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ child, projectedField.getType(), fieldById, false)) {
+ return true;
+ }
+ }
+ } else if (projectedType.isArrayType()) {
+ Preconditions.checkState(column.getChildren().size() == 1,
+ "Iceberg array column %s must have one child",
column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), ((ArrayType)
projectedType).getItemType(),
+ fieldById, false)) {
+ return true;
+ }
+ } else if (projectedType.isMapType()) {
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ MapType mapType = (MapType) projectedType;
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), mapType.getKeyType(),
fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(1),
mapType.getValueType(), fieldById, false)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, List<String> path, int pathIndex,
+ Map<Integer, NestedField> fieldById, boolean isTopLevel) {
+ if (requiresInitialDefaultMaterialization(column, fieldById,
isTopLevel)) {
+ return true;
+ }
+ if (pathIndex == path.size()) {
+ return requiresRecursiveInitialDefaultMaterialization(column,
fieldById);
+ }
+
+ String component = path.get(pathIndex);
+ if (AccessPathInfo.ACCESS_NULL.equals(component)
+ || AccessPathInfo.ACCESS_OFFSET.equals(component)) {
+ return false;
+ }
+ Preconditions.checkState(column.getChildren() != null,
+ "Iceberg access path continues below primitive column %s",
column.getName());
+
+ if (AccessPathInfo.ACCESS_ALL.equals(component)) {
+ if (column.getType().isArrayType()) {
+ Preconditions.checkState(column.getChildren().size() == 1,
+ "Iceberg array column %s must have one child",
column.getName());
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), path, pathIndex + 1,
fieldById, false);
+ }
+ Preconditions.checkState(column.getType().isMapType(),
+ "Unexpected Iceberg access-all path below column %s",
column.getName());
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ Column key = column.getChildren().get(0);
+ // element_at(map, key) reads the complete key subtree, while any
path after '*'
+ // describes only the selected value subtree.
+ if (requiresInitialDefaultMaterialization(key, fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(key,
fieldById)) {
+ return true;
+ }
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(1), path, pathIndex + 1,
fieldById, false);
+ }
+ if (column.getType().isMapType()) {
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ int childIndex;
+ if (AccessPathInfo.ACCESS_MAP_KEYS.equals(component)) {
+ childIndex = 0;
+ } else {
+
Preconditions.checkState(AccessPathInfo.ACCESS_MAP_VALUES.equals(component),
+ "Unexpected Iceberg map access path component %s",
component);
+ childIndex = 1;
+ }
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(childIndex), path, pathIndex + 1,
fieldById, false);
+ }
+
+ Column child = findAccessPathChild(column, component);
+ Preconditions.checkState(child != null,
+ "Iceberg access path child %s is absent from column %s",
component,
+ column.getName());
+ return requiresRecursiveInitialDefaultMaterialization(
+ child, path, pathIndex + 1, fieldById, false);
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, Map<Integer, NestedField> fieldById) {
+ if (column.getChildren() == null) {
+ return false;
+ }
+ for (Column child : column.getChildren()) {
+ if (requiresInitialDefaultMaterialization(child, fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(child,
fieldById)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresInitialDefaultMaterialization(
+ Column column, Map<Integer, NestedField> fieldById, boolean
isTopLevel) {
+ NestedField field = fieldById.get(column.getUniqueId());
+ return field != null && field.initialDefault() != null
Review Comment:
[P1] Fence the new required-field check during rolling upgrades
The new V1/V2 paths reject a physically missing required field with no
initial default, but this predicate is false whenever `initialDefault()` is
null (and it also excludes top-level primitive fields). Iceberg can create that
evolution explicitly via `allowIncompatibleChanges()`; if an older file
remains, a current BE now errors while a smooth-upgrade-source BE follows the
legacy missing-column path. Include selected required fields whose physical
presence is not guaranteed in the compatibility decision, or version/fence this
requiredness behavior separately.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -524,30 +567,354 @@ 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,
+ * so FE cannot first wait for their equality field IDs. Carry the latest
historical definition
+ * of every dropped primitive field; the field IDs referenced by any
planned equality delete are
+ * a subset of this bounded metadata. Iceberg field IDs are never reused.
+ */
+ @VisibleForTesting
+ List<NestedField> getSchemaFieldsForScan(
+ Schema scanSchema, boolean includeHistoricalEqualityFields) throws
UserException {
+ List<NestedField> fields = new ArrayList<>(scanSchema.columns());
+ if (isSystemTable || !includeHistoricalEqualityFields) {
+ return fields;
+ }
+
+ Set<Integer> carriedFieldIds = new HashSet<>(
+ TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ Preconditions.checkState(icebergTable instanceof HasTableOperations,
+ "Iceberg table does not expose metadata schema history: %s",
icebergTable.name());
+ List<Schema> schemaHistory = ((HasTableOperations) icebergTable)
+ .operations().current().schemas();
+ // 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);
+ addDroppedPrimitiveFields(fields, carriedFieldIds,
historicalSchema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null :
icebergTable.snapshot(parentId);
+ }
+ for (int index = schemaHistory.size() - 1; index >= 0; index--) {
+ addDroppedPrimitiveFields(fields, carriedFieldIds,
schemaHistory.get(index));
+ }
+ return fields;
+ }
+
+ private static void addDroppedPrimitiveFields(List<NestedField> fields,
+ Set<Integer> carriedFieldIds, Schema historicalSchema) {
+ for (NestedField field :
TypeUtil.indexById(historicalSchema.asStruct()).values()) {
+ if (field.type().isPrimitiveType() &&
carriedFieldIds.add(field.fieldId())) {
+ fields.add(field);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static boolean requiresRecursiveInitialDefaultMaterialization(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots) {
+ Map<Integer, NestedField> fieldById =
TypeUtil.indexById(scanSchema.asStruct());
+ Set<Integer> topLevelFieldIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelFieldIds.add(field.fieldId());
+ }
+ for (SlotDescriptor slot : projectedSlots) {
+ Column column = slot.getColumn();
+ List<ColumnAccessPath> accessPaths = slot.getAllAccessPaths();
+ if (accessPaths != null && !accessPaths.isEmpty()) {
+ for (ColumnAccessPath accessPath : accessPaths) {
+ List<String> path = accessPath.getPath();
+ Preconditions.checkState(!path.isEmpty(),
+ "Iceberg column access path must not be empty");
+
Preconditions.checkState(matchesAccessPathComponent(column, path.get(0)),
+ "Iceberg access path root %s does not match column
%s", path.get(0),
+ column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column, path, 1, fieldById,
+ topLevelFieldIds.contains(column.getUniqueId()))) {
+ return true;
+ }
+ }
+ } else if (requiresRecursiveInitialDefaultMaterialization(
+ column, slot.getType(), fieldById,
+ topLevelFieldIds.contains(column.getUniqueId()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, org.apache.doris.catalog.Type projectedType,
+ Map<Integer, NestedField> fieldById, boolean isTopLevel) {
+ if (requiresInitialDefaultMaterialization(column, fieldById,
isTopLevel)) {
+ return true;
+ }
+ if (column.getChildren() == null) {
+ return false;
+ }
+ if (projectedType.isStructType()) {
+ for (StructField projectedField : ((StructType)
projectedType).getFields()) {
+ Column child = findChildByName(column,
projectedField.getName());
+ Preconditions.checkState(child != null,
+ "Projected Iceberg child %s is absent from column %s",
+ projectedField.getName(), column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ child, projectedField.getType(), fieldById, false)) {
+ return true;
+ }
+ }
+ } else if (projectedType.isArrayType()) {
+ Preconditions.checkState(column.getChildren().size() == 1,
+ "Iceberg array column %s must have one child",
column.getName());
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), ((ArrayType)
projectedType).getItemType(),
+ fieldById, false)) {
+ return true;
+ }
+ } else if (projectedType.isMapType()) {
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ MapType mapType = (MapType) projectedType;
+ if (requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), mapType.getKeyType(),
fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(1),
mapType.getValueType(), fieldById, false)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, List<String> path, int pathIndex,
+ Map<Integer, NestedField> fieldById, boolean isTopLevel) {
+ if (requiresInitialDefaultMaterialization(column, fieldById,
isTopLevel)) {
+ return true;
+ }
+ if (pathIndex == path.size()) {
+ return requiresRecursiveInitialDefaultMaterialization(column,
fieldById);
+ }
+
+ String component = path.get(pathIndex);
+ if (AccessPathInfo.ACCESS_NULL.equals(component)
+ || AccessPathInfo.ACCESS_OFFSET.equals(component)) {
+ return false;
+ }
+ Preconditions.checkState(column.getChildren() != null,
+ "Iceberg access path continues below primitive column %s",
column.getName());
+
+ if (AccessPathInfo.ACCESS_ALL.equals(component)) {
+ if (column.getType().isArrayType()) {
+ Preconditions.checkState(column.getChildren().size() == 1,
+ "Iceberg array column %s must have one child",
column.getName());
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(0), path, pathIndex + 1,
fieldById, false);
+ }
+ Preconditions.checkState(column.getType().isMapType(),
+ "Unexpected Iceberg access-all path below column %s",
column.getName());
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ Column key = column.getChildren().get(0);
+ // element_at(map, key) reads the complete key subtree, while any
path after '*'
+ // describes only the selected value subtree.
+ if (requiresInitialDefaultMaterialization(key, fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(key,
fieldById)) {
+ return true;
+ }
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(1), path, pathIndex + 1,
fieldById, false);
+ }
+ if (column.getType().isMapType()) {
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ int childIndex;
+ if (AccessPathInfo.ACCESS_MAP_KEYS.equals(component)) {
+ childIndex = 0;
+ } else {
+
Preconditions.checkState(AccessPathInfo.ACCESS_MAP_VALUES.equals(component),
+ "Unexpected Iceberg map access path component %s",
component);
+ childIndex = 1;
+ }
+ return requiresRecursiveInitialDefaultMaterialization(
+ column.getChildren().get(childIndex), path, pathIndex + 1,
fieldById, false);
+ }
+
+ Column child = findAccessPathChild(column, component);
+ Preconditions.checkState(child != null,
+ "Iceberg access path child %s is absent from column %s",
component,
+ column.getName());
+ return requiresRecursiveInitialDefaultMaterialization(
+ child, path, pathIndex + 1, fieldById, false);
+ }
+
+ private static boolean requiresRecursiveInitialDefaultMaterialization(
+ Column column, Map<Integer, NestedField> fieldById) {
+ if (column.getChildren() == null) {
+ return false;
+ }
+ for (Column child : column.getChildren()) {
+ if (requiresInitialDefaultMaterialization(child, fieldById, false)
+ || requiresRecursiveInitialDefaultMaterialization(child,
fieldById)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresInitialDefaultMaterialization(
+ Column column, Map<Integer, NestedField> fieldById, boolean
isTopLevel) {
+ NestedField field = fieldById.get(column.getUniqueId());
+ return field != null && field.initialDefault() != null
+ && (!isTopLevel || field.type().isNestedType());
+ }
+
+ private static boolean matchesAccessPathComponent(Column column, String
component) {
+ return Integer.toString(column.getUniqueId()).equals(component)
+ || column.getName().equalsIgnoreCase(component);
+ }
+
+ private static Column findAccessPathChild(Column column, String component)
{
+ for (Column child : column.getChildren()) {
+ if (matchesAccessPathComponent(child, component)) {
+ return child;
+ }
+ }
+ return null;
+ }
+
+ private static Column findChildByName(Column column, String childName) {
+ for (Column child : column.getChildren()) {
+ if (child.getName().equalsIgnoreCase(childName)) {
+ return child;
+ }
+ }
+ return null;
+ }
+
+ @VisibleForTesting
+ boolean mayRequireEqualityDeleteIdentitySemantics() throws UserException {
+ Snapshot snapshot = createTableScan().snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ Map<String, String> summary = snapshot.summary();
+ String equalityDeleteCount = summary == null
+ ? null : summary.get(IcebergUtils.TOTAL_EQUALITY_DELETES);
+ if (equalityDeleteCount != null && Long.parseLong(equalityDeleteCount)
> 0) {
+ return true;
+ }
+ // total-equality-deletes counts records, so a live zero-row
equality-delete file still
+ // reports 0 while carrying field IDs that may require historical
schema metadata. Missing
+ // counters have the same ambiguity. A non-empty delete-manifest list
may contain only
+ // position deletes, but conservatively using the current semantics is
safe and avoids
+ // reading manifest entries before asynchronous batch splits.
+ return mayRequireEqualityDeleteIdentitySemantics(
+ equalityDeleteCount,
!snapshot.deleteManifests(icebergTable.io()).isEmpty());
+ }
+
+ @VisibleForTesting
+ static boolean mayRequireEqualityDeleteIdentitySemantics(
+ String equalityDeleteCount, boolean hasDeleteManifests) {
+ return hasDeleteManifests
Review Comment:
[P1] Keep position-delete-only scans available during upgrades
`hasDeleteManifests` is true for position-delete manifests too, so the
explicitly tested `(total-equality-deletes = 0, has manifests)` case activates
equality semantics and the caller rejects every smooth-upgrade-source BE.
Ordinary position-delete data scans already worked before this patch and need
none of the new equality-key/default behavior, so they now lose availability
for the full upgrade window. Treat an explicit zero equality count as no
equality semantics, or distinguish live equality entries under authentication
instead of gating on every delete manifest.
--
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]
