[
https://issues.apache.org/jira/browse/FLINK-38139?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18015564#comment-18015564
]
TemiChan edited comment on FLINK-38139 at 8/22/25 2:30 AM:
-----------------------------------------------------------
During the process of adding a field to the source table in CDC3.3 using pt-osc
or gh-ost, if data is inserted into the source table, the error "column size
does not match the data size" will be reported by CDC. However, if there is no
data change in the source table during the online field addition process, and
the data change is made to the source table after the field addition is
completed by the tool, no error will be reported.
1.while true insert data
{{while}} {{{}true{}}}{{{};{}}}{{{}do{}}} {{mysql --login-path=dba doris_sync
-Be }}{{{}"insert into test(ptest21,tcc10)select ptest21,tcc10 from
emp_balance_revoke_record limit 10"{}}}{{{};done;{}}}
{{2.add column}}
{{{}pt-online-schema-change -{-}charset=utf8mb4 --recursion-method=none
--no-version-check --skip-check-slave-lag --user=root
--password={-}{}}}{{{}"xx"{}}} {{-host=127.0.0.1-}}
{{{}u=chenqimi,P={}}}{{{}3006{}}}{{{},D=doris_sync,t={}}}{{{}test{}}}{{{},A=utf8mb4
--alter {{}}}}{{{}"add column tax_code VARCHAR(30) NULL "{}}} {{-execute}}
3.flinkcdc error
{{java.lang.IllegalStateException: Column size does not match the data size}}
{{ }}{{{}at
org.apache.flink.cdc.common.utils.Preconditions.checkState(Preconditions.java:{}}}{{{}161{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serializerRecord(DorisEventSerializer.java:{}}}{{{}121{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.applyDataChangeEvent(DorisEventSerializer.java:{}}}{{{}100{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serialize(DorisEventSerializer.java:{}}}{{{}71{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serialize(DorisEventSerializer.java:{}}}{{{}49{}}}{{{}){}}}
{{ }}{{{}at
org.apache.doris.flink.sink.batch.DorisBatchWriter.write(DorisBatchWriter.java:{}}}{{{}120{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:{}}}{{{}160{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.runtime.operators.sink.DataSinkWriterOperator.processElement(DataSinkWriterOperator.java:{}}}{{{}178{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:{}}}{{{}75{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:{}}}{{{}50{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:{}}}{{{}29{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:{}}}{{{}38{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:{}}}{{{}238{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:{}}}{{{}157{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:{}}}{{{}114{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:{}}}{{{}65{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:{}}}{{{}579{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:{}}}{{{}231{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:{}}}{{{}909{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:{}}}{{{}858{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:{}}}{{{}958{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:{}}}{{{}937{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:{}}}{{{}751{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:{}}}{{{}566{}}}{{{}){}}}
{{ }}{{{}at java.lang.Thread.run(Thread.java:{}}}{{{}748{}}}{{{}){}}}
{{}}
was (Author: JIRAUSER310778):
During the process of adding a field to the source table in CDC3.3 using pt-osc
or gh-ost, if data is inserted into the source table, the error "column size
does not match the data size" will be reported by CDC. However, if there is no
data change in the source table during the online field addition process, and
the data change is made to the source table after the field addition is
completed by the tool, no error will be reported.
1.while true insert data
{{while}} {{{}true{}}}{{{};{}}}{{{}do{}}} {{mysql --login-path=dba doris_sync
-Be }}{{{}"insert into test(ptest21,tcc10)select ptest21,tcc10 from
emp_balance_revoke_record limit 10"{}}}{{{};done;{}}}
{{}}
{{2.add column}}
{{{}pt-online-schema-change --charset=utf8mb4 --recursion-method=none
--no-version-check --skip-check-slave-lag --user=root
--password={}}}{{{}"xx"{}}} {{--host=127.0.0.1}}
{{{}u=chenqimi,P={}}}{{{}3006{}}}{{{},D=doris_sync,t={}}}{{{}test{}}}{{{},A=utf8mb4
--alter {}}}{{"add column tax_code VARCHAR(30) NULL "}} {{--execute}}
{{}}
3.flinkcdc error
{{java.lang.IllegalStateException: Column size does not match the data size}}
{{ }}{{{}at
org.apache.flink.cdc.common.utils.Preconditions.checkState(Preconditions.java:{}}}{{{}161{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serializerRecord(DorisEventSerializer.java:{}}}{{{}121{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.applyDataChangeEvent(DorisEventSerializer.java:{}}}{{{}100{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serialize(DorisEventSerializer.java:{}}}{{{}71{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serialize(DorisEventSerializer.java:{}}}{{{}49{}}}{{{}){}}}
{{ }}{{{}at
org.apache.doris.flink.sink.batch.DorisBatchWriter.write(DorisBatchWriter.java:{}}}{{{}120{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:{}}}{{{}160{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.cdc.runtime.operators.sink.DataSinkWriterOperator.processElement(DataSinkWriterOperator.java:{}}}{{{}178{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:{}}}{{{}75{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:{}}}{{{}50{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:{}}}{{{}29{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:{}}}{{{}38{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:{}}}{{{}238{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:{}}}{{{}157{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:{}}}{{{}114{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:{}}}{{{}65{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:{}}}{{{}579{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:{}}}{{{}231{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:{}}}{{{}909{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:{}}}{{{}858{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:{}}}{{{}958{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:{}}}{{{}937{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:{}}}{{{}751{}}}{{{}){}}}
{{ }}{{{}at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:{}}}{{{}566{}}}{{{}){}}}
{{ }}{{{}at java.lang.Thread.run(Thread.java:{}}}{{{}748{}}}{{{}){}}}
{{}}
> Schema changes still caused issues after enabling
> scan.parse.online.schema.changes.enabled
> ------------------------------------------------------------------------------------------
>
> Key: FLINK-38139
> URL: https://issues.apache.org/jira/browse/FLINK-38139
> Project: Flink
> Issue Type: Improvement
> Components: Flink CDC
> Affects Versions: cdc-3.4.0
> Reporter: Yanquan Lv
> Assignee: Yanquan Lv
> Priority: Major
> Labels: pull-request-available
>
> I am currently using version 3.4.0 of FlinkCDC.
> I have a synchronization job with MySQL as the source table. After
> configuring scan.parse.online.script.changes.enabled to true, I made changes
> to the table structure (using the pt osc tool), but the job still reported an
> error.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
