Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/23127#discussion_r236333530
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala
---
@@ -406,14 +415,62 @@ trait BlockingOperatorWithCodegen extends
CodegenSupport {
override def limitNotReachedChecks: Seq[String] = Nil
}
+/**
+ * Leaf codegen node reading from a single RDD.
+ */
+trait InputRDDCodegen extends CodegenSupport {
+
+ def inputRDD: RDD[InternalRow]
+
+ // If the input is an RDD of InternalRow which are potentially not
UnsafeRow,
+ // and there is no parent to consume it, it needs an UnsafeProjection.
+ protected val createUnsafeProjection: Boolean = (parent == null)
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ inputRDD :: Nil
+ }
+
+ override def doProduce(ctx: CodegenContext): String = {
+ // Inline mutable state since an InputRDDCodegen is used once in a
task for WholeStageCodegen
+ val input = ctx.addMutableState("scala.collection.Iterator", "input",
v => s"$v = inputs[0];",
+ forceInline = true)
+ val row = ctx.freshName("row")
+
+ val outputVars = if (createUnsafeProjection) {
+ // creating the vars will make the parent consume add an unsafe
projection.
+ ctx.INPUT_ROW = row
+ ctx.currentVars = null
+ output.zipWithIndex.map { case (a, i) =>
+ BoundReference(i, a.dataType, a.nullable).genCode(ctx)
+ }
+ } else {
+ null
+ }
+
+ val numOutputRowsCode = if (metrics.contains("numOutputRows")) {
--- End diff --
how about `updateNumOutputRowsMetrics` instead of `numOutputRowsCode`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
