[Enhancement] support incremental scan ranges deployment at FE side

[dirtysalt]

Why I'm doing:

What I'm doing:

CoordinatorPreprocessor

Added assignIncrementalScanRangesToFragmentInstances interface, which recalculates the scan ranges of all fragment instances below a fragment.

Currently the distribution of node2ScanRanges is supported, but more complex data distributions such as node + driver_seq -> scan range are not supported for incremental deployment.

ExecutionSchedule

We currently support two scheduling methods: all-at-once and phased. The difference is that they schedule different sets of fragment instances.

The modification to the code is that after each collection of fragment instances is scheduled, it immediately checks to see if there are any scan nodes left in the fragment instances that have not yet been scheduled. If there are, the scheduling process will continue.

The method used for subsequent scheduling is coordinator.assignIncrementalScanRangesToDeployStates(deployer, states);

Coordinator

Adds the assignIncrementalScanRangesToDeployStates interface, which checks the scan node for more scan ranges based on the deploy states (i.e., the set of locally dispatched fragment instances).

• Checks all the following fragment instances against each deploy state.
• According to the fragment instances corresponding to these fragment instances, see if there are still scan nodes with scan ranges.
• If so, call coordinator preprocess to recalculate node2ScanRanges under fragment instances.
• Based on these fragment instances, regenerate the rpc request, and put it under the deploy state.

Arch Diagram

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

[sonarcloud]

Quality Gate passed

Issues
14 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarCloud

[github-actions]

[BE Incremental Coverage Report]

✅ pass : 0 / 0 (0%)

[github-actions]

[Java-Extensions Incremental Coverage Report]

✅ pass : 0 / 0 (0%)

[github-actions]

[FE Incremental Coverage Report]

✅ pass : 158 / 163 (96.93%)

file detail

	path	covered_line	new_line	coverage	not_covered_line_detail
🔵	com/starrocks/planner/ScanNode.java	0	1	00.00%	[104]
🔵	com/starrocks/qe/scheduler/Coordinator.java	0	1	00.00%	[136]
🔵	com/starrocks/qe/HDFSBackendSelector.java	34	36	94.44%	[260, 267]
🔵	com/starrocks/qe/DefaultCoordinator.java	44	45	97.78%	[700]
🔵	com/starrocks/qe/scheduler/TFragmentInstanceFactory.java	8	8	100.00%	[]
🔵	com/starrocks/qe/SessionVariable.java	8	8	100.00%	[]
🔵	com/starrocks/qe/scheduler/dag/FragmentInstanceExecState.java	6	6	100.00%	[]
🔵	com/starrocks/qe/scheduler/Deployer.java	2	2	100.00%	[]
🔵	com/starrocks/qe/scheduler/dag/PhasedExecutionSchedule.java	9	9	100.00%	[]
🔵	com/starrocks/qe/scheduler/dag/AllAtOnceExecutionSchedule.java	10	10	100.00%	[]
🔵	com/starrocks/qe/CoordinatorPreprocessor.java	6	6	100.00%	[]
🔵	com/starrocks/qe/scheduler/dag/JobSpec.java	4	4	100.00%	[]
🔵	com/starrocks/qe/scheduler/assignment/FragmentAssignmentStrategyFactory.java	1	1	100.00%	[]
🔵	com/starrocks/qe/scheduler/assignment/BackendSelectorFactory.java	7	7	100.00%	[]
🔵	com/starrocks/qe/scheduler/assignment/LocalFragmentAssignmentStrategy.java	15	15	100.00%	[]
🔵	com/starrocks/planner/HdfsScanNode.java	4	4	100.00%	[]

[stephen-shelby]

I will get this error in the following session variables.

[dirtysalt]

I will get this error in the following session variables.

if you want to test, you can test this PR [Enhancement] support incremental scan ranges deployment at BE side by dirtysalt · Pull Request #50254 · StarRocks/starrocks
#50254

there need to be some modification on BE side too.

[Youngwb]

Could you do not send empty scan range to all BE, maybe we can send to the incremental scan range to the worker which already has fragment instance before?

[Youngwb]

How do you deal with the incremental partition in hiveTable toThfift

[dirtysalt]

How do you deal with the incremental partition in hiveTable toThfift

Don't handle it in this PR. for hive scan node, there is no need to handle incremental partition.

[dirtysalt]

Could you do not send empty scan range to all BE, maybe we can send to the incremental scan range to the worker which already has fragment instance before?

Yes. It will be more complicated, we can optimize in next PR. We can

• add a extra flag to fragment instances which means it already has fragment plan.
• in the end, if there is no more scan range, we have to send a empty scan range to fragment instance to notify it "all is done".

anyway, it's just an optimization.

[dirtysalt]

@Youngwb

I don't think the placeholder is good name, how about is_empty ?

I can not reply this comment. No problem, I can fix that.

[ZiheLiu]

Better to move to JobSpec::Builder::commonProperties and not expose setIncrementalScanRanges.

        public Builder commonProperties(ConnectContext context) {
            TWorkGroup newResourceGroup = prepareResourceGroup(
                    context, ResourceGroupClassifier.QueryType.fromTQueryType(instance.queryOptions.getQuery_type()));
            this.resourceGroup(newResourceGroup);

            this.enablePipeline(isEnablePipeline(context, instance.fragments));
            instance.connectContext = context;

            instance.enableQueue = isEnableQueue(context);
            instance.needQueued = needCheckQueue();
            instance.enableGroupLevelQueue = instance.enableQueue && GlobalVariable.isEnableGroupLevelQueryQueue();
            
+          instance.incrementalScanRanges =  connectContextgetSessionVariable().isEnableConnectorIncrementalScanRanges();           

            return this;
        }