[SPARK-51986] Support Parameterized SQL queries in sql API

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

This PR aims to support `Parameterized SQL queries` in `sql` API.

### Why are the changes needed?

For feature parity, we had better support this GA feature.

- apache/spark#38864 (Since Spark 3.4.0)
- apache/spark#40623 (Since Spark 3.4.0)
- apache/spark#41568 (Since Spark 3.5.0)
- apache/spark#48965 (GA Since Spark 4.0.0)

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

No.

### How was this patch tested?

Pass the CIs.

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

No.

Closes #103 from dongjoon-hyun/SPARK-51986.

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

Author: dongjoon-hyun

Sources/SparkConnect/DataFrame.swift

@@ -43,9 +43,19 @@ public actor DataFrame: Sendable {
   /// - Parameters:
   ///   - spark: A `SparkSession` instance to use.
   ///   - sqlText: A SQL statement.
-  init(spark: SparkSession, sqlText: String) async throws {
+  ///   - posArgs: An array of strings.
+  init(spark: SparkSession, sqlText: String, _ posArgs: [Sendable]? = nil) async throws {
     self.spark = spark
-    self.plan = sqlText.toSparkConnectPlan
+    if let posArgs {
+      self.plan = sqlText.toSparkConnectPlan(posArgs)
+    } else {
+      self.plan = sqlText.toSparkConnectPlan
+    }
+  }
+
+  init(spark: SparkSession, sqlText: String, _ args: [String: Sendable]) async throws {
+    self.spark = spark
+    self.plan = sqlText.toSparkConnectPlan(args)
   }
 
   public func getPlan() -> Sendable {

Sources/SparkConnect/Extension.swift

@@ -31,6 +31,74 @@ extension String {
     return plan
   }
 
+  func toSparkConnectPlan(_ posArguments: [Sendable]) -> Plan {
+    var sql = Spark_Connect_SQL()
+    sql.query = self
+    sql.posArguments = posArguments.map {
+      var literal = ExpressionLiteral()
+      switch $0 {
+      case let value as Bool:
+        literal.boolean = value
+      case let value as Int8:
+        literal.byte = Int32(value)
+      case let value as Int16:
+        literal.short = Int32(value)
+      case let value as Int32:
+        literal.integer = value
+      case let value as Int64:
+        literal.long = value
+      case let value as Int:
+        literal.long = Int64(value)
+      case let value as String:
+        literal.string = value
+      default:
+        literal.string = $0 as! String
+      }
+      var expr = Spark_Connect_Expression()
+      expr.literal = literal
+      return expr
+    }
+    var relation = Relation()
+    relation.sql = sql
+    var plan = Plan()
+    plan.opType = Plan.OneOf_OpType.root(relation)
+    return plan
+  }
+
+  func toSparkConnectPlan(_ namedArguments: [String: Sendable]) -> Plan {
+    var sql = Spark_Connect_SQL()
+    sql.query = self
+    sql.namedArguments = namedArguments.mapValues { value in
+      var literal = ExpressionLiteral()
+      switch value {
+      case let value as Bool:
+        literal.boolean = value
+      case let value as Int8:
+        literal.byte = Int32(value)
+      case let value as Int16:
+        literal.short = Int32(value)
+      case let value as Int32:
+        literal.integer = value
+      case let value as Int64:
+        literal.long = value
+      case let value as Int:
+        literal.long = Int64(value)
+      case let value as String:
+        literal.string = value
+      default:
+        literal.string = value as! String
+      }
+      var expr = Spark_Connect_Expression()
+      expr.literal = literal
+      return expr
+    }
+    var relation = Relation()
+    relation.sql = sql
+    var plan = Plan()
+    plan.opType = Plan.OneOf_OpType.root(relation)
+    return plan
+  }
+
   /// Get a `UserContext` instance from a string.
   var toUserContext: UserContext {
     var context = UserContext()

Sources/SparkConnect/SparkSession.swift

@@ -112,6 +112,26 @@ public actor SparkSession {
     return try await DataFrame(spark: self, sqlText: sqlText)
   }
 
+  /// Executes a SQL query substituting positional parameters by the given arguments, returning the
+  /// result as a `DataFrame`.
+  /// - Parameters:
+  ///   - sqlText: A SQL statement with positional parameters to execute.
+  ///   - args: ``Sendable`` values that can be converted to SQL literal expressions.
+  /// - Returns: A ``DataFrame``.
+  public func sql(_ sqlText: String, _ args: Sendable...) async throws -> DataFrame {
+    return try await DataFrame(spark: self, sqlText: sqlText, args)
+  }
+
+  /// Executes a SQL query substituting named parameters by the given arguments, returning the
+  /// result as a `DataFrame`.
+  /// - Parameters:
+  ///   - sqlText: A SQL statement with named parameters to execute.
+  ///   - args: A dictionary with key string and ``Sendable`` value.
+  /// - Returns: A ``DataFrame``.
+  public func sql(_ sqlText: String, args: [String: Sendable]) async throws -> DataFrame {
+    return try await DataFrame(spark: self, sqlText: sqlText, args)
+  }
+
   /// Returns a ``DataFrameReader`` that can be used to read non-streaming data in as a
   /// `DataFrame`
   public var read: DataFrameReader {

Sources/SparkConnect/TypeAliases.swift

@@ -28,6 +28,7 @@ typealias Drop = Spark_Connect_Drop
 typealias ExecutePlanRequest = Spark_Connect_ExecutePlanRequest
 typealias ExecutePlanResponse = Spark_Connect_ExecutePlanResponse
 typealias ExplainMode = AnalyzePlanRequest.Explain.ExplainMode
+typealias ExpressionLiteral = Spark_Connect_Expression.Literal
 typealias ExpressionString = Spark_Connect_Expression.ExpressionString
 typealias Filter = Spark_Connect_Filter
 typealias GroupType = Spark_Connect_Aggregate.GroupType

Tests/SparkConnectTests/SparkSessionTests.swift

@@ -76,6 +76,19 @@ struct SparkSessionTests {
     await spark.stop()
   }
 
+#if !os(Linux)
+  @Test
+  func sql() async throws {
+    let spark = try await SparkSession.builder.getOrCreate()
+    let expected = [Row(true, 1, "a")]
+    if await spark.version.starts(with: "4.") {
+      #expect(try await spark.sql("SELECT ?, ?, ?", true, 1, "a").collect() == expected)
+      #expect(try await spark.sql("SELECT :x, :y, :z", args: ["x": true, "y": 1, "z": "a"]).collect() == expected)
+    }
+    await spark.stop()
+  }
+#endif
+
   @Test
   func table() async throws {
     let tableName = "TABLE_" + UUID().uuidString.replacingOccurrences(of: "-", with: "")