[AMORO-1035][Flink] Support customizing mixed-format table source parallelism

[YesOrNo828]

Why are the changes needed?

Support customizing the parallelism of the filestore/logstore source via Flink SQL.

Close #1035 .

Brief change log

• update flink modules,
• update the filestore and logstore table source

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) yes
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) docs

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (bd58d1f) 32.64% compared to head (75b00ef) 30.76%.

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #1973      +/-   ##
============================================
- Coverage     32.64%   30.76%   -1.88%     
+ Complexity     4482     3896     -586     
============================================
  Files           599      553      -46     
  Lines         50321    45613    -4708     
  Branches       6691     6178     -513     
============================================
- Hits          16426    14035    -2391     
+ Misses        32578    30489    -2089     
+ Partials       1317     1089     -228     

Flag	Coverage Δ
core	30.76% <ø> (+<0.01%)	⬆️
trino	?

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.

[lklhdu]

LGTM.

[huyuanfeng2018]

Can you show the rendering of flink webui after the change? As far as I know, Flinksql has not yet supported custom source parallelism. In flinksql, all operators use the parallelism of the previous operator as the current parallelism, so After I customized the source.parallelism , I think the parallelism of subsequent tasks will not run according to the default parallelism of the task.

[YesOrNo828]

Can you show the rendering of flink webui after the change? As far as I know, Flinksql has not yet supported custom source parallelism. In flinksql, all operators use the parallelism of the previous operator as the current parallelism, so After I customized the source.parallelism , I think the parallelism of subsequent tasks will not run according to the default parallelism of the task.

@huyuanfeng2018 Thanks for your comment. Actually we could modify the source operater's parallelism through the Table API. In the ScanTableSource.getScanRuntimeProvider() method, we could modify the DataStreamSource's parallelism.

select * from 
log_data /*+OPTIONS('arctic.read.mode'='log','properties.group.id'='......','parallelism'='5' )*/ --The job parallelism is 10.

Specify source parallelism with SQL hint

[huyuanfeng2018]

Can you show the rendering of flink webui after the change? As far as I know, Flinksql has not yet supported custom source parallelism. In flinksql, all operators use the parallelism of the previous operator as the current parallelism, so After I customized the source.parallelism , I think the parallelism of subsequent tasks will not run according to the default parallelism of the task.

@huyuanfeng2018 Thanks for your comment. Actually we could modify the source operater's parallelism through the Table API. In the ScanTableSource.getScanRuntimeProvider() method, we could modify the DataStreamSource's parallelism.

```sql select * from log_data /*+OPTIONS('arctic.read.mode'='log','properties.group.id'='......','parallelism'='5' )*/ --The job parallelism is 10. ```

Specify source parallelism with SQL hint

What I am confused about is how the subsequent task changed back to 10 after the scan parallelism was changed? Let me give you an example. From the perspective of flink-table-planner, often a select statement in the source table will trigger Calc. However, the parallelism in Calc is the parallelism of the source Transformation used.:

I didn’t see this modification in this PR. I would like to know how this is done. In our internal practice, we need to make certain modifications to Flink-table-planner to achieve this effect.

[YesOrNo828]

Can you show the rendering of flink webui after the change? As far as I know, Flinksql has not yet supported custom source parallelism. In flinksql, all operators use the parallelism of the previous operator as the current parallelism, so After I customized the source.parallelism , I think the parallelism of subsequent tasks will not run according to the default parallelism of the task.

@huyuanfeng2018 Thanks for your comment. Actually we could modify the source operater's parallelism through the Table API. In the ScanTableSource.getScanRuntimeProvider() method, we could modify the DataStreamSource's parallelism.

select * from 
log_data /*+OPTIONS('arctic.read.mode'='log','properties.group.id'='......','parallelism'='5' )*/ --The job parallelism is 10.

Specify source parallelism with SQL hint

What I am confused about is how the subsequent task changed back to 10 after the scan parallelism was changed? Let me give you an example. From the perspective of flink-table-planner, often a select statement in the source table will trigger Calc. However, the parallelism in Calc is the parallelism of the source Transformation used.: I didn’t see this modification in this PR. I would like to know how this is done. In our internal practice, we need to make certain modifications to Flink-table-planner to achieve this effect.

Because the above example has a window aggregate operator, it can't be chained with the previous source and calc operators, so it will become 10 parallelism, and the source and calc operators are 5 parallelism. This PR does not change the parallelism of the source operator alone, but also the operators in its chain.

[huyuanfeng2018]

LGTM.

[YesOrNo828]

Fixed the conflicts after the spotless tool was added to the Flink module.

[CLAassistant]

All committers have signed the CLA.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[czy006]

LGTM

[zhoujinsong]

LGTM.