Implement update command for deltalake 3.3.x (#13014)

Close #12988 .

---------

Signed-off-by: liurenjie1024 <[email protected]>

Author: liurenjie1024

delta-lake/delta-33x/src/main/scala/com/nvidia/spark/rapids/delta/delta33x/Delta33xProvider.scala

@@ -22,7 +22,7 @@ import com.nvidia.spark.rapids.delta.DeltaIOProvider
 import org.apache.spark.sql.delta.DeltaParquetFileFormat
 import org.apache.spark.sql.delta.DeltaParquetFileFormat.{IS_ROW_DELETED_COLUMN_NAME, ROW_INDEX_COLUMN_NAME}
 import org.apache.spark.sql.delta.catalog.DeltaCatalog
-import org.apache.spark.sql.delta.commands.{DeleteCommand, MergeIntoCommand}
+import org.apache.spark.sql.delta.commands.{DeleteCommand, MergeIntoCommand, UpdateCommand}
 import org.apache.spark.sql.delta.rapids.DeltaRuntimeShim
 import org.apache.spark.sql.execution.FileSourceScanExec
 import org.apache.spark.sql.execution.command.RunnableCommand
@@ -39,6 +39,10 @@ object Delta33xProvider extends DeltaIOProvider {
           "Delete rows from a Delta Lake table",
           (a, conf, p, r) => new DeleteCommandMeta(a, conf, p, r))
         .disabledByDefault("Delta Lake delete support is experimental"),
+      GpuOverrides.runnableCmd[UpdateCommand](
+          "Update rows from a Delta Lake table",
+          (a, conf, p, r) => new UpdateCommandMeta(a, conf, p, r))
+        .disabledByDefault("Delta Lake update support is experimental"),
       GpuOverrides.runnableCmd[MergeIntoCommand](
           "Merge of a source query/table into a Delta Lake table",
           (a, conf, p, r) => new MergeIntoCommandMeta(a, conf, p, r))

delta-lake/delta-33x/src/main/scala/com/nvidia/spark/rapids/delta/delta33x/UpdateCommandMeta.scala

@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2025, 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.delta33x
+
+import com.nvidia.spark.rapids.{DataFromReplacementRule, RapidsConf, RapidsMeta, RunnableCommandMeta}
+import com.nvidia.spark.rapids.delta.RapidsDeltaUtils
+
+import org.apache.spark.sql.delta.commands.{DeletionVectorUtils, UpdateCommand}
+import org.apache.spark.sql.delta.rapids.GpuDeltaLog
+import org.apache.spark.sql.delta.rapids.delta33x.GpuUpdateCommand
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.execution.command.RunnableCommand
+
+class UpdateCommandMeta(
+    updateCmd: UpdateCommand,
+    conf: RapidsConf,
+    parent: Option[RapidsMeta[_, _, _]],
+    rule: DataFromReplacementRule)
+    extends RunnableCommandMeta[UpdateCommand](updateCmd, conf, parent, rule) {
+
+  override def tagSelfForGpu(): Unit = {
+    if (!conf.isDeltaWriteEnabled) {
+      willNotWorkOnGpu("Delta Lake output acceleration has been disabled. To enable set " +
+          s"${RapidsConf.ENABLE_DELTA_WRITE} to true")
+    }
+
+    val dvFeatureEnabled = DeletionVectorUtils.deletionVectorsWritable(
+      updateCmd.tahoeFileIndex.deltaLog.unsafeVolatileSnapshot)
+
+    if (dvFeatureEnabled && updateCmd.conf.getConf(
+      DeltaSQLConf.DELETE_USE_PERSISTENT_DELETION_VECTORS)) {
+      // https://github.com/NVIDIA/spark-rapids/issues/8554
+      willNotWorkOnGpu("Deletion vectors are not supported on GPU")
+    }
+
+    RapidsDeltaUtils.tagForDeltaWrite(this, updateCmd.target.schema,
+      Some(updateCmd.tahoeFileIndex.deltaLog),
+      Map.empty, updateCmd.tahoeFileIndex.spark)
+  }
+
+  override def convertToGpu(): RunnableCommand = {
+    GpuUpdateCommand(
+      new GpuDeltaLog(updateCmd.tahoeFileIndex.deltaLog, conf),
+      updateCmd.tahoeFileIndex,
+      updateCmd.catalogTable,
+      updateCmd.target,
+      updateCmd.updateExpressions,
+      updateCmd.condition
+    )
+  }
+}