Adaptive task sizing for fault tolerant execution

[linzebing]

Description

This PR adds adaptive task sizing for fault tolerant execution. Specifically:

• For arbitrary distribution: we will start small and gradually increase target partition size.
• For hash distribution: we now use a smaller target partition size for compute tasks and simple let small tasks coalesce (number of tasks are bounded by number of partitions); for write tasks, we adjust target partition size based on the total amount of input bytes to avoid creating massive amount of tasks.

This change dramatically improves small query latency on fault-tolerant execution. Preliminary testing on tpcds-sf100 shows 40%+ latency reduction.

Additional context and related issues

Fix #16103

Release notes

( ) 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.
(x) Release notes are required, with the following suggested text:

# Section
* Implement adaptive task sizing for fault-tolerant execution, which greatly reduces latency for small queries on a FTE cluster. ({issue}`16103`)

[losipiuk]

nit: It would be nice to ensure that this translates to number of splits which are multiply of task.max-drivers-per-task here.
One option - maybe good enough would be to assume that minTargetPartitionSizeInBytes satisfies this requirement and then ensure that calculated targetPartitionSizeInBytes is a multiply of minTargetPartitionSizeInBytes.

Then we can use adaptiveGrowthFactor less than 2.0 to have more fluent growth. 2.0 is pretty agressive.

cc: @arhimondr

[arhimondr]

One option - maybe good enough would be to assume that minTargetPartitionSizeInBytes satisfies this requirement and then ensure that calculated targetPartitionSizeInBytes is a multiply of minTargetPartitionSizeInBytes.

I was also thinking about something along these lines, basically try to round to the closest minTargetPartitionSizeInBytes

[linzebing]

Updated

[losipiuk]

LGTM % comments.
@arhimondr PTAL

[arhimondr]

Let's discuss what would be good defaults offline

[linzebing]

Changed defaults to:

    private int faultTolerantExecutionArbitraryDistributionComputeTaskTargetSizeGrowthPeriod = 64;
    private double faultTolerantExecutionArbitraryDistributionComputeTaskTargetSizeGrowthFactor = 1.2;
    private DataSize faultTolerantExecutionArbitraryDistributionComputeTaskTargetSizeMin = DataSize.of(512, MEGABYTE);
    private DataSize faultTolerantExecutionArbitraryDistributionComputeTaskTargetSizeMax = DataSize.of(50, GIGABYTE);

    private int faultTolerantExecutionArbitraryDistributionWriteTaskTargetSizeGrowthPeriod = 64;
    private double faultTolerantExecutionArbitraryDistributionWriteTaskTargetSizeGrowthFactor = 1.2;
    private DataSize faultTolerantExecutionArbitraryDistributionWriteTaskTargetSizeMin = DataSize.of(4, GIGABYTE);
    private DataSize faultTolerantExecutionArbitraryDistributionWriteTaskTargetSizeMax = DataSize.of(50, GIGABYTE);

    private DataSize faultTolerantExecutionHashDistributionComputeTaskTargetSize = DataSize.of(512, MEGABYTE);
    private DataSize faultTolerantExecutionHashDistributionWriteTaskTargetSize = DataSize.of(4, GIGABYTE);
    private int faultTolerantExecutionHashDistributionWriteTaskTargetMaxCount = 2000;

let me know your thoughts

[arhimondr]

One option - maybe good enough would be to assume that minTargetPartitionSizeInBytes satisfies this requirement and then ensure that calculated targetPartitionSizeInBytes is a multiply of minTargetPartitionSizeInBytes.

I was also thinking about something along these lines, basically try to round to the closest minTargetPartitionSizeInBytes

[losipiuk]

One option - maybe good enough would be to assume that minTargetPartitionSizeInBytes satisfies this requirement and then ensure that calculated targetPartitionSizeInBytes is a multiply of minTargetPartitionSizeInBytes.

I was also thinking about something along these lines, basically try to round to the closest minTargetPartitionSizeInBytes

This would not guarantee that we are actually having number of splits per task which is multiply of task.max-drivers-per-task. Probably explicitly counting splits is better.
Let's address that in separate PR. @linzebing you said you will create tracking issue. Did you maybe?

[linzebing]

Created #16805 and #16806

[linzebing]

@jhlodin : can you (or someone else) help me update documentation for this PR?