Add shims layer, Delete command base classes and refactor delta-33x Delete

delta-lake/common/src/main/delta-33x-40x/scala/org/apache/spark/sql/delta/rapids/DeltaCommandShims.scala

@@ -22,35 +22,57 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 
 /**
  * Trait to abstract version-specific Spark API differences between Delta 3.3.x and 4.0.x.
- * Mainly handles SparkSession type changes and Column/DataFrame creation APIs introduced
- * in Spark 4.0.
+ * 
+ * Key API differences handled:
+ * 1. SparkSession types: Spark 4.0 split SparkSession to support both classic and connect
+ *    execution modes. It introduced:
+ *    - org.apache.spark.sql.SparkSession (SqlSparkSession): unified API interface
+ *    - org.apache.spark.sql.classic.SparkSession (ClassicSparkSession): classic execution mode
+ *    - org.apache.spark.sql.connect.SparkSession: Spark Connect remote execution
+ *    Spark 3.x uses a single SparkSession type for all purposes.
+ * 2. DataFrame creation: Spark 3.x uses Dataset.ofRows(), Spark 4.0 uses
+ *    TrampolineConnectShims.createDataFrame().
+ * 3. Column creation: Spark 3.x uses new Column(expr), Spark 4.0 uses DFUDFShims.exprToColumn().
  */
 trait DeltaCommandShims {
 
   /**
-   * Type alias for the version-specific SparkSession type used in run() methods.
-   * Delta 3.3.x uses SparkSession, Delta 4.0.x uses SqlSparkSession.
+   * Type alias for the version-specific SparkSession type used in the shimming layer.
+   * This type is used when:
+   * 1. Casting the SparkSession parameter inside run() method body
+   * 2. As parameter type for shim methods (toOperationSparkSession, recacheByPlan)
+   * 
    */
-  type RunSparkSession <: SparkSession
+  type ShimSparkSession <: SparkSession
 
   /**
-   * Type alias for the version-specific SparkSession type used in operations.
-   * Delta 3.3.x uses SparkSession, Delta 4.0.x uses ClassicSparkSession.
+   * Type alias for the version-specific SparkSession type used for internal operations.
+   * This is the SparkSession type used within command execution for operations like:
+   * - Creating DataFrames from LogicalPlans (via createDataFrame method)
+   * - Utility operations (splitMetadataAndDataPredicates, createSetTransaction)
+   * 
+   * Version mapping:
+   * - Delta 3.3.x (Spark 3.x): org.apache.spark.sql.SparkSession
+   * - Delta 4.0.x (Spark 4.0): org.apache.spark.sql.classic.SparkSession (ClassicSparkSession)
    */
   type OperationSparkSession <: SparkSession
 
   /**
-   * Convert RunSparkSession to OperationSparkSession.
+   * Convert ShimSparkSession to OperationSparkSession.
+   * In Spark 4.0, this converts SqlSparkSession to ClassicSparkSession.
+   * In Spark 3.x, Both types are SparkSession.
    */
-  def toOperationSparkSession(spark: RunSparkSession): OperationSparkSession
+  def toOperationSparkSession(spark: ShimSparkSession): OperationSparkSession
 
   /**
    * Get the active OperationSparkSession.
    */
   def getActiveOperationSparkSession: OperationSparkSession
 
   /**
-   * Create a DataFrame from a LogicalPlan (version-specific API).
+   * Create a DataFrame from a LogicalPlan using version-specific API.
+   * - Spark 3.x: Dataset.ofRows(spark, logicalPlan)
+   * - Spark 4.0: TrampolineConnectShims.createDataFrame(spark, logicalPlan)
    */
   def createDataFrame(spark: OperationSparkSession, logicalPlan: LogicalPlan): DataFrame
 
@@ -62,5 +84,5 @@ trait DeltaCommandShims {
   /**
    * Recache by plan with the correct SparkSession type.
    */
-  def recacheByPlan(spark: RunSparkSession, plan: LogicalPlan): Unit
+  def recacheByPlan(spark: ShimSparkSession, plan: LogicalPlan): Unit
 }

delta-lake/common/src/main/delta-33x-40x/scala/org/apache/spark/sql/delta/rapids/GpuDeleteCommandBase.scala

@@ -75,12 +75,12 @@ abstract class GpuDeleteCommandBase(
     recordDeltaOperation(gpuDeltaLog.deltaLog, "delta.dml.delete") {
       gpuDeltaLog.withNewTransaction(catalogTable) { txn =>
         DeltaLog.assertRemovable(txn.snapshot)
-        if (hasBeenExecuted(txn, sparkSession.asInstanceOf[RunSparkSession])) {
+        if (hasBeenExecuted(txn, sparkSession.asInstanceOf[ShimSparkSession])) {
           sendDriverMetrics(sparkSession, metrics)
           return Seq.empty
         }
 
-        val opSpark = toOperationSparkSession(sparkSession.asInstanceOf[RunSparkSession])
+        val opSpark = toOperationSparkSession(sparkSession.asInstanceOf[ShimSparkSession])
         val (deleteActions, deleteMetrics) = performDelete(opSpark, deltaLog, txn)
         val commitVersion = txn.commitIfNeeded(
           actions = deleteActions,
@@ -94,7 +94,7 @@ abstract class GpuDeleteCommandBase(
       }
       // Re-cache all cached plans(including this relation itself, if it's cached) that refer to
       // this data source relation.
-      recacheByPlan(sparkSession.asInstanceOf[RunSparkSession], target)
+      recacheByPlan(sparkSession.asInstanceOf[ShimSparkSession], target)
     }
 
     // Adjust for deletes at partition boundaries. Deletes at partition boundaries is a metadata

delta-lake/delta-33x/src/main/scala/org/apache/spark/sql/delta/rapids/delta33x/Delta33xCommandShims.scala

@@ -26,10 +26,10 @@ import org.apache.spark.sql.delta.rapids.DeltaCommandShims
  * Uses the original Spark 3.x APIs.
  */
 trait Delta33xCommandShims extends DeltaCommandShims {
-  override type RunSparkSession = SparkSession
+  override type ShimSparkSession = SparkSession
   override type OperationSparkSession = SparkSession
 
-  override def toOperationSparkSession(spark: RunSparkSession): OperationSparkSession = spark
+  override def toOperationSparkSession(spark: ShimSparkSession): OperationSparkSession = spark
 
   override def getActiveOperationSparkSession: OperationSparkSession = SparkSession.active
 
@@ -41,7 +41,7 @@ trait Delta33xCommandShims extends DeltaCommandShims {
 
   override def exprToColumn(expr: Expression): Column = new Column(expr)
 
-  override def recacheByPlan(spark: RunSparkSession, plan: LogicalPlan): Unit = {
+  override def recacheByPlan(spark: ShimSparkSession, plan: LogicalPlan): Unit = {
     spark.sharedState.cacheManager.recacheByPlan(spark, plan)
   }