Allow forcing streaming exchange for Mark Distinct

[aweisberg]

Mark distinct is enabled by default and introduces a stage per distinct column in a query in order to redistribute the tracking of distinct values across the entire cluster. With materialized exchange the cost of these additional exchanges is higher. Enabling streaming exchange if there is sufficient memory available to perform the distincting will reduce the cost of these additional stages while still reducing the memory used by other by non-Mark Distinct stages.

== RELEASE NOTES ==

General Changes
* Allow forcing streaming exchange for Mark Distinct using 'query.use-streaming-exchange-for-mark-distinct'

[wenleix]

Hmm.. ideally we want to add the third option into exchange_materialization_strategy:

• ALL
• NONE
• ALL_BUT_DISTINCT

But I can totally see this would make operation / rollout totally a mess. So it makes sense to have this new stream_mark_distinct_with_materialized_exchange session property (name TBD...) :). Just want to make sure we discussed this :)

[aweisberg]

I don't think this is an exchange materialization strategy because we may have several different permutations of whether we want to stream or materialize. Enumerating the permutations would be difficult under one config option.

A strategy would be something more like "CBO_BASED_MATERIALIZATION" that makes smarter decisions.

[wenleix]

I don't think this is an exchange materialization strategy because we may have several different permutations of whether we want to stream or materialize. Enumerating the permutations would be difficult under one config option.

@aweisberg : By "permutation" I assume you mean there are several different orthogonal dimensions about whether do stream or materialize ? (e.g. based on operator ? :) )

In my opinion it's still an exchange materialization strategy, however, we decompose the strategy space into several orthogonal options, such as decision on Mark Distinct, etc.

[wenleix]

"Use static imports in TestAddExchangePlans". LGTM.

[rongrong]

Not related to this change, but a related topic. We should change the default for use_mark_distinct to false.

[wenleix]

"Add stream mark distinct with LBM session property". One question about the condition.

For the commit message subject, "LBM" is an Facebook internal term, what about :

Allow stream exchange for mark distinct when materialized exchange is set

When MarkDistinctOperator is to execute queries with many COUNT(DISTINCT)
it requires separate exchange for every distinct COUNT(DISTINCT), which is 
prohibitively expensive for materialized exchange. 

[wenleix]

nit: What about useStreamExchangeForMarkDistinct?

[aweisberg]

You are right technically that is what this does. I'll change to that.

[wenleix]

ditto, query.use-stream-exchange-for-mark-distinct

[wenleix]

nit: useStreamExchangeForMarkDitinct as variable name ? :)

[arhimondr]

nit: static import

[arhimondr]

nit: It should be fine to inline it after the static import is applied

[aweisberg]

@rongrong Regarding changing the default of use_mark_distinct. Where peak_task_memory is an issue use_mark_distinct can help. For T1 this is more interesting because we have more distributed memory than task memory.

[aweisberg]

@wenleix regarding the commit message. I was trying to keep the first line under 50 characters. That's 73.

[aweisberg]

@wenleix Also the description seems too specific WRT to COUNT(DISTINCT)? You can distinct the inputs to any aggregation column.

[aweisberg]

@wenleix Finished landing changes. I went with a slightly different commit message and useStreamingExchangeForMarkDistinct instead of useStreamExchangeForMarkDistinct.

[wenleix]

LGTM. Will merge it once Travis is green.

(A side note -- do we know if this helps? 😃 )

[wenleix]

Merged #14216, thanks for the contribution!