From dc217482a94425cabd30b16bba72fe70c8cf784e Mon Sep 17 00:00:00 2001 From: morningman Date: Wed, 21 Oct 2026 17:53:15 +0100 Subject: [PATCH 1/2] [fix](fe) Resolve dropped Iceberg equality-delete keys after a schema ID is reused ### What problem does this PR solve? Issue Number: None Related PR: #66538, #69394 Problem Summary: When an Iceberg table drops a column so that its schema becomes equal to an earlier one, Iceberg reuses the earlier schema ID instead of adding a new schema, and the metadata schema list keeps that schema at its original, earlier position. The equality-delete schema carrier in the Iceberg connector located the scan schema in that list by ID and searched only the schemas listed before it. A live equality delete keyed on the dropped column therefore could not find the schema that still defines the key, and every scan of the table failed with "Iceberg equality-delete fields are absent from schema history". This happens in steady state, not only during an upgrade. Reproduce: create a table with column id, add column k, write data and an equality delete keyed on k, drop k (the schema returns to the initial one and reuses its ID), then query the table. branch-4.1 (#66538) searches the scan snapshot's parent chain first, so the definition recorded on that lineage wins, and falls back to the whole metadata list, newest first, for schema-only changes and expired ancestors. This change ports that order to IcebergScanPlanProvider. Each lineage schema is indexed at most once and the walk stops as soon as every key is resolved. It also ports the branch-4.1 tests for a reused schema ID (with and without an expired parent) and for time travel past a later rename. The existing test equalityCarrierKeepsLatestDefinitionOfDroppedEqualityKey (#69394) renamed and then dropped the key with no write in between. It now writes a snapshot after each schema change, as the branch-4.1 test it was ported from does, and still expects k2. ### Release note Fix Iceberg queries that failed with "equality-delete fields are absent from schema history" after a column used as an equality-delete key was dropped and the table schema returned to an earlier one. ### Check List (For Author) - Test: Unit Test - fe-connector-iceberg: 4 new cases and 1 adjusted case in IcebergScanPlanProviderTest; the two reused-ID cases fail without the fix; the whole module passes - Behavior changed: Yes. Such queries no longer fail. When a dropped key has several historical definitions, the carrier now prefers the one recorded on the scan snapshot's lineage, as branch-4.1 does. - Does this need documentation: No Co-authored-by: daidai Co-Authored-By: Claude Opus 5.5 (1M context) --- .../iceberg/IcebergScanPlanProvider.java | 47 +++-- .../iceberg/IcebergScanPlanProviderTest.java | 161 +++++++++++++++++- 2 files changed, 189 insertions(+), 19 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java index d3245f604d1011..4125af104b4b86 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java @@ -2108,7 +2108,7 @@ public Map getScanNodeProperties( // every branch so the default and current field type match BE's read. if (hasApplicableEqualityDeletes) { List equalityFields = schemaForPotentialEqualityDeletes( - table, scanSchema, applicableEqualityDeleteFieldIds); + table, exactScan.snapshot(), scanSchema, applicableEqualityDeleteFieldIds); dict = IcebergSchemaUtils.encodeEqualitySchemaEvolutionProp( table, equalityFields, appendRowLineage, enableVarbinary, enableTimestampTz); @@ -2280,29 +2280,24 @@ private Set 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. * - *

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.

+ *

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.

