[AMORO-1812] Support spark-based external optimizer

[tcodehuber]

Why are the changes needed?

Close #1812 .

Brief change log

How was this patch tested?

• Add some test cases that check the changes thoroughly including negative and positive cases if possible

• Add screenshots for manual tests if appropriate

• Run test locally before making a pull request

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

[tcodehuber]

@wangtaohz @baiyangtx Can you give me some suggestions about this PR, thx?

[codecov]

Codecov Report

Attention: 5 lines in your changes are missing coverage. Please review.

Comparison is base (c2bdb64) 32.23% compared to head (2918ca9) 32.37%.
Report is 4 commits behind head on master.

Files	Patch %	Lines
...ase/arctic/optimizer/common/OptimizerExecutor.java	81.48%	4 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #2421      +/-   ##
============================================
+ Coverage     32.23%   32.37%   +0.14%     
- Complexity     4383     4403      +20     
============================================
  Files           589      590       +1     
  Lines         49892    50041     +149     
  Branches       6618     6619       +1     
============================================
+ Hits          16081    16200     +119     
- Misses        32539    32579      +40     
+ Partials       1272     1262      -10     

Flag	Coverage Δ
core	30.50% <87.50%> (+0.22%)	⬆️
trino	50.38% <ø> (-0.56%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[tcodehuber]

@baiyangtx Can you review it again?

[baiyangtx]

Recently, I discussed this PR with @zhoujinsong and overall agree with its implementation approach. However, I hope we can reuse the code in the Optimizer/Common package as much as possible. The refactoring should follow the logic below:

1. Have SparkOptimizer extend the common.Optimizer class to reuse its OptimizerToucher and OptimizerExecutor management capabilities.

2. Modify the part of the Optimizer constructor that initializes the executor array to use a protected newOptimizerExecutor() method instead of directly creating a new OptimizerExecutor object.

3. Implement the SparkOptimizerExecutor class under the Spark module, which extends the common.OptimizerExecutor class type and overrides the private OptimizingTaskResult executeTask(OptimizingTask task) method.

4. In the overridden OptimizingTaskResult executeTask(OptimizingTask task) method of SparkOptimizerExecutor, implement the logic currently found in SparkOptimizingTaskSubmitter.submitOptimizingTask.

In this way, we can reuse the code in the Optimizer/Common module as much as possible, and in the future, we can also implement an optimizer based on Flink batch in the Flink session using this framework.

[tcodehuber]

Recently, I discussed this PR with @zhoujinsong and overall agree with its implementation approach. However, I hope we can reuse the code in the Optimizer/Common package as much as possible. The refactoring should follow the logic below:

1. Have SparkOptimizer extend the common.Optimizer class to reuse its OptimizerToucher and OptimizerExecutor management capabilities.
2. Modify the part of the Optimizer constructor that initializes the executor array to use a protected newOptimizerExecutor() method instead of directly creating a new OptimizerExecutor object.
3. Implement the SparkOptimizerExecutor class under the Spark module, which extends the common.OptimizerExecutor class type and overrides the private OptimizingTaskResult executeTask(OptimizingTask task) method.
4. In the overridden OptimizingTaskResult executeTask(OptimizingTask task) method of SparkOptimizerExecutor, implement the logic currently found in SparkOptimizingTaskSubmitter.submitOptimizingTask.

In this way, we can reuse the code in the Optimizer/Common module as much as possible, and in the future, we can also implement an optimizer based on Flink batch in the Flink session using this framework.

OK, I will try to work on it later.

[baiyangtx]

There is no need to develop separate logic for the active exit of the Spark Optimizer. Therefore, the related code in this area should be unnecessary. I have already commented it out, and the other parts should be fine.

[zhoujinsong]

@tcodehuber All codes look fine to me expect a small issue should be fixed.

[zhoujinsong]

LGTM.
Validated in my local environment.