incorrect dynamic filter on partitioned join when file partitioned

[gene-bordegaray]

Describe the bug

When preserving file partitioning we claim "hash" partitioned on the columns values. Then when applying a dynamic filter for partitioned joins it creates a filter based on a true hash function, not key/value based like file partitioning. This causes partitions to be filtered incorrectly:

Declare file partitioned: Hash([partition_col], N)
Filter with hash-based routing:
  CASE hash(key) % N
    WHEN 0 THEN <partition 0 bounds>
    WHEN 1 THEN <partition 1 bounds>
    ...

Since file partitioned is using value based partitioning and dynamic filtering is using true hash based end of in this scenario:
  ┌─────┬──────────────────────┬───────────────┐
  │ Key │ Hive Partition Index │ hash(key) % 3 │
  ├─────┼──────────────────────┼───────────────┤
  │ "A" │ 0                    │ 2             │
  ├─────┼──────────────────────┼───────────────┤
  │ "B" │ 1                    │ 2             │
  ├─────┼──────────────────────┼───────────────┤
  │ "C" │ 2                    │ 2             │
  └─────┴──────────────────────┴───────────────┘

thus, when the filter evaluates row f_dkey=A, the filter computes it to map to partition 2 bounds even though it is actually in partition 0. So it doesn't match and is incorrectly filtered out.

To Reproduce

see PR: #20175

Expected behavior

For now dynamic filtering will be disabled for partitioned hash joins when preserve file partitions is on.

A long term fix is to introduce a new type of partitioning for the file partitioning to safely distinguish the two. Something like KeyPartitoned or ValuePartitioned is suiting.

Oracle calls this ListPartitioning

Additional context

No response

[gabotechs]

A long term fix is to introduce a new type of partitioning for the file partitioning to safely distinguish the two. Something like KeyPartitoned or ValuePartitioned is suiting.

I think the problem goes beyond that. Even if the two sides of a join are Partitioning::Hash because there was a RepartitionExec before, there is no guarantee that the partitioning strategy was the same in both. For example:

• What if both sides of the join where manually repartitioned by the user with a custom rule, and the random seed used to build hashes is different?
• What if in the future we want a new algorithm for RepartitionExec that is capable of adaptively increase or decrease the output partitions? still both sides need to match.

Following the same rule, for the same reason we introduce KeyPartitoned or similar, we could argue that more partitioning modes would need to be added, when all these partitioning methods match the current definition of "Hash Partitioned" (a bit of an unfortunate name).

Oracle calls this ListPartitioning

Note that this is referring to how data is laid out physically in a persistent storage. While the document you shared describes how to partition data storage, the problem here is how to partition read compute resources.

[gabotechs]

I imagine that there should be a relatively easy fix for this though, we can be in two situations:

1. There is a RepartitionExec in between the producer and consumer side of the dynamic filter => the expressions the producer side of the dynamic filter needs to account for it by building expressions that look like CASE hash(key) % N...
2. There is no RepartitionExec in between the producer and consumer side of the dynamic filter => it's fine to treat each partition in isolation, the producer side of the dynamic filter can work per partition and produce one expression for partition 0, that will be consumed by the partition 0 on the consumer side, same for all the other partitions

[gabotechs]

About KeyPartitioning:

Spark does have something similar: KeyGroupedPartitioning. It's pretty much like the current hash partitioning mechanism we have when leaf nodes are already partitioned, but with a different name.

[gene-bordegaray]

follow up work to be done in: #20195