HyukjinKwon commented on a change in pull request #23787: [SPARK-26830][SQL][R]
Vectorized R dapply() implementation
URL: https://github.com/apache/spark/pull/23787#discussion_r257096137
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/objects.scala
##########
@@ -193,6 +194,87 @@ case class MapPartitionsExec(
}
}
+/**
+ * Similar with [[MapPartitionsExec]] and
+ * [[org.apache.spark.sql.execution.r.MapPartitionsRWrapper]] but serializes
and deserializes
+ * input/output in Arrow format.
+ *
+ * This is somewhat similar with
[[org.apache.spark.sql.execution.python.ArrowEvalPythonExec]]
+ */
+case class MapPartitionsInRWithArrowExec(
+ func: Array[Byte],
+ packageNames: Array[Byte],
+ broadcastVars: Array[Broadcast[Object]],
+ inputSchema: StructType,
+ output: Seq[Attribute],
+ child: SparkPlan) extends UnaryExecNode {
+
+ private val batchSize = conf.arrowMaxRecordsPerBatch
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ child.execute().mapPartitionsInternal { iter =>
+ // Hyukjin: seems to convert from Arrow batches to Arrow batches
directly here
+ // causes segmentation fault intermittently. For instance, when child is
directly
+ // originated from Arrow batches. Input projection was added to avoid it
based on
+ // my speculation (and it does not happen anymore). It is potentially an
issue in Arrow
+ // optimization in my humble opinion.
Review comment:
I don't have an explicit evidence for this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
