Add Delta Lake 4.2 support with basic features

delta-lake/README.md

@@ -19,6 +19,7 @@ and directory contains the corresponding support code.
 | 3.3.x              | Spark 3.5.[3-]  | `delta-33x`        |
 | 4.0.x              | Spark 4.0.x     | `delta-40x`        |
 | 4.1.0              | Spark 4.1.0, 4.1.1 | `delta-41x`     |
+| 4.2.0              | Spark 4.0.1, 4.1.1 | `delta-42x`     |
 | Databricks 14.3    | Databricks 14.3 | `delta-spark350db143` |
 
 Delta Lake is not supported on all Spark versions, and for Spark versions where it is not

delta-lake/common/src/main/delta-20x-24x/scala/org/apache/spark/sql/delta/rapids/GpuWriteIntoDelta.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2023, NVIDIA CORPORATION.
+ * Copyright (c) 2022-2026, NVIDIA CORPORATION.
  *
  * This file was derived from WriteIntoDelta.scala
  * in the Delta Lake project at https://github.com/delta-io/delta.
@@ -30,7 +30,8 @@ import org.apache.spark.sql.execution.command.LeafRunnableCommand
 case class GpuWriteIntoDelta(
     gpuDeltaLog: GpuDeltaLog,
     cpuWrite: WriteIntoDelta)
