[SPARK-51708] Add CaseInsensitiveDictionary

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

This PR aims to add `CaseInsensitiveDictionary` and use it in the following classes.
- `DataFrameReader`
- `DataFrameWriter`

### Why are the changes needed?

For feature parity.

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

No, this is a new addition to the unreleased version.

### How was this patch tested?

Pass the CIs.

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

No.

Closes #40 from dongjoon-hyun/SPARK-51708.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

Sources/SparkConnect/CaseInsensitiveDictionary.swift

@@ -0,0 +1,63 @@
+//
+// 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.
+//
+import Foundation
+
+/// A  dictionary in which keys are case insensitive. The input dictionary can be
+/// accessed for cases where case-sensitive information is required.
+public struct CaseInsensitiveDictionary: Sendable {
+  public var originalDictionary: [String: Sendable]
+  private var keyLowerCasedDictionary: [String: Sendable] = [:]
+
+  init(_ originalDictionary: [String: Sendable] = [:]) {
+    self.originalDictionary = originalDictionary
+    for (key, value) in originalDictionary {
+      keyLowerCasedDictionary[key.lowercased()] = value
+    }
+  }
+
+  subscript(key: String) -> Sendable? {
+    get {
+      return keyLowerCasedDictionary[key.lowercased()]
+    }
+    set {
+      let lowerKey = key.lowercased()
+      if let newValue = newValue {
+        keyLowerCasedDictionary[lowerKey] = newValue
+      } else {
+        keyLowerCasedDictionary.removeValue(forKey: lowerKey)
+      }
+      originalDictionary = originalDictionary.filter { $0.key.lowercased() != lowerKey }
+      if let newValue = newValue {
+        originalDictionary[key] = newValue
+      }
+    }
+  }
+
+  public func toDictionary() -> [String: Sendable] {
+    return originalDictionary
+  }
+
+  public func toStringDictionary() -> [String: String] {
+    return originalDictionary.mapValues { String(describing: $0) }
+  }
+
+  public var count: Int {
+    return keyLowerCasedDictionary.count
+  }
+}

Sources/SparkConnect/DataFrameReader.swift

@@ -32,8 +32,7 @@ public actor DataFrameReader: Sendable {
 
   var paths: [String] = []
 
-  // TODO: Case-insensitive Map
-  var extraOptions: [String: String] = [:]
+  var extraOptions: CaseInsensitiveDictionary = CaseInsensitiveDictionary([:])
 
   let sparkSession: SparkSession
 
@@ -84,7 +83,7 @@ public actor DataFrameReader: Sendable {
     var dataSource = DataSource()
     dataSource.format = self.source
     dataSource.paths = self.paths
-    dataSource.options = self.extraOptions
+    dataSource.options = self.extraOptions.toStringDictionary()
 
     var read = Read()
     read.dataSource = dataSource

Sources/SparkConnect/DataFrameWriter.swift

@@ -32,8 +32,7 @@ public actor DataFrameWriter: Sendable {
 
   var saveMode: String = "default"
 
-  // TODO: Case-insensitive Map
-  var extraOptions: [String: String] = [:]
+  var extraOptions: CaseInsensitiveDictionary = CaseInsensitiveDictionary()
 
   var partitioningColumns: [String]? = nil
 
@@ -146,7 +145,7 @@ public actor DataFrameWriter: Sendable {
       write.bucketBy = bucketBy
     }
 
-    for option in self.extraOptions {
+    for option in self.extraOptions.toStringDictionary() {
       write.options[option.key] = option.value
     }
 

Tests/SparkConnectTests/CaseInsensitiveDictionaryTests.swift

@@ -0,0 +1,87 @@
+//
+// 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.
+//
+
+import Foundation
+import Testing
+
+@testable import SparkConnect
+
+/// A test suite for `CaseInsensitiveDictionary`
+struct CaseInsensitiveDictionaryTests {
+  @Test
+  func empty() async throws {
+    let dict = CaseInsensitiveDictionary([:])
+    #expect(dict.count == 0)
+  }
+
+  @Test
+  func originalDictionary() async throws {
+    let dict = CaseInsensitiveDictionary([
+      "key1": "value1",
+      "KEY1": "VALUE1",
+    ])
+    #expect(dict.count == 1)
+    #expect(dict.originalDictionary.count == 2)
+  }
+
+  @Test
+  func toDictionary() async throws {
+    let dict = CaseInsensitiveDictionary([
+      "key1": "value1",
+      "KEY1": "VALUE1",
+    ])
+    #expect(dict.toDictionary().count == 2)
+  }
+
+  @Test
+  func `subscript`() async throws {
+    var dict = CaseInsensitiveDictionary([:])
+    #expect(dict.count == 0)
+
+    dict["KEY1"] = "value1"
+    #expect(dict.count == 1)
+    #expect(dict["key1"] as! String == "value1")
+    #expect(dict["KEY1"] as! String == "value1")
+    #expect(dict["KeY1"] as! String == "value1")
+
+    dict["key2"] = false
+    #expect(dict.count == 2)
+    #expect(dict["kEy2"] as! Bool == false)
+
+    dict["key3"] = 2025
+    #expect(dict.count == 3)
+    #expect(dict["key3"] as! Int == 2025)
+  }
+
+  @Test
+  func updatedOriginalDictionary() async throws {
+    var dict = CaseInsensitiveDictionary([
+      "key1": "value1",
+      "KEY1": "VALUE1",
+    ])
+    #expect(dict.count == 1)
+    #expect(dict.originalDictionary.count == 2)
+
+    dict["KEY1"] = "Swift"
+    #expect(dict["KEY1"] as! String == "Swift")
+    #expect(dict.count == 1)
+    #expect(dict.originalDictionary.count == 1)
+    #expect(dict.toDictionary().count == 1)
+  }
+}