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

Jump to bottom

Add TypeSig checks for join keys and other special cases #3183

Merged
merged 9 commits into from
Aug 18, 2021
Merged

Add TypeSig checks for join keys and other special cases #3183

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
1429330
Add TypeSig checks for join keys
andygrove Aug 10, 2021
da3019c
Remove use of reflection
andygrove Aug 10, 2021
babe6c4
Use rowspan
andygrove Aug 10, 2021
92bd4b6
fix rowSpan issue
andygrove Aug 10, 2021
2b46a96
Add check for named expressions without associated ExecChecks
andygrove Aug 11, 2021
76afb74
add checks for Spark311dbShims joins
andygrove Aug 11, 2021
64fbc33
Implement extended checks for WindowExec
andygrove Aug 11, 2021
f3613c5
Support struct in join keys
andygrove Aug 17, 2021
ca48030
Check in updated generated docs
andygrove Aug 17, 2021
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
353 changes: 271 additions & 82 deletions docs/supported_ops.md
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 3 additions & 0 deletions ...k301/src/main/scala/com/nvidia/spark/rapids/shims/spark301/GpuBroadcastHashJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ class GpuBroadcastHashJoinMeta(
val condition: Option[BaseExprMeta[_]] =
join.condition.map(GpuOverrides.wrapExpr(_, conf, Some(this)))

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override def tagPlanForGpu(): Unit = {
Expand Down
3 changes: 3 additions & 0 deletions ...rk301/src/main/scala/com/nvidia/spark/rapids/shims/spark301/GpuShuffledHashJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,9 @@ class GpuShuffledHashJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
}
Expand Down
3 changes: 3 additions & 0 deletions ...spark301/src/main/scala/com/nvidia/spark/rapids/shims/spark301/GpuSortMergeJoinMeta.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,9 @@ class GpuSortMergeJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
// Use conditions from Hash Join
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
Expand Down
28 changes: 14 additions & 14 deletions shims/spark301/src/main/scala/com/nvidia/spark/rapids/shims/spark301/SparkBaseShims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,31 +232,31 @@ abstract class SparkBaseShims extends SparkShims {
"Sort merge join, replacing with shuffled hash join",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
Comment on lines -235 to -237
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just so I am not misunderstanding, are these unrelated changes and more like and audit that you ran into while making the doc change?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Prior to this PR we relied on the "ps notes" to document the fact that we don't support map/struct/array as join keys, but we now have specific lines in the supported_ops table for leftKeys and rightKeys for the join operators, showing which types are supported, so we no longer need these ps notes.

.nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuSortMergeJoinMeta(join, conf, p, r)),
GpuOverrides.exec[BroadcastHashJoinExec](
"Implementation of join using broadcast data",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 +
TypeSig.ARRAY + TypeSig.STRUCT + TypeSig.MAP)
.nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuBroadcastHashJoinMeta(join, conf, p, r)),
GpuOverrides.exec[ShuffledHashJoinExec](
"Implementation of join using hashed shuffled data",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuShuffledHashJoinMeta(join, conf, p, r)),
GpuOverrides.exec[ArrowEvalPythonExec](
"The backend of the Scalar Pandas UDFs. Accelerates the data transfer between the" +
Expand Down
4 changes: 4 additions & 0 deletions ...db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuBroadcastHashJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,10 @@ class GpuBroadcastHashJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)


override def tagPlanForGpu(): Unit = {
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
val Seq(leftChild, rightChild) = childPlans
Expand Down
3 changes: 3 additions & 0 deletions ...1db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuShuffledHashJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@ class GpuShuffledHashJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
}
Expand Down
3 changes: 3 additions & 0 deletions ...k301db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuSortMergeJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,9 @@ class GpuSortMergeJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
// Use conditions from Hash Join
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
Expand Down
25 changes: 12 additions & 13 deletions .../spark301db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/Spark301dbShims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,32 +183,31 @@ class Spark301dbShims extends Spark301Shims {
"Sort merge join, replacing with shuffled hash join",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuSortMergeJoinMeta(join, conf, p, r)),
GpuOverrides.exec[BroadcastHashJoinExec](
"Implementation of join using broadcast data",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT +
TypeSig.DECIMAL_64)
, TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuBroadcastHashJoinMeta(join, conf, p, r)),
GpuOverrides.exec[ShuffledHashJoinExec](
"Implementation of join using hashed shuffled data",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuShuffledHashJoinMeta(join, conf, p, r)),
GpuOverrides.exec[ArrowEvalPythonExec](
"The backend of the Scalar Pandas UDFs. Accelerates the data transfer between the" +
Expand Down
3 changes: 3 additions & 0 deletions ...k311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuBroadcastHashJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,9 @@ class GpuBroadcastHashJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
val Seq(leftChild, rightChild) = childPlans
Expand Down
3 changes: 3 additions & 0 deletions ...rk311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuShuffledHashJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@ class GpuShuffledHashJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
}
Expand Down
3 changes: 3 additions & 0 deletions ...spark311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuSortMergeJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,9 @@ class GpuSortMergeJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
// Use conditions from Hash Join
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
Expand Down
25 changes: 12 additions & 13 deletions shims/spark311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/Spark311Shims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -393,32 +393,31 @@ class Spark311Shims extends Spark301Shims {
"Sort merge join, replacing with shuffled hash join",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuSortMergeJoinMeta(join, conf, p, r)),
GpuOverrides.exec[BroadcastHashJoinExec](
"Implementation of join using broadcast data",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT +
TypeSig.DECIMAL_64)
, TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuBroadcastHashJoinMeta(join, conf, p, r)),
GpuOverrides.exec[ShuffledHashJoinExec](
"Implementation of join using hashed shuffled data",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuShuffledHashJoinMeta(join, conf, p, r))
).map(r => (r.getClassFor.asSubclass(classOf[SparkPlan]), r))
}
Expand Down
3 changes: 3 additions & 0 deletions ...db/src/main/scala/com/nvidia/spark/rapids/shims/spark311db/GpuBroadcastHashJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,9 @@ class GpuBroadcastHashJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
val Seq(leftChild, rightChild) = childPlans
Expand Down
3 changes: 3 additions & 0 deletions ...1db/src/main/scala/com/nvidia/spark/rapids/shims/spark311db/GpuShuffledHashJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@ class GpuShuffledHashJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
}
Expand Down
3 changes: 3 additions & 0 deletions ...k311db/src/main/scala/com/nvidia/spark/rapids/shims/spark311db/GpuSortMergeJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,9 @@ class GpuSortMergeJoinMeta(

override val childExprs: Seq[BaseExprMeta[_]] = leftKeys ++ rightKeys ++ condition

override val namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] =
Map("leftKeys" -> leftKeys, "rightKeys" -> rightKeys)