-    extends LeafRunnableCommand {
+    extends LeafRunnableCommand
+      with GpuWriteIntoDeltaLike {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     gpuDeltaLog.withNewTransaction { txn =>

delta-lake/delta-40x/src/main/scala/org/apache/spark/sql/delta/rapids/DeltaMdcShims.scala → delta-lake/common/src/main/delta-20x-24x/scala/org/apache/spark/sql/delta/rapids/GpuWriteIntoDeltaLike.scala

@@ -16,9 +16,9 @@
 
 package org.apache.spark.sql.delta.rapids
 
-import org.apache.spark.internal.{LogKey, MDC}
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 
-object DeltaMdcShims {
-  def mdc(logKey: AnyRef, value: Any): MDC =
-    MDC.of(logKey.asInstanceOf[LogKey], value)
-}
+/**
+ * GPU counterpart of WriteIntoDeltaLike.
+ */
+trait GpuWriteIntoDeltaLike extends LeafRunnableCommand

delta-lake/common/src/main/delta-33x-41x/scala/com/nvidia/spark/rapids/delta/common/DeltaCDFRelationStrategy.scala

@@ -16,46 +16,11 @@
 
 package com.nvidia.spark.rapids.delta.common
 
-import org.apache.spark.sql.catalyst.expressions.{Alias, And, Literal}
-import org.apache.spark.sql.catalyst.planning.PhysicalOperation
-import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
+import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.delta.commands.cdc.CDCReader.DeltaCDFRelation
-import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
-import org.apache.spark.sql.execution.datasources.LogicalRelation
 
-/**
- * Plans the internal DataFrame of an OSS Delta batch CDF relation directly.
- *
- * DeltaCDFRelation.buildScan returns the internal DataFrame as RDD[Row]. Spark wraps that RDD in a
- * RowDataSourceScanExec, introducing a row boundary around file scans that can otherwise remain
- * columnar. Replanning the internal logical plan exposes those scans to the regular Spark and
- * RAPIDS planning rules.
- */
-object DeltaCDFRelationStrategy extends SparkStrategy {
-
-  override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
-    case PhysicalOperation(projects, filters, relation: LogicalRelation)
-        if relation.relation.isInstanceOf[DeltaCDFRelation] =>
-      val cdf = relation.relation.asInstanceOf[DeltaCDFRelation]
-      if (cdf.startingVersion.isEmpty) {
-        Nil
-      } else {
-        val spark = cdf.sqlContext.sparkSession
-        val changes = DeltaCDFRelationShim.changesToBatchDF(cdf)
-
-        val changesByName = changes.queryExecution.analyzed.output.map(a => a.name -> a).toMap
-        val relationOutput = relation.output.map { attr =>
-          Alias(changesByName(attr.name), attr.name)(
-            exprId = attr.exprId,
-            qualifier = attr.qualifier,
-            explicitMetadata = Some(attr.metadata))
-        }
-        val filter = filters.reduceOption(And).getOrElse(Literal.TrueLiteral)
-        val rewritten = Project(projects,
-          Filter(filter, Project(relationOutput, changes.queryExecution.analyzed)))
+object DeltaCDFRelationStrategy extends DeltaCDFRelationStrategyBase {
 
-        Seq(planLater(spark.sessionState.optimizer.execute(rewritten)))
-      }
-    case _ => Nil
-  }
+  override protected def changesToBatchDF(cdf: DeltaCDFRelation): DataFrame =
+    DeltaCDFRelationShim.changesToBatchDF(cdf)
 }

delta-lake/common/src/main/delta-33x-41x/scala/com/nvidia/spark/rapids/delta/common/DeltaReorgTableCommandMeta.scala

@@ -23,9 +23,8 @@ import com.nvidia.spark.rapids.delta.RapidsDeltaUtils
 
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.delta.{IcebergCompat, RowTracking, UniversalFormat}
-import org.apache.spark.sql.delta.commands.{DeltaCommand, DeltaReorgTableCommand,
-  DeltaReorgTableMode}
-import org.apache.spark.sql.delta.rapids.GpuDeltaReorgTableCommand
+import org.apache.spark.sql.delta.commands.{DeltaReorgTableCommand, DeltaReorgTableMode}
+import org.apache.spark.sql.delta.rapids.{GpuDeltaCommandLike, GpuDeltaReorgTableCommand}
 import org.apache.spark.sql.execution.command.RunnableCommand
 
 object DeltaReorgTableCommandMeta {
@@ -47,9 +46,9 @@ class DeltaReorgTableCommandMeta(
     conf: RapidsConf,
     parent: Option[RapidsMeta[_, _, _]],
     rule: DataFromReplacementRule)
-  extends RunnableCommandMeta[DeltaReorgTableCommand](cmd, conf, parent, rule) {
+  extends DeltaReorgTableCommandMetaBase(cmd, conf, parent, rule) {
 
-  private object DeltaCmdProxy extends DeltaCommand
+  private object DeltaCmdProxy extends GpuDeltaCommandLike
 
   override def tagSelfForGpu(): Unit = {
     if (!conf.isDeltaWriteEnabled) {

delta-lake/common/src/main/delta-33x-41x/scala/com/nvidia/spark/rapids/delta/DeltaWriteUtils.scala → delta-lake/common/src/main/delta-33x-42x/scala/com/nvidia/spark/rapids/delta/DeltaWriteUtils.scala

@@ -22,6 +22,10 @@ import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.internal.SQLConf
 
 object DeltaWriteUtils {
+  def toBooleanOption(enabled: Boolean): Option[Boolean] = {
+    if (enabled) Some(true) else None
+  }
+
   // scalastyle:off line.size.limit
   /**
    * Optimized writes can be enabled/disabled through the following order:

delta-lake/common/src/main/delta-33x-41x/scala/com/nvidia/spark/rapids/delta/GpuDeltaCatalogBase.scala → delta-lake/common/src/main/delta-33x-42x/scala/com/nvidia/spark/rapids/delta/GpuDeltaCatalogBase.scala

@@ -38,7 +38,8 @@ import org.apache.spark.sql.delta.{ColumnWithDefaultExprUtils, DeltaConfigs, Del
 import org.apache.spark.sql.delta.catalog.DeltaCatalog
 import org.apache.spark.sql.delta.commands.{TableCreationModes, WriteIntoDelta}
 import org.apache.spark.sql.delta.metering.DeltaLogging
-import org.apache.spark.sql.delta.rapids.{DeltaTrampoline, GpuDeltaLog, GpuWriteIntoDelta}
+import org.apache.spark.sql.delta.rapids.{DeltaRuntimeShim, DeltaTrampoline, GpuDeltaLog,
+  GpuWriteIntoDeltaLike}
 import org.apache.spark.sql.delta.sources.{DeltaSourceUtils, DeltaSQLConf}
 import org.apache.spark.sql.delta.stats.StatisticsCollection
 import org.apache.spark.sql.execution.datasources.DataSource
@@ -64,11 +65,48 @@ abstract class GpuDeltaCatalogBase(
       withDb: CatalogTable,
       existingTableOpt: Option[CatalogTable],
       mode: SaveMode,
-      writer: Option[GpuWriteIntoDelta],
+      writer: Option[GpuWriteIntoDeltaLike],
       operation: TableCreationModes.CreationMode,
       isByPath: Boolean,
       tableCreateFunc: Option[CatalogTable => Unit]): Unit
 
+  /**
+   * Converts a V2 catalog identifier to a V1 table identifier.
+   *
+   * @param ident the V2 identifier containing the table name and namespace
+   * @return a V1 identifier whose table name is `ident.name()` and whose optional database is the
+   *         last element of `ident.namespace()`
+   */
+  protected def getTableIdentifier(ident: Identifier): TableIdentifier = {
+    TableIdentifier(ident.name(), ident.namespace().lastOption)
+  }
+
+  /**
+   * Finds the catalog metadata for a table that may already exist.
+   *
+   * @param table the V1 table identifier derived from `ident`
+   * @param ident the original V2 identifier
+   * @param operation the requested creation mode
+   * @return the existing table metadata when an applicable catalog lookup finds the table, or
+   *         `None` otherwise
+   */
+  protected def getExistingTableIfExists(
+      table: TableIdentifier,
+      ident: Identifier,
+      operation: TableCreationModes.CreationMode): Option[CatalogTable]
+
+  /**
+   * Determines whether table metadata should be created through the Delta catalog or
+   * Spark's session catalog.
+   *
+   * @param sourceQuery the data produced by a CTAS or RTAS query, or `None` when the table
+   *                    creation or replacement has no `AS SELECT` clause
+   * @return `true` to create the table metadata through the Delta catalog, `false` otherwise.
+   */
+  protected def useCatalogCreateTable(sourceQuery: Option[DataFrame]): Boolean = {
+    isUnityCatalog && sourceQuery.isEmpty
+  }
+
   /** copied from trait SupportsPathIdentifier */
   private def supportSQLOnFile: Boolean = spark.sessionState.conf.runSQLonFile
 
@@ -201,11 +239,9 @@ abstract class GpuDeltaCatalogBase(
     } else {
       Option(allTableProperties.get("location"))
     }
-    val id = {
-      TableIdentifier(ident.name(), ident.namespace().lastOption)
-    }
+    val id = getTableIdentifier(ident)
     val locUriOpt = location.map(CatalogUtils.stringToURI)
-    val existingTableOpt = cpuCatalog.getExistingTableIfExists(id)
+    val existingTableOpt = getExistingTableIfExists(id, ident, operation)
     // PROP_IS_MANAGED_LOCATION indicates that the table location is not user-specified but
     // system-generated. The table should be created as managed table in this case.
     val isManagedLocation = Option(allTableProperties.get(TableCatalog.PROP_IS_MANAGED_LOCATION))
@@ -258,7 +294,7 @@ abstract class GpuDeltaCatalogBase(
         Some(tableDesc),
         schemaInCatalog = if (newSchema != schema) Some(newSchema) else None)
       val gpuDeltaLog = new GpuDeltaLog(deltaLog, rapidsConf)
-      GpuWriteIntoDelta(gpuDeltaLog, cpuWriter)
+      DeltaRuntimeShim.createGpuWrite(gpuDeltaLog, cpuWriter)
     }
 
 
@@ -268,7 +304,7 @@ abstract class GpuDeltaCatalogBase(
     // TODO: Spark `V2SessionCatalog` mistakenly treat tables with location as EXTERNAL table.
     //       Before this bug is fixed, we should only call the catalog plugin API to create tables
     //       if UC is enabled to replace `V2SessionCatalog`.
-    val tableCreateFunc = if (isUnityCatalog && sourceQuery.isEmpty) {
+    val tableCreateFunc = if (useCatalogCreateTable(sourceQuery)) {
       Some[CatalogTable => Unit](v1Table => {
         val t = DeltaTrampoline.getV1Table(v1Table)
         cpuCatalog.createTable(ident, t.columns(), t.partitioning, t.properties)

delta-lake/common/src/main/delta-33x-42x/scala/com/nvidia/spark/rapids/delta/common/DeltaCDFRelationStrategyBase.scala

@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+
+package com.nvidia.spark.rapids.delta.common
+
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Literal}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
+import org.apache.spark.sql.delta.commands.cdc.CDCReader.DeltaCDFRelation
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+
+/**
+ * Shared planning logic for exposing the internal DataFrame of an OSS Delta batch CDF relation.
+ * Concrete strategies provide the version-specific Delta API call that builds the DataFrame.
+ */
+abstract class DeltaCDFRelationStrategyBase extends SparkStrategy {
+
+  protected def changesToBatchDF(cdf: DeltaCDFRelation): DataFrame
+
+  override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+    case PhysicalOperation(projects, filters, relation: LogicalRelation)
+        if relation.relation.isInstanceOf[DeltaCDFRelation] =>
+      val cdf = relation.relation.asInstanceOf[DeltaCDFRelation]
+      if (cdf.startingVersion.isEmpty) {
+        Nil
+      } else {
+        val spark = cdf.sqlContext.sparkSession
+        val changes = changesToBatchDF(cdf)
+
+        val changesByName = changes.queryExecution.analyzed.output.map(a => a.name -> a).toMap
+        val relationOutput = relation.output.map { attr =>
+          Alias(changesByName(attr.name), attr.name)(
+            exprId = attr.exprId,
+            qualifier = attr.qualifier,
+            explicitMetadata = Some(attr.metadata))
+        }
+        val filter = filters.reduceOption(And).getOrElse(Literal.TrueLiteral)
+        val rewritten = Project(projects,
+          Filter(filter, Project(relationOutput, changes.queryExecution.analyzed)))
+
+        Seq(planLater(spark.sessionState.optimizer.execute(rewritten)))
+      }
+    case _ => Nil
+  }
+}

delta-lake/common/src/main/delta-33x-41x/scala/com/nvidia/spark/rapids/delta/common/DeltaDynamicPartitionOverwriteCommandMetaBase.scala → delta-lake/common/src/main/delta-33x-42x/scala/com/nvidia/spark/rapids/delta/common/DeltaDynamicPartitionOverwriteCommandMetaBase.scala

@@ -52,7 +52,7 @@ class DeltaDynamicPartitionOverwriteCommandMetaBase(
 
     RapidsDeltaUtils.tagForDeltaWrite(this, overwriteCommand.table.schema,
       Some(overwriteCommand.deltaTable.deltaLog),
-      Map.empty, overwriteCommand.deltaTable.spark)
+      overwriteCommand.writeOptions, overwriteCommand.deltaTable.spark)
   }
 
   override def convertToGpu(): RunnableCommand = {

delta-lake/common/src/main/delta-33x-41x/scala/com/nvidia/spark/rapids/delta/common/DeltaProviderBase.scala → delta-lake/common/src/main/delta-33x-42x/scala/com/nvidia/spark/rapids/delta/common/DeltaProviderBase.scala

@@ -71,8 +71,10 @@ case class GpuIncrementMetric(cpuInc: IncrementMetric, override val child: Expre
 
 abstract class DeltaProviderBase extends DeltaIOProvider {
 
+  protected def getCDFRelationStrategy: SparkStrategy
+
   override def getStrategyRules: Seq[SparkStrategy] =
-    DeltaCDFRelationStrategy +: super.getStrategyRules
+    getCDFRelationStrategy +: super.getStrategyRules
 
   override def getCreatableRelationRules: Map[Class[_ <: CreatableRelationProvider],
       CreatableRelationProviderRule[_ <: CreatableRelationProvider]] = {

delta-lake/common/src/main/delta-33x-42x/scala/com/nvidia/spark/rapids/delta/common/DeltaReorgTableCommandMetaBase.scala

@@ -0,0 +1,29 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+
+package com.nvidia.spark.rapids.delta.common
+
+import com.nvidia.spark.rapids.{DataFromReplacementRule, RapidsConf, RapidsMeta,
+  RunnableCommandMeta}
+
+import org.apache.spark.sql.delta.commands.DeltaReorgTableCommand
+
+abstract class DeltaReorgTableCommandMetaBase(
+    cmd: DeltaReorgTableCommand,
+    conf: RapidsConf,
+    parent: Option[RapidsMeta[_, _, _]],
+    rule: DataFromReplacementRule)
+  extends RunnableCommandMeta[DeltaReorgTableCommand](cmd, conf, parent, rule)

delta-lake/common/src/main/delta-33x-41x/scala/com/nvidia/spark/rapids/delta/common/RapidsRowIndexFilters.scala → delta-lake/common/src/main/delta-33x-42x/scala/com/nvidia/spark/rapids/delta/common/RapidsRowIndexFilters.scala

@@ -127,4 +127,3 @@ trait RapidsRowIndexMarkingFiltersBuilder {
     }
   }
 }
-

delta-lake/common/src/main/delta-33x-41x/scala/org/apache/spark/sql/GpuDeltaDynamicPartitionOverwriteCommand.scala → delta-lake/common/src/main/delta-33x-42x/scala/org/apache/spark/sql/GpuDeltaDynamicPartitionOverwriteCommand.scala

@@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.delta.DeltaOptions
 import org.apache.spark.sql.delta.catalog.DeltaTableV2
 import org.apache.spark.sql.delta.commands.WriteIntoDelta
-import org.apache.spark.sql.delta.rapids.{DeltaCommandShims, GpuDeltaLog, GpuWriteIntoDelta}
+import org.apache.spark.sql.delta.rapids.{DeltaCommandShims, DeltaRuntimeShim, GpuDeltaLog}
 import org.apache.spark.sql.execution.command.RunnableCommand
 
 case class GpuDeltaDynamicPartitionOverwriteCommand(
@@ -62,8 +62,8 @@ case class GpuDeltaDynamicPartitionOverwriteCommand(
 
     val operationSession = shims.toOperationSparkSession(
       sparkSession.asInstanceOf[shims.ShimSparkSession])
-    
-    GpuWriteIntoDelta(
+
+    DeltaRuntimeShim.createGpuWrite(
       gpuDeltaLog,
       WriteIntoDelta(
         gpuDeltaLog.deltaLog,

delta-lake/common/src/main/delta-33x-41x/scala/org/apache/spark/sql/delta/hooks/GpuAutoCompact.scala → delta-lake/common/src/main/delta-33x-42x/scala/org/apache/spark/sql/delta/hooks/GpuAutoCompact.scala

@@ -36,6 +36,8 @@ import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.stats.AutoCompactPartitionStats
 
 trait GpuAutoCompactBase extends AutoCompactBase {
+  protected def getTableId(deltaLog: DeltaLog): String
+
   /**
    * Execute a prepared auto-compaction request. Version-specific shims are responsible for
    * constructing the request with the correct Delta API.
@@ -48,11 +50,10 @@ trait GpuAutoCompactBase extends AutoCompactBase {
       opType: String,
       maxDeletedRowsRatio: Option[Double]
   ): Seq[OptimizeMetrics] = {
-    val tableId = deltaLog.tableId
+    val tableId = getTableId(deltaLog)
     if (autoCompactRequest.shouldCompact) {
       try {
-        val metrics = GpuAutoCompact
-          .compact(
+        val metrics = compact(
             spark,
             deltaLog,
             catalogTable,

delta-lake/common/src/main/delta-33x-41x/scala/org/apache/spark/sql/delta/rapids/DeltaTrampoline.scala → delta-lake/common/src/main/delta-33x-42x/scala/org/apache/spark/sql/delta/rapids/DeltaTrampoline.scala

@@ -62,4 +62,4 @@ object DeltaTrampoline {
 
     (identityCols.toSeq, bucketSpec, clusterBySpec)
   }
-}
+}