Spark 3.4: IcebergSource extends SessionConfigSupport

docs/docs/spark-configuration.md

@@ -154,6 +154,10 @@ spark.read
     .table("catalog.db.table")
 ```
 
+Iceberg 1.8.0 and later support setting read options by Spark session configuration `spark.datasource.iceberg.<key>=<value>`
+when using DataFrame to read Iceberg tables, for example: `spark.datasource.iceberg.split-size=512m`, it has lower priority
+than options explicitly passed to DataFrameReader.
+
 | Spark option    | Default               | Description                                                                               |
 | --------------- | --------------------- | ----------------------------------------------------------------------------------------- |
 | snapshot-id     | (latest)              | Snapshot ID of the table snapshot to read                                                 |
@@ -167,16 +171,20 @@ spark.read
 
 ### Write options
 
-Spark write options are passed when configuring the DataFrameWriter, like this:
+Spark write options are passed when configuring the DataFrameWriterV2, like this:
 
 ```scala
 // write with Avro instead of Parquet
-df.write
+df.writeTo("catalog.db.table")
     .option("write-format", "avro")
     .option("snapshot-property.key", "value")
-    .insertInto("catalog.db.table")
+    .append()
 ```
 
+Iceberg 1.8.0 and later support setting write options by Spark session configuration `spark.datasource.iceberg.<key>=<value>`
+when using DataFrame to write Iceberg tables, for example: `spark.datasource.iceberg.write-format=orc`, it has lower priority
+than options explicitly passed to DataFrameWriterV2.
+
 | Spark option           | Default                    | Description                                                  |
 | ---------------------- | -------------------------- | ------------------------------------------------------------ |
 | write-format           | Table write.format.default | File format to use for this write operation; parquet, avro, or orc |

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java

@@ -38,6 +38,7 @@
 import org.apache.spark.sql.connector.catalog.CatalogManager;
 import org.apache.spark.sql.connector.catalog.CatalogPlugin;
 import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.SessionConfigSupport;
 import org.apache.spark.sql.connector.catalog.SupportsCatalogOptions;
 import org.apache.spark.sql.connector.catalog.Table;
 import org.apache.spark.sql.connector.catalog.TableCatalog;
@@ -61,7 +62,8 @@
  * <p>The above list is in order of priority. For example: a matching catalog will take priority
  * over any namespace resolution.
  */
-public class IcebergSource implements DataSourceRegister, SupportsCatalogOptions {
+public class IcebergSource
+    implements DataSourceRegister, SupportsCatalogOptions, SessionConfigSupport {
   private static final String DEFAULT_CATALOG_NAME = "default_iceberg";
   private static final String DEFAULT_CACHE_CATALOG_NAME = "default_cache_iceberg";
   private static final String DEFAULT_CATALOG = "spark.sql.catalog." + DEFAULT_CATALOG_NAME;
@@ -80,6 +82,11 @@ public String shortName() {
     return "iceberg";
   }
 
+  @Override
+  public String keyPrefix() {
+    return shortName();
+  }
+
   @Override
   public StructType inferSchema(CaseInsensitiveStringMap options) {
     return null;

spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java

@@ -2184,6 +2184,51 @@ private void testWithFilter(String filterExpr, TableIdentifier tableIdentifier)
     assertThat(actual).as("Rows must match").containsExactlyInAnyOrderElementsOf(expected);
   }
 
+  @Test
+  public void testSessionConfigSupport() {
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("id").build();
+    TableIdentifier tableIdentifier = TableIdentifier.of("db", "session_config_table");
+    Table table = createTable(tableIdentifier, SCHEMA, spec);
+
+    List<SimpleRecord> initialRecords =
+        Lists.newArrayList(
+            new SimpleRecord(1, "a"), new SimpleRecord(2, "b"), new SimpleRecord(3, "c"));
+
+    Dataset<Row> df = spark.createDataFrame(initialRecords, SimpleRecord.class);
+
+    df.select("id", "data")
+        .write()
+        .format("iceberg")
+        .mode(SaveMode.Append)
+        .save(loadLocation(tableIdentifier));
+
+    long s1 = table.currentSnapshot().snapshotId();
+
+    withSQLConf(
+        // set write option through session configuration
+        ImmutableMap.of("spark.datasource.iceberg.snapshot-property.foo", "bar"),
+        () -> {
+          df.select("id", "data")
+              .write()
+              .format("iceberg")
+              .mode(SaveMode.Append)
+              .save(loadLocation(tableIdentifier));
+        });
+
+    table.refresh();
+    Assert.assertEquals("bar", table.currentSnapshot().summary().get("foo"));
+
+    withSQLConf(
+        // set read option through session configuration
+        ImmutableMap.of("spark.datasource.iceberg.snapshot-id", String.valueOf(s1)),
+        () -> {
+          Dataset<Row> result = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
+          List<SimpleRecord> actual = result.as(Encoders.bean(SimpleRecord.class)).collectAsList();
+          Assert.assertEquals("Number of rows should match", initialRecords.size(), actual.size());
+          Assert.assertEquals("Result rows should match", initialRecords, actual);
+        });
+  }
+
   private GenericData.Record manifestRecord(
       Table manifestTable, Long referenceSnapshotId, ManifestFile manifest) {
     GenericRecordBuilder builder =