[FLINK-35014] SqlNode to operation conversion for models

[lihaosky]

What is the purpose of the change

Add SqlNode to SqlOperation conversion for models

Brief change log

• Support Model CRUD node to operation conversion

Verifying this change

This change added tests and can be verified as follows:

• Added unit test and integration test in TableEnvironmentTest

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): (yes)
• 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? (JavaDocs)

[lihaosky]

cc @twalthr to review. Thanks!

[flinkbot]

CI report:

• 92fa1f1 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[lihaosky]

2024-12-20T21:31:47.4797792Z Dec 20 21:31:47 21:31:47.478 [INFO] Running org.apache.flink.table.planner.plan.rules.physical.stream.WatermarkAssignerChangelogNormalizeTransposeRuleTest
2024-12-20T21:31:47.7285003Z Dec 20 21:31:47 21:31:47.726 [ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 2.865 s <<< FAILURE! -- in org.apache.flink.architecture.rules.BanJunit4Rules
2024-12-20T21:31:47.7287086Z Dec 20 21:31:47 21:31:47.726 [ERROR] BanJunit4Rules.NO_NEW_ADDED_JUNIT4_TEST_RULE -- Time elapsed: 2.860 s <<< FAILURE!
2024-12-20T21:31:47.7288102Z Dec 20 21:31:47 java.lang.AssertionError: 
2024-12-20T21:31:47.7289605Z Dec 20 21:31:47 Architecture Violation [Priority: MEDIUM] - Rule 'Junit4 is forbidden, please use Junit5 instead' was violated (2 times):
2024-12-20T21:31:47.7291813Z Dec 20 21:31:47 Method <org.apache.flink.table.planner.operations.SqlDdlToOperationConverterTest.testCreateModel()> calls method <org.junit.Assert.assertNotNull(java.lang.Object)> in (SqlDdlToOperationConverterTest.java:296)
2024-12-20T21:31:47.7294137Z Dec 20 21:31:47 Method <org.apache.flink.table.planner.operations.SqlDdlToOperationConverterTest.testCreateModel()> calls method <org.junit.Assert.assertNotNull(java.lang.Object)> in (SqlDdlToOperationConverterTest.java:303)
2024-12-20T21:31:47.7295939Z Dec 20 21:31:47 	at com.tngtech.archunit.lang.ArchRule$Assertions.assertNoViolation(ArchRule.java:94)
2024-12-20T21:31:47.7297056Z Dec 20 21:31:47 	at com.tngtech.archunit.lang.ArchRule$Assertions.check(ArchRule.java:86)
2024-12-20T21:31:47.7298210Z Dec 20 21:31:47 	at com.tngtech.archunit.library.freeze.FreezingArchRule.check(FreezingArchRule.java:97)
2024-12-20T21:31:47.7301989Z Dec 20 21:31:47 	at com.tngtech.archunit.junit.internal.ArchUnitTestDescriptor$ArchUnitRuleDescriptor.execute(ArchUnitTestDescriptor.java:166)
2024-12-20T21:31:47.7303496Z Dec 20 21:31:47 	at com.tngtech.archunit.junit.internal.ArchUnitTestDescriptor$ArchUnitRuleDescriptor.execute(ArchUnitTestDescriptor.java:149)
2024-12-20T21:31:47.7317722Z Dec 20 21:31:47 	at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
2024-12-20T21:31:47.7319141Z Dec 20 21:31:47 	at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
2024-12-20T21:31:47.7320302Z Dec 20 21:31:47 	at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)