From 8601e86fb431ebcf03b1d77c2932cd847367dd33 Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Mon, 23 Dec 2024 17:42:19 -0800 Subject: [PATCH 1/4] Temporary fix for PythonRunner metrics Signed-off-by: Gera Shegalov --- .../rapids/execution/python/shims/GpuBasePythonRunner.scala | 3 ++- tools/generated_files/400/supportedExprs.csv | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/sql-plugin/src/main/spark341db/scala/org/apache/spark/sql/rapids/execution/python/shims/GpuBasePythonRunner.scala b/sql-plugin/src/main/spark341db/scala/org/apache/spark/sql/rapids/execution/python/shims/GpuBasePythonRunner.scala index 8493fe596b9..725bc16a1e9 100644 --- a/sql-plugin/src/main/spark341db/scala/org/apache/spark/sql/rapids/execution/python/shims/GpuBasePythonRunner.scala +++ b/sql-plugin/src/main/spark341db/scala/org/apache/spark/sql/rapids/execution/python/shims/GpuBasePythonRunner.scala @@ -33,4 +33,5 @@ abstract class GpuBasePythonRunner[IN]( evalType: Int, argOffsets: Array[Array[Int]], jobArtifactUUID: Option[String] -) extends BasePythonRunner[IN, ColumnarBatch](funcs, evalType, argOffsets, jobArtifactUUID) +) extends BasePythonRunner[IN, ColumnarBatch](funcs, evalType, argOffsets, jobArtifactUUID, + metrics = Map.empty) diff --git a/tools/generated_files/400/supportedExprs.csv b/tools/generated_files/400/supportedExprs.csv index d3f1f1b851a..926bb4f6c36 100644 --- a/tools/generated_files/400/supportedExprs.csv +++ b/tools/generated_files/400/supportedExprs.csv @@ -666,7 +666,7 @@ WindowExpression,S, ,None,window,result,S,S,S,S,S,S,S,S,PS,S,S,S,S,S,PS,PS,PS,S, WindowSpecDefinition,S, ,None,project,partition,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS,NS,NS WindowSpecDefinition,S, ,None,project,value,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS,NS,NS WindowSpecDefinition,S, ,None,project,result,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS,NS,NS -XxHash64,S,`xxhash64`,None,project,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,NS,NS,NS,NS +XxHash64,S,`xxhash64`,None,project,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,PS,PS,NS,NS,NS XxHash64,S,`xxhash64`,None,project,result,NA,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA Year,S,`year`,None,project,input,NA,NA,NA,NA,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA Year,S,`year`,None,project,result,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA From 138f5422dfc4c160fcbbbe344e65aece48f81930 Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Tue, 24 Dec 2024 01:42:59 -0800 Subject: [PATCH 2/4] Eager RSM init Signed-off-by: Gera Shegalov --- .../spark/sql/rapids/GpuShuffleEnv.scala | 35 +++++++------ .../spark/rapids/RapidsShuffleManager.scala | 2 - .../shims/ShuffleManagerShimUtils.scala | 49 +++++++++++++++++++ .../spark/rapids/RapidsShuffleManager.scala | 32 ++++++++++++ .../shims/ShuffleManagerShimUtils.scala | 25 ++++++++++ 5 files changed, 126 insertions(+), 17 deletions(-) create mode 100644 sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala create mode 100644 sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/RapidsShuffleManager.scala create mode 100644 sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala diff --git a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuShuffleEnv.scala b/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuShuffleEnv.scala index 7f8733b9e00..3e0e2afa9c7 100644 --- a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuShuffleEnv.scala +++ b/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuShuffleEnv.scala @@ -19,6 +19,7 @@ package org.apache.spark.sql.rapids import java.util.Locale import com.nvidia.spark.rapids._ +import com.nvidia.spark.rapids.shims.ShuffleManagerShimUtils import org.apache.spark.SparkEnv import org.apache.spark.internal.Logging @@ -105,21 +106,25 @@ object GpuShuffleEnv extends Logging { // def initShuffleManager(): Unit = { val shuffleManager = SparkEnv.get.shuffleManager - shuffleManager match { - case rapidsShuffleManager: RapidsShuffleManagerLike => - rapidsShuffleManager.initialize - case _ => - val rsmLoaderViaShuffleManager = shuffleManager.getClass.getSuperclass.getInterfaces - .collectFirst { - case c if c.getName == classOf[RapidsShuffleManagerLike].getName => c.getClassLoader - } - val rsmLoaderDirect = classOf[RapidsShuffleManagerLike].getClassLoader - - throw new IllegalStateException(s"Cannot initialize the RAPIDS Shuffle Manager " + - s"${shuffleManager}! Expected: an instance of RapidsShuffleManagerLike loaded by " + - s"${rsmLoaderDirect}. Actual: ${shuffleManager} tagged with RapidsShuffleManagerLike " + - s"loaded by: ${rsmLoaderViaShuffleManager}" - ) + if (ShuffleManagerShimUtils.eagerlyInitialized) { + // skip deferred init + } else { + shuffleManager match { + case rapidsShuffleManager: RapidsShuffleManagerLike => + rapidsShuffleManager.initialize + case _ => + val rsmLoaderViaShuffleManager = shuffleManager.getClass.getSuperclass.getInterfaces + .collectFirst { + case c if c.getName == classOf[RapidsShuffleManagerLike].getName => c.getClassLoader + } + val rsmLoaderDirect = classOf[RapidsShuffleManagerLike].getClassLoader + + throw new IllegalStateException(s"Cannot initialize the RAPIDS Shuffle Manager " + + s"${shuffleManager}! Expected: an instance of RapidsShuffleManagerLike loaded by " + + s"${rsmLoaderDirect}. Actual: ${shuffleManager} tagged with RapidsShuffleManagerLike " + + s"loaded by: ${rsmLoaderViaShuffleManager}" + ) + } } } diff --git a/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/RapidsShuffleManager.scala b/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/RapidsShuffleManager.scala index 155eb35a516..e0ca739463f 100644 --- a/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/RapidsShuffleManager.scala +++ b/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/RapidsShuffleManager.scala @@ -37,11 +37,9 @@ {"spark": "343"} {"spark": "344"} {"spark": "350"} -{"spark": "350db143"} {"spark": "351"} {"spark": "352"} {"spark": "353"} -{"spark": "400"} spark-rapids-shim-json-lines ***/ package com.nvidia.spark.rapids.$_spark.version.classifier_ diff --git a/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala b/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala new file mode 100644 index 00000000000..4c04377ff4c --- /dev/null +++ b/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala @@ -0,0 +1,49 @@ +/* + * Copyright (c) 2024, 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. + */ + + +/*** spark-rapids-shim-json-lines +{"spark": "320"} +{"spark": "321"} +{"spark": "321cdh"} +{"spark": "322"} +{"spark": "323"} +{"spark": "324"} +{"spark": "330"} +{"spark": "330cdh"} +{"spark": "330db"} +{"spark": "331"} +{"spark": "332"} +{"spark": "332cdh"} +{"spark": "332db"} +{"spark": "333"} +{"spark": "334"} +{"spark": "340"} +{"spark": "341"} +{"spark": "341db"} +{"spark": "342"} +{"spark": "343"} +{"spark": "344"} +{"spark": "350"} +{"spark": "351"} +{"spark": "352"} +{"spark": "353"} +spark-rapids-shim-json-lines ***/ +package com.nvidia.spark.rapids.shims + +object ShuffleManagerShimUtils { + def eagerlyInitialized = false +} diff --git a/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/RapidsShuffleManager.scala b/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/RapidsShuffleManager.scala new file mode 100644 index 00000000000..ca4a9a1dc4b --- /dev/null +++ b/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/RapidsShuffleManager.scala @@ -0,0 +1,32 @@ +/* + * Copyright (c) 2024, 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. + */ + +/*** spark-rapids-shim-json-lines +{"spark": "350db143"} +{"spark": "400"} +spark-rapids-shim-json-lines ***/ +package com.nvidia.spark.rapids.$_spark.version.classifier_ + +import org.apache.spark.SparkConf +import org.apache.spark.sql.rapids.ProxyRapidsShuffleInternalManagerBase + +/** A shuffle manager optimized for the RAPIDS Plugin for Apache Spark. */ +sealed class RapidsShuffleManager( + conf: SparkConf, + isDriver: Boolean +) extends ProxyRapidsShuffleInternalManagerBase(conf, isDriver) { + initialize +} diff --git a/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala b/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala new file mode 100644 index 00000000000..4bab458acad --- /dev/null +++ b/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala @@ -0,0 +1,25 @@ +/* + * Copyright (c) 2024, 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. + */ + +/*** spark-rapids-shim-json-lines +{"spark": "350db143"} +{"spark": "400"} +spark-rapids-shim-json-lines ***/ +package com.nvidia.spark.rapids.shims + +object ShuffleManagerShimUtils { + def eagerlyInitialized = false +} From 68c79b43a7a7bb8a498f4727a7d429fb51d19ed7 Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Tue, 24 Dec 2024 02:29:30 -0800 Subject: [PATCH 3/4] Revert "Temporary fix for PythonRunner metrics" This reverts commit 8601e86fb431ebcf03b1d77c2932cd847367dd33. --- .../rapids/execution/python/shims/GpuBasePythonRunner.scala | 3 +-- tools/generated_files/400/supportedExprs.csv | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/sql-plugin/src/main/spark341db/scala/org/apache/spark/sql/rapids/execution/python/shims/GpuBasePythonRunner.scala b/sql-plugin/src/main/spark341db/scala/org/apache/spark/sql/rapids/execution/python/shims/GpuBasePythonRunner.scala index 725bc16a1e9..8493fe596b9 100644 --- a/sql-plugin/src/main/spark341db/scala/org/apache/spark/sql/rapids/execution/python/shims/GpuBasePythonRunner.scala +++ b/sql-plugin/src/main/spark341db/scala/org/apache/spark/sql/rapids/execution/python/shims/GpuBasePythonRunner.scala @@ -33,5 +33,4 @@ abstract class GpuBasePythonRunner[IN]( evalType: Int, argOffsets: Array[Array[Int]], jobArtifactUUID: Option[String] -) extends BasePythonRunner[IN, ColumnarBatch](funcs, evalType, argOffsets, jobArtifactUUID, - metrics = Map.empty) +) extends BasePythonRunner[IN, ColumnarBatch](funcs, evalType, argOffsets, jobArtifactUUID) diff --git a/tools/generated_files/400/supportedExprs.csv b/tools/generated_files/400/supportedExprs.csv index 926bb4f6c36..d3f1f1b851a 100644 --- a/tools/generated_files/400/supportedExprs.csv +++ b/tools/generated_files/400/supportedExprs.csv @@ -666,7 +666,7 @@ WindowExpression,S, ,None,window,result,S,S,S,S,S,S,S,S,PS,S,S,S,S,S,PS,PS,PS,S, WindowSpecDefinition,S, ,None,project,partition,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS,NS,NS WindowSpecDefinition,S, ,None,project,value,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS,NS,NS WindowSpecDefinition,S, ,None,project,result,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS,NS,NS -XxHash64,S,`xxhash64`,None,project,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,PS,PS,NS,NS,NS +XxHash64,S,`xxhash64`,None,project,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,NS,NS,NS,NS XxHash64,S,`xxhash64`,None,project,result,NA,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA Year,S,`year`,None,project,input,NA,NA,NA,NA,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA Year,S,`year`,None,project,result,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA From 33c064aaec791ec7fec15e6ec8f67a8e110186a4 Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Tue, 24 Dec 2024 02:30:24 -0800 Subject: [PATCH 4/4] Flip eager init for 14.3 --- .../com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala b/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala index 4bab458acad..1ceb7902042 100644 --- a/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala +++ b/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/ShuffleManagerShimUtils.scala @@ -21,5 +21,5 @@ spark-rapids-shim-json-lines ***/ package com.nvidia.spark.rapids.shims object ShuffleManagerShimUtils { - def eagerlyInitialized = false + def eagerlyInitialized = true }