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..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 @@ -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,26 @@ 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. 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( 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,8 +2307,25 @@ private static List schemaForPotentialEqualityDeletes( } List fields = new ArrayList<>(scanSchema.columns()); - for (int index = lastRelevantIndex; index >= 0 && !missing.isEmpty(); index--) { - addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index)); + 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(); + 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, widestTypes); + } + for (int index = metadataSchemas.size() - 1; index >= 0 && !missing.isEmpty(); index--) { + addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index), widestTypes); } if (!missing.isEmpty()) { throw new IllegalStateException( @@ -2327,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) { @@ -2345,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 b0a6866de3ad98..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; @@ -108,6 +109,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 +1927,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 +1941,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 +2076,188 @@ 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 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())); + 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();