You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by sz...@apache.org on 2026/06/05 23:56:18 UTC
(iceberg) branch main updated: Spark: Fix time-travel filter on renamed columns in distributed planning mode (#16523)
This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 4057f59dbe Spark: Fix time-travel filter on renamed columns in distributed planning mode (#16523)
4057f59dbe is described below
commit 4057f59dbee6878b3986f7cc39b1f30a59ee9ee3
Author: sanshi <43...@users.noreply.github.com>
AuthorDate: Sat Jun 6 07:56:12 2026 +0800
Spark: Fix time-travel filter on renamed columns in distributed planning mode (#16523)
---
.../apache/iceberg/BaseDistributedDataScan.java | 2 +-
.../apache/iceberg/SparkDistributedDataScan.java | 21 ++++--
.../org/apache/iceberg/spark/sql/TestSelect.java | 75 ++++++++++++++++++++++
.../apache/iceberg/SparkDistributedDataScan.java | 21 ++++--
.../org/apache/iceberg/spark/sql/TestSelect.java | 75 ++++++++++++++++++++++
.../apache/iceberg/SparkDistributedDataScan.java | 21 ++++--
.../org/apache/iceberg/spark/sql/TestSelect.java | 75 ++++++++++++++++++++++
7 files changed, 268 insertions(+), 22 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java b/core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java
index e951ae8307..027a3d2298 100644
--- a/core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java
+++ b/core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java
@@ -397,7 +397,7 @@ abstract class BaseDistributedDataScan
private <R> Map<Integer, R> specCache(Function<PartitionSpec, R> load) {
Map<Integer, R> cache = Maps.newHashMap();
- table().specs().forEach((specId, spec) -> cache.put(specId, load.apply(spec)));
+ specs().forEach((specId, spec) -> cache.put(specId, load.apply(spec)));
return cache;
}
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
index 43ce2a303e..700257a794 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
@@ -132,7 +132,7 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
JavaRDD<DataFile> dataFileRDD =
sparkContext
.parallelize(toBeans(dataManifests), dataManifests.size())
- .flatMap(new ReadDataManifest(tableBroadcast(), context(), withColumnStats));
+ .flatMap(new ReadDataManifest(tableBroadcast(), specs(), context(), withColumnStats));
List<List<DataFile>> dataFileGroups = collectPartitions(dataFileRDD);
int matchingFilesCount = dataFileGroups.stream().mapToInt(List::size).sum();
@@ -159,14 +159,14 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
List<DeleteFile> deleteFiles =
sparkContext
.parallelize(toBeans(deleteManifests), deleteManifests.size())
- .flatMap(new ReadDeleteManifest(tableBroadcast(), context()))
+ .flatMap(new ReadDeleteManifest(tableBroadcast(), specs(), context()))
.collect();
int skippedFilesCount = liveFilesCount(deleteManifests) - deleteFiles.size();
scanMetrics().skippedDeleteFiles().increment(skippedFilesCount);
return DeleteFileIndex.builderFor(deleteFiles)
- .specsById(table().specs())
+ .specsById(specs())
.caseSensitive(isCaseSensitive())
.scanMetrics(scanMetrics())
.build();
@@ -221,12 +221,18 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
private static class ReadDataManifest implements FlatMapFunction<ManifestFileBean, DataFile> {
private final Broadcast<Table> table;
+ private final Map<Integer, PartitionSpec> specs;
private final Expression filter;
private final boolean withStats;
private final boolean isCaseSensitive;
- ReadDataManifest(Broadcast<Table> table, TableScanContext context, boolean withStats) {
+ ReadDataManifest(
+ Broadcast<Table> table,
+ Map<Integer, PartitionSpec> specs,
+ TableScanContext context,
+ boolean withStats) {
this.table = table;
+ this.specs = specs;
this.filter = context.rowFilter();
this.withStats = withStats;
this.isCaseSensitive = context.caseSensitive();
@@ -235,7 +241,6 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
@Override
public Iterator<DataFile> call(ManifestFileBean manifest) throws Exception {
FileIO io = table.value().io();
- Map<Integer, PartitionSpec> specs = table.value().specs();
return new ClosingIterator<>(
ManifestFiles.read(manifest, io, specs)
.select(withStats ? SCAN_WITH_STATS_COLUMNS : SCAN_COLUMNS)
@@ -248,11 +253,14 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
private static class ReadDeleteManifest implements FlatMapFunction<ManifestFileBean, DeleteFile> {
private final Broadcast<Table> table;
+ private final Map<Integer, PartitionSpec> specs;
private final Expression filter;
private final boolean isCaseSensitive;
- ReadDeleteManifest(Broadcast<Table> table, TableScanContext context) {
+ ReadDeleteManifest(
+ Broadcast<Table> table, Map<Integer, PartitionSpec> specs, TableScanContext context) {
this.table = table;
+ this.specs = specs;
this.filter = context.rowFilter();
this.isCaseSensitive = context.caseSensitive();
}
@@ -260,7 +268,6 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
@Override
public Iterator<DeleteFile> call(ManifestFileBean manifest) throws Exception {
FileIO io = table.value().io();
- Map<Integer, PartitionSpec> specs = table.value().specs();
return new ClosingIterator<>(
ManifestFiles.readDeleteManifest(manifest, io, specs)
.select(DELETE_SCAN_WITH_STATS_COLUMNS)
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
index 21d5afcf1d..c637937d49 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
@@ -696,6 +696,81 @@ public class TestSelect extends CatalogTestBase {
sql("DROP TABLE IF EXISTS %s", nestedStructTable);
}
+ @TestTemplate
+ public void testTimeTravelFilterOnRenamedColumn() {
+ String ttTableName = tableName("tt_rename_table");
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ sql(
+ "CREATE TABLE %s (id BIGINT, col DOUBLE) USING iceberg TBLPROPERTIES ("
+ + "'read.data-planning-mode'='distributed',"
+ + "'read.delete-planning-mode'='distributed')",
+ ttTableName);
+ sql("INSERT INTO %s VALUES (1, 100.0), (2, 200.0), (3, 0.0)", ttTableName);
+
+ TableIdentifier ttTableIdent = TableIdentifier.of(tableIdent.namespace(), "tt_rename_table");
+ long snapshotId = validationCatalog.loadTable(ttTableIdent).currentSnapshot().snapshotId();
+
+ sql("ALTER TABLE %s RENAME COLUMN col TO value", ttTableName);
+ sql("INSERT INTO %s VALUES (4, 400.0)", ttTableName);
+
+ Dataset<Row> df =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.VERSION_AS_OF, snapshotId)
+ .load(ttTableName);
+
+ assertThat(df.columns()).containsExactly("id", "col");
+
+ List<Object[]> results =
+ rowsToJava(df.filter(df.col("col").gt(0)).orderBy("id").collectAsList());
+ assertEquals(
+ "Should return rows where col > 0",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ results);
+
+ List<Object[]> sqlResults =
+ sql("SELECT * FROM %s VERSION AS OF %s WHERE col > 0 ORDER BY id", ttTableName, snapshotId);
+ assertEquals(
+ "SQL time-travel filter should also work",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ sqlResults);
+
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ }
+
+ @TestTemplate
+ public void testTimeTravelFilterOnRenamedColumnWithDeleteFiles() {
+ String ttTableName = tableName("tt_rename_delete_table");
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ sql(
+ "CREATE TABLE %s (id BIGINT, col DOUBLE) USING iceberg PARTITIONED BY (col) TBLPROPERTIES ("
+ + "'format-version'='2',"
+ + "'write.delete.mode'='merge-on-read',"
+ + "'read.data-planning-mode'='distributed',"
+ + "'read.delete-planning-mode'='distributed')",
+ ttTableName);
+ sql("INSERT INTO %s VALUES (1, 100.0), (2, 200.0), (3, 0.0)", ttTableName);
+
+ sql("DELETE FROM %s WHERE id = 3", ttTableName);
+
+ TableIdentifier ttTableIdent =
+ TableIdentifier.of(tableIdent.namespace(), "tt_rename_delete_table");
+ long snapshotId = validationCatalog.loadTable(ttTableIdent).currentSnapshot().snapshotId();
+
+ sql("ALTER TABLE %s RENAME COLUMN col TO value", ttTableName);
+ sql("INSERT INTO %s VALUES (4, 400.0)", ttTableName);
+
+ List<Object[]> results =
+ sql("SELECT * FROM %s VERSION AS OF %s WHERE col > 0 ORDER BY id", ttTableName, snapshotId);
+ assertEquals(
+ "Should return rows where col > 0, excluding deleted row",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ results);
+
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ }
+
@TestTemplate
public void simpleTypesInFilter() {
String tableName = tableName("simple_types_table");
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
index 43ce2a303e..700257a794 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
@@ -132,7 +132,7 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
JavaRDD<DataFile> dataFileRDD =
sparkContext
.parallelize(toBeans(dataManifests), dataManifests.size())
- .flatMap(new ReadDataManifest(tableBroadcast(), context(), withColumnStats));
+ .flatMap(new ReadDataManifest(tableBroadcast(), specs(), context(), withColumnStats));
List<List<DataFile>> dataFileGroups = collectPartitions(dataFileRDD);
int matchingFilesCount = dataFileGroups.stream().mapToInt(List::size).sum();
@@ -159,14 +159,14 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
List<DeleteFile> deleteFiles =
sparkContext
.parallelize(toBeans(deleteManifests), deleteManifests.size())
- .flatMap(new ReadDeleteManifest(tableBroadcast(), context()))
+ .flatMap(new ReadDeleteManifest(tableBroadcast(), specs(), context()))
.collect();
int skippedFilesCount = liveFilesCount(deleteManifests) - deleteFiles.size();
scanMetrics().skippedDeleteFiles().increment(skippedFilesCount);
return DeleteFileIndex.builderFor(deleteFiles)
- .specsById(table().specs())
+ .specsById(specs())
.caseSensitive(isCaseSensitive())
.scanMetrics(scanMetrics())
.build();
@@ -221,12 +221,18 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
private static class ReadDataManifest implements FlatMapFunction<ManifestFileBean, DataFile> {
private final Broadcast<Table> table;
+ private final Map<Integer, PartitionSpec> specs;
private final Expression filter;
private final boolean withStats;
private final boolean isCaseSensitive;
- ReadDataManifest(Broadcast<Table> table, TableScanContext context, boolean withStats) {
+ ReadDataManifest(
+ Broadcast<Table> table,
+ Map<Integer, PartitionSpec> specs,
+ TableScanContext context,
+ boolean withStats) {
this.table = table;
+ this.specs = specs;
this.filter = context.rowFilter();
this.withStats = withStats;
this.isCaseSensitive = context.caseSensitive();
@@ -235,7 +241,6 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
@Override
public Iterator<DataFile> call(ManifestFileBean manifest) throws Exception {
FileIO io = table.value().io();
- Map<Integer, PartitionSpec> specs = table.value().specs();
return new ClosingIterator<>(
ManifestFiles.read(manifest, io, specs)
.select(withStats ? SCAN_WITH_STATS_COLUMNS : SCAN_COLUMNS)
@@ -248,11 +253,14 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
private static class ReadDeleteManifest implements FlatMapFunction<ManifestFileBean, DeleteFile> {
private final Broadcast<Table> table;
+ private final Map<Integer, PartitionSpec> specs;
private final Expression filter;
private final boolean isCaseSensitive;
- ReadDeleteManifest(Broadcast<Table> table, TableScanContext context) {
+ ReadDeleteManifest(
+ Broadcast<Table> table, Map<Integer, PartitionSpec> specs, TableScanContext context) {
this.table = table;
+ this.specs = specs;
this.filter = context.rowFilter();
this.isCaseSensitive = context.caseSensitive();
}
@@ -260,7 +268,6 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
@Override
public Iterator<DeleteFile> call(ManifestFileBean manifest) throws Exception {
FileIO io = table.value().io();
- Map<Integer, PartitionSpec> specs = table.value().specs();
return new ClosingIterator<>(
ManifestFiles.readDeleteManifest(manifest, io, specs)
.select(DELETE_SCAN_WITH_STATS_COLUMNS)
diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
index 78a6718a9f..57457c0c3f 100644
--- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
+++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
@@ -740,6 +740,81 @@ public class TestSelect extends CatalogTestBase {
sql("DROP TABLE IF EXISTS %s", tableName);
}
+ @TestTemplate
+ public void testTimeTravelFilterOnRenamedColumn() {
+ String ttTableName = tableName("tt_rename_table");
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ sql(
+ "CREATE TABLE %s (id BIGINT, col DOUBLE) USING iceberg TBLPROPERTIES ("
+ + "'read.data-planning-mode'='distributed',"
+ + "'read.delete-planning-mode'='distributed')",
+ ttTableName);
+ sql("INSERT INTO %s VALUES (1, 100.0), (2, 200.0), (3, 0.0)", ttTableName);
+
+ TableIdentifier ttTableIdent = TableIdentifier.of(tableIdent.namespace(), "tt_rename_table");
+ long snapshotId = validationCatalog.loadTable(ttTableIdent).currentSnapshot().snapshotId();
+
+ sql("ALTER TABLE %s RENAME COLUMN col TO value", ttTableName);
+ sql("INSERT INTO %s VALUES (4, 400.0)", ttTableName);
+
+ Dataset<Row> df =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.VERSION_AS_OF, snapshotId)
+ .load(ttTableName);
+
+ assertThat(df.columns()).containsExactly("id", "col");
+
+ List<Object[]> results =
+ rowsToJava(df.filter(df.col("col").gt(0)).orderBy("id").collectAsList());
+ assertEquals(
+ "Should return rows where col > 0",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ results);
+
+ List<Object[]> sqlResults =
+ sql("SELECT * FROM %s VERSION AS OF %s WHERE col > 0 ORDER BY id", ttTableName, snapshotId);
+ assertEquals(
+ "SQL time-travel filter should also work",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ sqlResults);
+
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ }
+
+ @TestTemplate
+ public void testTimeTravelFilterOnRenamedColumnWithDeleteFiles() {
+ String ttTableName = tableName("tt_rename_delete_table");
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ sql(
+ "CREATE TABLE %s (id BIGINT, col DOUBLE) USING iceberg PARTITIONED BY (col) TBLPROPERTIES ("
+ + "'format-version'='2',"
+ + "'write.delete.mode'='merge-on-read',"
+ + "'read.data-planning-mode'='distributed',"
+ + "'read.delete-planning-mode'='distributed')",
+ ttTableName);
+ sql("INSERT INTO %s VALUES (1, 100.0), (2, 200.0), (3, 0.0)", ttTableName);
+
+ sql("DELETE FROM %s WHERE id = 3", ttTableName);
+
+ TableIdentifier ttTableIdent =
+ TableIdentifier.of(tableIdent.namespace(), "tt_rename_delete_table");
+ long snapshotId = validationCatalog.loadTable(ttTableIdent).currentSnapshot().snapshotId();
+
+ sql("ALTER TABLE %s RENAME COLUMN col TO value", ttTableName);
+ sql("INSERT INTO %s VALUES (4, 400.0)", ttTableName);
+
+ List<Object[]> results =
+ sql("SELECT * FROM %s VERSION AS OF %s WHERE col > 0 ORDER BY id", ttTableName, snapshotId);
+ assertEquals(
+ "Should return rows where col > 0, excluding deleted row",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ results);
+
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ }
+
@TestTemplate
public void variantTypeInFilter() {
assumeThat(validationCatalog)
diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
index 569bf7adda..1028cbcdd0 100644
--- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
+++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java
@@ -133,7 +133,7 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
JavaRDD<DataFile> dataFileRDD =
sparkContext
.parallelize(toBeans(dataManifests), dataManifests.size())
- .flatMap(new ReadDataManifest(tableBroadcast(), context(), withColumnStats));
+ .flatMap(new ReadDataManifest(tableBroadcast(), specs(), context(), withColumnStats));
List<List<DataFile>> dataFileGroups = collectPartitions(dataFileRDD);
int matchingFilesCount = dataFileGroups.stream().mapToInt(List::size).sum();
@@ -160,7 +160,7 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
List<DeleteFile> deleteFiles =
sparkContext
.parallelize(toBeans(deleteManifests), deleteManifests.size())
- .flatMap(new ReadDeleteManifest(tableBroadcast(), context()))
+ .flatMap(new ReadDeleteManifest(tableBroadcast(), specs(), context()))
.collect();
int skippedFilesCount = liveFilesCount(deleteManifests) - deleteFiles.size();
@@ -168,7 +168,7 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
return DeleteFileIndex.builderFor(deleteFiles)
.schemasById(schemas())
- .specsById(table().specs())
+ .specsById(specs())
.caseSensitive(isCaseSensitive())
.scanMetrics(scanMetrics())
.build();
@@ -223,12 +223,18 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
private static class ReadDataManifest implements FlatMapFunction<ManifestFileBean, DataFile> {
private final Broadcast<Table> table;
+ private final Map<Integer, PartitionSpec> specs;
private final Expression filter;
private final boolean withStats;
private final boolean isCaseSensitive;
- ReadDataManifest(Broadcast<Table> table, TableScanContext context, boolean withStats) {
+ ReadDataManifest(
+ Broadcast<Table> table,
+ Map<Integer, PartitionSpec> specs,
+ TableScanContext context,
+ boolean withStats) {
this.table = table;
+ this.specs = specs;
this.filter = context.rowFilter();
this.withStats = withStats;
this.isCaseSensitive = context.caseSensitive();
@@ -237,7 +243,6 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
@Override
public Iterator<DataFile> call(ManifestFileBean manifest) throws Exception {
FileIO io = table.value().io();
- Map<Integer, PartitionSpec> specs = table.value().specs();
return new ClosingIterator<>(
ManifestFiles.read(manifest, io, specs)
.select(withStats ? SCAN_WITH_STATS_COLUMNS : SCAN_COLUMNS)
@@ -250,11 +255,14 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
private static class ReadDeleteManifest implements FlatMapFunction<ManifestFileBean, DeleteFile> {
private final Broadcast<Table> table;
+ private final Map<Integer, PartitionSpec> specs;
private final Expression filter;
private final boolean isCaseSensitive;
- ReadDeleteManifest(Broadcast<Table> table, TableScanContext context) {
+ ReadDeleteManifest(
+ Broadcast<Table> table, Map<Integer, PartitionSpec> specs, TableScanContext context) {
this.table = table;
+ this.specs = specs;
this.filter = context.ignoreResiduals() ? Expressions.alwaysTrue() : context.rowFilter();
this.isCaseSensitive = context.caseSensitive();
}
@@ -262,7 +270,6 @@ public class SparkDistributedDataScan extends BaseDistributedDataScan {
@Override
public Iterator<DeleteFile> call(ManifestFileBean manifest) throws Exception {
FileIO io = table.value().io();
- Map<Integer, PartitionSpec> specs = table.value().specs();
return new ClosingIterator<>(
ManifestFiles.readDeleteManifest(manifest, io, specs)
.select(DELETE_SCAN_WITH_STATS_COLUMNS)
diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
index 6270296b01..2364cea11d 100644
--- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
+++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
@@ -740,6 +740,81 @@ public class TestSelect extends CatalogTestBase {
sql("DROP TABLE IF EXISTS %s", tableName);
}
+ @TestTemplate
+ public void testTimeTravelFilterOnRenamedColumn() {
+ String ttTableName = tableName("tt_rename_table");
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ sql(
+ "CREATE TABLE %s (id BIGINT, col DOUBLE) USING iceberg TBLPROPERTIES ("
+ + "'read.data-planning-mode'='distributed',"
+ + "'read.delete-planning-mode'='distributed')",
+ ttTableName);
+ sql("INSERT INTO %s VALUES (1, 100.0), (2, 200.0), (3, 0.0)", ttTableName);
+
+ TableIdentifier ttTableIdent = TableIdentifier.of(tableIdent.namespace(), "tt_rename_table");
+ long snapshotId = validationCatalog.loadTable(ttTableIdent).currentSnapshot().snapshotId();
+
+ sql("ALTER TABLE %s RENAME COLUMN col TO value", ttTableName);
+ sql("INSERT INTO %s VALUES (4, 400.0)", ttTableName);
+
+ Dataset<Row> df =
+ spark
+ .read()
+ .format("iceberg")
+ .option(SparkReadOptions.VERSION_AS_OF, snapshotId)
+ .load(ttTableName);
+
+ assertThat(df.columns()).containsExactly("id", "col");
+
+ List<Object[]> results =
+ rowsToJava(df.filter(df.col("col").gt(0)).orderBy("id").collectAsList());
+ assertEquals(
+ "Should return rows where col > 0",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ results);
+
+ List<Object[]> sqlResults =
+ sql("SELECT * FROM %s VERSION AS OF %s WHERE col > 0 ORDER BY id", ttTableName, snapshotId);
+ assertEquals(
+ "SQL time-travel filter should also work",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ sqlResults);
+
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ }
+
+ @TestTemplate
+ public void testTimeTravelFilterOnRenamedColumnWithDeleteFiles() {
+ String ttTableName = tableName("tt_rename_delete_table");
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ sql(
+ "CREATE TABLE %s (id BIGINT, col DOUBLE) USING iceberg PARTITIONED BY (col) TBLPROPERTIES ("
+ + "'format-version'='2',"
+ + "'write.delete.mode'='merge-on-read',"
+ + "'read.data-planning-mode'='distributed',"
+ + "'read.delete-planning-mode'='distributed')",
+ ttTableName);
+ sql("INSERT INTO %s VALUES (1, 100.0), (2, 200.0), (3, 0.0)", ttTableName);
+
+ sql("DELETE FROM %s WHERE id = 3", ttTableName);
+
+ TableIdentifier ttTableIdent =
+ TableIdentifier.of(tableIdent.namespace(), "tt_rename_delete_table");
+ long snapshotId = validationCatalog.loadTable(ttTableIdent).currentSnapshot().snapshotId();
+
+ sql("ALTER TABLE %s RENAME COLUMN col TO value", ttTableName);
+ sql("INSERT INTO %s VALUES (4, 400.0)", ttTableName);
+
+ List<Object[]> results =
+ sql("SELECT * FROM %s VERSION AS OF %s WHERE col > 0 ORDER BY id", ttTableName, snapshotId);
+ assertEquals(
+ "Should return rows where col > 0, excluding deleted row",
+ ImmutableList.of(row(1L, 100.0), row(2L, 200.0)),
+ results);
+
+ sql("DROP TABLE IF EXISTS %s", ttTableName);
+ }
+
@TestTemplate
public void variantTypeInFilter() {
assumeThat(validationCatalog)