[SPARK-51874][CORE][SQL] Add TypedConfigBuilder for Scala Enumeration

### What changes were proposed in this pull request?

This PR introduces TypedConfigBuilder for Scala Enumeration and leverages it for existing configurations that use Enumeration as parameters.

Before this PR, we need to change them from Enumeration to string, string to Enumeration, back and forth... We also need to do upper-case transformation, .checkValues validation one by one.

After this PR, those steps are centralized.

### Why are the changes needed?

better support for Enumeration-like configurations

### Does this PR introduce _any_ user-facing change?
no

### How was this patch tested?
new tests

### Was this patch authored or co-authored using generative AI tooling?
no

Closes #50674 from yaooqinn/enum.

Authored-by: Kent Yao <[email protected]>
Signed-off-by: Kent Yao <[email protected]>

Author: yaooqinn

common/utils/src/main/scala/org/apache/spark/internal/config/ConfigBuilder.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.internal.config
 
+import java.util.Locale
 import java.util.concurrent.TimeUnit
 import java.util.regex.PatternSyntaxException
 
@@ -46,6 +47,16 @@ private object ConfigHelpers {
     }
   }
 
+  def toEnum[E <: Enumeration](s: String, enumClass: E, key: String): enumClass.Value = {
+    try {
+      enumClass.withName(s.trim.toUpperCase(Locale.ROOT))
+    } catch {
+      case _: NoSuchElementException =>
+        throw new IllegalArgumentException(
+          s"$key should be one of ${enumClass.values.mkString(", ")}, but was $s")
+    }
+  }
+
   def stringToSeq[T](str: String, converter: String => T): Seq[T] = {
     SparkStringUtils.stringToSeq(str).map(converter)
   }
@@ -271,6 +282,11 @@ private[spark] case class ConfigBuilder(key: String) {
     new TypedConfigBuilder(this, v => v)
   }
 
+  def enumConf(e: Enumeration): TypedConfigBuilder[e.Value] = {
+    checkPrependConfig
+    new TypedConfigBuilder(this, toEnum(_, e, key))
+  }
+
   def timeConf(unit: TimeUnit): TypedConfigBuilder[Long] = {
     checkPrependConfig
     new TypedConfigBuilder(this, timeFromString(_, unit), timeToString(_, unit))

core/src/test/scala/org/apache/spark/internal/config/ConfigEntrySuite.scala

@@ -387,4 +387,25 @@ class ConfigEntrySuite extends SparkFunSuite {
     ConfigBuilder(testKey("oc5")).onCreate(_ => onCreateCalled = true).fallbackConf(fallback)
     assert(onCreateCalled)
   }
+
+
+  test("SPARK-51874: Add Enum support to ConfigBuilder") {
+    object MyTestEnum extends Enumeration {
+      val X, Y, Z = Value
+    }
+    val conf = new SparkConf()
+    val enumConf = ConfigBuilder("spark.test.enum.key")
+      .enumConf(MyTestEnum)
+      .createWithDefault(MyTestEnum.X)
+    assert(conf.get(enumConf) === MyTestEnum.X)
+    conf.set(enumConf, MyTestEnum.Y)
+    assert(conf.get(enumConf) === MyTestEnum.Y)
+    conf.set(enumConf.key, "Z")
+    assert(conf.get(enumConf) === MyTestEnum.Z)
+    val e = intercept[IllegalArgumentException] {
+      conf.set(enumConf.key, "A")
+      conf.get(enumConf)
+    }
+    assert(e.getMessage === s"${enumConf.key} should be one of X, Y, Z, but was A")
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CTESubstitution.scala

@@ -84,7 +84,7 @@ object CTESubstitution extends Rule[LogicalPlan] {
 
     val cteDefs = ArrayBuffer.empty[CTERelationDef]
     val (substituted, firstSubstituted) =
-      LegacyBehaviorPolicy.withName(conf.getConf(LEGACY_CTE_PRECEDENCE_POLICY)) match {
+      conf.getConf(LEGACY_CTE_PRECEDENCE_POLICY) match {
         case LegacyBehaviorPolicy.EXCEPTION =>
           assertNoNameConflictsInCTE(plan)
           traverseAndSubstituteCTE(plan, forceInline, Seq.empty, cteDefs, None)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala

@@ -426,7 +426,7 @@ class ResolverGuard(catalogManager: CatalogManager) extends SQLConfHelper {
       Some("hiveCaseSensitiveInferenceMode")
     } else if (conf.getConf(SQLConf.LEGACY_INLINE_CTE_IN_COMMANDS)) {
       Some("legacyInlineCTEInCommands")
-    } else if (LegacyBehaviorPolicy.withName(conf.getConf(SQLConf.LEGACY_CTE_PRECEDENCE_POLICY)) !=
+    } else if (conf.getConf(SQLConf.LEGACY_CTE_PRECEDENCE_POLICY) !=
       LegacyBehaviorPolicy.CORRECTED) {
       Some("legacyCTEPrecedencePolicy")
     } else {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CodeGeneratorWithInterpretedFallback.scala

@@ -38,7 +38,7 @@ abstract class CodeGeneratorWithInterpretedFallback[IN, OUT] extends Logging {
 
   def createObject(in: IN): OUT = {
     // We are allowed to choose codegen-only or no-codegen modes if under tests.
-    val fallbackMode = CodegenObjectFactoryMode.withName(SQLConf.get.codegenFactoryMode)
+    val fallbackMode = SQLConf.get.codegenFactoryMode
 
     fallbackMode match {
       case CodegenObjectFactoryMode.CODEGEN_ONLY =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ToStringBase.scala

@@ -457,7 +457,7 @@ trait ToStringBase { self: UnaryExpression with TimeZoneAwareExpression =>
 object ToStringBase {
   def getBinaryFormatter: BinaryFormatter = {
     val style = SQLConf.get.getConf(SQLConf.BINARY_OUTPUT_STYLE)
-    style.map(BinaryOutputStyle.withName) match {
+    style match {
       case Some(BinaryOutputStyle.UTF8) =>
         (array: Array[Byte]) => UTF8String.fromBytes(array)
       case Some(BinaryOutputStyle.BASIC) =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ArrayBasedMapBuilder.scala

@@ -84,9 +84,9 @@ class ArrayBasedMapBuilder(keyType: DataType, valueType: DataType) extends Seria
       keys.append(keyNormalized)
       values.append(value)
     } else {
-      if (mapKeyDedupPolicy == SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
+      if (mapKeyDedupPolicy == SQLConf.MapKeyDedupPolicy.EXCEPTION) {
         throw QueryExecutionErrors.duplicateMapKeyFoundError(key)
-      } else if (mapKeyDedupPolicy == SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+      } else if (mapKeyDedupPolicy == SQLConf.MapKeyDedupPolicy.LAST_WIN) {
         // Overwrite the previous value, as the policy is last wins.
         values(index) = value
       } else {