Skip to content

Commit dc21748

Browse files
morningmanhubgeterclaude
committed
[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 <[email protected]> Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
1 parent 2524893 commit dc21748

2 files changed

Lines changed: 189 additions & 19 deletions

File tree

‎fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java‎

Lines changed: 29 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -2108,7 +2108,7 @@ public Map<String, String> getScanNodeProperties(
21082108
// every branch so the default and current field type match BE's read.
21092109
if (hasApplicableEqualityDeletes) {
21102110
List<NestedField> equalityFields = schemaForPotentialEqualityDeletes(
2111-
table, scanSchema, applicableEqualityDeleteFieldIds);
2111+
table, exactScan.snapshot(), scanSchema, applicableEqualityDeleteFieldIds);
21122112
dict = IcebergSchemaUtils.encodeEqualitySchemaEvolutionProp(
21132113
table, equalityFields, appendRowLineage,
21142114
enableVarbinary, enableTimestampTz);
@@ -2280,37 +2280,48 @@ private Set<Integer> cachedApplicableEqualityDeleteFieldIds(Table table, TableSc
22802280
* delete manifests does not enumerate data files or byte-split tasks, and retained state is bounded by the
22812281
* number of equality keys rather than the table's entire schema history.
22822282
*
2283-
* <p>The metadata schema list is searched in reverse chronology up to the selected schema (schema IDs are
2284-
* identifiers, not a sequence), so the latest definition of each stable field ID wins. Current fields remain
2285-
* first, allowing a dropped/re-added name to resolve the projected field by name while a live historical
2286-
* equality key resolves by its stable field ID.</p>
2283+
* <p>Schema IDs are identifiers, not a sequence: Iceberg reuses an ID when evolution returns to an earlier
2284+
* schema, so a schema's position in the metadata list does not tell when it was last active, and the list may
2285+
* also hold schemas committed after a time-travel or branch target. The scan snapshot's parent chain is
2286+
* therefore searched first, so the definition active on that lineage wins; the whole metadata list, newest
2287+
* first, is the fallback for schema-only changes and expired ancestors. Current fields remain first, allowing
2288+
* a dropped/re-added name to resolve the projected field by name while a live historical equality key
2289+
* resolves by its stable field ID.</p>
22872290
*/
22882291
@VisibleForTesting
22892292
static List<NestedField> schemaForPotentialEqualityDeletes(
22902293
Table table, TableScan scan, Schema scanSchema) {
22912294
return schemaForPotentialEqualityDeletes(
2292-
table, scanSchema, applicableEqualityDeleteFieldIds(table, scan));
2295+
table, scan.snapshot(), scanSchema, applicableEqualityDeleteFieldIds(table, scan));
22932296
}
22942297

2295-
private static List<NestedField> schemaForPotentialEqualityDeletes(
2296-
Table table, Schema scanSchema, Set<Integer> equalityFieldIds) {
2297-
List<Schema> metadataSchemas = metadataSchemaHistory(table);
2298-
int selectedSchemaIndex = -1;
2299-
for (int index = 0; index < metadataSchemas.size(); index++) {
2300-
if (metadataSchemas.get(index).schemaId() == scanSchema.schemaId()) {
2301-
selectedSchemaIndex = index;
2302-
}
2303-
}
2304-
int lastRelevantIndex = selectedSchemaIndex >= 0
2305-
? selectedSchemaIndex : metadataSchemas.size() - 1;
2298+
@VisibleForTesting
2299+
static List<NestedField> schemaForPotentialEqualityDeletes(
2300+
Table table, Snapshot scanSnapshot, Schema scanSchema, Set<Integer> equalityFieldIds) {
23062301
Set<Integer> missing = new HashSet<>(equalityFieldIds);
23072302
missing.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
23082303
if (missing.isEmpty()) {
23092304
return scanSchema.columns();
23102305
}
23112306

23122307
List<NestedField> fields = new ArrayList<>(scanSchema.columns());
2313-
for (int index = lastRelevantIndex; index >= 0 && !missing.isEmpty(); index--) {
2308+
// Snapshots on one lineage mostly share a schema, so each schema is indexed at most once.
2309+
Set<Integer> lineageSchemaIds = new HashSet<>();
2310+
for (Snapshot snapshot = scanSnapshot; snapshot != null && !missing.isEmpty();
2311+
snapshot = snapshot.parentId() == null ? null : table.snapshot(snapshot.parentId())) {
2312+
Integer schemaId = snapshot.schemaId();
2313+
if (schemaId == null || !lineageSchemaIds.add(schemaId)) {
2314+
continue;
2315+
}
2316+
Schema lineageSchema = table.schemas().get(schemaId);
2317+
if (lineageSchema == null) {
2318+
throw new IllegalStateException(
2319+
"Iceberg snapshot schema " + schemaId + " is absent from table metadata");
2320+
}
2321+
addHistoricalEqualityFields(fields, missing, lineageSchema);
2322+
}
2323+
List<Schema> metadataSchemas = metadataSchemaHistory(table);
2324+
for (int index = metadataSchemas.size() - 1; index >= 0 && !missing.isEmpty(); index--) {
23142325
addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index));
23152326
}
23162327
if (!missing.isEmpty()) {

‎fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java‎

Lines changed: 160 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,7 @@
108108
import java.util.HashMap;
109109
import java.util.IdentityHashMap;
110110
import java.util.Iterator;
111+
import java.util.LinkedHashMap;
111112
import java.util.List;
112113
import java.util.Map;
113114
import java.util.NoSuchElementException;
@@ -1925,7 +1926,8 @@ null, new IcebergTableHandle("db1", "dropped_eq_key"),
19251926
@Test
19261927
public void equalityCarrierKeepsLatestDefinitionOfDroppedEqualityKey() throws Exception {
19271928
// BE fills an equality key that an old data file lacks from the key's carrier entry, so a dropped key must
1928-
// resolve to its latest historical definition: its last name, its optionality and its initial default.
1929+
// resolve to the latest definition on the scan's lineage: its last name, its optionality and its initial
1930+
// default. Each schema change is followed by a write, as on a table that keeps ingesting.
19291931
Schema schema = new Schema(
19301932
Types.NestedField.required(1, "id", Types.IntegerType.get()),
19311933
Types.NestedField.optional("k").withId(2).ofType(Types.IntegerType.get())
@@ -1938,7 +1940,11 @@ public void equalityCarrierKeepsLatestDefinitionOfDroppedEqualityKey() throws Ex
19381940
table.newRowDelta().addDeletes(equalityDeleteFile(
19391941
"s3://b/db/renamed_dropped_eq_key/eq.parquet", FileFormat.PARQUET, keyId)).commit();
19401942
table.updateSchema().renameColumn("k", "k2").commit();
1943+
table.newAppend().appendFile(dataFile(table.spec(),
1944+
"s3://b/db/renamed_dropped_eq_key/f2.parquet", 1024, null, null)).commit();
19411945
table.updateSchema().deleteColumn("k2").commit();
1946+
table.newAppend().appendFile(dataFile(table.spec(),
1947+
"s3://b/db/renamed_dropped_eq_key/f3.parquet", 1024, null, null)).commit();
19421948

19431949
IcebergScanPlanProvider provider = providerOver(table);
19441950
Map<String, String> props = provider.getScanNodeProperties(
@@ -2069,6 +2075,159 @@ public void equalityCarrierSizeDoesNotGrowWithUnrelatedSchemaHistory() throws Ex
20692075
Assertions.assertEquals(table.schema().columns(), carrier);
20702076
}
20712077

2078+
@Test
2079+
public void equalityCarrierResolvesDroppedKeyWhenSchemaIdIsReused() throws Exception {
2080+
Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
2081+
Table table = createTable("reused_schema_id_eq_key", schema, PartitionSpec.unpartitioned(),
2082+
Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
2083+
int initialSchemaId = table.schema().schemaId();
2084+
table.updateSchema().addColumn("k", Types.StringType.get()).commit();
2085+
int keyId = table.schema().findField("k").fieldId();
2086+
table.newAppend().appendFile(dataFile(table.spec(),
2087+
"s3://b/db/reused_schema_id_eq_key/f1.parquet", 1024, null, null)).commit();
2088+
table.newRowDelta().addDeletes(equalityDeleteFile(
2089+
"s3://b/db/reused_schema_id_eq_key/eq.parquet", FileFormat.PARQUET, keyId)).commit();
2090+
// Dropping k makes the schema equal to the initial one, so Iceberg reuses the initial schema ID, which the
2091+
// metadata list holds before the schema that still defines the equality key. The next snapshot carries the
2092+
// reused ID; only its parent records the schema with k.
2093+
table.updateSchema().deleteColumn("k").commit();
2094+
table.newAppend().appendFile(dataFile(table.spec(),
2095+
"s3://b/db/reused_schema_id_eq_key/f2.parquet", 1024, null, null)).commit();
2096+
Assertions.assertEquals(initialSchemaId, table.schema().schemaId());
2097+
Assertions.assertEquals(initialSchemaId, table.currentSnapshot().schemaId());
2098+
2099+
List<NestedField> carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
2100+
table, table.newScan(), table.schema());
2101+
Assertions.assertEquals(2, carrier.size(), "the dropped equality key must stay resolvable: " + carrier);
2102+
Assertions.assertEquals(keyId, carrier.get(1).fieldId());
2103+
Assertions.assertEquals("k", carrier.get(1).name());
2104+
2105+
IcebergScanPlanProvider provider = providerOver(table);
2106+
Map<String, String> props = provider.getScanNodeProperties(
2107+
null, new IcebergTableHandle("db1", "reused_schema_id_eq_key"),
2108+
Collections.singletonList(new IcebergColumnHandle("id", 1)), Optional.empty());
2109+
TFileScanRangeParams params = new TFileScanRangeParams();
2110+
provider.populateScanLevelParams(params, props);
2111+
List<TFieldPtr> fields = params.getHistorySchemaInfo().get(0).getRootField().getFields();
2112+
Assertions.assertTrue(fields.stream().anyMatch(field -> field.getFieldPtr().getId() == keyId),
2113+
"the scan dictionary must carry the dropped equality key after the schema ID is reused");
2114+
}
2115+
2116+
@Test
2117+
public void equalityCarrierResolvesReusedSchemaIdAfterParentSnapshotsExpire() throws Exception {
2118+
Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
2119+
Table table = createTable("reused_schema_id_expired_parent", schema, PartitionSpec.unpartitioned(),
2120+
Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
2121+
table.updateSchema().addColumn("k", Types.StringType.get()).commit();
2122+
int keyId = table.schema().findField("k").fieldId();
2123+
table.newAppend().appendFile(dataFile(table.spec(),
2124+
"s3://b/db/reused_schema_id_expired_parent/f1.parquet", 1024, null, null)).commit();
2125+
long appendSnapshotId = table.currentSnapshot().snapshotId();
2126+
table.newRowDelta().addDeletes(equalityDeleteFile(
2127+
"s3://b/db/reused_schema_id_expired_parent/eq.parquet", FileFormat.PARQUET, keyId)).commit();
2128+
long deleteSnapshotId = table.currentSnapshot().snapshotId();
2129+
table.updateSchema().deleteColumn("k").commit();
2130+
table.newAppend().appendFile(dataFile(table.spec(),
2131+
"s3://b/db/reused_schema_id_expired_parent/f2.parquet", 1024, null, null)).commit();
2132+
// With the snapshots that recorded k expired, the scan lineage ends at the reused-ID snapshot, so the key
2133+
// can only come from the metadata schema list, where it sits after the reused schema.
2134+
table.expireSnapshots().expireSnapshotId(appendSnapshotId).expireSnapshotId(deleteSnapshotId)
2135+
.cleanExpiredFiles(false).commit();
2136+
Assertions.assertNull(table.snapshot(table.currentSnapshot().parentId()));
2137+
2138+
List<NestedField> carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
2139+
table, table.newScan(), table.schema());
2140+
Assertions.assertEquals(2, carrier.size(), "the dropped equality key must stay resolvable: " + carrier);
2141+
Assertions.assertEquals(keyId, carrier.get(1).fieldId());
2142+
Assertions.assertEquals("k", carrier.get(1).name());
2143+
}
2144+
2145+
@Test
2146+
public void equalityCarrierPrefersScanLineageOverLaterRenameForTimeTravel() {
2147+
// The metadata list also holds schemas committed after a time-travel target. A later schema that renames
2148+
// the key must not win over the definition recorded on the target's own lineage.
2149+
NestedField id = Types.NestedField.required(1, "id", Types.LongType.get());
2150+
Schema schemaWithEqualityKey = new Schema(100, Arrays.asList(id, intKeyWithDefault("k")));
2151+
Schema timeTravelSchema = new Schema(1, Collections.singletonList(id));
2152+
Schema laterSchema = new Schema(2, Arrays.asList(id, intKeyWithDefault("k2")));
2153+
Snapshot timeTravelSnapshot = snapshotOf(1001L, 1000L, timeTravelSchema);
2154+
Table table = tableWithHistory(Arrays.asList(schemaWithEqualityKey, timeTravelSchema, laterSchema),
2155+
snapshotOf(1000L, null, schemaWithEqualityKey), timeTravelSnapshot);
2156+
2157+
List<NestedField> carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
2158+
table, timeTravelSnapshot, timeTravelSchema, Collections.singleton(7));
2159+
Assertions.assertEquals(2, carrier.size());
2160+
Assertions.assertEquals(7, carrier.get(1).fieldId());
2161+
Assertions.assertEquals("k", carrier.get(1).name());
2162+
}
2163+
2164+
@Test
2165+
public void equalityCarrierKeepsDefaultWhenTimeTravelParentExpiredBeforeLaterRename() {
2166+
// Once the lineage that recorded the key has expired, the metadata list is the only source left. Its newest
2167+
// definition may come from a later rename, but the field ID and its initial default stay the same.
2168+
NestedField id = Types.NestedField.required(1, "id", Types.LongType.get());
2169+
Schema schemaWithEqualityKey = new Schema(100, Arrays.asList(id, intKeyWithDefault("k")));
2170+
Schema timeTravelSchema = new Schema(1, Collections.singletonList(id));
2171+
Schema laterSchema = new Schema(2, Arrays.asList(id, intKeyWithDefault("k2")));
2172+
Snapshot timeTravelSnapshot = snapshotOf(1001L, 1000L, timeTravelSchema);
2173+
Table table = tableWithHistory(Arrays.asList(schemaWithEqualityKey, timeTravelSchema, laterSchema),
2174+
timeTravelSnapshot);
2175+
2176+
List<NestedField> carrier = IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
2177+
table, timeTravelSnapshot, timeTravelSchema, Collections.singleton(7));
2178+
Assertions.assertEquals(2, carrier.size());
2179+
Assertions.assertEquals(7, carrier.get(1).fieldId());
2180+
Assertions.assertEquals(7, carrier.get(1).initialDefault());
2181+
}
2182+
2183+
private static NestedField intKeyWithDefault(String name) {
2184+
return Types.NestedField.optional(name).withId(7).ofType(Types.IntegerType.get())
2185+
.withInitialDefault(7).build();
2186+
}
2187+
2188+
/** A snapshot double exposing only its lineage and the ID of the schema it was written with. */
2189+
private static Snapshot snapshotOf(long snapshotId, Long parentId, Schema schema) {
2190+
return (Snapshot) Proxy.newProxyInstance(Snapshot.class.getClassLoader(), new Class<?>[] {Snapshot.class},
2191+
(proxy, method, args) -> {
2192+
switch (method.getName()) {
2193+
case "snapshotId":
2194+
return snapshotId;
2195+
case "parentId":
2196+
return parentId;
2197+
case "schemaId":
2198+
return schema.schemaId();
2199+
default:
2200+
throw new UnsupportedOperationException(method.getName());
2201+
}
2202+
});
2203+
}
2204+
2205+
/**
2206+
* A table double exposing only schema history, in metadata-list order, and the retained snapshots. It allows
2207+
* schema histories that the SDK cannot build, such as a later schema that renames a dropped field ID.
2208+
*/
2209+
private static Table tableWithHistory(List<Schema> schemas, Snapshot... snapshots) {
2210+
Map<Integer, Schema> schemasById = new LinkedHashMap<>();
2211+
for (Schema schema : schemas) {
2212+
schemasById.put(schema.schemaId(), schema);
2213+
}
2214+
Map<Long, Snapshot> snapshotsById = new HashMap<>();
2215+
for (Snapshot snapshot : snapshots) {
2216+
snapshotsById.put(snapshot.snapshotId(), snapshot);
2217+
}
2218+
return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(), new Class<?>[] {Table.class},
2219+
(proxy, method, args) -> {
2220+
switch (method.getName()) {
2221+
case "schemas":
2222+
return schemasById;
2223+
case "snapshot":
2224+
return snapshotsById.get((Long) args[0]);
2225+
default:
2226+
throw new UnsupportedOperationException(method.getName());
2227+
}
2228+
});
2229+
}
2230+
20722231
@Test
20732232
public void partitionPrunedEqualityDeleteConservativelyRequiresCurrentBackendSemantics() {
20742233
PartitionSpec spec = PartitionSpec.builderFor(PART_SCHEMA).identity("p").build();

0 commit comments

Comments
 (0)