Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Unshim many SparkShim interfaces [databricks] #5031

Merged
merged 17 commits into from
Mar 24, 2022
Merged

Unshim many SparkShim interfaces [databricks] #5031

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
26a953b
Unshim Arrow
jlowe Mar 18, 2022
e509622
Unshim Alluxio
jlowe Mar 18, 2022
818f9d0
Unshim Kryo
jlowe Mar 18, 2022
77bf1c6
Unshim getGpuColumnarToRowTransition
jlowe Mar 18, 2022
1ce0790
Unshim createTable
jlowe Mar 18, 2022
fbbb7bf
Unshim hasSeparateINT96RebaseConf
jlowe Mar 18, 2022
e318d81
Unshim getScalaUDFAsExpression
jlowe Mar 21, 2022
168606a
Unshim getMapSizesByExecutorId
jlowe Mar 21, 2022
f734a3d
Unshim getFileSourceMaxMetadataValueLength
jlowe Mar 21, 2022
e52dcf3
Unshim getGpuShuffleExchangeExec
jlowe Mar 21, 2022
e9332c0
Unshim SortOrder
jlowe Mar 21, 2022
3e98201
Remove HadoopFSUtils shim
jlowe Mar 21, 2022
e229f6a
Unshim getLegacyComplexTypeToString
jlowe Mar 21, 2022
6eea393
Unshim shouldFailDivByZero
jlowe Mar 21, 2022
ed2267f
Unshim shouldFailOnElementNotExists
jlowe Mar 21, 2022
d335573
Remove registerKryo from Spark31XdbShims
jlowe Mar 23, 2022
9c92f4f
Unshim shouldFallbackOnAnsiTimestamp
jlowe Mar 23, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 0 additions & 25 deletions sql-plugin/src/main/311+-all/scala/org/apache/spark/sql/rapids/shims/HadoopFSUtilsShim.scala
View file
Open in desktop

This file was deleted.

4 changes: 2 additions & 2 deletions ...ala/org/apache/spark/sql/rapids/execution/python/shims/GpuFlatMapGroupsInPandasExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ package org.apache.spark.sql.rapids.execution.python.shims

import com.nvidia.spark.rapids._
import com.nvidia.spark.rapids.python.PythonWorkerSemaphore
import com.nvidia.spark.rapids.shims.{ShimUnaryExecNode, SparkShimImpl}
import com.nvidia.spark.rapids.shims.ShimUnaryExecNode

import org.apache.spark.TaskContext
import org.apache.spark.api.python.{ChainedPythonFunctions, PythonEvalType}
Expand Down Expand Up @@ -96,7 +96,7 @@ case class GpuFlatMapGroupsInPandasExec(
}

override def requiredChildOrdering: Seq[Seq[SortOrder]] =
Seq(groupingAttributes.map(SparkShimImpl.sortOrder(_, Ascending)))
Seq(groupingAttributes.map(SortOrder(_, Ascending)))

private val pandasFunction = func.asInstanceOf[GpuPythonUDF].func

Expand Down
9 changes: 3 additions & 6 deletions ...n/src/main/311until320-all/scala/org/apache/spark/rapids/shims/ShuffledBatchRDDUtil.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,6 @@

package org.apache.spark.rapids.shims

import com.nvidia.spark.rapids.shims.SparkShimImpl

import org.apache.spark.{MapOutputTrackerMaster, Partition, ShuffleDependency, SparkEnv, TaskContext}
import org.apache.spark.shuffle.ShuffleReader
import org.apache.spark.sql.execution.{CoalescedPartitionSpec, PartialMapperPartitionSpec, PartialReducerPartitionSpec}
Expand Down Expand Up @@ -57,7 +55,6 @@ object ShuffledBatchRDDUtil {
dependency: ShuffleDependency[Int, ColumnarBatch, ColumnarBatch],
sqlMetricsReporter: SQLShuffleReadMetricsReporter):
(ShuffleReader[Nothing, Nothing], Long) = {
val shim = SparkShimImpl
split.asInstanceOf[ShuffledBatchRDDPartition].spec match {
case CoalescedPartitionSpec(startReducerIndex, endReducerIndex) =>
val reader = SparkEnv.get.shuffleManager.getReader(
Expand All @@ -66,7 +63,7 @@ object ShuffledBatchRDDUtil {
endReducerIndex,
context,
sqlMetricsReporter)
val blocksByAddress = shim.getMapSizesByExecutorId(
val blocksByAddress = SparkEnv.get.mapOutputTracker.getMapSizesByExecutorId(
dependency.shuffleHandle.shuffleId, 0, Int.MaxValue, startReducerIndex, endReducerIndex)
val partitionSize = blocksByAddress.flatMap(_._2).map(_._2).sum
(reader, partitionSize)
Expand All @@ -80,7 +77,7 @@ object ShuffledBatchRDDUtil {
reducerIndex + 1,
context,
sqlMetricsReporter)
val blocksByAddress = shim.getMapSizesByExecutorId(
val blocksByAddress = SparkEnv.get.mapOutputTracker.getMapSizesByExecutorId(
dependency.shuffleHandle.shuffleId, 0, Int.MaxValue, reducerIndex,
reducerIndex + 1)
val partitionSize = blocksByAddress.flatMap(_._2)
Expand All @@ -96,7 +93,7 @@ object ShuffledBatchRDDUtil {
endReducerIndex,
context,
sqlMetricsReporter)
val blocksByAddress = shim.getMapSizesByExecutorId(
val blocksByAddress = SparkEnv.get.mapOutputTracker.getMapSizesByExecutorId(
dependency.shuffleHandle.shuffleId, 0, Int.MaxValue, startReducerIndex, endReducerIndex)
val partitionSize = blocksByAddress.flatMap(_._2)
.filter(_._3 == mapIndex)
Expand Down
Loading