Skip to content

[fix](iceberg) Fix historical scans after schema evolution #67479

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Gabriel39 wants to merge 1 commit into
base: branch-4.1
Choose a base branch
from
Open

[fix](iceberg) Fix historical scans after schema evolution #67479

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/dlf/DLFTableOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ public DLFTableOperations(Configuration conf,
String catalogName,
String database,
String table) {
super(conf, metaClients, fileIO, catalogName, database, table);
// DLF does not configure an Iceberg KMS client; null preserves the existing unencrypted behavior.
super(conf, metaClients, fileIO, null, catalogName, database, table);
}
}
63 changes: 49 additions & 14 deletions fe/fe-core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,9 @@
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expression;
Expand Down Expand Up @@ -64,7 +67,7 @@
* DataFile)} or {@link #forEntry(ManifestEntry)} to get the delete files to apply to a given data
* file.
*
* Copied from https://github.com/apache/iceberg/blob/apache-iceberg-1.9.1/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
* Copied from https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
* Change DeleteFileIndex and some methods to public.
*/
public class DeleteFileIndex {
Expand Down Expand Up @@ -371,6 +374,7 @@ public static class Builder {
private final Iterable<DeleteFile> deleteFiles;
private long minSequenceNumber = 0L;
private Map<Integer, PartitionSpec> specsById = null;
private Map<Integer, Schema> schemasById = null;
private Expression dataFilter = Expressions.alwaysTrue();
private Expression partitionFilter = Expressions.alwaysTrue();
private PartitionSet partitionSet = null;
Expand All @@ -396,6 +400,11 @@ Builder afterSequenceNumber(long seq) {
return this;
}

Builder schemasById(Map<Integer, Schema> newSchemasById) {

@github-actions github-actions Bot Sep 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Expose and wire the historical-schema map

The only Doris caller of the public iterable builder is in another package, so it cannot call this new package-private setter and currently supplies only icebergTable.specs(). Iceberg 1.11 added schemasById specifically because current specs no longer contain an equality-delete field after that field is dropped; fieldLookup then returns null and forDataFile throws. Doris catches that in the manifest-cache planner and reruns native planning, so every affected scan loses the enabled cache and logs a failure. Please make this hook public, pass the frozen table's full schemas() map, and cover a dropped equality key through the cache planner.

this.schemasById = newSchemasById;
return this;
}

public Builder specsById(Map<Integer, PartitionSpec> newSpecsById) {
this.specsById = newSpecsById;
return this;
Expand Down Expand Up @@ -459,8 +468,14 @@ private Collection<DeleteFile> loadDeleteFiles() {
try (CloseableIterable<ManifestEntry<DeleteFile>> reader = deleteFile) {
for (ManifestEntry<DeleteFile> entry : reader) {
if (entry.dataSequenceNumber() > minSequenceNumber) {
DeleteFile file = entry.file();
// keep minimum stats to avoid memory pressure
Set<Integer> columns =
file.content() == FileContent.POSITION_DELETES
? Collections.singleton(MetadataColumns.DELETE_FILE_PATH.fieldId())
: Sets.newHashSet(file.equalityFieldIds());
// copy with stats for better filtering against data file stats
files.add(entry.file().copy());
files.add(ContentFileUtil.copy(file, true, columns));
}
}
} catch (IOException e) {
Expand All @@ -470,10 +485,21 @@ private Collection<DeleteFile> loadDeleteFiles() {
return files;
}

private Collection<Schema> schemas() {
if (schemasById != null) {
return schemasById.values();
} else {
return specsById.values().stream().map(PartitionSpec::schema).collect(Collectors.toList());
}
}

public DeleteFileIndex build() {
// Equality deletes may reference fields from historical schemas, so index every known field ID.
Map<Integer, Types.NestedField> fieldsById = Schema.indexFields(schemas());
Function<Integer, Types.NestedField> fieldLookup = fieldsById::get;
Iterable<DeleteFile> files = deleteFiles != null ? filterDeleteFiles() : loadDeleteFiles();

EqualityDeletes globalDeletes = new EqualityDeletes();
EqualityDeletes globalDeletes = new EqualityDeletes(fieldLookup);
PartitionMap<EqualityDeletes> eqDeletesByPartition = PartitionMap.create(specsById);
PartitionMap<PositionDeletes> posDeletesByPartition = PartitionMap.create(specsById);
Map<String, PositionDeletes> posDeletesByPath = Maps.newHashMap();
Expand All @@ -489,7 +515,7 @@ public DeleteFileIndex build() {
}
break;
case EQUALITY_DELETES:
add(globalDeletes, eqDeletesByPartition, file);
add(globalDeletes, eqDeletesByPartition, file, fieldLookup);
break;
default:
throw new UnsupportedOperationException("Unsupported content: " + file.content());
Expand Down Expand Up @@ -536,7 +562,8 @@ private void add(
private void add(
EqualityDeletes globalDeletes,
PartitionMap<EqualityDeletes> deletesByPartition,
DeleteFile file) {
DeleteFile file,
Function<Integer, Types.NestedField> fieldLookup) {
PartitionSpec spec = specsById.get(file.specId());

EqualityDeletes deletes;
Expand All @@ -545,10 +572,11 @@ private void add(
} else {
int specId = spec.specId();
StructLike partition = file.partition();
deletes = deletesByPartition.computeIfAbsent(specId, partition, EqualityDeletes::new);
Supplier<EqualityDeletes> initEqDeletes = () -> new EqualityDeletes(fieldLookup);
deletes = deletesByPartition.computeIfAbsent(specId, partition, initEqDeletes);
}

deletes.add(spec, file);
deletes.add(file);
}

private Iterable<CloseableIterable<ManifestEntry<DeleteFile>>> deleteManifestReaders() {
Expand Down Expand Up @@ -725,16 +753,22 @@ static class EqualityDeletes {
Comparator.comparingLong(EqualityDeleteFile::applySequenceNumber);
private static final EqualityDeleteFile[] EMPTY_EQUALITY_DELETES = new EqualityDeleteFile[0];

private final Function<Integer, Types.NestedField> fieldLookup;

// indexed state
private long[] seqs = null;
private EqualityDeleteFile[] files = null;

// a buffer that is used to hold files before indexing
private volatile List<EqualityDeleteFile> buffer = Lists.newArrayList();

public void add(PartitionSpec spec, DeleteFile file) {
EqualityDeletes(Function<Integer, Types.NestedField> fieldLookup) {
this.fieldLookup = fieldLookup;
}

public void add(DeleteFile file) {
Preconditions.checkState(buffer != null, "Can't add files upon indexing");
buffer.add(new EqualityDeleteFile(spec, file));
buffer.add(new EqualityDeleteFile(fieldLookup, file));
}

public DeleteFile[] filter(long seq, DataFile dataFile) {
Expand Down Expand Up @@ -800,15 +834,15 @@ private static long[] indexSeqs(EqualityDeleteFile[] files) {
// an equality delete file wrapper that caches the converted boundaries for faster boundary checks
// this class is not meant to be exposed beyond the delete file index
private static class EqualityDeleteFile {
private final PartitionSpec spec;
private final Function<Integer, Types.NestedField> fieldLookup;
private final DeleteFile wrapped;
private final long applySequenceNumber;
private volatile List<Types.NestedField> equalityFields = null;
private volatile Map<Integer, Object> convertedLowerBounds = null;
private volatile Map<Integer, Object> convertedUpperBounds = null;

EqualityDeleteFile(PartitionSpec spec, DeleteFile file) {
this.spec = spec;
EqualityDeleteFile(Function<Integer, Types.NestedField> fieldLookup, DeleteFile file) {
this.fieldLookup = fieldLookup;
this.wrapped = file;
this.applySequenceNumber = wrapped.dataSequenceNumber() - 1;
}
Expand All @@ -827,7 +861,8 @@ public List<Types.NestedField> equalityFields() {
if (equalityFields == null) {
List<Types.NestedField> fields = Lists.newArrayList();
for (int id : wrapped.equalityFieldIds()) {
Types.NestedField field = spec.schema().findField(id);
Types.NestedField field = fieldLookup.apply(id);
Preconditions.checkArgument(field != null, "Cannot find field for ID %s", id);
fields.add(field);
}
this.equalityFields = fields;
Expand Down Expand Up @@ -890,7 +925,7 @@ private Map<Integer, Object> convertBounds(Map<Integer, ByteBuffer> bounds) {
if (bounds != null) {
for (Types.NestedField field : equalityFields()) {
int id = field.fieldId();
Type type = spec.schema().findField(id).type();
Type type = field.type();
if (type.isPrimitiveType()) {
ByteBuffer bound = bounds.get(id);
if (bound != null) {
Expand Down
57 changes: 57 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2938,6 +2938,63 @@ public void testHistoricalPredicateUsesSelectedScanSchema() throws Exception {
Mockito.verify(scan).filter(Mockito.argThat(expression -> expression.toString().contains("old_name")));
}

@Test
public void testHistoricalPredicatePlansAfterColumnRename() throws Exception {
assertHistoricalPredicatePlansAfterSchemaEvolution(false);
}

@Test
public void testHistoricalPredicatePlansAfterColumnDrop() throws Exception {
assertHistoricalPredicatePlansAfterSchemaEvolution(true);
}

private void assertHistoricalPredicatePlansAfterSchemaEvolution(boolean dropColumn) throws Exception {
Schema historicalSchema = new Schema(
Types.NestedField.optional(1, "x", Types.IntegerType.get()),
Types.NestedField.optional(2, "y", Types.IntegerType.get()),
Types.NestedField.optional(3, "part", Types.IntegerType.get()));
HadoopTables tables = new HadoopTables(new Configuration());
String tableLocation = temporaryFolder.getRoot().toPath()
.resolve("historical_predicate_after_" + (dropColumn ? "drop" : "rename")).toUri().toString();
Table table = tables.create(
historicalSchema, PartitionSpec.unpartitioned(), SortOrder.unsorted(),
ImmutableMap.of(TableProperties.FORMAT_VERSION, "2"), tableLocation);
DataFile historicalDataFile = DataFiles.builder(table.spec())
.withPath(tableLocation + "/data/historical.parquet")
.withFormat(FileFormat.PARQUET)
.withFileSizeInBytes(10)
.withRecordCount(2)
.build();
table.newFastAppend().appendFile(historicalDataFile).commit();
long historicalSnapshotId = table.currentSnapshot().snapshotId();
int historicalSchemaId = table.currentSnapshot().schemaId();

if (dropColumn) {
table.updateSchema().deleteColumn("x").commit();
} else {
table.updateSchema().renameColumn("x", "renamed_x").commit();
}
DataFile currentDataFile = DataFiles.builder(table.spec())

@github-actions github-actions Bot Sep 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Cover schema-only evolution before advancing the snapshot

Iceberg schema commits do not create a new snapshot, so immediately after this rename/drop historicalSnapshotId is still the table's current snapshot. In Iceberg 1.11, useSnapshot selects that snapshot's old schema, but SnapshotScan.specs() skips rebinding when the selected and current snapshot IDs are equal and returns specs bound to the renamed/dropped current schema. The old-name predicate then still fails in Projections. This append makes the IDs differ, so both tests avoid the unresolved case while the release note claims it is fixed. Please assert planning before this append and fix that path as well.

.withPath(tableLocation + "/data/current.parquet")
.withFormat(FileFormat.PARQUET)
.withFileSizeInBytes(10)
.withRecordCount(1)
.build();
table.newFastAppend().appendFile(currentDataFile).commit();

// Historical filters must be resolved with the snapshot schema after later schema evolution.
TableScan scan = table.newScan()
.useSnapshot(historicalSnapshotId)
.project(table.schemas().get(historicalSchemaId));
BinaryPredicate conjunct = new BinaryPredicate(BinaryPredicate.Operator.EQ,
new SlotRef(new TableName(), "x"), new IntLiteral(1, Type.INT));
org.apache.iceberg.expressions.Expression predicate =
IcebergUtils.convertToIcebergExpr(conjunct, scan.schema());
Assert.assertNotNull(predicate);
scan = scan.filter(predicate);
Assert.assertEquals(1, materializeTasks(scan).size());

@github-actions github-actions Bot Sep 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Exercise the Doris planner, not only native planFiles

This assertion calls Iceberg's TableScan.planFiles() directly, but production reaches IcebergScanNode.isBatchMode() first and batch mode defaults on. That preflight calls getMatchingManifest(..., icebergTable.specs(), scan.filter()); these specs use the current renamed/dropped schema, so the historical old-name predicate fails while Projections binds it, and the catch at isBatchMode() rethrows. Thus these tests can pass while the default production query still fails. Rebind the preflight/custom-planner specs to scan.schema() (matching Iceberg 1.11's snapshot-spec behavior) and run this regression through the Doris node.

}

@Test
public void testPinnedBranchUsesFrozenSnapshotWithCurrentSchema() throws Exception {
Schema snapshotSchema = new Schema(11, ImmutableList.of(
Expand Down
4 changes: 2 additions & 2 deletions fe/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,7 @@ under the License.
<module>fe-authentication</module>
</modules>
<properties>
<doris.hive.catalog.shade.version>3.1.2</doris.hive.catalog.shade.version>
<doris.hive.catalog.shade.version>3.1.3-ICEBERG-SNAPSHOT</doris.hive.catalog.shade.version>

@github-actions github-actions Bot Sep 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Pin an immutable shade release before merging

This changes packaged/runtime consumers to 3.1.3-ICEBERG-SNAPSHOT, and this POM configures snapshots with updatePolicy=always. The coordinate currently resolves to timestamped build 3.1.3-ICEBERG-20260903.063704-1, but a later build can silently replace it and repository cleanup can make the same Doris revision stop resolving; there is no stable 3.1.3 artifact yet, and the PR description calls this temporary. Please publish the merged Doris Shade change and pin its immutable release here before merging.

<!-- iceberg 1.9.1 depends avro on 1.12 -->
<avro.version>1.12.1</avro.version>
<parquet.version>1.17.0</parquet.version>
Expand Down Expand Up @@ -333,7 +333,7 @@ under the License.
<!-- ATTN: avro version must be consistent with Iceberg version -->
<!-- Please modify iceberg.version and avro.version together,
you can find avro version info in iceberg mvn repository -->
<iceberg.version>1.10.1</iceberg.version>
<iceberg.version>1.11.0</iceberg.version>
<lance.version>9.1.0-beta.3</lance.version>
<substrait.version>0.40.0</substrait.version>
<!-- 0.56.1 has bug that "SplitMode" in query response may not be set-->
Expand Down
Loading