[FLINK-39196][pipeline-connector][oracle] Support change column nullable without data type

[FLINK-39196][pipeline-connector][oracle] Support change column nullable without data type

Author: zml1206

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/source/parser/OracleAlterTableParserListener.java

@@ -28,7 +28,6 @@
 import io.debezium.ddl.parser.oracle.generated.PlSqlParser;
 import io.debezium.relational.Column;
 import io.debezium.relational.ColumnEditor;
-import io.debezium.relational.Table;
 import io.debezium.relational.TableEditor;
 import io.debezium.relational.TableId;
 import io.debezium.text.ParsingException;
@@ -88,7 +87,7 @@ public OracleAlterTableParserListener(
     @Override
     public void enterAlter_table(PlSqlParser.Alter_tableContext ctx) {
         TableId tableId = new TableId(catalogName, schemaName, getTableName(ctx.tableview_name()));
-        tableEditor = Table.editor().tableId(tableId);
+        tableEditor = parser.databaseTables().editTable(tableId);
         if (tableEditor == null) {
             throw new ParsingException(
                     null,
@@ -174,7 +173,7 @@ public void enterModify_column_clauses(PlSqlParser.Modify_column_clausesContext
                     columnEditors = new ArrayList<>(columns.size());
                     for (PlSqlParser.Modify_col_propertiesContext column : columns) {
                         String columnName = getColumnName(column.column_name());
-                        Column existingColumn = Column.editor().name(columnName).create();
+                        Column existingColumn = tableEditor.create().columnWithName(columnName);
                         if (existingColumn != null) {
                             ColumnEditor columnEditor = existingColumn.edit();
                             columnDefinitionParserListener =

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java

@@ -1727,13 +1727,20 @@ private List<Event> executeAlterAndProvideExpected(TableId tableId, Statement st
 
         statement.execute(
                 String.format(
-                        "ALTER TABLE %s.products ADD DESC1 VARCHAR(45) DEFAULT NULL", "debezium"));
+                        "ALTER TABLE %s.products ADD DESC1 VARCHAR(45) DEFAULT '' NOT NULL",
+                        "debezium"));
         expected.add(
                 new AddColumnEvent(
                         tableId,
                         Collections.singletonList(
                                 new AddColumnEvent.ColumnWithPosition(
-                                        Column.physicalColumn("DESC1", DataTypes.VARCHAR(45))))));
+                                        Column.physicalColumn(
+                                                "DESC1", DataTypes.VARCHAR(45).notNull())))));
+
+        statement.execute(String.format("ALTER TABLE %s.products MODIFY DESC1 NULL", "DEBEZIUM"));
+        expected.add(
+                new AlterColumnTypeEvent(
+                        tableId, Collections.singletonMap("DESC1", DataTypes.VARCHAR(45))));
 
         statement.execute(
                 String.format(