[Bug Fix]: Deem hash repartition unnecessary when input and output has 1 partition

[mustafasrepo]

Which issue does this PR close?

Closes#9928.

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Yes

Are there any user-facing changes?

[mustafasrepo]

cc @echai58, this fix should solve the problem in the issue. Feel free to review, if you have time.

[mustafasrepo]

@korowa thanks again for the analysis which helped a lot in understanding the issue and replicating it in datafusion.

[ozankabak]

LGTM, thank you

[echai58]

Thanks for the fix!

[alamb]

Thanks @mustafasrepo and @ozankabak

[korowa]

@mustafasrepo good catch on UNION -- I previously suggested that this kind of plan (with partitions > config.target_partitions) is "illegal" in DF.

[mustafasrepo]

@mustafasrepo good catch on UNION -- I previously suggested that this kind of plan (with partitions > config.target_partitions) is "illegal" in DF.

I agree, this behaviour is a bit counter intuitive. However, with current implementation of the UnionExec it is hard to make sure partitions<=config.target_partitions always true. If this behaviour prone to errors. Maybe we can insert RepartitionExec on top UnionExecs if their output partition number > config.target_partitions. By this way, we can guarantee this violation wouldn't propagate to other operators.

[alamb]

I think this query is non determisitic and fails sometimes on CI as it doesn't have an ORDER BY and isn't annotated with rowsort. Here is a PR to fix that: #10120

[korowa]

Maybe we can insert RepartitionExec on top UnionExecs if their output partition number > config.target_partitions. By this way, we can guarantee this violation wouldn't propagate to other operators.

Probably better solution would be planning union inputs execution according to total available partitions -- e.g

select l_linenumber as f
from lineitem
union allselect l_orderkey as f
from lineitem

with target_partitions = 4, could plan 2 threads for each ParquetExec (ideally we could also use byte/row statistics and plan according to them -- not only 2-2, but probably 1-3 if there is significant data skew across inputs/files).

Currently, with target_partitions = 4, it's planned as 4 threads per ParquetExec, and 8 output partitions for UNION.

And on top of it, when target_partitions is less then number of UNION inputs (e.g. UNION has 10 inputs, target_partitions = 4, and we need at least 1 thread for each input) there could be RepartitionExec.

[mustafasrepo]

Maybe we can insert RepartitionExec on top UnionExecs if their output partition number > config.target_partitions. By this way, we can guarantee this violation wouldn't propagate to other operators.

Probably better solution would be planning union inputs execution according to total available partitions -- e.g

select l_linenumber as f
from lineitem
union allselect l_orderkey as f
from lineitem

with target_partitions = 4, could plan 2 threads for each ParquetExec (ideally we could also use byte/row statistics and plan according to them -- not only 2-2, but probably 1-3 if there is significant data skew across inputs/files).

Currently, with target_partitions = 4, it's planned as 4 threads per ParquetExec, and 8 output partitions for UNION.

And on top of it, when target_partitions is less then number of UNION inputs (e.g. UNION has 10 inputs, target_partitions = 4, and we need at least 1 thread for each input) there could be RepartitionExec.

That might work. However, this approach cannot solve all cases I guess. For the following query

select*from table
union allselect*from table
union allselect*from table

when target_partitions=2, there is no way to change input partitions to generate 2 partitions after union. We would generate at least 3 partitions(assuming each input generates single partition). Hence, this approach may not solve all use cases. If I am not mistaken.

[echai58]

Hi guys, not familiar with datafusion's release process - is there a estimate of when this will be released in a new datafusion version?

[alamb]

Hi guys, not familiar with datafusion's release process - is there a estimate of when this will be released in a new datafusion version?

This should be included in 38.0.0 -- I just filed #10255 to track that release if you want to watch it @echai58