Databricks 12.2 Support [databricks]

aggregator/pom.xml

@@ -505,6 +505,23 @@
                 </dependency>
             </dependencies>
         </profile>
+        <profile>
+            <id>release332db</id>
+            <activation>
+                <property>
+                    <name>buildver</name>
+                    <value>332db</value>
+                </property>
+            </activation>
+            <dependencies>
+                <dependency>
+                    <groupId>com.nvidia</groupId>
+                    <artifactId>rapids-4-spark-delta-spark332db_${scala.binary.version}</artifactId>
+                    <version>${project.version}</version>
+                    <classifier>${spark.version.classifier}</classifier>
+                </dependency>
+            </dependencies>
+        </profile>
         <profile>
             <id>release333</id>
             <activation>

delta-lake/README.md

@@ -16,6 +16,7 @@ and directory contains the corresponding support code.
 | 2.2.x              | Spark 3.3.x     | `delta-22x`        |
 | Databricks 10.4    | Databricks 10.4 | `delta-spark321db` |
 | Databricks 11.3    | Databricks 11.3 | `delta-spark330db` |
+| Databricks 12.2    | Databricks 12.2 | `delta-spark332db` |
 
 Delta Lake is not supported on all Spark versions, and for Spark versions where it is not
 supported the `delta-stub` project is used.

delta-lake/common/src/main/databricks/scala/com/nvidia/spark/rapids/delta/MergeIntoCommandMeta.scala

@@ -19,6 +19,7 @@ package com.nvidia.spark.rapids.delta
 import com.databricks.sql.transaction.tahoe.commands.{MergeIntoCommand, MergeIntoCommandEdge}
 import com.databricks.sql.transaction.tahoe.rapids.{GpuDeltaLog, GpuMergeIntoCommand}
 import com.nvidia.spark.rapids.{DataFromReplacementRule, RapidsConf, RapidsMeta, RunnableCommandMeta}
+import com.nvidia.spark.rapids.delta.shims.MergeIntoCommandMetaShim
 
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.execution.command.RunnableCommand
@@ -35,6 +36,7 @@ class MergeIntoCommandMeta(
       willNotWorkOnGpu("Delta Lake output acceleration has been disabled. To enable set " +
           s"${RapidsConf.ENABLE_DELTA_WRITE} to true")
     }
+    MergeIntoCommandMetaShim.tagForGpu(this, mergeCmd)
     val targetSchema = mergeCmd.migratedSchema.getOrElse(mergeCmd.target.schema)
     val deltaLog = mergeCmd.targetFileIndex.deltaLog
     RapidsDeltaUtils.tagForDeltaWrite(this, targetSchema, deltaLog, Map.empty, SparkSession.active)
@@ -64,6 +66,7 @@ class MergeIntoCommandEdgeMeta(
       willNotWorkOnGpu("Delta Lake output acceleration has been disabled. To enable set " +
           s"${RapidsConf.ENABLE_DELTA_WRITE} to true")
     }
+    MergeIntoCommandMetaShim.tagForGpu(this, mergeCmd)
     val targetSchema = mergeCmd.migratedSchema.getOrElse(mergeCmd.target.schema)
     val deltaLog = mergeCmd.targetFileIndex.deltaLog
     RapidsDeltaUtils.tagForDeltaWrite(this, targetSchema, deltaLog, Map.empty, SparkSession.active)

delta-lake/common/src/main/databricks/scala/com/nvidia/spark/rapids/delta/RapidsDeltaUtils.scala

@@ -18,6 +18,7 @@ package com.nvidia.spark.rapids.delta
 
 import com.databricks.sql.transaction.tahoe.{DeltaConfigs, DeltaLog, DeltaOptions, DeltaParquetFileFormat}
 import com.nvidia.spark.rapids.{DeltaFormatType, FileFormatChecks, GpuOverrides, GpuParquetFileFormat, RapidsMeta, TypeSig, WriteFileOp}
+import com.nvidia.spark.rapids.delta.shims.DeltaLogShim
 
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.execution.datasources.DataSourceUtils
@@ -32,7 +33,7 @@ object RapidsDeltaUtils {
       options: Map[String, String],
       spark: SparkSession): Unit = {
     FileFormatChecks.tag(meta, schema, DeltaFormatType, WriteFileOp)
-    deltaLog.fileFormat() match {
+    DeltaLogShim.fileFormat(deltaLog) match {
       case _: DeltaParquetFileFormat =>
         GpuParquetFileFormat.tagGpuSupport(meta, spark, options, schema)
       case f =>
@@ -65,7 +66,7 @@ object RapidsDeltaUtils {
         orderableTypeSig.isSupportedByPlugin(t)
       }
       if (unorderableTypes.nonEmpty) {
-        val metadata = deltaLog.snapshot.metadata
+        val metadata = DeltaLogShim.getMetadata(deltaLog)
         val hasPartitioning = metadata.partitionColumns.nonEmpty ||
             options.get(DataSourceUtils.PARTITIONING_COLUMNS_KEY).exists(_.nonEmpty)
         if (!hasPartitioning) {

delta-lake/delta-spark321db/src/main/scala/com/nvidia/spark/rapids/delta/shims/DeltaLogShim.scala

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2023, 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.shims
+
+import com.databricks.sql.transaction.tahoe.DeltaLog
+import com.databricks.sql.transaction.tahoe.actions.Metadata
+
+import org.apache.spark.sql.execution.datasources.FileFormat
+
+object DeltaLogShim {
+  def fileFormat(deltaLog: DeltaLog): FileFormat = {
+    deltaLog.fileFormat()
+  }
+
+  def getMetadata(deltaLog: DeltaLog): Metadata = {
+    deltaLog.snapshot.metadata
+  }
+}

delta-lake/delta-spark321db/src/main/scala/com/nvidia/spark/rapids/delta/shims/MergeIntoCommandMetaShim.scala

@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2023, 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.shims
+
+import com.databricks.sql.transaction.tahoe.commands.{MergeIntoCommand, MergeIntoCommandEdge}
+import com.nvidia.spark.rapids.delta.{MergeIntoCommandEdgeMeta, MergeIntoCommandMeta}
+
+object MergeIntoCommandMetaShim {
+  def tagForGpu(meta: MergeIntoCommandMeta, mergeCmd: MergeIntoCommand): Unit = {}
+  def tagForGpu(meta: MergeIntoCommandEdgeMeta, mergeCmd: MergeIntoCommandEdge): Unit = {}
+}