morningman commented on code in PR #69536:
URL: https://github.com/apache/doris/pull/69536#discussion_r4363772474
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -2280,37 +2280,48 @@ private Set<Integer>
cachedApplicableEqualityDeleteFieldIds(Table table, TableSc
* delete manifests does not enumerate data files or byte-split tasks, and
retained state is bounded by the
* number of equality keys rather than the table's entire schema history.
*
- * <p>The metadata schema list is searched in reverse chronology up to the
selected schema (schema IDs are
- * identifiers, not a sequence), so the latest definition of each stable
field ID wins. Current fields remain
- * first, allowing a dropped/re-added name to resolve the projected field
by name while a live historical
- * equality key resolves by its stable field ID.</p>
+ * <p>Schema IDs are identifiers, not a sequence: Iceberg reuses an ID
when evolution returns to an earlier
+ * schema, so a schema's position in the metadata list does not tell when
it was last active, and the list may
+ * also hold schemas committed after a time-travel or branch target. The
scan snapshot's parent chain is
+ * therefore searched first, so the definition active on that lineage
wins; the whole metadata list, newest
+ * first, is the fallback for schema-only changes and expired ancestors.
Current fields remain first, allowing
+ * a dropped/re-added name to resolve the projected field by name while a
live historical equality key
+ * resolves by its stable field ID.</p>
*/
@VisibleForTesting
static List<NestedField> schemaForPotentialEqualityDeletes(
Table table, TableScan scan, Schema scanSchema) {
return schemaForPotentialEqualityDeletes(
- table, scanSchema, applicableEqualityDeleteFieldIds(table,
scan));
+ table, scan.snapshot(), scanSchema,
applicableEqualityDeleteFieldIds(table, scan));
}
- private static List<NestedField> schemaForPotentialEqualityDeletes(
- Table table, Schema scanSchema, Set<Integer> equalityFieldIds) {
- 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;
+ @VisibleForTesting
+ static List<NestedField> schemaForPotentialEqualityDeletes(
+ Table table, Snapshot scanSnapshot, Schema scanSchema,
Set<Integer> equalityFieldIds) {
Set<Integer> missing = new HashSet<>(equalityFieldIds);
missing.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
if (missing.isEmpty()) {
return scanSchema.columns();
}
List<NestedField> fields = new ArrayList<>(scanSchema.columns());
- for (int index = lastRelevantIndex; index >= 0 && !missing.isEmpty();
index--) {
+ // Snapshots on one lineage mostly share a schema, so each schema is
indexed at most once.
+ Set<Integer> lineageSchemaIds = new HashSet<>();
+ for (Snapshot snapshot = scanSnapshot; snapshot != null &&
!missing.isEmpty();
+ snapshot = snapshot.parentId() == null ? null :
table.snapshot(snapshot.parentId())) {
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId == null || !lineageSchemaIds.add(schemaId)) {
+ continue;
+ }
+ Schema lineageSchema = table.schemas().get(schemaId);
+ if (lineageSchema == null) {
+ throw new IllegalStateException(
+ "Iceberg snapshot schema " + schemaId + " is absent
from table metadata");
+ }
+ addHistoricalEqualityFields(fields, missing, lineageSchema);
Review Comment:
Confirmed, and fixed in 71f51edf754. With Iceberg 1.12 a pending `RowDelta`
is stamped with the schema that is current when it commits
(`SnapshotProducer.apply` → `base.currentSchemaId()`), and nothing validates
its equality field IDs against that schema, so a key promoted by a schema-only
change and dropped before the delete commits has no snapshot on the lineage
that records the promoted type. FileScannerV2 compares equality keys in the
carrier's type, so the INT carrier narrowed the LONG tombstone to NULL
(non-strict) and matched NULL keys; the V1 reader compares in the delete file's
own type and was not affected.
The carrier now keeps the lineage definition of a dropped key (name,
optionality, initial default) but takes the widest type its field ID has in any
retained schema, where wider means an Iceberg promotion
(`TypeUtil.isPromotionAllowed`: INT→LONG, FLOAT→DOUBLE, decimal precision).
Field IDs are never reused and promotion is monotonic and lossless, so that
type covers every file written for the field, and widening is harmless when it
comes from a schema after a time-travel target.
`equalityCarrierKeepsPromotedTypeOfKeyDroppedBeforeItsDeleteCommits` replays
the sequence with the SDK (promote, prepare the delete, drop, commit) and fails
without the widening (`expected: <long> but was: <int>`).
branch-4.2 resolves keys the same way
(`IcebergScanNode.getSchemaFieldsForScan`) and has the same FileScannerV2
comparison, so it carries the same defect; it will be passed to the 4.x
maintainers. One residual case is shared by every version:
`expireSnapshots().cleanExpiredMetadata(true)` can remove a schema-only schema
that no retained snapshot references, after which no metadata records the
promoted type; closing that needs a BE-side guard when a delete file's key type
is wider than the carrier, which is outside this PR.
--
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]
