[
https://issues.apache.org/jira/browse/SPARK-59114?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-59114:
-----------------------------------
Labels: pull-request-available (was: )
> java.lang.ClassCastException: class
> org.apache.spark.sql.catalyst.expressions.UnsafeRow cannot be cast to class
> org.apache.spark.sql.vectorized.ColumnarBatch
> --------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-59114
> URL: https://issues.apache.org/jira/browse/SPARK-59114
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0, 4.3.0, 4.1.3
> Reporter: Asif
> Priority: Major
> Labels: pull-request-available
>
> When a spark plan using a FileSourceScanExec as leaf, reading a parquet
> format file, with default config (i.e parquet vector read enabled) is cached,
> the SparkPlan created is something like
> WholeStageCodegenExec -> ColumnToRowExec -> InputAdapter ->
> FileSourceScanExec.
> The presence of ColumntoRowExec, implies that the child is going to output
> ColumnarBatch as the RDD.
> This SparkPlan is stored in the CacheManager, and since the
> CachedRDDBuilder's buffers are lazy init, the spark plan is not executed at
> the time of caching.
>
> Now after cached entry is created, and a query is executed involving the
> cached plan, with SQLConf having
> SQLConf.PARQUET_VECTORIZED_READER_ENABLED -> false,
>
> then the cached plan is executed, with the this property, and as a result,
> the FileSourceScanExec no longer returns RDD[ColumnarBatch] ( it is the RDD
> of Row now).
> While the CachedRDDBuilder code is expecting ColumnarBatch.
>
> The fix proposed though is trivial, that is in the ColumnToRowExec.inputRDDs
> , explictly set the conf property of
> SQLConf.PARQUET_VECTORIZED_READER_ENABLED as true.
> The reasoning is that , since the plan contains ColumnToRowExec, that itself
> implies that the child RDD should be returning a ColumnarBatch and not Row.
> However , I am not sure if this is sufficient and may be
> ColumnToRowExec.execute etc may also need to be tackled.
>
> It is also not an elegant solution , more of work around.
> The holistic solution should I suppose audit all the SQLConf properties which
> otherwise can impact the Plan being used , such that it renders it totally
> incompatible, May be such properties should be intialized in the Plan's
> constructir itself at the time of creation and it should over-ride any
> existing SQLConf.
>
> I am exploring if its possible to store the nature of RDD return type ( Row
> or Batch), as part of FileSourceScanExec constructor itself, so that its
> unaffected by the subsequent change in conf.
>
> A PR ( or may be two possible approaches fix will be filed shortly).
>
>
>
>
>
> Job aborted due to stage failure: Task 0 in stage 2.0 failed 1 times, most
> recent failure: Lost task 0.0 in stage 2.0 (TID 3) (localhost executor
> driver): java.lang.ClassCastException: class
> org.apache.spark.sql.catalyst.expressions.UnsafeRow cannot be cast to class
> org.apache.spark.sql.vectorized.ColumnarBatch
> (org.apache.spark.sql.catalyst.expressions.UnsafeRow and
> org.apache.spark.sql.vectorized.ColumnarBatch are in unnamed module of loader
> 'app')
> at
> org.apache.spark.sql.execution.FileSourceScanExec$$anon$1.next(DataSourceScanExec.scala:832)
> at
> org.apache.spark.sql.execution.FileSourceScanExec$$anon$1.next(DataSourceScanExec.scala:821)
> at
> org.apache.spark.sql.execution.ColumnarToRowExec$.advanceBatch(Columnar.scala:233)
> at
> org.apache.spark.sql.execution.ColumnarToRowExec.advanceBatch(Columnar.scala)
> at
> org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.columnartorow_nextBatch_0$(Unknown
> Source)
> at
> org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
> Source)
> at
> org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:44)
> at
> org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:50)
> at
> org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:178)
> at
> org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anon$2.hasNext(InMemoryRelation.scala:445)
> at
> org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:231)
> at
> org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:318)
> at
> org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1882)
> at
> org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1806)
> at
> org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1872)
> at
> org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1642)
> at
> org.apache.spark.storage.BlockManager.getOrElseUpdateRDDBlock(BlockManager.scala:1594)
> at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:439)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:336)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)
> at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:223)
> at org.apache.spark.scheduler.Task.run(Task.scala:147)
> at
> org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$5(Executor.scala:912)
> at
> org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:86)
> at
> org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:83)
> at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:97)
> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:915)
> at
> java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
> at
> java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
> at java.base/java.lang.Thread.run(Thread.java:1583)
>
> Driver stacktrace:
> org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in
> stage 2.0 failed 1 times, most recent failure: Lost task 0.0 in stage 2.0
> (TID 3) (localhost executor driver): java.lang.ClassCastException: class
> org.apache.spark.sql.catalyst.expressions.UnsafeRow cannot be cast to class
> org.apache.spark.sql.vectorized.ColumnarBatch
> (org.apache.spark.sql.catalyst.expressions.UnsafeRow and
> org.apache.spark.sql.vectorized.ColumnarBatch are in unnamed module of loader
> 'app')
> at
> org.apache.spark.sql.execution.FileSourceScanExec$$anon$1.next(DataSourceScanExec.scala:832)
> at
> org.apache.spark.sql.execution.FileSourceScanExec$$anon$1.next(DataSourceScanExec.scala:821)
> at
> org.apache.spark.sql.execution.ColumnarToRowExec$.advanceBatch(Columnar.scala:233)
> at
> org.apache.spark.sql.execution.ColumnarToRowExec.advanceBatch(Columnar.scala)
> at
> org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.columnartorow_nextBatch_0$(Unknown
> Source)
> at
> org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
> Source)
> at
> org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:44)
> at
> org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:50)
> at
> org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:178)
> at
> org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anon$2.hasNext(InMemoryRelation.scala:445)
> at
> org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:231)
> at
> org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:318)
> at
> org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1882)
> at
> org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1806)
> at
> org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1872)
> at
> org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1642)
> at
> org.apache.spark.storage.BlockManager.getOrElseUpdateRDDBlock(BlockManager.scala:1594)
> at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:439)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:336)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:57)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
> at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)
> at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:223)
> at org.apache.spark.scheduler.Task.run(Task.scala:147)
> at
> org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$5(Executor.scala:912)
> at
> org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:86)
> at
> org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:83)
> at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:97)
> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:915)
> at
> java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
> at
> java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
> at java.base/java.lang.Thread.run(Thread.java:1583)
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
