diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/dlf/DLFTableOperations.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/dlf/DLFTableOperations.java
index 2aab8e754ca2ea..33bf0deccfeef7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/dlf/DLFTableOperations.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/dlf/DLFTableOperations.java
@@ -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);
     }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/iceberg/DeleteFileIndex.java b/fe/fe-core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
index 36cf36b556d098..630f19c81a47a4 100644
--- a/fe/fe-core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
+++ b/fe/fe-core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
@@ -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;
@@ -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 {
@@ -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;
@@ -396,6 +400,11 @@ Builder afterSequenceNumber(long seq) {
       return this;
     }
 
+    Builder schemasById(Map<Integer, Schema> newSchemasById) {
+      this.schemasById = newSchemasById;
+      return this;
+    }
+
     public Builder specsById(Map<Integer, PartitionSpec> newSpecsById) {
       this.specsById = newSpecsById;
       return this;
@@ -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) {
@@ -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();
@@ -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());
@@ -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;
@@ -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() {
@@ -725,6 +753,8 @@ 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;
@@ -732,9 +762,13 @@ static class EqualityDeletes {
     // 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) {
@@ -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;
     }
@@ -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;
@@ -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) {
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
index 941f30c4dd7019..8bb684ce1b8f8a 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
@@ -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())
+                .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());
+    }
+
     @Test
     public void testPinnedBranchUsesFrozenSnapshotWithCurrentSchema() throws Exception {
         Schema snapshotSchema = new Schema(11, ImmutableList.of(
diff --git a/fe/pom.xml b/fe/pom.xml
index 9e63c98bdac252..07e0762b3ec619 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -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>
         <!-- iceberg 1.9.1 depends avro on 1.12 -->
         <avro.version>1.12.1</avro.version>
         <parquet.version>1.17.0</parquet.version>
@@ -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-->