Preparation for the coming Kudo support #11667
Conversation
Some code refactor for easily adding in the Kudo support in Shuffle coalesce and join execs. Signed-off-by: Firestarman <[email protected]>
Signed-off-by: Firestarman <[email protected]>
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
…alesceExec.scala Co-authored-by: Renjie Liu <[email protected]>
…alesceExec.scala Co-authored-by: Renjie Liu <[email protected]>
Signed-off-by: Firestarman <[email protected]>
|
build |
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Show resolved
Hide resolved
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Outdated
Show resolved
Hide resolved
|
|
||
| object CoalesceReadOption { | ||
| def apply(conf: RapidsConf): CoalesceReadOption = { | ||
| // TODO get the value from conf |
There was a problem hiding this comment.
It is better to have.
It will be quite easy to add in new fields that can be got from the conf. And you do not need to update all the places where the CoalesceReadOption(conf) is called.
| metricsMap: Map[String, GpuMetric], | ||
| prefetchFirstBatch: Boolean = false): Iterator[ColumnarBatch] = { | ||
| val hostIter = if (readOption.kudoEnabled) { | ||
| // TODO replace with the actual Kudo host iterator |
There was a problem hiding this comment.
Can we please throw an exception instead? If this is true we have problems and making the data disappear feels wrong to me.
There was a problem hiding this comment.
Good suggestion, done
| def getCoalescedBufferSize(concated: AnyRef): Long = concated match { | ||
| // TODO add the Kudo case | ||
| case c: HostConcatResult => c.getTableHeader.getDataLen | ||
| case g => GpuColumnVector.getTotalDeviceMemoryUsed(g.asInstanceOf[ColumnarBatch]) |
There was a problem hiding this comment.
nit: can we please have some better error messages here or something? I don't like that we have g.asInstanceOf, when the match should do that for us. It also means that the error message if something goes wrong will just be a class cast exception instead of a cleaner message
There was a problem hiding this comment.
This function is removed.
|
Thx for review, but I will drop this PR since it is not necessary for only the table operator part. @liurenjie1024 Seems it would be better to have a single PR for the Kudo support. |
Signed-off-by: Firestarman <[email protected]>
Signed-off-by: Firestarman <[email protected]>
|
build |
Signed-off-by: Firestarman <[email protected]>
|
build |
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffleCoalesceExec.scala
Show resolved
Hide resolved
|
@revans2 Could you help take a look ? Thanks. |
contributes to #11590
This is some code refactor for easily adding in the Kudo support in Shuffle coalesce and join execs. The main idea is to abstract the batch operations (
SerializedTableOperator) from the coalesce read process.