Spark 3.5: Support camel case session configs and options (apache#10310)

Author: aokolnychyi

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkConfParser.java

@@ -270,13 +270,23 @@ protected T parse(Function<String, T> conversion, T defaultValue) {
         if (optionValue != null) {
           return conversion.apply(optionValue);
         }
+
+        String sparkOptionValue = options.get(toCamelCase(optionName));
+        if (sparkOptionValue != null) {
+          return conversion.apply(sparkOptionValue);
+        }
       }
 
       if (sessionConfName != null) {
         String sessionConfValue = sessionConf.get(sessionConfName, null);
         if (sessionConfValue != null) {
           return conversion.apply(sessionConfValue);
         }
+
+        String sparkSessionConfValue = sessionConf.get(toCamelCase(sessionConfName), null);
+        if (sparkSessionConfValue != null) {
+          return conversion.apply(sparkSessionConfValue);
+        }
       }
 
       if (tablePropertyName != null) {
@@ -288,5 +298,23 @@ protected T parse(Function<String, T> conversion, T defaultValue) {
 
       return defaultValue;
     }
+
+    private String toCamelCase(String key) {
+      StringBuilder transformedKey = new StringBuilder();
+      boolean capitalizeNext = false;
+
+      for (char character : key.toCharArray()) {
+        if (character == '-') {
+          capitalizeNext = true;
+        } else if (capitalizeNext) {
+          transformedKey.append(Character.toUpperCase(character));
+          capitalizeNext = false;
+        } else {
+          transformedKey.append(character);
+        }
+      }
+
+      return transformedKey.toString();
+    }
   }
 }

spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java

@@ -88,6 +88,32 @@ public void testOptionCaseInsensitive() {
     assertThat(parsedValue).isEqualTo("value");
   }
 
+  @TestTemplate
+  public void testCamelCaseSparkSessionConf() {
+    Table table = validationCatalog.loadTable(tableIdent);
+    String confName = "spark.sql.iceberg.some-int-conf";
+    String sparkConfName = "spark.sql.iceberg.someIntConf";
+
+    withSQLConf(
+        ImmutableMap.of(sparkConfName, "1"),
+        () -> {
+          SparkConfParser parser = new SparkConfParser(spark, table, ImmutableMap.of());
+          Integer value = parser.intConf().sessionConf(confName).parseOptional();
+          assertThat(value).isEqualTo(1);
+        });
+  }
+
+  @TestTemplate
+  public void testCamelCaseSparkOption() {
+    Table table = validationCatalog.loadTable(tableIdent);
+    String option = "some-int-option";
+    String sparkOption = "someIntOption";
+    Map<String, String> options = ImmutableMap.of(sparkOption, "1");
+    SparkConfParser parser = new SparkConfParser(spark, table, options);
+    Integer value = parser.intConf().option(option).parseOptional();
+    assertThat(value).isEqualTo(1);
+  }
+
   @TestTemplate
   public void testDurationConf() {
     Table table = validationCatalog.loadTable(tableIdent);