[FLINK-30365][table-planner] New dynamic partition pruning strategy to support more dpp patterns

[swuferhong]

What is the purpose of the change

This pr is aims to introduce a new dynamic partition pruning strategy named FlinkDynamicPartitionPruningConverterProgram to instead of the old dpp rule DynamicPartitionPruningRule to support more dpp patterns. Now, the old dpp rule is tightly coupled with the join reorder rules, which will affect the result of join reorder. At the same time, the dpp rule don't support these patterns like union node/ agg node in fact side、dim side is a sub-query etc. This pr will fix this two problem together.

Brief change log

• Introduce a new dynamic partition pruning strategy named FlinkDynamicPartitionPruningConverterProgram to instead of the old dpp rule DynamicPartitionPruningRule.
• Add more UT tests in FlinkDynamicPartitionPruningConverterProgramTest.

Verifying this change

• Add more UT tests in FlinkDynamicPartitionPruningConverterProgramTest.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? no docs

[flinkbot]

CI report:

• 9656978 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[godfreyhe]

Thanks for the contribution @swuferhong , I left some comments

[godfreyhe]

Thanks for the update, LGTM

[godfreyhe]

nit: why we need new ArrayList instance here?

[swuferhong]

nit: why we need new ArrayList instance here?

The list variable tables have an add operation in the for loop. If we don't new one new ArrayList, a concurrent modification exception will be thrown.

[godfreyhe]

It's better we can add this optimization in flink-tpcds-test module, we can do it in another pr.

[swuferhong]

It's better we can add this optimization in flink-tpcds-test module, we can do it in another pr.
Ok. I will make a jira issue to track this work.