[
https://issues.apache.org/jira/browse/SPARK-59114?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18109598#comment-18109598
]
Asif commented on SPARK-59114:
------------------------------
Initially I debugged the issue in my private fork of 4.1, where the fix was
trivial.
But in current master, the fix is non trivial, though the overall changes are
simple, but they span multiple files.
> 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]