*/ @VisibleForTesting static List schemaForPotentialEqualityDeletes( Table table, TableScan scan, Schema scanSchema) { return schemaForPotentialEqualityDeletes( - table, scanSchema, applicableEqualityDeleteFieldIds(table, scan)); + table, scan.snapshot(), scanSchema, applicableEqualityDeleteFieldIds(table, scan)); } - private static List schemaForPotentialEqualityDeletes( - Table table, Schema scanSchema, Set equalityFieldIds) { - List 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 schemaForPotentialEqualityDeletes( + Table table, Snapshot scanSnapshot, Schema scanSchema, Set equalityFieldIds) { Set missing = new HashSet<>(equalityFieldIds); missing.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet()); if (missing.isEmpty()) { @@ -2310,7 +2305,23 @@ private static List schemaForPotentialEqualityDeletes( } List 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 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); + } + List metadataSchemas = metadataSchemaHistory(table); + for (int index = metadataSchemas.size() - 1; index >= 0 && !missing.isEmpty(); index--) { addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index)); } if (!missing.isEmpty()) { diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java index b0a6866de3ad98..2ccdd913a6583c 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java @@ -108,6 +108,7 @@ import java.util.HashMap; import java.util.IdentityHashMap; import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.NoSuchElementException; @@ -1925,7 +1926,8 @@ null, new IcebergTableHandle("db1", "dropped_eq_key"), @Test public void equalityCarrierKeepsLatestDefinitionOfDroppedEqualityKey() throws Exception { // BE fills an equality key that an old data file lacks from the key's carrier entry, so a dropped key must - // resolve to its latest historical definition: its last name, its optionality and its initial default. + // resolve to the latest definition on the scan's lineage: its last name, its optionality and its initial + // default. Each schema change is followed by a write, as on a table that keeps ingesting. Schema schema = new Schema( Types.NestedField.required(1, "id", Types.IntegerType.get()), Types.NestedField.optional("k").withId(2).ofType(Types.IntegerType.get()) @@ -1938,7 +1940,11 @@ public void equalityCarrierKeepsLatestDefinitionOfDroppedEqualityKey() throws Ex table.newRowDelta().addDeletes(equalityDeleteFile( "s3://b/db/renamed_dropped_eq_key/eq.parquet", FileFormat.PARQUET, keyId)).commit(); table.updateSchema().renameColumn("k", "k2").commit(); + table.newAppend().appendFile(dataFile(table.spec(), + "s3://b/db/renamed_dropped_eq_key/f2.parquet", 1024, null, null)).commit(); table.updateSchema().deleteColumn("k2").commit(); + table.newAppend().appendFile(dataFile(table.spec(), + "s3://b/db/renamed_dropped_eq_key/f3.parquet", 1024, null, null)).commit(); IcebergScanPlanProvider provider = providerOver(table); Map props = provider.getScanNodeProperties( @@ -2069,6 +2075,159 @@ public void equalityCarrierSizeDoesNotGrowWithUnrelatedSchemaHistory() throws Ex Assertions.assertEquals(table.schema().columns(), carrier); } + @Test + public void equalityCarrierResolvesDroppedKeyWhenSchemaIdIsReused() throws Exception { + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); + Table table = createTable("reused_schema_id_eq_key", schema, PartitionSpec.unpartitioned(), + Collections.singletonMap(TableProperties.FORMAT_VERSION, "2")); + int initialSchemaId = table.schema().schemaId(); + table.updateSchema().addColumn("k", Types.StringType.get()).commit(); + int keyId = table.schema().findField("k").fieldId(); + table.newAppend().appendFile(dataFile(table.spec(), + "s3://b/db/reused_schema_id_eq_key/f1.parquet", 1024, null, null)).commit(); + table.newRowDelta().addDeletes(equalityDeleteFile( + "s3://b/db/reused_schema_id_eq_key/eq.parquet", FileFormat.PARQUET, keyId)).commit(); + // Dropping k makes the schema equal to the initial one, so Iceberg reuses the initial schema ID, which the + // metadata list holds before the schema that still defines the equality key. The next snapshot carries the + // reused ID; only its parent records the schema with k. + table.updateSchema().deleteColumn("k").commit(); + table.newAppend().appendFile(dataFile(table.spec(), + "s3://b/db/reused_schema_id_eq_key/f2.parquet", 1024, null, null)).commit(); + Assertions.assertEquals(initialSchemaId, table.schema().schemaId()); + Assertions.assertEquals(initialSchemaId, table.currentSnapshot().schemaId()); + + List carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes( + table, table.newScan(), table.schema()); + Assertions.assertEquals(2, carrier.size(), "the dropped equality key must stay resolvable: " + carrier); + Assertions.assertEquals(keyId, carrier.get(1).fieldId()); + Assertions.assertEquals("k", carrier.get(1).name()); + + IcebergScanPlanProvider provider = providerOver(table); + Map props = provider.getScanNodeProperties( + null, new IcebergTableHandle("db1", "reused_schema_id_eq_key"), + Collections.singletonList(new IcebergColumnHandle("id", 1)), Optional.empty()); + TFileScanRangeParams params = new TFileScanRangeParams(); + provider.populateScanLevelParams(params, props); + List fields = params.getHistorySchemaInfo().get(0).getRootField().getFields(); + Assertions.assertTrue(fields.stream().anyMatch(field -> field.getFieldPtr().getId() == keyId), + "the scan dictionary must carry the dropped equality key after the schema ID is reused"); + } + + @Test + public void equalityCarrierResolvesReusedSchemaIdAfterParentSnapshotsExpire() throws Exception { + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); + Table table = createTable("reused_schema_id_expired_parent", schema, PartitionSpec.unpartitioned(), + Collections.singletonMap(TableProperties.FORMAT_VERSION, "2")); + table.updateSchema().addColumn("k", Types.StringType.get()).commit(); + int keyId = table.schema().findField("k").fieldId(); + table.newAppend().appendFile(dataFile(table.spec(), + "s3://b/db/reused_schema_id_expired_parent/f1.parquet", 1024, null, null)).commit(); + long appendSnapshotId = table.currentSnapshot().snapshotId(); + table.newRowDelta().addDeletes(equalityDeleteFile( + "s3://b/db/reused_schema_id_expired_parent/eq.parquet", FileFormat.PARQUET, keyId)).commit(); + long deleteSnapshotId = table.currentSnapshot().snapshotId(); + table.updateSchema().deleteColumn("k").commit(); + table.newAppend().appendFile(dataFile(table.spec(), + "s3://b/db/reused_schema_id_expired_parent/f2.parquet", 1024, null, null)).commit(); + // With the snapshots that recorded k expired, the scan lineage ends at the reused-ID snapshot, so the key + // can only come from the metadata schema list, where it sits after the reused schema. + table.expireSnapshots().expireSnapshotId(appendSnapshotId).expireSnapshotId(deleteSnapshotId) + .cleanExpiredFiles(false).commit(); + Assertions.assertNull(table.snapshot(table.currentSnapshot().parentId())); + + List carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes( + table, table.newScan(), table.schema()); + Assertions.assertEquals(2, carrier.size(), "the dropped equality key must stay resolvable: " + carrier); + Assertions.assertEquals(keyId, carrier.get(1).fieldId()); + Assertions.assertEquals("k", carrier.get(1).name()); + } + + @Test + public void equalityCarrierPrefersScanLineageOverLaterRenameForTimeTravel() { + // The metadata list also holds schemas committed after a time-travel target. A later schema that renames + // the key must not win over the definition recorded on the target's own lineage. + NestedField id = Types.NestedField.required(1, "id", Types.LongType.get()); + Schema schemaWithEqualityKey = new Schema(100, Arrays.asList(id, intKeyWithDefault("k"))); + Schema timeTravelSchema = new Schema(1, Collections.singletonList(id)); + Schema laterSchema = new Schema(2, Arrays.asList(id, intKeyWithDefault("k2"))); + Snapshot timeTravelSnapshot = snapshotOf(1001L, 1000L, timeTravelSchema); + Table table = tableWithHistory(Arrays.asList(schemaWithEqualityKey, timeTravelSchema, laterSchema), + snapshotOf(1000L, null, schemaWithEqualityKey), timeTravelSnapshot); + + List carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes( + table, timeTravelSnapshot, timeTravelSchema, Collections.singleton(7)); + Assertions.assertEquals(2, carrier.size()); + Assertions.assertEquals(7, carrier.get(1).fieldId()); + Assertions.assertEquals("k", carrier.get(1).name()); + } + + @Test + public void equalityCarrierKeepsDefaultWhenTimeTravelParentExpiredBeforeLaterRename() { + // Once the lineage that recorded the key has expired, the metadata list is the only source left. Its newest + // definition may come from a later rename, but the field ID and its initial default stay the same. + NestedField id = Types.NestedField.required(1, "id", Types.LongType.get()); + Schema schemaWithEqualityKey = new Schema(100, Arrays.asList(id, intKeyWithDefault("k"))); + Schema timeTravelSchema = new Schema(1, Collections.singletonList(id)); + Schema laterSchema = new Schema(2, Arrays.asList(id, intKeyWithDefault("k2"))); + Snapshot timeTravelSnapshot = snapshotOf(1001L, 1000L, timeTravelSchema); + Table table = tableWithHistory(Arrays.asList(schemaWithEqualityKey, timeTravelSchema, laterSchema), + timeTravelSnapshot); + + List carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes( + table, timeTravelSnapshot, timeTravelSchema, Collections.singleton(7)); + Assertions.assertEquals(2, carrier.size()); + Assertions.assertEquals(7, carrier.get(1).fieldId()); + Assertions.assertEquals(7, carrier.get(1).initialDefault()); + } + + private static NestedField intKeyWithDefault(String name) { + return Types.NestedField.optional(name).withId(7).ofType(Types.IntegerType.get()) + .withInitialDefault(7).build(); + } + + /** A snapshot double exposing only its lineage and the ID of the schema it was written with. */ + private static Snapshot snapshotOf(long snapshotId, Long parentId, Schema schema) { + return (Snapshot) Proxy.newProxyInstance(Snapshot.class.getClassLoader(), new Class[] {Snapshot.class}, + (proxy, method, args) -> { + switch (method.getName()) { + case "snapshotId": + return snapshotId; + case "parentId": + return parentId; + case "schemaId": + return schema.schemaId(); + default: + throw new UnsupportedOperationException(method.getName()); + } + }); + } + + /** + * A table double exposing only schema history, in metadata-list order, and the retained snapshots. It allows + * schema histories that the SDK cannot build, such as a later schema that renames a dropped field ID. + */ + private static Table tableWithHistory(List schemas, Snapshot... snapshots) { + Map schemasById = new LinkedHashMap<>(); + for (Schema schema : schemas) { + schemasById.put(schema.schemaId(), schema); + } + Map snapshotsById = new HashMap<>(); + for (Snapshot snapshot : snapshots) { + snapshotsById.put(snapshot.snapshotId(), snapshot); + } + return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(), new Class[] {Table.class}, + (proxy, method, args) -> { + switch (method.getName()) { + case "schemas": + return schemasById; + case "snapshot": + return snapshotsById.get((Long) args[0]); + default: + throw new UnsupportedOperationException(method.getName()); + } + }); + } + @Test public void partitionPrunedEqualityDeleteConservativelyRequiresCurrentBackendSemantics() { PartitionSpec spec = PartitionSpec.builderFor(PART_SCHEMA).identity("p").build(); From 71f51edf754b1d8088aa6a7cacb547bbf54b1e8d Mon Sep 17 00:00:00 2001 From: morningman Date: Fri, 23 Oct 2026 22:45:10 +0800 Subject: [PATCH 2/2] [fix](iceberg) Keep the promoted type of a dropped equality-delete key ### What problem does this PR solve? Issue Number: None Related PR: #66538 Problem Summary: Review follow-up. The previous commit resolves an equality-delete key that the scan schema no longer has from the scan snapshot's parent chain first. With Iceberg 1.12 a pending RowDelta is stamped with the schema that is current when it commits, and nothing checks its equality field IDs against that schema. So for a key k promoted from INT to LONG by a schema-only change (no snapshot), a writer that wrote LONG tombstones, and k dropped before that delete commits, the delete snapshot records the schema without k and its parent the INT one: no snapshot on the lineage records LONG. The carrier then said INT, FileScannerV2 compares equality keys in the carrier's type, and a tombstone such as 2147483648 was cast to NULL in non-strict mode and deleted every row whose k is NULL (strict mode failed the query). The previous metadata-list lookup found LONG. A dropped key now keeps its lineage definition (name, optionality, initial default) but takes the widest type its field ID has in any retained schema. Field IDs are never reused and promotion is monotonic and lossless, so that type covers every file written for the field, also when it comes from a schema after a time-travel target. ### Release note None ### Check List (For Author) - Test: Unit Test - IcebergScanPlanProviderTest#equalityCarrierKeepsPromotedTypeOfKeyDroppedBeforeItsDeleteCommits replays the sequence with the SDK; it fails without the widening - fe-connector-iceberg module tests - Behavior changed: No - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 (1M context) --- .../iceberg/IcebergScanPlanProvider.java | 53 +++++++++++++++---- .../iceberg/IcebergScanPlanProviderTest.java | 30 +++++++++++ 2 files changed, 74 insertions(+), 9 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java index 4125af104b4b86..c749f3472fbcd0 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java @@ -2284,9 +2284,11 @@ private Set cachedApplicableEqualityDeleteFieldIds(Table table, TableSc * 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.

+ * first, is the fallback for schema-only changes and expired ancestors. A dropped key nevertheless takes the + * widest type it has in any retained schema: a delete committed after a schema-only promotion holds values of + * the promoted type even when no snapshot records it. 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.

*/ @VisibleForTesting static List schemaForPotentialEqualityDeletes( @@ -2305,6 +2307,8 @@ static List schemaForPotentialEqualityDeletes( } List fields = new ArrayList<>(scanSchema.columns()); + List metadataSchemas = metadataSchemaHistory(table); + Map widestTypes = widestPromotedTypes(missing, metadataSchemas); // Snapshots on one lineage mostly share a schema, so each schema is indexed at most once. Set lineageSchemaIds = new HashSet<>(); for (Snapshot snapshot = scanSnapshot; snapshot != null && !missing.isEmpty(); @@ -2318,11 +2322,10 @@ static List schemaForPotentialEqualityDeletes( throw new IllegalStateException( "Iceberg snapshot schema " + schemaId + " is absent from table metadata"); } - addHistoricalEqualityFields(fields, missing, lineageSchema); + addHistoricalEqualityFields(fields, missing, lineageSchema, widestTypes); } - List metadataSchemas = metadataSchemaHistory(table); for (int index = metadataSchemas.size() - 1; index >= 0 && !missing.isEmpty(); index--) { - addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index)); + addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index), widestTypes); } if (!missing.isEmpty()) { throw new IllegalStateException( @@ -2338,8 +2341,40 @@ private static List metadataSchemaHistory(Table table) { return new ArrayList<>(table.schemas().values()); } - private static void addHistoricalEqualityFields( - List fields, Set missingFieldIds, Schema historicalSchema) { + /** + * Field IDs are never reused and a field's type only changes by promotion, which is monotonic, so the widest + * type a field has in any retained schema covers every file written for it. + */ + private static Map widestPromotedTypes(Set fieldIds, List schemas) { + Map widestTypes = new HashMap<>(); + for (Schema schema : schemas) { + for (Integer fieldId : fieldIds) { + Type type = schema.findType(fieldId); + if (type == null) { + continue; + } + Type widest = widestTypes.get(fieldId); + if (widest == null || TypeUtil.isPromotionAllowed(widest, type.asPrimitiveType())) { + widestTypes.put(fieldId, type); + } + } + } + return widestTypes; + } + + private static List withWidestTypes(List fields, Map widestTypes) { + List widened = new ArrayList<>(fields.size()); + for (NestedField field : fields) { + Type type = field.type().isStructType() + ? Types.StructType.of(withWidestTypes(field.type().asStructType().fields(), widestTypes)) + : widestTypes.getOrDefault(field.fieldId(), field.type()); + widened.add(type.equals(field.type()) ? field : Types.NestedField.from(field).ofType(type).build()); + } + return widened; + } + + private static void addHistoricalEqualityFields(List fields, Set missingFieldIds, + Schema historicalSchema, Map widestTypes) { Map historicalFields = TypeUtil.indexById(historicalSchema.asStruct()); Set selectedFieldIds = new HashSet<>(); for (Integer fieldId : missingFieldIds) { @@ -2356,7 +2391,7 @@ private static void addHistoricalEqualityFields( return; } Schema selectedSchema = TypeUtil.select(historicalSchema, selectedFieldIds); - mergeHistoricalEqualityFields(fields, selectedSchema.columns()); + mergeHistoricalEqualityFields(fields, withWidestTypes(selectedSchema.columns(), widestTypes)); missingFieldIds.removeAll(selectedFieldIds); } diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java index 2ccdd913a6583c..1df6bf8578510e 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java @@ -62,6 +62,7 @@ import org.apache.iceberg.Metrics; import org.apache.iceberg.PartitionData; import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotSummary; @@ -2113,6 +2114,35 @@ null, new IcebergTableHandle("db1", "reused_schema_id_eq_key"), "the scan dictionary must carry the dropped equality key after the schema ID is reused"); } + @Test + public void equalityCarrierKeepsPromotedTypeOfKeyDroppedBeforeItsDeleteCommits() throws Exception { + // A writer started after a schema-only promotion of k to LONG writes LONG tombstones, and k is dropped before + // its delete commits. The delete snapshot records the schema without k and its parent the INT one, so only + // the metadata list knows LONG; an INT carrier would narrow 2147483648 to NULL and match NULL keys. + Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "k", Types.IntegerType.get())); + Table table = createTable("promoted_dropped_eq_key", schema, PartitionSpec.unpartitioned(), + Collections.singletonMap(TableProperties.FORMAT_VERSION, "2")); + table.newAppend().appendFile(dataFile(table.spec(), + "s3://b/db/promoted_dropped_eq_key/f1.parquet", 1024, null, null)).commit(); + int intSchemaId = table.currentSnapshot().schemaId(); + table.updateSchema().updateColumn("k", Types.LongType.get()).commit(); + RowDelta pendingDelete = table.newRowDelta().addDeletes(equalityDeleteFile( + "s3://b/db/promoted_dropped_eq_key/eq.parquet", FileFormat.PARQUET, 2)); + table.updateSchema().deleteColumn("k").commit(); + pendingDelete.commit(); + Assertions.assertNull(table.schemas().get(table.currentSnapshot().schemaId()).findField(2)); + Assertions.assertEquals(intSchemaId, table.snapshot(table.currentSnapshot().parentId()).schemaId()); + + List carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes( + table, table.newScan(), table.schema()); + Assertions.assertEquals(2, carrier.size(), "the dropped equality key must stay resolvable: " + carrier); + Assertions.assertEquals(2, carrier.get(1).fieldId()); + Assertions.assertEquals("k", carrier.get(1).name()); + Assertions.assertEquals(Types.LongType.get(), carrier.get(1).type()); + } + @Test public void equalityCarrierResolvesReusedSchemaIdAfterParentSnapshotsExpire() throws Exception { Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));