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/07/14 21:11:52 UTC
(iceberg) branch main updated: Spark: Backport split-size session conf to 3.5 and 4.0 (#17199)
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 f2875fdcce Spark: Backport split-size session conf to 3.5 and 4.0 (#17199)
f2875fdcce is described below
commit f2875fdccecfdc0e8ca6b09b248deb1b66bccb08
Author: Gera Shegalov <gs...@nvidia.com>
AuthorDate: Tue Jul 14 14:11:46 2026 -0700
Spark: Backport split-size session conf to 3.5 and 4.0 (#17199)
---
.../org/apache/iceberg/spark/SparkReadConf.java | 7 +++-
.../apache/iceberg/spark/SparkSQLProperties.java | 3 ++
.../apache/iceberg/spark/TestSparkReadConf.java | 38 ++++++++++++++++++++
.../apache/iceberg/spark/source/TestSparkScan.java | 41 ++++++++++++++++++++++
.../org/apache/iceberg/spark/SparkReadConf.java | 7 +++-
.../apache/iceberg/spark/SparkSQLProperties.java | 3 ++
.../apache/iceberg/spark/TestSparkReadConf.java | 38 ++++++++++++++++++++
.../apache/iceberg/spark/source/TestSparkScan.java | 41 ++++++++++++++++++++++
8 files changed, 176 insertions(+), 2 deletions(-)
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
index a5de652f80..5bdcd570d3 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
@@ -192,13 +192,18 @@ public class SparkReadConf {
}
public Long splitSizeOption() {
- return confParser.longConf().option(SparkReadOptions.SPLIT_SIZE).parseOptional();
+ return confParser
+ .longConf()
+ .option(SparkReadOptions.SPLIT_SIZE)
+ .sessionConf(SparkSQLProperties.READ_SPLIT_SIZE)
+ .parseOptional();
}
public long splitSize() {
return confParser
.longConf()
.option(SparkReadOptions.SPLIT_SIZE)
+ .sessionConf(SparkSQLProperties.READ_SPLIT_SIZE)
.tableProperty(TableProperties.SPLIT_SIZE)
.defaultValue(TableProperties.SPLIT_SIZE_DEFAULT)
.parse();
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
index 3d3287e471..4b9509fc8b 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
@@ -113,6 +113,9 @@ public class SparkSQLProperties {
public static final String READ_ADAPTIVE_SPLIT_SIZE_ENABLED =
"spark.sql.iceberg.read.adaptive-split-size.enabled";
+ // Overrides the split target size for scan planning
+ public static final String READ_SPLIT_SIZE = "spark.sql.iceberg.read.split-size";
+
// Overrides the parallelism used for adaptive split sizing. When unset, the parallelism
// defaults to max(spark.default.parallelism, spark.sql.shuffle.partitions).
public static final String READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM =
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java
index c3fc69c8b2..4589d74448 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java
@@ -67,6 +67,44 @@ public class TestSparkReadConf extends TestBaseWithCatalog {
});
}
+ @TestTemplate
+ public void testSplitSizeSessionConf() {
+ Table table = validationCatalog.loadTable(tableIdent);
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_SPLIT_SIZE, "42"),
+ () -> {
+ SparkReadConf conf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty());
+ assertThat(conf.splitSizeOption()).isEqualTo(42L);
+ assertThat(conf.splitSize()).isEqualTo(42L);
+ });
+ }
+
+ @TestTemplate
+ public void testSplitSizeCamelCaseSessionConf() {
+ Table table = validationCatalog.loadTable(tableIdent);
+ withSQLConf(
+ ImmutableMap.of("spark.sql.iceberg.read.splitSize", "42"),
+ () -> {
+ SparkReadConf conf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty());
+ assertThat(conf.splitSizeOption()).isEqualTo(42L);
+ assertThat(conf.splitSize()).isEqualTo(42L);
+ });
+ }
+
+ @TestTemplate
+ public void testSplitSizeReadOptionOverridesSessionConf() {
+ Table table = validationCatalog.loadTable(tableIdent);
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_SPLIT_SIZE, "24"),
+ () -> {
+ CaseInsensitiveStringMap options =
+ new CaseInsensitiveStringMap(ImmutableMap.of(SparkReadOptions.SPLIT_SIZE, "42"));
+ SparkReadConf conf = new SparkReadConf(spark, table, options);
+ assertThat(conf.splitSizeOption()).isEqualTo(42L);
+ assertThat(conf.splitSize()).isEqualTo(42L);
+ });
+ }
+
@TestTemplate
public void testSplitParallelismRejectsZero() {
Table table = validationCatalog.loadTable(tableIdent);
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 417a84d827..2eb61407b0 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
@@ -36,6 +36,7 @@ import org.apache.iceberg.GenericStatisticsFile;
import org.apache.iceberg.Parameter;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
+import org.apache.iceberg.ScanTask;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -144,6 +145,37 @@ public class TestSparkScan extends TestBaseWithCatalog {
assertThat(stats.numRows().getAsLong()).isEqualTo(10000L);
}
+ @TestTemplate
+ public void testSessionSplitSizeSkipsAdaptiveSplitSizeAdjustment() {
+ sql("CREATE TABLE %s (id BIGINT) USING iceberg", tableName);
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ long splitSize = TableProperties.SPLIT_SIZE_DEFAULT;
+ List<ScanTask> tasks = ImmutableList.of(scanTaskWithSize(16L * 1024 * 1024));
+
+ withSQLConf(
+ ImmutableMap.of(SQLConf.SHUFFLE_PARTITIONS().key(), "200"),
+ () -> {
+ SparkScan scan =
+ (SparkScan)
+ new SparkScanBuilder(spark, table, CaseInsensitiveStringMap.empty()).build();
+ assertThat(scan.adjustSplitSize(tasks, splitSize)).isLessThan(splitSize);
+ });
+
+ withSQLConf(
+ ImmutableMap.of(
+ SQLConf.SHUFFLE_PARTITIONS().key(),
+ "200",
+ SparkSQLProperties.READ_SPLIT_SIZE,
+ String.valueOf(splitSize)),
+ () -> {
+ SparkScan scan =
+ (SparkScan)
+ new SparkScanBuilder(spark, table, CaseInsensitiveStringMap.empty()).build();
+ assertThat(scan.adjustSplitSize(tasks, splitSize)).isEqualTo(splitSize);
+ });
+ }
+
@TestTemplate
public void testTableWithoutColStats() throws NoSuchTableException {
sql("CREATE TABLE %s (id int, data string) USING iceberg", tableName);
@@ -1022,6 +1054,15 @@ public class TestSparkScan extends TestBaseWithCatalog {
assertThat(scan.planInputPartitions()).hasSize(4);
}
+ private ScanTask scanTaskWithSize(long sizeBytes) {
+ return new ScanTask() {
+ @Override
+ public long sizeBytes() {
+ return sizeBytes;
+ }
+ };
+ }
+
private SparkScanBuilder scanBuilder() throws Exception {
Table table = Spark3Util.loadIcebergTable(spark, tableName);
CaseInsensitiveStringMap options =
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
index da1093627f..e36da1105a 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
@@ -192,13 +192,18 @@ public class SparkReadConf {
}
public Long splitSizeOption() {
- return confParser.longConf().option(SparkReadOptions.SPLIT_SIZE).parseOptional();
+ return confParser
+ .longConf()
+ .option(SparkReadOptions.SPLIT_SIZE)
+ .sessionConf(SparkSQLProperties.READ_SPLIT_SIZE)
+ .parseOptional();
}
public long splitSize() {
return confParser
.longConf()
.option(SparkReadOptions.SPLIT_SIZE)
+ .sessionConf(SparkSQLProperties.READ_SPLIT_SIZE)
.tableProperty(TableProperties.SPLIT_SIZE)
.defaultValue(TableProperties.SPLIT_SIZE_DEFAULT)
.parse();
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
index b0ac0c055b..eb4d90292b 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
@@ -116,6 +116,9 @@ public class SparkSQLProperties {
public static final String READ_ADAPTIVE_SPLIT_SIZE_ENABLED =
"spark.sql.iceberg.read.adaptive-split-size.enabled";
+ // Overrides the split target size for scan planning
+ public static final String READ_SPLIT_SIZE = "spark.sql.iceberg.read.split-size";
+
// Overrides the parallelism used for adaptive split sizing. When unset, the parallelism
// defaults to max(spark.default.parallelism, spark.sql.shuffle.partitions).
public static final String READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM =
diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java
index c3fc69c8b2..4589d74448 100644
--- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java
+++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java
@@ -67,6 +67,44 @@ public class TestSparkReadConf extends TestBaseWithCatalog {
});
}
+ @TestTemplate
+ public void testSplitSizeSessionConf() {
+ Table table = validationCatalog.loadTable(tableIdent);
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_SPLIT_SIZE, "42"),
+ () -> {
+ SparkReadConf conf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty());
+ assertThat(conf.splitSizeOption()).isEqualTo(42L);
+ assertThat(conf.splitSize()).isEqualTo(42L);
+ });
+ }
+
+ @TestTemplate
+ public void testSplitSizeCamelCaseSessionConf() {
+ Table table = validationCatalog.loadTable(tableIdent);
+ withSQLConf(
+ ImmutableMap.of("spark.sql.iceberg.read.splitSize", "42"),
+ () -> {
+ SparkReadConf conf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty());
+ assertThat(conf.splitSizeOption()).isEqualTo(42L);
+ assertThat(conf.splitSize()).isEqualTo(42L);
+ });
+ }
+
+ @TestTemplate
+ public void testSplitSizeReadOptionOverridesSessionConf() {
+ Table table = validationCatalog.loadTable(tableIdent);
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.READ_SPLIT_SIZE, "24"),
+ () -> {
+ CaseInsensitiveStringMap options =
+ new CaseInsensitiveStringMap(ImmutableMap.of(SparkReadOptions.SPLIT_SIZE, "42"));
+ SparkReadConf conf = new SparkReadConf(spark, table, options);
+ assertThat(conf.splitSizeOption()).isEqualTo(42L);
+ assertThat(conf.splitSize()).isEqualTo(42L);
+ });
+ }
+
@TestTemplate
public void testSplitParallelismRejectsZero() {
Table table = validationCatalog.loadTable(tableIdent);
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 417a84d827..2eb61407b0 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
@@ -36,6 +36,7 @@ import org.apache.iceberg.GenericStatisticsFile;
import org.apache.iceberg.Parameter;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
+import org.apache.iceberg.ScanTask;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -144,6 +145,37 @@ public class TestSparkScan extends TestBaseWithCatalog {
assertThat(stats.numRows().getAsLong()).isEqualTo(10000L);
}
+ @TestTemplate
+ public void testSessionSplitSizeSkipsAdaptiveSplitSizeAdjustment() {
+ sql("CREATE TABLE %s (id BIGINT) USING iceberg", tableName);
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ long splitSize = TableProperties.SPLIT_SIZE_DEFAULT;
+ List<ScanTask> tasks = ImmutableList.of(scanTaskWithSize(16L * 1024 * 1024));
+
+ withSQLConf(
+ ImmutableMap.of(SQLConf.SHUFFLE_PARTITIONS().key(), "200"),
+ () -> {
+ SparkScan scan =
+ (SparkScan)
+ new SparkScanBuilder(spark, table, CaseInsensitiveStringMap.empty()).build();
+ assertThat(scan.adjustSplitSize(tasks, splitSize)).isLessThan(splitSize);
+ });
+
+ withSQLConf(
+ ImmutableMap.of(
+ SQLConf.SHUFFLE_PARTITIONS().key(),
+ "200",
+ SparkSQLProperties.READ_SPLIT_SIZE,
+ String.valueOf(splitSize)),
+ () -> {
+ SparkScan scan =
+ (SparkScan)
+ new SparkScanBuilder(spark, table, CaseInsensitiveStringMap.empty()).build();
+ assertThat(scan.adjustSplitSize(tasks, splitSize)).isEqualTo(splitSize);
+ });
+ }
+
@TestTemplate
public void testTableWithoutColStats() throws NoSuchTableException {
sql("CREATE TABLE %s (id int, data string) USING iceberg", tableName);
@@ -1022,6 +1054,15 @@ public class TestSparkScan extends TestBaseWithCatalog {
assertThat(scan.planInputPartitions()).hasSize(4);
}
+ private ScanTask scanTaskWithSize(long sizeBytes) {
+ return new ScanTask() {
+ @Override
+ public long sizeBytes() {
+ return sizeBytes;
+ }
+ };
+ }
+
private SparkScanBuilder scanBuilder() throws Exception {
Table table = Spark3Util.loadIcebergTable(spark, tableName);
CaseInsensitiveStringMap options =