[Refactor] support incremental scan ranges deployment

[dirtysalt]

Why I'm doing:

What I'm doing:

Fixes #50196

What type of PR is this:

• BugFix
• Feature
• Enhancement
• Refactor
• UT
• Doc
• Tool

Does this PR entail a change in behavior?

• Yes, this PR will result in a change in behavior.
• No, this PR will not result in a change in behavior.

If yes, please specify the type of change:

• Interface/UI changes: syntax, type conversion, expression evaluation, display information
• Parameter changes: default values, similar parameters but with different default values
• Policy changes: use new policy to replace old one, functionality automatically enabled
• Feature removed
• Miscellaneous: upgrade & downgrade compatibility, etc.

Checklist:

• I have added test cases for my bug fix or my new feature
• This pr needs user documentation (for new or modified features or behaviors)
  • I have added documentation for my new feature or new function
• This is a backport pr

Bugfix cherry-pick branch check:

• I have checked the version labels which the pr will be auto-backported to the target branch
  • 3.3
  • 3.2
  • 3.1
  • 3.0
  • 2.5

[mergify]

⚠️ The sha of the head commit of this PR conflicts with #50189. Mergify cannot evaluate rules on this PR. ⚠️

[starrocks-cr]

The most risky bug in this code is:
Incorrect handling of while (true) loop when deployState becomes empty, leading to a potential infinite loop.

You can modify the code like this:

    while (deployState != null && !deployState.isEmpty()) {
        deployState = coordinator.assignIncrementalScanRangesToDeployStates(deployer, deployState);
        for (DeployState state : deployState) {
            deployer.deployFragments(state);
        }
    }

[starrocks-cr]

The most risky bug in this code is:
List.of() creates an immutable list, which may cause issues if later modifications are attempted on the scan ranges.

You can modify the code like this:

@@ -239,7 +245,20 @@ private void assignScanRangesToNormalFragmentInstancePerWorker(ExecutionFragment
         final int parallelExecInstanceNum = fragment.getParallelExecNum();
         final int pipelineDop = fragment.getPipelineDop();
 
-        execFragment.getScanRangeAssignment().forEach((workerId, scanRangesPerWorker) -> {
+        FragmentScanRangeAssignment assignment = execFragment.getScanRangeAssignment();
+
+        if (useIncrementalScanRanges) {
+            for (ScanNode scanNode : execFragment.getScanNodes()) {
+                if (scanNode.isIncrementalScanRangesSupported()) {
+                    // TODO(yan): put some scan range for ending.
+                    for (ComputeNode computeNode : workerProvider.getAllWorkers()) {
+                        assignment.putAll(computeNode.getId(), scanNode.getId().asInt(), new ArrayList<>());
+                    }
+                }
+            }
+        }
+
+        assignment.forEach((workerId, scanRangesPerWorker) -> {
             // 1. Handle normal scan node firstly
             scanRangesPerWorker.forEach((scanId, scanRangesOfNode) -> {
                 if (replicatedScanIds.contains(scanId)) {

Replace List.of() with new ArrayList<>() to create a mutable list.

[starrocks-cr]

The most risky bug in this code is:
The assignIncrementalScanRangesToDeployStates method could produce IndexOutOfBoundsException.

You can modify the code like this:

@Override
public List<DeployState> assignIncrementalScanRangesToDeployStates(Deployer deployer, List<DeployState> deployStates)
        throws UserException {
    List<DeployState> updatedStates = new ArrayList<>();
    if (!jobSpec.isIncrementalScanRanges()) {
        return updatedStates;
    }
    for (DeployState state : deployStates) {

        Set<PlanFragmentId> planFragmentIds = new HashSet<>();
        for (List<FragmentInstanceExecState> fragmentInstanceExecStates : state.getThreeStageExecutionsToDeploy()) {
            for (FragmentInstanceExecState execState : fragmentInstanceExecStates) {
                planFragmentIds.add(execState.getFragmentId());
            }
        }

        Set<PlanFragmentId> updatedPlanFragmentIds = new HashSet<>();
        for (PlanFragmentId fragmentId : planFragmentIds) {
            boolean hasMoreScanRanges = false;
            ExecutionFragment fragment = executionDAG.getFragment(fragmentId);
            for (ScanNode scanNode : fragment.getScanNodes()) {
                if (scanNode.hasMoreScanRanges()) {
                    hasMoreScanRanges = true;
                }
            }
            if (hasMoreScanRanges) {
                coordinatorPreprocessor.assignIncrementalScanRangesToFragmentInstances(fragment);
                updatedPlanFragmentIds.add(fragmentId);
            }
        }

        if (updatedPlanFragmentIds.isEmpty()) {
            continue;
        }

        DeployState newState = new DeployState();
        updatedStates.add(newState);
        int index = 0;
        for (List<FragmentInstanceExecState> fragmentInstanceExecStates : state.getThreeStageExecutionsToDeploy()) {
            // Ensure there are enough elements
            while (newState.getThreeStageExecutionsToDeploy().size() <= index) {
                newState.getThreeStageExecutionsToDeploy().add(new ArrayList<>());
            }
            List<FragmentInstanceExecState> res = newState.getThreeStageExecutionsToDeploy().get(index);
            index += 1;
            for (FragmentInstanceExecState execState : fragmentInstanceExecStates) {
                if (!updatedPlanFragmentIds.contains(execState.getFragmentId())) {
                    continue;
                }
                FragmentInstance instance = execState.getFragmentInstance();
                TExecPlanFragmentParams request = deployer.createIncrementalScanRangesRequest(instance);
                execState.setRequestToDeploy(request);
                res.add(execState);
            }
        }
    }
    return updatedStates;
}