Schedule dynamic filtering collecting task immediately

[sopel39]

In case of plan
J1
/
J2 S3
/
S1 S2

It might happen that that dynamic filtering dependencies are:
S3 => S3 => S1

With phased scheduler source stage consisting of
(J1, J2, S1) won't be scheduler until stages with S3 and S2
have finished split enumaration. However, it might happen
that S2 is waiting for dynamic filters for S3. In that case,
S3 will never complete because DFs for S3 are collected in
stage (J1, J2, S1).

This commit makes scheduling of DF collecting task immediately
which will prevent queries from deadlock.

[dain]

The scheduler stuff and the tests look good to me. I don't understand the implications of the change to DynamicFilterService.

[sopel39]

ac @raunaqmorarka @dain . I've narrowed down when phased scheduler won't start join stage immediately to only non-fixed source stages. I've also improved tests with regards to source stage partitioning

[dain]

looks good

[arhimondr]

It doesn't feel like I really understand the problem. Added some questions.

It would be great if you could extract refactoring and improvements that are unrelated to have a commit that only does what's needed to address the problem.

Additionally if you could elaborate more on a problem in the commit message it would be great. Maybe you can provide an example of a query that triggers a deadlock with a description of what join distributions are, where are the stage boundaries and how dynamic filters interact with each other.

[sopel39]

@arhimondr I've simplified this PR (removed refactor), keeping just needed parts. i cannot answer some outdated comments

This changes the logic from If there's any dynamic filters produced and consumed within a stage to If there's any dynamic filters produced by a stage. Am i correct? Could you please explain why is it necessary? Is it needed to cover a mix of broadcast and partitioned joins?

I've changed to logic to create collecting task if there is any lazy DF produced by stage. This is needed because there might be consumers of that lazy DF outside of stage.

From what I understand it is to prevent a non source distributed stage running a replicated join from being executed lazily. Is it correct? I wonder why is it necessary.

There won't be extra collecting task for partitioned stages. Hence, if there are any lazy DFs produced by that stage, then stage needs to be scheduled without delay

[sopel39]

Thanks for review