override def tagPlanForGpu(): Unit = {
// Use conditions from Hash Join
GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
Expand Down
27 changes: 13 additions & 14 deletions .../spark311db/src/main/scala/com/nvidia/spark/rapids/shims/spark311db/Spark311dbShims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,33 +170,32 @@ class Spark311dbShims extends Spark311Shims {
GpuOverrides.exec[SortMergeJoinExec](
"Sort merge join, replacing with shuffled hash join",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
TypeSig.STRUCT + TypeSig.MAP),
.nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuSortMergeJoinMeta(join, conf, p, r)),
GpuOverrides.exec[BroadcastHashJoinExec](
"Implementation of join using broadcast data",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT +
TypeSig.DECIMAL_64)
, TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuBroadcastHashJoinMeta(join, conf, p, r)),
GpuOverrides.exec[ShuffledHashJoinExec](
"Implementation of join using hashed shuffled data",
ExecChecks((TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 + TypeSig.ARRAY +
TypeSig.STRUCT + TypeSig.MAP)
.withPsNote(TypeEnum.ARRAY, "Cannot be used as join key")
.withPsNote(TypeEnum.STRUCT, "Cannot be used as join key")
.withPsNote(TypeEnum.MAP, "Cannot be used as join key")
.nested(TypeSig.commonCudfTypes + TypeSig.NULL +
TypeSig.DECIMAL_64), TypeSig.all),
TypeSig.DECIMAL_64),
Map("leftKeys" -> TypeSig.joinKeyTypes,
"rightKeys" -> TypeSig.joinKeyTypes),
TypeSig.all),
(join, conf, p, r) => new GpuShuffledHashJoinMeta(join, conf, p, r)),
GpuOverrides.exec[ArrowEvalPythonExec](
"The backend of the Scalar Pandas UDFs. Accelerates the data transfer between the" +
Expand Down
6 changes: 2 additions & 4 deletions sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3127,10 +3127,8 @@ object GpuOverrides {
"Window-operator backend",
ExecChecks(
(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64 +
TypeSig.STRUCT + TypeSig.ARRAY + TypeSig.MAP).nested() +
TypeSig.psNote(TypeEnum.MAP, "Not supported as a partition by key") +
TypeSig.psNote(TypeEnum.STRUCT, "Not supported as a partition by key") +
TypeSig.psNote(TypeEnum.ARRAY, "Not supported as a partition by key"),
TypeSig.STRUCT + TypeSig.ARRAY + TypeSig.MAP).nested(),
Map("partitionSpec" -> (TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_64)),
TypeSig.all),
(windowOp, conf, p, r) =>
new GpuWindowExecMeta(windowOp, conf, p, r)
Expand Down
15 changes: 5 additions & 10 deletions sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuWindowExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,23 +86,18 @@ abstract class GpuBaseWindowExecMeta[WindowExecType <: SparkPlan] (windowExec: W
lazy val inputFields: Seq[BaseExprMeta[Attribute]] =
windowExec.children.head.output.map(GpuOverrides.wrapExpr(_, conf, Some(this)))


override def namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] = Map(
"partitionSpec" -> partitionSpec
)

override def tagPlanForGpu(): Unit = {
// Implementation depends on receiving a `NamedExpression` wrapped WindowExpression.
windowExpressions.map(meta => meta.wrapped)
.filter(expr => !expr.isInstanceOf[NamedExpression])
.foreach(_ => willNotWorkOnGpu("Unexpected query plan with Windowing functions; " +
"cannot convert for GPU execution. " +
"(Detail: WindowExpression not wrapped in `NamedExpression`.)"))
val unsupportedKeys = partitionSpec.map(_.wrapped.dataType).exists {
case _: StructType => true
case _: ArrayType => true
case _: MapType => true
case _ => false
}

if (unsupportedKeys) {
willNotWorkOnGpu(s"nested partition by keys are not supported")
}
}

override def convertToGpu(): GpuExec = {
Expand Down
2 changes: 2 additions & 0 deletions sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsMeta.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -562,6 +562,8 @@ abstract class SparkPlanMeta[INPUT <: SparkPlan](plan: INPUT,
override val childParts: Seq[PartMeta[_]] = Seq.empty
override val childDataWriteCmds: Seq[DataWritingCommandMeta[_]] = Seq.empty

def namedChildExprs: Map[String, Seq[BaseExprMeta[_]]] = Map.empty

var cpuCost: Double = 0
var gpuCost: Double = 0
var estimatedOutputRows: Option[BigInt] = None
Expand Down
Loading