Repository: apache/iceberg (language: python)

Issue:
Stale column stats getting reported when reading puffin files generated by Presto with Spark engine
### Apache Iceberg version

1.6.0 (latest release)

### Query engine

Spark

### Please describe the bug 🐞

As we work with multiple engines like Spark, Presto, Flink, etc., interoperability is crucial.

Recently, PRs \[#10288\](https://github.com/apache/iceberg/pull/10288) and \[#10659\](https://github.com/apache/iceberg/pull/10659) added Columnar stats support, specifically NDV (Number of Distinct Values) for Spark. Based on this, we can generate the Puffin (\`.stat\`) file, which can be used by Spark for better query planning.

The Puffin file generation logic was already developed internally by Presto, which creates the Puffin file when running the \`ANALYZE\` query.

According to the interoperability and Puffin specification, any Puffin file generated by one engine should be usable by another engine. However, we encountered issues with Spark. The Puffin file generated by Presto contains another \`blob-metadata\` entry, i.e., \`"data\_size"\` of type \`presto-sum-data-size-bytes-v1\`, while \`ndv\` is of type \`apache-datasketches-theta-v1\`.

When we used the Puffin file generated by Presto in Spark, the \`ndv\` stats were being overwritten to \`null\` for columns where the \`"data\_size"\` blob was present, particularly for string type fields.

To resolve this, we need to ensure that Puffin files generated by Presto can be correctly utilized by Spark.

### Willingness to contribute

- [ ] I can contribute a fix for this bug independently
- [X] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time


Discussion / hints:

Fix patch:
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 8b88cf49c692..019f3919dc57 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
@@ -198,25 +198,31 @@ protected Statistics estimateStatistics(Snapshot snapshot) {
       if (!files.isEmpty()) {
         List<BlobMetadata> metadataList = (files.get(0)).blobMetadata();
 
-        for (BlobMetadata blobMetadata : metadataList) {
-          int id = blobMetadata.fields().get(0);
-          String colName = table.schema().findColumnName(id);
+        Map<Integer, List<BlobMetadata>> groupedByField =
+            metadataList.stream()
+                .collect(
+                    Collectors.groupingBy(
+                        metadata -> metadata.fields().get(0), Collectors.toList()));
+
+        for (Map.Entry<Integer, List<BlobMetadata>> entry : groupedByField.entrySet()) {
+          String colName = table.schema().findColumnName(entry.getKey());
           NamedReference ref = FieldReference.column(colName);
-
           Long ndv = null;
-          if (blobMetadata
-              .type()
-              .equals(org.apache.iceberg.puffin.StandardBlobTypes.APACHE_DATASKETCHES_THETA_V1)) {
-            String ndvStr = blobMetadata.properties().get(NDV_KEY);
-            if (!Strings.isNullOrEmpty(ndvStr)) {
-              ndv = Long.parseLong(ndvStr);
+
+          for (BlobMetadata blobMetadata : entry.getValue()) {
+            if (blobMetadata
+                .type()
+                .equals(org.apache.iceberg.puffin.StandardBlobTypes.APACHE_DATASKETCHES_THETA_V1)) {
+              String ndvStr = blobMetadata.properties().get(NDV_KEY);
+              if (!Strings.isNullOrEmpty(ndvStr)) {
+                ndv = Long.parseLong(ndvStr);
+              } else {
+                LOG.debug("{} is not set in BlobMetadata for column {}", NDV_KEY, colName);
+              }
             } else {
-              LOG.debug("ndv is not set in BlobMetadata for column {}", colName);
+              LOG.debug("Blob type {} is not supported yet", blobMetadata.type());
             }
-          } else {
-            LOG.debug("DataSketch blob is not available for column {}", colName);
           }
-
           ColumnStatistics colStats =
               new SparkColumnStatistics(ndv, null, null, null, null, null, null);
 


Test patch:
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 7d5475ff919e..af144fe4bf3d 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
@@ -82,6 +82,8 @@
 @ExtendWith(ParameterizedTestExtension.class)
 public class TestSparkScan extends TestBaseWithCatalog {
 
+  private static final String DUMMY_BLOB_TYPE = "sum-data-size-bytes-v1";
+
   @Parameter(index = 3)
   private String format;
 
@@ -178,6 +180,59 @@ public void testTableWithoutColStats() throws NoSuchTableException {
         reportColStatsEnabled, () -> checkColStatisticsReported(scan, 4L, Maps.newHashMap()));
   }
 
+  @TestTemplate
+  public void testTableWithoutApacheDatasketchColStat() throws NoSuchTableException {
+    sql("CREATE TABLE %s (id int, data string) USING iceberg", tableName);
+
+    List<SimpleRecord> records =
+        Lists.newArrayList(
+            new SimpleRecord(1, "a"),
+            new SimpleRecord(2, "b"),
+            new SimpleRecord(3, "a"),
+            new SimpleRecord(4, "b"));
+    spark
+        .createDataset(records, Encoders.bean(SimpleRecord.class))
+        .coalesce(1)
+        .writeTo(tableName)
+        .append();
+
+    Table table = validationCatalog.loadTable(tableIdent);
+    long snapshotId = table.currentSnapshot().snapshotId();
+
+    SparkScanBuilder scanBuilder =
+        new SparkScanBuilder(spark, table, CaseInsensitiveStringMap.empty());
+    SparkScan scan = (SparkScan) scanBuilder.build();
+
+    Map<String, String> reportColStatsDisabled =
+        ImmutableMap.of(
+            SQLConf.CBO_ENABLED().key(), "true", SparkSQLProperties.REPORT_COLUMN_STATS, "false");
+
+    Map<String, String> reportColStatsEnabled =
+        ImmutableMap.of(SQLConf.CBO_ENABLED().key(), "true");
+
+    GenericStatisticsFile statisticsFile =
+        new GenericStatisticsFile(
+            snapshotId,
+            "/test/statistics/file.puffin",
+            100,
+            42,
+            ImmutableList.of(
+                new GenericBlobMetadata(
+                    DUMMY_BLOB_TYPE,
+                    snapshotId,
+                    1,
+                    ImmutableList.of(1),
+                    ImmutableMap.of("data_size", "4"))));
+
+    table.updateStatistics().setStatistics(snapshotId, statisticsFile).commit();
+
+    checkColStatisticsNotReported(scan, 4L);
+    withSQLConf(reportColStatsDisabled, () -> checkColStatisticsNotReported(scan, 4L));
+    // The expected col NDVs are nulls
+    withSQLConf(
+        reportColStatsEnabled, () -> checkColStatisticsReported(scan, 4L, Maps.newHashMap()));
+  }
+
   @TestTemplate
   public void testTableWithOneColStats() throws NoSuchTableException {
     sql("CREATE TABLE %s (id int, data string) USING iceberg", tableName);
@@ -232,6 +287,67 @@ public void testTableWithOneColStats() throws NoSuchTableException {
     withSQLConf(reportColStatsEnabled, () -> checkColStatisticsReported(scan, 4L, expectedOneNDV));
   }
 
+  @TestTemplate
+  public void testTableWithOneApacheDatasketchColStatAndOneDifferentColStat()
+      throws NoSuchTableException {
+    sql("CREATE TABLE %s (id int, data string) USING iceberg", tableName);
+
+    List<SimpleRecord> records =
+        Lists.newArrayList(
+            new SimpleRecord(1, "a"),
+            new SimpleRecord(2, "b"),
+            new SimpleRecord(3, "a"),
+            new SimpleRecord(4, "b"));
+    spark
+        .createDataset(records, Encoders.bean(SimpleRecord.class))
+        .coalesce(1)
+        .writeTo(tableName)
+        .append();
+
+    Table table = validationCatalog.loadTable(tableIdent);
+    long snapshotId = table.currentSnapshot().snapshotId();
+
+    SparkScanBuilder scanBuilder =
+        new SparkScanBuilder(spark, table, CaseInsensitiveStringMap.empty());
+    SparkScan scan = (SparkScan) scanBuilder.build();
+
+    Map<String, String> reportColStatsDisabled =
+        ImmutableMap.of(
+            SQLConf.CBO_ENABLED().key(), "true", SparkSQLProperties.REPORT_COLUMN_STATS, "false");
+
+    Map<String, String> reportColStatsEnabled =
+        ImmutableMap.of(SQLConf.CBO_ENABLED().key(), "true");
+
+    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")),
+                new GenericBlobMetadata(
+                    DUMMY_BLOB_TYPE,
+                    snapshotId,
+                    1,
+                    ImmutableList.of(1),
+                    ImmutableMap.of("data_size", "2"))));
+
+    table.updateStatistics().setStatistics(snapshotId, statisticsFile).commit();
+
+    checkColStatisticsNotReported(scan, 4L);
+    withSQLConf(reportColStatsDisabled, () -> checkColStatisticsNotReported(scan, 4L));
+
+    Map<String, Long> expectedOneNDV = Maps.newHashMap();
+    expectedOneNDV.put("id", 4L);
+    withSQLConf(reportColStatsEnabled, () -> checkColStatisticsReported(scan, 4L, expectedOneNDV));
+  }
+
   @TestTemplate
   public void testTableWithTwoColStats() throws NoSuchTableException {
     sql("CREATE TABLE %s (id int, data string) USING iceberg", tableName);
@@ -911,9 +1027,15 @@ private void checkColStatisticsReported(
     assertThat(stats.numRows().getAsLong()).isEqualTo(expectedRowCount);
 
     Map<NamedReference, ColumnStatistics> columnStats = stats.columnStats();
-    for (Map.Entry<String, Long> entry : expectedNDVs.entrySet()) {
-      assertThat(columnStats.get(FieldReference.column(entry.getKey())).distinctCount().getAsLong())
-          .isEqualTo(entry.getValue());
+    if (expectedNDVs.isEmpty()) {
+      assertThat(columnStats.values().stream().allMatch(value -> value.distinctCount().isEmpty()))
+          .isTrue();
+    } else {
+      for (Map.Entry<String, Long> entry : expectedNDVs.entrySet()) {
+        assertThat(
+                columnStats.get(FieldReference.column(entry.getKey())).distinctCount().getAsLong())
+            .isEqualTo(entry.getValue());
+      }
     }
   }