commetns

Author: richardc-db

sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala

@@ -333,21 +333,24 @@ case class OneRowRelationExec() extends LeafExecNode
 
   override val output: Seq[Attribute] = Nil
 
-  private val rdd: RDD[InternalRow] = session.sparkContext.parallelize(Seq(InternalRow.empty), 1)
+  private val rdd: RDD[InternalRow] = {
+    val numOutputRows = longMetric("numOutputRows")
+    session
+      .sparkContext
+      .parallelize(Seq(InternalRow()), 1)
+      .mapPartitionsInternal { _ =>
+        val proj = UnsafeProjection.create(Seq.empty[Expression])
+        Iterator(proj.apply(InternalRow.empty)).map { r =>
+          numOutputRows += 1
+          r
+        }
+      }
+  }
 
   override lazy val metrics = Map(
     "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"))
 
-  protected override def doExecute(): RDD[InternalRow] = {
-    val numOutputRows = longMetric("numOutputRows")
-    rdd.mapPartitionsWithIndexInternal { (_, iter) =>
-      val proj = UnsafeProjection.create(schema)
-      iter.map { r =>
-        numOutputRows += 1
-        proj(r)
-      }
-    }
-  }
+  protected override def doExecute(): RDD[InternalRow] = rdd
 
   override def simpleString(maxFields: Int): String = s"$nodeName[]"