Github user kiszk commented on a diff in the pull request:
https://github.com/apache/spark/pull/17302#discussion_r107090057
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala ---
@@ -70,7 +70,20 @@ object RDDConversions {
object ExternalRDD {
def apply[T: Encoder](rdd: RDD[T], session: SparkSession): LogicalPlan =
{
- val externalRdd = ExternalRDD(CatalystSerde.generateObjAttr[T],
rdd)(session)
+ val attr = {
+ val attr = CatalystSerde.generateObjAttr[T]
--- End diff --
When we called `deserializer.nullable` at `CatalystSerde.generateObjAttr`
with the following change in the calling context of `ExternalRDD.apply()`, the
following error occurs. I think that `ExternalRDD.apply()` calls
`CatalystSerde.generateObjAttr` at very early phase where a plan has not been
resolved yet.
```java
object CatalystSerde {
def deserialize[T : Encoder](child: LogicalPlan): DeserializeToObject = {
val deserializer = UnresolvedDeserializer(encoderFor[T].deserializer)
DeserializeToObject(deserializer, generateObjAttr[T], child)
}
def serialize[T : Encoder](child: LogicalPlan): SerializeFromObject = {
SerializeFromObject(encoderFor[T].namedExpressions, child)
}
def generateObjAttr[T : Encoder]: Attribute = {
val deserializer = encoderFor[T].deserializer
AttributeReference("obj", deserializer.dataType,
deserializer.nullable)()
}
}
```
```java
Invalid call to nullable on unresolved object, tree: getcolumnbyordinal(0,
IntegerType)
org.apache.spark.sql.catalyst.analysis.UnresolvedException: Invalid call to
nullable on unresolved object, tree: getcolumnbyordinal(0, IntegerType)
at
org.apache.spark.sql.catalyst.analysis.GetColumnByOrdinal.nullable(unresolved.scala:399)
at
org.apache.spark.sql.catalyst.expressions.UnaryExpression.nullable(Expression.scala:314)
at
org.apache.spark.sql.catalyst.expressions.objects.InvokeLike$$anonfun$needNullCheck$1.apply(objects.scala:44)
at
org.apache.spark.sql.catalyst.expressions.objects.InvokeLike$$anonfun$needNullCheck$1.apply(objects.scala:44)
at
scala.collection.LinearSeqOptimized$class.exists(LinearSeqOptimized.scala:93)
at scala.collection.immutable.List.exists(List.scala:84)
at
org.apache.spark.sql.catalyst.expressions.objects.InvokeLike$class.needNullCheck(objects.scala:44)
at
org.apache.spark.sql.catalyst.expressions.objects.NewInstance.needNullCheck$lzycompute(objects.scala:290)
at
org.apache.spark.sql.catalyst.expressions.objects.NewInstance.needNullCheck(objects.scala:290)
at
org.apache.spark.sql.catalyst.expressions.objects.NewInstance.nullable(objects.scala:298)
at
org.apache.spark.sql.catalyst.plans.logical.CatalystSerde$.generateObjAttr(object.scala:45)
at
org.apache.spark.sql.execution.ExternalRDD$.apply(ExistingRDD.scala:76)
at
org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:471)
at org.apache.spark.sql.SQLContext.createDataset(SQLContext.scala:393)
at
org.apache.spark.sql.SQLImplicits.rddToDatasetHolder(SQLImplicits.scala:238)
at
org.apache.spark.sql.DataFrameImplicitsSuite$$anonfun$6.apply$mcV$sp(DataFrameImplicitsSuite.scala:56)
at
org.apache.spark.sql.DataFrameImplicitsSuite$$anonfun$6.apply(DataFrameImplicitsSuite.scala:55)
at
org.apache.spark.sql.DataFrameImplicitsSuite$$anonfun$6.apply(DataFrameImplicitsSuite.scala:55)
at
org.scalatest.Transformer$$anonfun$apply$1.apply$mcV$sp(Transformer.scala:22)
at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
at org.scalatest.Transformer.apply(Transformer.scala:22)
at org.scalatest.Transformer.apply(Transformer.scala:20)
at org.scalatest.FunSuiteLike$$anon$1.apply(FunSuiteLike.scala:166)
at org.apache.spark.SparkFunSuite.withFixture(SparkFunSuite.scala:68)
at
org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:163)
at
org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
at
org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:175)
at
org.apache.spark.sql.DataFrameImplicitsSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(DataFrameImplicitsSuite.scala:22)
at
org.scalatest.BeforeAndAfterEach$class.runTest(BeforeAndAfterEach.scala:255)
at
org.apache.spark.sql.DataFrameImplicitsSuite.runTest(DataFrameImplicitsSuite.scala:22)
...
```
Here is an example program with plans.
```java
val dfInt = sparkContext.parallelize(Seq[java.lang.Integer](0, null, 2),
1).toDF
dfInt.explain(true)
assert(dfInt.collect === Array(Row(0), Row(null), Row(2)))
```
```
== Parsed Logical Plan ==
SerializeFromObject [input[0, java.lang.Integer, true].intValue AS value#2]
+- ExternalRDD [obj#1]
== Analyzed Logical Plan ==
value: int
SerializeFromObject [input[0, java.lang.Integer, true].intValue AS value#2]
+- ExternalRDD [obj#1]
== Optimized Logical Plan ==
SerializeFromObject [input[0, java.lang.Integer, true].intValue AS value#2]
+- ExternalRDD [obj#1]
== Physical Plan ==
*SerializeFromObject [input[0, java.lang.Integer, true].intValue AS value#2]
+- Scan ExternalRDDScan[obj#1]
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
