[SPARK-52098] Support JDBC in DataFrameReader and Writer

Sources/SparkConnect/DataFrameReader.swift

@@ -258,4 +258,21 @@ public actor DataFrameReader: Sendable {
     self.source = "parquet"
     return load(paths)
   }
+
+  /// Construct a `DataFrame` representing the database table accessible via JDBC URL url named
+  /// table and connection properties.
+  /// - Parameters:
+  ///   - url: The JDBC URL of the form `jdbc:subprotocol:subname` to connect to.
+  ///   - table: The JDBC table that should be read from or written into.
+  ///   - properties: A string-string dictionary for connection properties.
+  /// - Returns: A `DataFrame`.
+  public func jdbc(_ url: String, _ table: String, _ properties: [String: String] = [:]) -> DataFrame {
+    for (key, value) in properties {
+      self.extraOptions[key] = value
+    }
+    self.extraOptions["url"] = url
+    self.extraOptions["dbtable"] = table
+    self.source = "jdbc"
+    return load()
+  }
 }

Sources/SparkConnect/DataFrameWriter.swift

@@ -228,4 +228,21 @@ public actor DataFrameWriter: Sendable {
     self.source = "text"
     return try await save(path)
   }
+
+  /// Saves the content of the `DataFrame` to an external database table via JDBC. In the case the
+  /// table already exists in the external database, behavior of this function depends on the save
+  /// mode, specified by the `mode` function (default to throwing an exception).
+  /// - Parameters:
+  ///   - url: The JDBC URL of the form `jdbc:subprotocol:subname` to connect to.
+  ///   - table: Name of the table in the external database.
+  ///   - properties:JDBC database connection arguments, a list of arbitrary string tag/value.
+  public func jdbc(_ url: String, _ table: String, _ properties: [String: String] = [:]) async throws {
+    for (key, value) in properties {
+      self.extraOptions[key] = value
+    }
+    self.extraOptions["url"] = url
+    self.extraOptions["dbtable"] = table
+    self.source = "jdbc"
+    return try await save()
+  }
 }