DDL Parsing Error When using multiple quotes

[leesenlen]

Describe the bug(Please use English)

MySQLCDC, DataStreamAPI

DDL Parsing Error When using multiple quotes ( '....' ) after DEFAULT VALUE

Expected 0 as default value , but get string 0'COLLATE'utf8_general_ci

DDL SQL:
create table test {
...
is_delete TINYINT DEFAULT VALUE '0' COLLATE 'utf8_general_ci'
...
}

ExceptionTrace：
Caused by: java.lang.NumberFormatException: For input string: "0'COLLATE'utf8_general_ci"
at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
at java.lang.Integer.parseInt(Integer.java:580)
at java.lang.Short.parseShort(Short.java:118)
at java.lang.Short.valueOf(Short.java:174)
at java.lang.Short.valueOf(Short.java:200)
at io.debezium.jdbc.JdbcValueConverters.lambda$convertSmallInt$40(JdbcValueConverters.java:865)
at io.debezium.jdbc.JdbcValueConverters.convertValue(JdbcValueConverters.java:1288)
at io.debezium.jdbc.JdbcValueConverters.convertSmallInt(JdbcValueConverters.java:853)
at io.debezium.jdbc.JdbcValueConverters.lambda$converter$4(JdbcValueConverters.java:293)
at io.debezium.connector.mysql.MySqlDefaultValueConverter.setColumnDefaultValue(MySqlDefaultValueConverter.java:405)
at io.debezium.connector.mysql.antlr.listener.CreateTableParserListener.convertDefaultValueToSchemaType(CreateTableParserListener.java:147)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
at io.debezium.connector.mysql.antlr.listener.CreateTableParserListener.lambda$exitColumnCreateTable$1(CreateTableParserListener.java:75)
at io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser.runIfNotNull(MySqlAntlrDdlParser.java:357)
at io.debezium.connector.mysql.antlr.listener.CreateTableParserListener.exitColumnCreateTable(CreateTableParserListener.java:55)
at io.debezium.ddl.parser.mysql.generated.MySqlParser$ColumnCreateTableContext.exitRule(MySqlParser.java:4733)
at io.debezium.antlr.ProxyParseTreeListenerUtil.delegateExitRule(ProxyParseTreeListenerUtil.java:64)
at io.debezium.connector.mysql.antlr.listener.MySqlAntlrDdlParserListener.exitEveryRule(MySqlAntlrDdlParserListener.java:106)
at com.ververica.cdc.connectors.shaded.org.antlr.v4.runtime.tree.ParseTreeWalker.exitRule(ParseTreeWalker.java:48)
at com.ververica.cdc.connectors.shaded.org.antlr.v4.runtime.tree.ParseTreeWalker.walk(ParseTreeWalker.java:30)
at com.ververica.cdc.connectors.shaded.org.antlr.v4.runtime.tree.ParseTreeWalker.walk(ParseTreeWalker.java:28)
at com.ververica.cdc.connectors.shaded.org.antlr.v4.runtime.tree.ParseTreeWalker.walk(ParseTreeWalker.java:28)
at com.ververica.cdc.connectors.shaded.org.antlr.v4.runtime.tree.ParseTreeWalker.walk(ParseTreeWalker.java:28)
at com.ververica.cdc.connectors.shaded.org.antlr.v4.runtime.tree.ParseTreeWalker.walk(ParseTreeWalker.java:28)
at io.debezium.antlr.AntlrDdlParser.parse(AntlrDdlParser.java:87)
at io.debezium.connector.mysql.MySqlDatabaseSchema.parseDdl(MySqlDatabaseSchema.java:213)
at io.debezium.connector.mysql.MySqlDatabaseSchema.parseStreamingDdl(MySqlDatabaseSchema.java:200)
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.handleQueryEvent(MySqlStreamingChangeEventSource.java:574)
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.handleEvent(MySqlStreamingChangeEventSource.java:352)

Environment :

• Flink version : 1.14
• Flink CDC version: 2.2
• Database and version: MySQL 5.7

To Reproduce
Steps to reproduce the behavior:

1. The test data :
2. The test code :
3. The error :

Additional Description
If applicable, add screenshots to help explain your problem.

[leonardBang]

@ruanhang1993 Would you like to take a look this issue?

[ruanhang1993]

OK, I will take a look at this problem.

[ruanhang1993]

@leesenlen I have written a test about this, but this error does not occur. Could you help to check whether there is something wrong in my test.
I used the branch release-2.2 and the sql and test as following.
SQL:

CREATE TABLE self_customers1 (
  id INTEGER NOT NULL PRIMARY KEY,
  name VARCHAR(255) NOT NULL DEFAULT 'flink',
  is_delete TINYINT DEFAULT '0' COLLATE 'utf8_general_ci'
);


INSERT INTO self_customers1
VALUES (101,"user_1",1),
       (102,"user_2",0);

