From a6ab5f7c17fbe22f07019280327b0a361e3dab8d Mon Sep 17 00:00:00 2001
From: Yuya Ebihara <ebyhry@gmail.com>
Date: Thu, 27 Aug 2026 16:18:09 +0900
Subject: [PATCH] Core: Discard changes when suppressing historical snapshots
 in CatalogHandlers

---
 .../apache/iceberg/rest/CatalogHandlers.java  |  1 +
 .../apache/iceberg/rest/TestRESTCatalog.java  | 71 +++++++++++++++++++
 2 files changed, 72 insertions(+)

diff --git a/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java b/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java
index 975cb960096b..7f81f181f8fa 100644
--- a/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java
+++ b/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java
@@ -528,6 +528,7 @@ public static LoadTableResponse loadTable(
               TableMetadata.buildFrom(loadedMetadata)
                   .withMetadataLocation(loadedMetadata.metadataFileLocation())
                   .suppressHistoricalSnapshots()
+                  .discardChanges()
                   .build();
           break;
         default:
diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
index cb100437e419..dd1f142733a0 100644
--- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
@@ -62,6 +62,8 @@
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
 import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.HistoryEntry;
 import org.apache.iceberg.MetadataUpdate;
@@ -1123,6 +1125,75 @@ public void testTableSnapshotLoading() {
             any());
   }
 
+  @Test
+  public void testTableSnapshotLoadingWithStatisticsOnHistoricalSnapshot() {
+    RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
+
+    RESTCatalog catalog =
+        new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
+    catalog.initialize(
+        "test",
+        ImmutableMap.of(
+            CatalogProperties.URI,
+            "ignored",
+            CatalogProperties.FILE_IO_IMPL,
+            "org.apache.iceberg.inmemory.InMemoryFileIO",
+            RESTCatalogProperties.SNAPSHOT_LOADING_MODE,
+            SnapshotMode.REFS.name()));
+
+    if (requiresNamespaceCreate()) {
+      catalog.createNamespace(TABLE.namespace());
+    }
+
+    Table table = catalog.createTable(TABLE, SCHEMA);
+    table
+        .newFastAppend()
+        .appendFile(
+            DataFiles.builder(PartitionSpec.unpartitioned())
+                .withPath("/path/to/data-a.parquet")
+                .withFileSizeInBytes(10)
+                .withRecordCount(2)
+                .build())
+        .commit();
+
+    long firstSnapshotId = table.currentSnapshot().snapshotId();
+
+    // add a statistics file to the first snapshot
+    table
+        .updateStatistics()
+        .setStatistics(
+            new GenericStatisticsFile(
+                firstSnapshotId,
+                "/path/to/stats.puffin",
+                100,
+                42,
+                ImmutableList.of(
+                    new GenericBlobMetadata(
+                        "some-stats", firstSnapshotId, 1, ImmutableList.of(1), ImmutableMap.of()))))
+        .commit();
+
+    // second snapshot makes the first snapshot historical (unreferenced)
+    table
+        .newFastAppend()
+        .appendFile(
+            DataFiles.builder(PartitionSpec.unpartitioned())
+                .withPath("/path/to/data-b.parquet")
+                .withFileSizeInBytes(10)
+                .withRecordCount(2)
+                .build())
+        .commit();
+
+    Table refsTable = catalog.loadTable(TABLE);
+
+    assertThat(((BaseTable) refsTable).operations().current())
+        .extracting("snapshots")
+        .asInstanceOf(InstanceOfAssertFactories.list(Snapshot.class))
+        .hasSize(1);
+
+    assertThat(refsTable.currentSnapshot()).isEqualTo(table.currentSnapshot());
+    assertThat(refsTable.statisticsFiles()).isEmpty();
+  }
+
   @ParameterizedTest
   @ValueSource(strings = {"1", "2"})
   public void testTableSnapshotLoadingWithDivergedBranches(String formatVersion) {