Implement adaptive partitioning for FTE based on input

[linzebing]

Description

Extend adaptive partitioning to fault-tolerant execution. Testing on tpcds-sf100 shows 4% improvement on wall time and 8% improvement on CPU time. Testing on tpcds-sf1000 shows 4.2% improvement on wall time and 1.9% improvement on CPU time.

Additional context and related issues

#15489

Release notes

(x) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Section
* Fix some things. ({issue}`issuenumber`)

[linzebing]

Adaptive partitioning is disabled for writes in streaming execution, is this a problem in FTE as well? @arhimondr ?

[losipiuk]

Yes - it makes sense to keep lots of partitons if we are writing to avoid "too many open writers" problem. Not sure if we need upper bound always.

[linzebing]

Also, do we need additional fte-specific configs of optimizer.min-input-size-per-task, optimizer.min-input-rows-per-task?

[losipiuk]

Nice. LGTM.

[losipiuk]

Also, do we need additional fte-specific configs of optimizer.min-input-size-per-task, optimizer.min-input-rows-per-task?

Technically we do not support mixed workloads - so maybe not. @arhimondr WDYT?

[losipiuk]

testPlanWhenRetryPolicyIsTask fails as number of partitions changed.

[linzebing]

Failure is unrelated

[linzebing]

@losipiuk @arhimondr : do you think it makes sense to introduce a flag to enable adaptive partitioning for writing? It's meaningful in the fact that we can enable this flag by setting fault_tolerant_execution_min_partition_count to a relative high number like 50.

[losipiuk]

@losipiuk @arhimondr : do you think it makes sense to introduce a flag to enable adaptive partitioning for writing? It's meaningful in the fact that we can enable this flag by setting fault_tolerant_execution_min_partition_count to a relative high number like 50.

IDK how important it is in practice. But if sth I would rather allow for configuring different minimum for writer and non-writer queries.