github-actions[bot] commented on code in PR #69536:
URL: https://github.com/apache/doris/pull/69536#discussion_r4359014047
##########
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:
[P1] Preserve promoted types when recovering a dropped equality key
Snapshot lineage can omit the type used by a pending writer. Start with
optional `k INT` and an existing row where `k IS NULL`; promote `k` to LONG
without a snapshot, prepare a LONG equality-delete file for `k=2147483648`,
drop `k`, then commit the pending `RowDelta`. Use unpartitioned files without
column metrics (including null counts and bounds). The SDK permits this:
[`SnapshotProducer.apply()`](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/SnapshotProducer.java)
refreshes metadata and stamps the current schema, and
[`BaseRowDelta`](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/BaseRowDelta.java)
does not reject the changed key schema. The delete snapshot therefore records
the dropped schema while its parent still records INT.
This call resolves the ID from that INT ancestor and removes it from
`missing`, so the retained schema-only LONG definition is never considered. V2
`_resolve_equality_delete_fields` takes the carrier's INT as its comparison
type; `_load_equality_delete_file` then casts the LONG tombstone to INT.
Overflow becomes NULL in non-strict mode, and NULL-safe equality incorrectly
deletes the existing NULL-key row (strict mode fails the query). The previous
metadata-history lookup selected LONG and preserved that row. Preserve the
applicable promoted comparison type when choosing lineage definitions, and
cover the promotion/drop/pending-delete sequence in tests.
--
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]
