Spark 4.1: Add session configs for adaptive split sizing and parallelism (#16088)

Author: karuppayya

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java

@@ -279,6 +279,7 @@ public boolean aggregatePushDownEnabled() {
   public boolean adaptiveSplitSizeEnabled() {
     return confParser
         .booleanConf()
+        .sessionConf(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_ENABLED)
         .tableProperty(TableProperties.ADAPTIVE_SPLIT_SIZE_ENABLED)
         .defaultValue(TableProperties.ADAPTIVE_SPLIT_SIZE_ENABLED_DEFAULT)
         .parse();
@@ -290,6 +291,17 @@ public int parallelism() {
     return Math.max(defaultParallelism, numShufflePartitions);
   }
 
+  public int splitParallelism() {
+    int parallelism =
+        confParser
+            .intConf()
+            .sessionConf(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM)
+            .defaultValue(parallelism())
+            .parse();
+    Preconditions.checkArgument(parallelism > 0, "Split parallelism must be > 0: %s", parallelism);
+    return parallelism;
+  }
+
   public boolean distributedPlanningEnabled() {
     return table instanceof SupportsDistributedScanPlanning distributed
         && distributed.allowDistributedPlanning()

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java

@@ -110,6 +110,15 @@ private SparkSQLProperties() {}
   // Prefix for custom snapshot properties
   public static final String SNAPSHOT_PROPERTY_PREFIX = "spark.sql.iceberg.snapshot-property.";
 
+  // Controls whether adaptive split sizing is enabled
+  public static final String READ_ADAPTIVE_SPLIT_SIZE_ENABLED =
+      "spark.sql.iceberg.read.adaptive-split-size.enabled";
+
+  // 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 =
+      "spark.sql.iceberg.read.adaptive-split-size.parallelism";
+
   // Controls whether to enable async micro batch planning for session
   public static final String ASYNC_MICRO_BATCH_PLANNING_ENABLED =
       "spark.sql.iceberg.async-micro-batch-planning-enabled";

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java

@@ -369,8 +369,17 @@ public CustomMetric[] supportedCustomMetrics() {
   protected long adjustSplitSize(List<? extends ScanTask> tasks, long splitSize) {
     if (readConf.splitSizeOption() == null && readConf.adaptiveSplitSizeEnabled()) {
       long scanSize = tasks.stream().mapToLong(ScanTask::sizeBytes).sum();
-      int parallelism = readConf.parallelism();
-      return TableScanUtil.adjustSplitSize(scanSize, parallelism, splitSize);
+      int parallelism = readConf.splitParallelism();
+      long adjustedSplitSize = TableScanUtil.adjustSplitSize(scanSize, parallelism, splitSize);
+      if (adjustedSplitSize != splitSize) {
+        LOG.debug(
+            "Adjusted split size from {} to {} for table {} with parallelism {}",
+            splitSize,
+            adjustedSplitSize,
+            table().name(),
+            parallelism);
+      }
+      return adjustedSplitSize;
     } else {
       return splitSize;
     }

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkReadConf.java

@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.spark.sql.internal.SQLConf;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestSparkReadConf extends TestBaseWithCatalog {
+
+  @BeforeEach
+  public void before() {
+    super.before();
+    sql("CREATE TABLE %s (id BIGINT, data STRING) USING iceberg", tableName);
+  }
+
+  @AfterEach
+  public void after() {
+    sql("DROP TABLE IF EXISTS %s", tableName);
+  }
+
+  @TestTemplate
+  public void testSplitParallelismDefault() {
+    Table table = validationCatalog.loadTable(tableIdent);
+    SparkReadConf conf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty());
+    assertThat(conf.splitParallelism()).isEqualTo(conf.parallelism());
+  }
+
+  @TestTemplate
+  public void testSplitParallelismSessionConf() {
+    Table table = validationCatalog.loadTable(tableIdent);
+    withSQLConf(
+        ImmutableMap.of(
+            SQLConf.SHUFFLE_PARTITIONS().key(),
+            "999",
+            SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM,
+            "42"),
+        () -> {
+          SparkReadConf conf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty());
+          assertThat(conf.splitParallelism()).isEqualTo(42);
+        });
+  }
+
+  @TestTemplate
+  public void testSplitParallelismRejectsZero() {
+    Table table = validationCatalog.loadTable(tableIdent);
+    withSQLConf(
+        ImmutableMap.of(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM, "0"),
+        () -> {
+          SparkReadConf conf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty());
+          assertThatIllegalArgumentException()
+              .isThrownBy(conf::splitParallelism)
+              .withMessageContaining("Split parallelism must be > 0");
+        });
+  }
+
+  @TestTemplate
+  public void testSplitParallelismRejectsNegative() {
+    Table table = validationCatalog.loadTable(tableIdent);
+    withSQLConf(
+        ImmutableMap.of(SparkSQLProperties.READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM, "-5"),
+        () -> {
+          SparkReadConf conf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty());
+          assertThatIllegalArgumentException()
+              .isThrownBy(conf::splitParallelism)
+              .withMessageContaining("Split parallelism must be > 0");
+        });
+  }
+}