Spark SQL physical plan doesn't reuse exchange

user_14355369 0
Viewed 284

I'm trying to optimize the physical plan for the following transformation.

  1. Read data from 'pad' and 'pi'
  2. Find rows in 'pad' that have a reference in 'pi' and transform some columns.
  3. Find rows in 'pad' that don't have a reference in 'pi' and transform some columns.
  4. Merge rows from 2 and 3.
val pad_in_pi = pad
  .join(
    pi
    , $"pad.ReferenceKeyCode" === $"pi.PurchaseInvoiceKeyCode"
    , "inner"
  )
  .selectExpr(
    "pad.AccountingDocumentKeyCode"
    , "pad.RegionId"
    , "pi.PurchaseInvoiceLineNumber as DocumentLineNumber"
    , "pi.CodingBlockSequentialNumber"
  )
  
val pad_not_in_pi = pad
  .join(
    pi
    , $"pad.ReferenceKeyCode" === $"pi.PurchaseInvoiceKeyCode"
    , "anti"
  )
  .selectExpr(
    "pad.AccountingDocumentKeyCode"
    , "pad.RegionId"
    , "pad.AccountingDocumentLineNumber as DocumentLineNumber"
    , "0001 as CodingBlockSequentialNumber"
  )
  
  pad_in_pi.union(pad_not_in_pi)

Branch 2 and 3 use the same join expression, and can thus reuse the exchange. The current physical plan doesn't. What could be the reason?

== Physical Plan ==
Union
:- *(3) Project [AccountingDocumentKeyCode#491, RegionId#539, PurchaseInvoiceLineNumber#205 AS DocumentLineNumber#954, CodingBlockSequentialNumber#203]
:  +- *(3) SortMergeJoin [ReferenceKeyCode#538], [PurchaseInvoiceKeyCode#235], Inner
:     :- Sort [ReferenceKeyCode#538 ASC NULLS FIRST], false, 0
:     :  +- Exchange hashpartitioning(ReferenceKeyCode#538, 200), true, [id=#684]
:     :     +- *(1) Project [AccountingDocumentKeyCode#491, ReferenceKeyCode#538, RegionId#539]
:     :        +- *(1) Filter ((isnotnull(RegionId#539) AND (RegionId#539 = R)) AND isnotnull(ReferenceKeyCode#538))
:     :           +- *(1) ColumnarToRow
:     :              +- FileScan parquet default.purchaseaccountingdocument_delta[AccountingDocumentKeyCode#491,ReferenceKeyCode#538,RegionId#539] Batched: true, DataFilters: [isnotnull(RegionId#539), (RegionId#539 = R), isnotnull(ReferenceKeyCode#538)], Format: Parquet, Location: PreparedDeltaFileIndex[dbfs:..., PartitionFilters: [], PushedFilters: [IsNotNull(RegionId), EqualTo(RegionId,R), IsNotNull(ReferenceKeyCode)], ReadSchema: struct<AccountingDocumentKeyCode:string,ReferenceKeyCode:string,RegionId:string>
:     +- Sort [PurchaseInvoiceKeyCode#235 ASC NULLS FIRST], false, 0
:        +- Exchange hashpartitioning(PurchaseInvoiceKeyCode#235, 200), true, [id=#692]
:           +- *(2) Project [CodingBlockSequentialNumber#203, PurchaseInvoiceLineNumber#205, PurchaseInvoiceKeyCode#235]
:              +- *(2) Filter ((isnotnull(RegionId#207) AND (RegionId#207 = R)) AND isnotnull(PurchaseInvoiceKeyCode#235))
:                 +- *(2) ColumnarToRow
:                    +- FileScan parquet default.purchaseinvoice_delta[CodingBlockSequentialNumber#203,PurchaseInvoiceLineNumber#205,RegionID#207,PurchaseInvoiceKeyCode#235] Batched: true, DataFilters: [isnotnull(RegionID#207), (RegionID#207 = R), isnotnull(PurchaseInvoiceKeyCode#235)], Format: Parquet, Location: PreparedDeltaFileIndex[dbfs:..., PartitionFilters: [], PushedFilters: [IsNotNull(RegionID), EqualTo(RegionID,R), IsNotNull(PurchaseInvoiceKeyCode)], ReadSchema: struct<CodingBlockSequentialNumber:string,PurchaseInvoiceLineNumber:string,RegionID:string,Purcha...
+- *(6) Project [AccountingDocumentKeyCode#491, RegionId#539, AccountingDocumentLineNumber#492 AS DocumentLineNumber#1208, 1 AS CodingBlockSequentialNumber#1210]
   +- SortMergeJoin [ReferenceKeyCode#538], [PurchaseInvoiceKeyCode#235], LeftAnti
      :- Sort [ReferenceKeyCode#538 ASC NULLS FIRST], false, 0
      :  +- Exchange hashpartitioning(ReferenceKeyCode#538, 200), true, [id=#703]
      :     +- *(4) Project [AccountingDocumentKeyCode#491, AccountingDocumentLineNumber#492, ReferenceKeyCode#538, RegionId#539]
      :        +- *(4) Filter (isnotnull(RegionId#539) AND (RegionId#539 = R))
      :           +- *(4) ColumnarToRow
      :              +- FileScan parquet default.purchaseaccountingdocument_delta[AccountingDocumentKeyCode#491,AccountingDocumentLineNumber#492,ReferenceKeyCode#538,RegionId#539] Batched: true, DataFilters: [isnotnull(RegionId#539), (RegionId#539 = R)], Format: Parquet, Location: PreparedDeltaFileIndex[dbfs:..., PartitionFilters: [], PushedFilters: [IsNotNull(RegionId), EqualTo(RegionId,R)], ReadSchema: struct<AccountingDocumentKeyCode:string,AccountingDocumentLineNumber:string,ReferenceKeyCode:stri...
      +- Sort [PurchaseInvoiceKeyCode#235 ASC NULLS FIRST], false, 0
         +- Exchange hashpartitioning(PurchaseInvoiceKeyCode#235, 200), true, [id=#710]
            +- *(5) Project [PurchaseInvoiceKeyCode#235]
               +- *(5) Filter ((isnotnull(RegionId#207) AND (RegionId#207 = R)) AND isnotnull(PurchaseInvoiceKeyCode#235))
                  +- *(5) ColumnarToRow
                     +- FileScan parquet default.purchaseinvoice_delta[RegionID#207,PurchaseInvoiceKeyCode#235] Batched: true, DataFilters: [isnotnull(RegionID#207), (RegionID#207 = R), isnotnull(PurchaseInvoiceKeyCode#235)], Format: Parquet, Location: PreparedDeltaFileIndex[dbfs:..., PartitionFilters: [], PushedFilters: [IsNotNull(RegionID), EqualTo(RegionID,R), IsNotNull(PurchaseInvoiceKeyCode)], ReadSchema: struct<RegionID:string,PurchaseInvoiceKeyCode:string>
apache-spark
1 Answers

Not directly answering about exchange reuse, but try a left outer join to get rid of a union:

pad.join(
    pi
    , $"pad.ReferenceKeyCode" === $"pi.PurchaseInvoiceKeyCode"
    , "left_outer"
  )
  .selectExpr(
    "pad.AccountingDocumentKeyCode"
    , "pad.RegionId"
    , "coalesce(pi.PurchaseInvoiceLineNumber, pad.AccountingDocumentLineNumber) as DocumentLineNumber"
    , "coalesce(pi.CodingBlockSequentialNumber, '0001') as CodingBlockSequentialNumber"
  )
user_28900930
Related