[
https://issues.apache.org/jira/browse/SPARK-24661?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hyukjin Kwon updated SPARK-24661:
---------------------------------
Component/s: (was: PySpark)
> Window API - using multiple fields for partitioning with WindowSpec API and
> dataset that is cached causes
> org.apache.spark.sql.catalyst.errors.package$TreeNodeException
> ------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-24661
> URL: https://issues.apache.org/jira/browse/SPARK-24661
> Project: Spark
> Issue Type: Bug
> Components: DStreams, Java API
> Affects Versions: 2.3.0
> Reporter: David Mavashev
> Priority: Major
>
> Steps to reproduce:
> Creating a data set:
>
> {code:java}
> List simpleWindowColumns = new ArrayList();
> simpleWindowColumns.add("column1");
> simpleWindowColumns.add("column2");
> Map<String, String> expressionsWithAliasesEntrySet = new HashMap<String,
> String>);
> expressionsWithAliasesEntrySet.put("count(id)", "count_column");
> DataFrameReader reader = sparkSession.read().format("csv");
> Dataset<Row> sparkDataSet = reader.option("header",
> "true").load("/path/to/data/data.csv");
> //Invoking cached:
> sparkDataSet = sparkDataSet.cache()
> //Creating window spec with 2 columns:
> WindowSpec window =
> Window.partitionBy(JavaConverters.asScalaIteratorConverter(simpleWindowColumns.stream().map(item->sparkDataSet.col(item)).iterator()).asScala().toSeq());
> sparkDataSet =
> sparkDataSet.withColumns(JavaConverters.asScalaIteratorConverter(expressionsWithAliasesEntrySet.stream().map(item->item.getKey()).collect(Collectors.toList()).iterator()).asScala().toSeq(),
>
> JavaConverters.asScalaIteratorConverter(expressionsWithAliasesEntrySet.stream().map(item->new
>
> Column(item.getValue()).over(finalWindow)).collect(Collectors.toList()).iterator()).asScala().toSeq());
> sparkDataSet.show();{code}
> Expected:
>
> Results are shown
>
>
> Actual: the following exception is thrown
> {code:java}
> org.apache.spark.sql.catalyst.errors.package$TreeNodeException: makeCopy,
> tree: windowspecdefinition(O003#3, O006#6, specifiedwindowframe(RowFrame,
> unboundedpreceding$(), unboundedfollowing$())) at
> org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:56) at
> org.apache.spark.sql.catalyst.trees.TreeNode.makeCopy(TreeNode.scala:385) at
> org.apache.spark.sql.catalyst.trees.TreeNode.withNewChildren(TreeNode.scala:244)
> at
> org.apache.spark.sql.catalyst.expressions.Expression.canonicalized$lzycompute(Expression.scala:190)
> at
> org.apache.spark.sql.catalyst.expressions.Expression.canonicalized(Expression.scala:188)
> at
> org.apache.spark.sql.catalyst.expressions.Expression$$anonfun$1.apply(Expression.scala:189)
> at
> org.apache.spark.sql.catalyst.expressions.Expression$$anonfun$1.apply(Expression.scala:189)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at scala.collection.immutable.List.foreach(List.scala:381) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:245) at
> scala.collection.immutable.List.map(List.scala:285) at
> org.apache.spark.sql.catalyst.expressions.Expression.canonicalized$lzycompute(Expression.scala:189)
> at
> org.apache.spark.sql.catalyst.expressions.Expression.canonicalized(Expression.scala:188)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$.normalizeExprId(QueryPlan.scala:288)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$doCanonicalize$1.apply(QueryPlan.scala:232)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$doCanonicalize$1.apply(QueryPlan.scala:226)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.transformExpression$1(QueryPlan.scala:106)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.org$apache$spark$sql$catalyst$plans$QueryPlan$$recursiveTransform$1(QueryPlan.scala:116)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$org$apache$spark$sql$catalyst$plans$QueryPlan$$recursiveTransform$1$1.apply(QueryPlan.scala:120)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:245) at
> scala.collection.AbstractTraversable.map(Traversable.scala:104) at
> org.apache.spark.sql.catalyst.plans.QueryPlan.org$apache$spark$sql$catalyst$plans$QueryPlan$$recursiveTransform$1(QueryPlan.scala:120)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$1.apply(QueryPlan.scala:125)
> at
> org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.mapExpressions(QueryPlan.scala:125)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.doCanonicalize(QueryPlan.scala:226)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized$lzycompute(QueryPlan.scala:210)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized(QueryPlan.scala:209)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at scala.collection.immutable.List.foreach(List.scala:381) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:245) at
> scala.collection.immutable.List.map(List.scala:285) at
> org.apache.spark.sql.catalyst.plans.QueryPlan.doCanonicalize(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized$lzycompute(QueryPlan.scala:210)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized(QueryPlan.scala:209)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at scala.collection.immutable.List.foreach(List.scala:381) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:245) at
> scala.collection.immutable.List.map(List.scala:285) at
> org.apache.spark.sql.catalyst.plans.QueryPlan.doCanonicalize(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized$lzycompute(QueryPlan.scala:210)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized(QueryPlan.scala:209)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at scala.collection.immutable.List.foreach(List.scala:381) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:245) at
> scala.collection.immutable.List.map(List.scala:285) at
> org.apache.spark.sql.catalyst.plans.QueryPlan.doCanonicalize(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized$lzycompute(QueryPlan.scala:210)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized(QueryPlan.scala:209)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at scala.collection.immutable.List.foreach(List.scala:381) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:245) at
> scala.collection.immutable.List.map(List.scala:285) at
> org.apache.spark.sql.catalyst.plans.QueryPlan.doCanonicalize(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized$lzycompute(QueryPlan.scala:210)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized(QueryPlan.scala:209)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$2.apply(QueryPlan.scala:224)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at scala.collection.immutable.List.foreach(List.scala:381) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:245) at
> scala.collection.immutable.List.map(List.scala:285) at
> org.apache.spark.sql.catalyst.plans.QueryPlan.doCanonicalize(QueryPlan.scala:224)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized$lzycompute(QueryPlan.scala:210)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized(QueryPlan.scala:209)
> at
> org.apache.spark.sql.catalyst.plans.QueryPlan.sameResult(QueryPlan.scala:257)
> at
> org.apache.spark.sql.execution.CacheManager$$anonfun$lookupCachedData$2$$anonfun$apply$3.apply(CacheManager.scala:166)
> at
> org.apache.spark.sql.execution.CacheManager$$anonfun$lookupCachedData$2$$anonfun$apply$3.apply(CacheManager.scala:166)
> at scala.collection.Iterator$class.find(Iterator.scala:803) at
> scala.collection.AbstractIterator.find(Iterator.scala:1202) at
> scala.collection.IterableLike$class.find(IterableLike.scala:79) at
> scala.collection.AbstractIterable.find(Iterable.scala:54) at
> org.apache.spark.sql.execution.CacheManager$$anonfun$lookupCachedData$2.apply(CacheManager.scala:166)
> at
> org.apache.spark.sql.execution.CacheManager$$anonfun$lookupCachedData$2.apply(CacheManager.scala:166)
> at
> org.apache.spark.sql.execution.CacheManager.readLock(CacheManager.scala:58)
> at
> org.apache.spark.sql.execution.CacheManager.lookupCachedData(CacheManager.scala:165)
> at
> org.apache.spark.sql.execution.CacheManager$$anonfun$1.applyOrElse(CacheManager.scala:180)
> at
> org.apache.spark.sql.execution.CacheManager$$anonfun$1.applyOrElse(CacheManager.scala:171)
> at
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
> at
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
> at
> org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
> at
> org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266)
> at
> org.apache.spark.sql.execution.CacheManager.useCachedData(CacheManager.scala:171)
> at
> org.apache.spark.sql.execution.QueryExecution.withCachedData$lzycompute(QueryExecution.scala:63)
> at
> org.apache.spark.sql.execution.QueryExecution.withCachedData(QueryExecution.scala:60)
> at
> org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:66)
> at
> org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:66)
> at
> org.apache.spark.sql.execution.QueryExecution.sparkPlan$lzycompute(QueryExecution.scala:72)
> at
> org.apache.spark.sql.execution.QueryExecution.sparkPlan(QueryExecution.scala:68)
> at
> org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:77)
> at
> org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:77)
> at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3248) at
> org.apache.spark.sql.Dataset.head(Dataset.scala:2484) at
> org.apache.spark.sql.Dataset.take(Dataset.scala:2698) at
> org.apache.spark.sql.Dataset.showString(Dataset.scala:254) at
> com.bottomline.pa.integration.infrastructure.spark.api.impl.SimpleSparkSQLAggregatingWindowTransformerTest.swiftTest(SimpleSparkSQLAggregatingWindowTransformerTest.java:441)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498) at
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325) at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290) at
> org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71) at
> org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288) at
> org.junit.runners.ParentRunner.access$000(ParentRunner.java:58) at
> org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268) at
> org.junit.runners.ParentRunner.run(ParentRunner.java:363) at
> org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:86)
> at
> org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38)
> at
> org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:538)
> at
> org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:760)
> at
> org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:460)
> at
> org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:206)
> Caused by: org.apache.spark.sql.catalyst.errors.package$TreeNodeException:
> Failed to copy node. Is otherCopyArgs specified correctly for
> WindowSpecDefinition. Exception message: argument type mismatch ctor: public
> org.apache.spark.sql.catalyst.expressions.WindowSpecDefinition(scala.collection.Seq,scala.collection.Seq,org.apache.spark.sql.catalyst.expressions.WindowFrame)?
> types: class scala.collection.immutable.Stream$Cons, class
> scala.collection.immutable.Nil$, class
> org.apache.spark.sql.catalyst.expressions.AttributeReference args:
> Stream(none#3, ?), List(), none#6 , tree: windowspecdefinition(O003#3,
> O006#6, specifiedwindowframe(RowFrame, unboundedpreceding$(),
> unboundedfollowing$())) at
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$makeCopy$1.apply(TreeNode.scala:415)
> at
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$makeCopy$1.apply(TreeNode.scala:385)
> at
> org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:52)
> ... 137 more
> {code}
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
