Deduplicate outputs in luigi.task.flatten_output (#3106)

[starhel]

Description

I've changed luigi.task.flatten_output implementation to deduplicate outputs from tasks which are required by more than one wrapper task. Implementation is slightly changed (from DFS to BFS), but there's no guaranteed order in docs.

Motivation and Context

Fixes #3106

Have you tested this? If so, how?

I've added unit tests for standard use cases and diamond problem.

[lallea]

We often use fine-grained tasks, resulting in large DAGs. The time it takes for Luigi to process the graph is sometimes a bottleneck. I'd appreciate if we did not choose a slower implementation, even if I appreciate a deterministic output order.

Could we either use an ordered set (https://pypi.org/project/ordered-set/) or an OrderedDict to emulate a set? Or perhaps use a set and sort in the output routine?

Thanks for doing the deduplication, BTW. :-)

[starhel]

We often use fine-grained tasks, resulting in large DAGs.

Just remember that flatten_output just gets outputs of subgraph - a list of outputs from expanding wrapper tasks, now the whole DAG.

Could we either use an ordered set (https://pypi.org/project/ordered-set/) or an OrderedDict to emulate a set? Or perhaps use a set and sort in the output routine?

I don't want to add a new dependency for a small feature. I moved the implementation to use an OrderedDict, it's equally fast as a set.

[lallea]

Ah, ok, thanks for the clarification. And for the OrderedDict. It seems like a good tradeoff point.

[dlstadther]

Thanks for going with the OrderedDict solution! LGTM