(iceberg) branch main updated: Spark 3.5: Use snapshot schema for CBO column statistics (#18459)

[russellspitzer]

This is an automated email from the ASF dual-hosted git repository.

RussellSpitzer 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 e5f8ab19c8 Spark 3.5: Use snapshot schema for CBO column statistics 
(#18459)
e5f8ab19c8 is described below

commit e5f8ab19c82fcf24be430cc6c0cecf9a23f5afe4
Author: Rajarshi Sarkar <rajarshisar...@google.com>
AuthorDate: Mon Oct 26 22:17:50 2026 +0530

    Spark 3.5: Use snapshot schema for CBO column statistics (#18459)
---
 .../org/apache/iceberg/spark/source/SparkScan.java |  4 +-
 .../apache/iceberg/spark/source/TestSparkScan.java | 52 ++++++++++++++++++++++
 2 files changed, 55 insertions(+), 1 deletion(-)

diff --git 
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java 
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
index 6cf12f6e76..820461ff87 100644
--- 
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
+++ 
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
@@ -214,12 +214,14 @@ abstract class SparkScan implements Scan, 
SupportsReportStatistics {
 
         Map<Integer, List<BlobMetadata>> groupedByField =
             metadataList.stream()
+                .filter(metadata -> !metadata.fields().isEmpty())
                 .collect(
                     Collectors.groupingBy(
                         metadata -> metadata.fields().get(0), 
Collectors.toList()));
 
+        Schema snapshotSchema = SnapshotUtil.schemaFor(table, 
snapshot.snapshotId());
         for (Map.Entry<Integer, List<BlobMetadata>> entry : 
groupedByField.entrySet()) {
-          String colName = table.schema().findColumnName(entry.getKey());
+          String colName = snapshotSchema.findColumnName(entry.getKey());
           NamedReference ref = FieldReference.column(colName);
           Long ndv = null;
 
diff --git 
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
 
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
index 2eb61407b0..251ad8f23a 100644
--- 
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
+++ 
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
@@ -28,6 +28,7 @@ import static 
org.apache.iceberg.spark.SystemFunctionPushDownHelper.timestampStr
 import static org.apache.spark.sql.functions.date_add;
 import static org.apache.spark.sql.functions.expr;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 
 import java.util.List;
 import java.util.Map;
@@ -468,6 +469,57 @@ public class TestSparkScan extends TestBaseWithCatalog {
     withSQLConf(reportColStatsEnabled, () -> checkColStatisticsReported(scan, 
4L, expectedTwoNDVs));
   }
 
+  @TestTemplate
+  public void testStatisticsMappedWithSnapshotSchema() throws 
NoSuchTableException {
+    sql("CREATE TABLE %s (id int, data string) USING iceberg", tableName);
+    sql("INSERT INTO %s VALUES (1, 'a')", tableName);
+
+    Table table = validationCatalog.loadTable(tableIdent);
+    long snapshotId = table.currentSnapshot().snapshotId();
+
+    GenericStatisticsFile statisticsFile =
+        new GenericStatisticsFile(
+            snapshotId,
+            "/test/statistics/file.puffin",
+            100,
+            42,
+            ImmutableList.of(
+                new GenericBlobMetadata(
+                    APACHE_DATASKETCHES_THETA_V1,
+                    snapshotId,
+                    1,
+                    ImmutableList.of(1),
+                    ImmutableMap.of("ndv", "4")),
+                // Blob for field ID 2 ('data') which is in the snapshot 
schema but later dropped
+                // from table
+                new GenericBlobMetadata(
+                    APACHE_DATASKETCHES_THETA_V1,
+                    snapshotId,
+                    1,
+                    ImmutableList.of(2),
+                    ImmutableMap.of("ndv", "2")),
+                // Blob with empty fields (e.g. table-level or non-field blob)
+                new GenericBlobMetadata(
+                    DUMMY_BLOB_TYPE, snapshotId, 1, ImmutableList.of(), 
ImmutableMap.of())));
+
+    table.updateStatistics().setStatistics(statisticsFile).commit();
+
+    // Drop column 'data' (field ID 2) to simulate real schema evolution
+    table.updateSchema().deleteColumn("data").commit();
+
+    Table updatedTable = validationCatalog.loadTable(tableIdent);
+    SparkScanBuilder scanBuilder =
+        new SparkScanBuilder(spark, updatedTable, 
CaseInsensitiveStringMap.empty());
+    SparkScan scan = (SparkScan) scanBuilder.build();
+
+    Map<String, String> reportColStatsEnabled =
+        ImmutableMap.of(SQLConf.CBO_ENABLED().key(), "true");
+
+    withSQLConf(
+        reportColStatsEnabled,
+        () -> 
assertThatCode(scan::estimateStatistics).doesNotThrowAnyException());
+  }
+
   @TestTemplate
   public void testUnpartitionedYears() throws Exception {
     createUnpartitionedTable(spark, tableName);