fix: use inputRDD to get outputPartitions in CometScanExec #1162
Conversation
|
@viirya could you review? |
| @@ -132,7 +132,7 @@ case class CometScanExec( | |||
| lazy val bucketedScan: Boolean = wrapped.bucketedScan | |||
|
|
|||
| override lazy val (outputPartitioning, outputOrdering): (Partitioning, Seq[SortOrder]) = | |||
| (wrapped.outputPartitioning, wrapped.outputOrdering) | |||
| (UnknownPartitioning(wrapped.inputRDD.getNumPartitions), wrapped.outputOrdering) | |||
There was a problem hiding this comment.
What the output partitioning is when it fails the test? I assume that the outputPartitioning should have correct partition number but not?
There was a problem hiding this comment.
We were seeing zero partitions. Here is some debug output from testing:
firstNonBroadcastPlan = Some((CometScan parquet [_1#6,_2#7] Batched: true, DataFilters: [isnotnull(_1#6)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/spark-a21845bf-72dc-49d1-94f0-f785bb6f2f18], PartitionFilters: [], PushedFilters: [IsNotNull(_1)], ReadSchema: struct<_1:int,_2:int>
,0))
firstNonBroadcastPlanNumPartitions = Some(0)
There was a problem hiding this comment.
Correct. wrapped.outputPartitioning was zero
There was a problem hiding this comment.
It is somehow weird. It means the outputPartitioning info is not correct in original ScanExec and not matched the output RDD's partition number. 🤔
| @@ -132,7 +132,7 @@ case class CometScanExec( | |||
| lazy val bucketedScan: Boolean = wrapped.bucketedScan | |||
|
|
|||
| override lazy val (outputPartitioning, outputOrdering): (Partitioning, Seq[SortOrder]) = | |||
| (wrapped.outputPartitioning, wrapped.outputOrdering) | |||
| (UnknownPartitioning(wrapped.inputRDD.getNumPartitions), wrapped.outputOrdering) | |||
There was a problem hiding this comment.
Btw, I think we should keep original partition instead a hard-coded UnknownPartitioning?
There was a problem hiding this comment.
I did this based on what we do in CometNativeScanExec as well as the original output partition. The original wrapped.outputPartitioning for CometScanExec inherits from FileSourceScanLike and this returns a hardcoded UnknownPartitioning(0) for non-bucketed scan.
outputPartitioning for CometScanExec was incorrectly returning 0 causing unit tests to fail