Implement adaptive remote task request size

[yansun7]

Currently HttpRemotTask will send all pending splits in one taskUpdateRequest, sometimes causing large request for each task in coordinator.

This change will enable the ability to split large taskUpdateRequest into several smaller batch dynamically to send.

Introduced 4 parameters regarding to this feature:

query.remote-task.enable-adaptive-request-size

• session: enable_adaptive_remote_task_request_size
• to control the enable/disable of the feature

query.remote-task.max-request-size

• session: max_remote_task_request_size
• the max size of taskUpdateRequest

query.remote-task.request-size-headroom

• session: remote_task_request_size_headroom
• the expected mean value of the taskUpdateRequest will be between (max_remote_task_request_size - remote_task_request_size_headroom) and (max_remote_task_request_size)

query.remote-task.guaranteed-splits-per-task

• session: remote_task_guaranteed_splits_per_request
• guaranteed splits that will be sent in taskUpdateRequest for each task to prevent deadlock, regardless of max_remote_task_request_size

[arhimondr]

sometimes causing large request for each task in coordinator.

How does this problem manifest itself? Could you please elaborate more on the problem you are trying to solve?

[JunhyungSong]

@arhimondr
The problematic situation:
Table has 3884 columns.
Table schema has a “parameters” field with the whole table schema string(~ 300k characters)
Size of each split object(in task update request) is big due to large strings in table properties.
If taskUpdateRequest contains a lot of huge splits like above and send it to the worker through http, it causes coordinator crash due to out of memory.
So, this PR tried to limit the size of taskUpdateRequest not only using max_unacknowledged_splits_per_task, but also using max_remote_task_request_size.

@yansun7 can you fix the merge conflicts, please?

[JunhyungSong]

Why isn't it 'true'?

[JunhyungSong]

There could be lots of splits. In this case, there will be lots of map accesses. It would be better to do like the below,

            .map(source -> {
                long currentMaxAcknowledgedSplit = maxAcknowledgedSplitByPlanNode.getOrDefault(source.getPlanNodeId(), Long.MIN_VALUE);
                return new TaskSource(
                        source.getPlanNodeId(),
                        source.getSplits().stream()
                                .filter(scheduledSplit -> scheduledSplit.getSequenceId() > currentMaxAcknowledgedSplit)
                                .collect(Collectors.toSet()),
                        // Like splits, noMoreSplitsForLifespan could be pruned so that only new items will be processed.
                        // This is not happening here because correctness won't be compromised due to duplicate events for noMoreSplitsForLifespan.
                        source.getNoMoreSplitsForLifespan(),
                        source.isNoMoreSplits());
            })

[JunhyungSong]

When taskSource.getSplits().size() == 0, it can 'continue;'.

[JunhyungSong]

If taskSource.getSplits().size() == 0 is already checked, .orElse(Long.MIN_VALUE) is not needed. Instead of that, just '.getAsLong()'.

[JunhyungSong]

If taskSource.getSplits().size() == 0 is already checked, it should throw when val >= maxAcknowledgedSplit.

e.g.

            maxAcknowledgedSplitByPlanNode.compute(planNodeId, (key, val) -> {
                if (val == null) {
                    return maxAcknowledgedSplit;
                }
                if (val >= maxAcknowledgedSplit) {
                    throw new IllegalStateException(format("%s - splits are out of order? planNodeId=%s, newMax=%d, currentMax=%d", taskId, planNodeId, maxAcknowledgedSplit, val));
                }
                return maxAcknowledgedSplit;
            });

[JunhyungSong]

If taskSource.getSplits().size() == 0 is already checked, this is not needed.

[JunhyungSong]

log.info?

[JunhyungSong]

It should find max only for partitioned sources.

[JunhyungSong]

With colocation joins with bucketed tables, there could be more than one partitioned sources. In that case, this logic will become awkward. Please apply this logic only when there is one partitioned source.

[bitsondatadev]

👋 @yansun7 - this PR is inactive and doesn't seem to be under development. If you'd like to continue work on this at any point in the future, feel free to re-open.