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 dfd7ef3b71 Spark 3.5, 4.0, 4.1, 4.2: Use scan schema for CBO column
statistics (#18655)
dfd7ef3b71 is described below
commit dfd7ef3b711cd1977b93f4e6f86cb4148712acd3
Author: Rajarshi Sarkar <[email protected]>
AuthorDate: Fri Oct 30 22:39:00 2026 +0530
Spark 3.5, 4.0, 4.1, 4.2: Use scan schema for CBO column statistics (#18655)
---
.../org/apache/iceberg/spark/source/SparkScan.java | 7 +-
.../apache/iceberg/spark/source/TestSparkScan.java | 40 +++++++++--
.../org/apache/iceberg/spark/source/SparkScan.java | 7 +-
.../apache/iceberg/spark/source/TestSparkScan.java | 80 ++++++++++++++++++++++
.../org/apache/iceberg/spark/source/SparkScan.java | 7 +-
.../apache/iceberg/spark/source/TestSparkScan.java | 80 ++++++++++++++++++++++
.../org/apache/iceberg/spark/source/SparkScan.java | 7 +-
.../apache/iceberg/spark/source/TestSparkScan.java | 80 ++++++++++++++++++++++
8 files changed, 297 insertions(+), 11 deletions(-)
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 820461ff87..55ab122997 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
@@ -219,9 +219,12 @@ abstract class SparkScan implements Scan,
SupportsReportStatistics {
Collectors.groupingBy(
metadata -> metadata.fields().get(0),
Collectors.toList()));
- Schema snapshotSchema = SnapshotUtil.schemaFor(table,
snapshot.snapshotId());
+ Schema scanSchema = expectedSchema();
for (Map.Entry<Integer, List<BlobMetadata>> entry :
groupedByField.entrySet()) {
- String colName = snapshotSchema.findColumnName(entry.getKey());
+ String colName = scanSchema.findColumnName(entry.getKey());
+ if (colName == null) {
+ continue;
+ }
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 251ad8f23a..5790f87ff6 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
@@ -470,7 +470,7 @@ public class TestSparkScan extends TestBaseWithCatalog {
}
@TestTemplate
- public void testStatisticsMappedWithSnapshotSchema() throws
NoSuchTableException {
+ public void testStatisticsMappedWithScanSchema() throws NoSuchTableException
{
sql("CREATE TABLE %s (id int, data string) USING iceberg", tableName);
sql("INSERT INTO %s VALUES (1, 'a')", tableName);
@@ -490,8 +490,7 @@ public class TestSparkScan extends TestBaseWithCatalog {
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
+ // Blob for field ID 2 ('data')
new GenericBlobMetadata(
APACHE_DATASKETCHES_THETA_V1,
snapshotId,
@@ -504,8 +503,12 @@ public class TestSparkScan extends TestBaseWithCatalog {
table.updateStatistics().setStatistics(statisticsFile).commit();
- // Drop column 'data' (field ID 2) to simulate real schema evolution
- table.updateSchema().deleteColumn("data").commit();
+ // Rename 'data' to 'data_renamed' and add new column 'data' (field ID 3)
without new snapshot
+ table.updateSchema().renameColumn("data", "data_renamed").commit();
+ table
+ .updateSchema()
+ .addColumn("data", org.apache.iceberg.types.Types.StringType.get())
+ .commit();
Table updatedTable = validationCatalog.loadTable(tableIdent);
SparkScanBuilder scanBuilder =
@@ -515,9 +518,34 @@ public class TestSparkScan extends TestBaseWithCatalog {
Map<String, String> reportColStatsEnabled =
ImmutableMap.of(SQLConf.CBO_ENABLED().key(), "true");
+ Map<String, Long> expectedNDVs = Maps.newHashMap();
+ expectedNDVs.put("id", 4L);
+ expectedNDVs.put("data_renamed", 2L);
+
withSQLConf(
reportColStatsEnabled,
- () ->
assertThatCode(scan::estimateStatistics).doesNotThrowAnyException());
+ () -> {
+ assertThatCode(scan::estimateStatistics).doesNotThrowAnyException();
+ checkColStatisticsReported(scan, 1L, expectedNDVs);
+ });
+
+ // Drop column 'data_renamed' (field ID 2) to simulate dropped column
handling
+ table.updateSchema().deleteColumn("data_renamed").commit();
+
+ Table droppedColTable = validationCatalog.loadTable(tableIdent);
+ SparkScanBuilder droppedColScanBuilder =
+ new SparkScanBuilder(spark, droppedColTable,
CaseInsensitiveStringMap.empty());
+ SparkScan droppedColScan = (SparkScan) droppedColScanBuilder.build();
+
+ Map<String, Long> expectedDroppedNDVs = Maps.newHashMap();
+ expectedDroppedNDVs.put("id", 4L);
+
+ withSQLConf(
+ reportColStatsEnabled,
+ () -> {
+
assertThatCode(droppedColScan::estimateStatistics).doesNotThrowAnyException();
+ checkColStatisticsReported(droppedColScan, 1L, expectedDroppedNDVs);
+ });
}
@TestTemplate
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
index 6cf12f6e76..55ab122997 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
@@ -214,12 +214,17 @@ 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 scanSchema = expectedSchema();
for (Map.Entry<Integer, List<BlobMetadata>> entry :
groupedByField.entrySet()) {
- String colName = table.schema().findColumnName(entry.getKey());
+ String colName = scanSchema.findColumnName(entry.getKey());
+ if (colName == null) {
+ continue;
+ }
NamedReference ref = FieldReference.column(colName);
Long ndv = null;
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
index 2eb61407b0..5790f87ff6 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
+++
b/spark/v4.0/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,85 @@ public class TestSparkScan extends TestBaseWithCatalog {
withSQLConf(reportColStatsEnabled, () -> checkColStatisticsReported(scan,
4L, expectedTwoNDVs));
}
+ @TestTemplate
+ public void testStatisticsMappedWithScanSchema() 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')
+ 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();
+
+ // Rename 'data' to 'data_renamed' and add new column 'data' (field ID 3)
without new snapshot
+ table.updateSchema().renameColumn("data", "data_renamed").commit();
+ table
+ .updateSchema()
+ .addColumn("data", org.apache.iceberg.types.Types.StringType.get())
+ .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");
+
+ Map<String, Long> expectedNDVs = Maps.newHashMap();
+ expectedNDVs.put("id", 4L);
+ expectedNDVs.put("data_renamed", 2L);
+
+ withSQLConf(
+ reportColStatsEnabled,
+ () -> {
+ assertThatCode(scan::estimateStatistics).doesNotThrowAnyException();
+ checkColStatisticsReported(scan, 1L, expectedNDVs);
+ });
+
+ // Drop column 'data_renamed' (field ID 2) to simulate dropped column
handling
+ table.updateSchema().deleteColumn("data_renamed").commit();
+
+ Table droppedColTable = validationCatalog.loadTable(tableIdent);
+ SparkScanBuilder droppedColScanBuilder =
+ new SparkScanBuilder(spark, droppedColTable,
CaseInsensitiveStringMap.empty());
+ SparkScan droppedColScan = (SparkScan) droppedColScanBuilder.build();
+
+ Map<String, Long> expectedDroppedNDVs = Maps.newHashMap();
+ expectedDroppedNDVs.put("id", 4L);
+
+ withSQLConf(
+ reportColStatsEnabled,
+ () -> {
+
assertThatCode(droppedColScan::estimateStatistics).doesNotThrowAnyException();
+ checkColStatisticsReported(droppedColScan, 1L, expectedDroppedNDVs);
+ });
+ }
+
@TestTemplate
public void testUnpartitionedYears() throws Exception {
createUnpartitionedTable(spark, tableName);
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
index ee61523d80..22340408b0 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
@@ -234,12 +234,17 @@ 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 scanSchema = schema();
for (Map.Entry<Integer, List<BlobMetadata>> entry :
groupedByField.entrySet()) {
- String colName = table.schema().findColumnName(entry.getKey());
+ String colName = scanSchema.findColumnName(entry.getKey());
+ if (colName == null) {
+ continue;
+ }
NamedReference ref = FieldReference.column(colName);
Long ndv = null;
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
index b226990e69..e600c29a94 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
+++
b/spark/v4.1/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;
@@ -470,6 +471,85 @@ public class TestSparkScan extends TestBaseWithCatalog {
withSQLConf(reportColStatsEnabled, () -> checkColStatisticsReported(scan,
4L, expectedTwoNDVs));
}
+ @TestTemplate
+ public void testStatisticsMappedWithScanSchema() 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')
+ 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();
+
+ // Rename 'data' to 'data_renamed' and add new column 'data' (field ID 3)
without new snapshot
+ table.updateSchema().renameColumn("data", "data_renamed").commit();
+ table
+ .updateSchema()
+ .addColumn("data", org.apache.iceberg.types.Types.StringType.get())
+ .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");
+
+ Map<String, Long> expectedNDVs = Maps.newHashMap();
+ expectedNDVs.put("id", 4L);
+ expectedNDVs.put("data_renamed", 2L);
+
+ withSQLConf(
+ reportColStatsEnabled,
+ () -> {
+ assertThatCode(scan::estimateStatistics).doesNotThrowAnyException();
+ checkColStatisticsReported(scan, 1L, expectedNDVs);
+ });
+
+ // Drop column 'data_renamed' (field ID 2) to simulate dropped column
handling
+ table.updateSchema().deleteColumn("data_renamed").commit();
+
+ Table droppedColTable = validationCatalog.loadTable(tableIdent);
+ SparkScanBuilder droppedColScanBuilder =
+ new SparkScanBuilder(spark, droppedColTable,
CaseInsensitiveStringMap.empty());
+ SparkScan droppedColScan = (SparkScan) droppedColScanBuilder.build();
+
+ Map<String, Long> expectedDroppedNDVs = Maps.newHashMap();
+ expectedDroppedNDVs.put("id", 4L);
+
+ withSQLConf(
+ reportColStatsEnabled,
+ () -> {
+
assertThatCode(droppedColScan::estimateStatistics).doesNotThrowAnyException();
+ checkColStatisticsReported(droppedColScan, 1L, expectedDroppedNDVs);
+ });
+ }
+
@TestTemplate
public void testUnpartitionedYears() throws Exception {
createUnpartitionedTable(spark, tableName);
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
index ee61523d80..22340408b0 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
@@ -234,12 +234,17 @@ 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 scanSchema = schema();
for (Map.Entry<Integer, List<BlobMetadata>> entry :
groupedByField.entrySet()) {
- String colName = table.schema().findColumnName(entry.getKey());
+ String colName = scanSchema.findColumnName(entry.getKey());
+ if (colName == null) {
+ continue;
+ }
NamedReference ref = FieldReference.column(colName);
Long ndv = null;
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
index b226990e69..e600c29a94 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkScan.java
+++
b/spark/v4.2/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;
@@ -470,6 +471,85 @@ public class TestSparkScan extends TestBaseWithCatalog {
withSQLConf(reportColStatsEnabled, () -> checkColStatisticsReported(scan,
4L, expectedTwoNDVs));
}
+ @TestTemplate
+ public void testStatisticsMappedWithScanSchema() 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')
+ 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();
+
+ // Rename 'data' to 'data_renamed' and add new column 'data' (field ID 3)
without new snapshot
+ table.updateSchema().renameColumn("data", "data_renamed").commit();
+ table
+ .updateSchema()
+ .addColumn("data", org.apache.iceberg.types.Types.StringType.get())
+ .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");
+
+ Map<String, Long> expectedNDVs = Maps.newHashMap();
+ expectedNDVs.put("id", 4L);
+ expectedNDVs.put("data_renamed", 2L);
+
+ withSQLConf(
+ reportColStatsEnabled,
+ () -> {
+ assertThatCode(scan::estimateStatistics).doesNotThrowAnyException();
+ checkColStatisticsReported(scan, 1L, expectedNDVs);
+ });
+
+ // Drop column 'data_renamed' (field ID 2) to simulate dropped column
handling
+ table.updateSchema().deleteColumn("data_renamed").commit();
+
+ Table droppedColTable = validationCatalog.loadTable(tableIdent);
+ SparkScanBuilder droppedColScanBuilder =
+ new SparkScanBuilder(spark, droppedColTable,
CaseInsensitiveStringMap.empty());
+ SparkScan droppedColScan = (SparkScan) droppedColScanBuilder.build();
+
+ Map<String, Long> expectedDroppedNDVs = Maps.newHashMap();
+ expectedDroppedNDVs.put("id", 4L);
+
+ withSQLConf(
+ reportColStatsEnabled,
+ () -> {
+
assertThatCode(droppedColScan::estimateStatistics).doesNotThrowAnyException();
+ checkColStatisticsReported(droppedColScan, 1L, expectedDroppedNDVs);
+ });
+ }
+
@TestTemplate
public void testUnpartitionedYears() throws Exception {
createUnpartitionedTable(spark, tableName);
