Core: Fix failure when not finding a column during time travel (#14438)

Author: ebyhr

core/src/main/java/org/apache/iceberg/SnapshotScan.java

@@ -95,7 +95,7 @@ protected Map<Integer, PartitionSpec> specs() {
     ImmutableMap.Builder<Integer, PartitionSpec> newSpecs =
         ImmutableMap.builderWithExpectedSize(specs.size());
     for (Map.Entry<Integer, PartitionSpec> entry : specs.entrySet()) {
-      newSpecs.put(entry.getKey(), entry.getValue().toUnbound().bind(snapshotSchema));
+      newSpecs.put(entry.getKey(), entry.getValue().toUnbound().bind(snapshotSchema, true));
     }
 
     return newSpecs.build();

core/src/test/java/org/apache/iceberg/TestScansAndSchemaEvolution.java

@@ -134,6 +134,45 @@ public void testPartitionSourceRename() throws IOException {
     assertThat(tasks).hasSize(1);
   }
 
+  @TestTemplate
+  public void testPartitionSourceAdd() throws IOException {
+    Table table = TestTables.create(temp, "test", SCHEMA, SPEC, formatVersion);
+
+    DataFile fileOne = createDataFile("one");
+    DataFile fileTwo = createDataFile("two");
+
+    table.newAppend().appendFile(fileOne).appendFile(fileTwo).commit();
+    long firstSnapshotId = table.currentSnapshot().snapshotId();
+
+    List<FileScanTask> tasks =
+        Lists.newArrayList(table.newScan().filter(Expressions.equal("part", "one")).planFiles());
+
+    assertThat(tasks).hasSize(1);
+
+    // add a new partition column
+    table.updateSchema().addColumn("hour", Types.IntegerType.get()).commit();
+    table.updateSpec().addField("hour").commit();
+
+    // plan the scan using the new column in a filter
+    tasks = Lists.newArrayList(table.newScan().filter(Expressions.isNull("hour")).planFiles());
+
+    assertThat(tasks).hasSize(2);
+
+    // create a new commit
+    table.newAppend().appendFile(createDataFile("three")).commit();
+
+    // plan the scan using the existing column in a filter
+    tasks =
+        Lists.newArrayList(
+            table
+                .newScan()
+                .useSnapshot(firstSnapshotId)
+                .filter(Expressions.equal("part", "one"))
+                .planFiles());
+
+    assertThat(tasks).hasSize(1);
+  }
+
   @TestTemplate
   public void testPartitionSourceDrop() throws IOException {
     Table table = TestTables.create(temp, "test", SCHEMA, SPEC, formatVersion);
@@ -179,6 +218,37 @@ public void testPartitionSourceDrop() throws IOException {
                 .collect(Collectors.toList()));
   }
 
+  @TestTemplate
+  public void testColumnAdd() throws IOException {
+    Table table = TestTables.create(temp, "test", SCHEMA, SPEC, formatVersion);
+
+    DataFile fileOne = createDataFile("one");
+    DataFile fileTwo = createDataFile("two");
+
+    table.newAppend().appendFile(fileOne).appendFile(fileTwo).commit();
+    long firstSnapshotId = table.currentSnapshot().snapshotId();
+
+    table.updateSchema().addColumn("hour", Types.IntegerType.get()).commit();
+
+    DataFile fileThree = createDataFile("three", table.schema(), table.spec());
+    table.newAppend().appendFile(fileThree).commit();
+
+    // plan the scan using the new column in a filter
+    List<FileScanTask> tasks =
+        Lists.newArrayList(table.newScan().filter(Expressions.isNull("hour")).planFiles());
+    assertThat(tasks).hasSize(3);
+
+    // plan the scan using the existing column in a filter
+    tasks =
+        Lists.newArrayList(
+            table
+                .newScan()
+                .useSnapshot(firstSnapshotId)
+                .filter(Expressions.equal("data", "xyz"))
+                .planFiles());
+    assertThat(tasks).hasSize(2);
+  }
+
   @TestTemplate
   public void testColumnRename() throws IOException {
     Table table = TestTables.create(temp, "test", SCHEMA, SPEC, formatVersion);