andygrove
commented
5 days ago The reason that each plan appears to be planned twice is that we split the ShuffleWriterExec from the rest of the plan (as noted in https://github.com/apache/datafusion-comet/issues/977).
executePlan() stage 18 partition 28 of 29: planning took 666.045344ms
Comet native query plan:
SortExec: TopK(fetch=10), expr=[col_1@1 DESC NULLS LAST, col_2@2 ASC], preserve_partitioning=[false]
CopyExec [UnpackOrDeepCopy]
ScanExec: source=[], schema=[col_0: Int64, col_1: Decimal128(34, 4), col_2: Date32, col_3: Int32]
executePlan() stage 18 partition 28 of 29: planning took 689.940558ms
Comet native query plan:
ShuffleWriterExec: partitioning=UnknownPartitioning(1)
ScanExec: source=[], schema=[col_0: Int64, col_1: Decimal128(34, 4), col_2: Date32, col_3: Int32]
What is the problem the feature request solves?
For each query stage, the serialized query plan is sent to the executor with each task. Each task deserializes the protobuf and then creates a
PhysicalPlannerand builds a native query plan. The query plan for each partition in a stage is essentially identical, except for the scan input JNI references, so we are duplicating this query planning work across each partition.In some cases, planning is very expensive, TPC-H q3 stage 18 seems to take around 90 seconds. Here is partial debug output. Note that each partition seems to create the query plan twice, which needs further investigation.
Here is another example where planning is relatively cheap, but repeated many times, resulting in 1.76 seconds total planning time.
Questions:
I used the following code to pass the partition numbers to the native code:
Describe the potential solution
No response
Additional context
No response