test in MySqlSourceITCase:

    @Test
    public void testDefaultColumn1() throws Exception {
        customDatabase.createAndInitialize();
        System.out.println(MYSQL_CONTAINER.getDatabasePort() + "/" + customDatabase.getDatabaseName());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(200L);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));

        // The sleeping source will sleep awhile after send per record
        MySqlSource<RowData> sleepingSource = buildSource();
        DataStreamSource<RowData> source =
                env.fromSource(sleepingSource, WatermarkStrategy.noWatermarks(), "selfSource");
        TypeSerializer<RowData> serializer =
                source.getTransformation().getOutputType().createSerializer(env.getConfig());
        String accumulatorName = "dataStreamCollect_" + UUID.randomUUID();
        CollectSinkOperatorFactory<RowData> factory =
                new CollectSinkOperatorFactory(serializer, accumulatorName);
        CollectSinkOperator<RowData> operator = (CollectSinkOperator) factory.getOperator();
        CollectResultIterator<RowData> iterator =
                new CollectResultIterator(
                        operator.getOperatorIdFuture(),
                        serializer,
                        accumulatorName,
                        env.getCheckpointConfig());
        CollectStreamSink<RowData> sink = new CollectStreamSink(source, factory);
        sink.name("Data stream collect sink");
        env.addOperator(sink.getTransformation());
        JobClient jobClient = env.executeAsync("snapshotSplitTest");
        iterator.setJobClient(jobClient);
        while (iterator.hasNext()) {
            int i = 11;
        }
    }

    private MySqlSource<RowData> buildSource() {
        ResolvedSchema physicalSchema =
                new ResolvedSchema(
                        Arrays.asList(
                                Column.physical("id", DataTypes.BIGINT().notNull()),
                                Column.physical("name", DataTypes.STRING()),
                                Column.physical("is_delete", DataTypes.TINYINT())),
                        new ArrayList<>(),
                        UniqueConstraint.primaryKey("pk", Collections.singletonList("id")));
        RowType physicalDataType =
                (RowType) physicalSchema.toPhysicalRowDataType().getLogicalType();
        MetadataConverter[] metadataConverters = new MetadataConverter[0];
        final TypeInformation<RowData> typeInfo = InternalTypeInfo.of(physicalDataType);

        RowDataDebeziumDeserializeSchema deserializer =
                RowDataDebeziumDeserializeSchema.newBuilder()
                        .setPhysicalRowType(physicalDataType)
                        .setMetadataConverters(metadataConverters)
                        .setResultTypeInfo(typeInfo)
                        .setServerTimeZone(ZoneId.of("UTC"))
                        .setUserDefinedConverterFactory(
                                MySqlDeserializationConverterFactory.instance())
                        .build();
        return MySqlSource.<RowData>builder()
                .hostname(MYSQL_CONTAINER.getHost())
                .port(MYSQL_CONTAINER.getDatabasePort())
                .databaseList(customDatabase.getDatabaseName())
                .tableList(customDatabase.getDatabaseName() + ".self_customers1")
                .username(customDatabase.getUsername())
                .password(customDatabase.getPassword())
                .serverTimeZone("UTC")
                .serverId(getServerId())
                .splitSize(8096)
                .splitMetaGroupSize(1000)
                .distributionFactorUpper(1000.0d)
                .distributionFactorLower(0.05d)
                .fetchSize(1024)
                .connectTimeout(Duration.ofSeconds(30))
                .connectMaxRetries(3)
                .connectionPoolSize(20)
                .debeziumProperties(new Properties())
                .startupOptions(StartupOptions.initial())
                .deserializer(deserializer)
                .scanNewlyAddedTableEnabled(false)
                .jdbcProperties(new Properties())
                .heartbeatInterval(Duration.ofSeconds(30))
                .build();
    }

After the test running, I alter the table using the sql.

alter table self_customers1 add column `state`  smallint(1) unsigned DEFAULT 0 COMMENT '禁用';

But my job does not throw an exception.

Please correct my mistake, Thanks~

[leesenlen]

@ruanhang1993

Try the ddl below

alter table self_customers1 add column `state_new`  smallint(1) DEFAULT '0' COLLATE 'utf8_general_ci' COMMENT '禁用';

This is happend when alter table with COLLATE after DEFAULT value with quotes(')

[ruanhang1993]

The default value is wrong when parsing the alter sql. Maybe it is a bug in the debezium. I will find a way to fix it.

[leesenlen]

@ruanhang1993 Thanks for helping.
Is there a possible timeline for this problem?
It often encounters in our situation. We are looking for a solution.

[leonardBang]

fixed in #1522

[ruanhang1993]

@leonardBang This is another problem that has not be fixed. We should reopen it.

[ruanhang1993]

@ruanhang1993 Thanks for helping. Is there a possible timeline for this problem? It often encounters in our situation. We are looking for a solution.

Maybe you could cherry pick this PR as a quick fix version.