Skip to content

Support optional filter in Join #2509

Closed

Description

@yjshen

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

It would be necessary to support filters in the join operator, instead of a join operator followed by a filter, the necessity comes from two folds:

  1. semi-join that columns from one table would be removed after the join, which would make a filter that references both sides impossible.
  2. filter out records earlier to reduce the need for constructing more records(batches).

Describe the solution you'd like

  1. pub type FilterOn = (Vec<Column>, Vec<Column>, Arc<dyn PhysicalExpr>); and Option<FilterOn> for JoinExec.
  2. generate record batch using arrays in the filter expr, but rebind the expr to point to the right columns of the newly generated batch.

Describe alternatives you've considered

  1. pub type FilterOn = Vec<(Column, Column, datafusion_expr::Operator)>;
  2. Normalize each filter into two sides of a binary op. like: t1.a + t2.b > 100 to t1.a > 100 - t2.b. evaluates a , 100-b separately as two columns and apply binary expr calculation logic.

But the approach would be quite limited since it greatly limits the expressions that could be used in a join filter.

Additional context

Consider Part of TPC-DS query-95's SparkSQL plan as an example:

+- SortMergeJoin [ws_order_number#251], [ws_order_number#285], Inner, NOT (ws_warehouse_sk#249 = ws_warehouse_sk#283)
   :- Sort [ws_order_number#251 ASC NULLS FIRST], false, 0
   :  +- CustomShuffleReader coalesced
   :     +- ShuffleQueryStage 4
   :        +- ReusedExchange [ws_warehouse_sk#249, ws_order_number#251], ArrowShuffleExchange hashpartitioning(ws_order_number#125, 200), true, [id=#226]
   +- Sort [ws_order_number#285 ASC NULLS FIRST], false, 0
      +- CustomShuffleReader coalesced
         +- ShuffleQueryStage 5
            +- ReusedExchange [ws_warehouse_sk#283, ws_order_number#285], ArrowShuffleExchange hashpartitioning(ws_order_number#125, 200), true, [id=#226]

Activity

  1. yjshen commented on May 11, 2022

    @yjshen
    MemberAuthor

    A related issue #2496.

  2. changed the title [-]Support optional filter in Join[/-] [+]Support optional filter in SortMergeJoin[/+] on May 26, 2022
  3. reopened this on May 26, 2022
  4. yjshen commented on May 26, 2022

    @yjshen
    MemberAuthor

    Reopened and renamed to track sort-merge join filter as well.

  5. alamb commented on May 26, 2022

    @alamb
    Contributor

    @yjshen do you mind if I close this ticket and reopen another describing the support needed for Sort-merge? I think it might be clearer to a future reader that we just needed to extend the support in HashJoin to MergeJoin whereas the description of this ticket now may confuse people as it talks about differing implementation possibilities

  6. yjshen commented on May 27, 2022

    @yjshen
    MemberAuthor

    Get it; I will open a new issue instead.

  7. changed the title [-]Support optional filter in SortMergeJoin[/-] [+]Support optional filter in Join[/+] on May 27, 2022
  8. alamb commented on May 27, 2022

    @alamb
    Contributor

    Get it; I will open a new issue instead.

    Thanks @yjshen !

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    enhancementNew feature or request

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions