You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "chunji96 (via GitHub)" <gi...@apache.org> on 2024/07/30 07:05:02 UTC
[I] [Bug] fix `float` type convert error in DebeziumSchemaUtils#fromDebeziumType [paimon]
chunji96 opened a new issue, #3846:
URL: https://github.com/apache/paimon/issues/3846
### Search before asking
- [X] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.
### Paimon version
commit id : 6da33cb6072ce90ad311450e0143aa9bf069d624
### Compute Engine
Flink
### Minimal reproduce step
when use [KafkaSyncDatabaseAction](https://paimon.apache.org/docs/master/api/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction) ,if field type in debezium-json format is `float` e.g.
```
{"type":"float","optional":false,"default":1,"field":"c1"}
```
will cause error
### What doesn't meet your expectations?
firstly, we use Kafka CDC to synchronize data from kafka to paimon and kafka format is debezium-json.
in piamon table the field c1 type is `float`,in kafka the c1 type is float too,but cause the follow error :
```
Caused by: java.lang.UnsupportedOperationException: Cannot convert field c1 from type FLOAT to STRING of Paimon table testdb.test_table.
at org.apache.paimon.flink.sink.cdc.UpdatedDataFieldsProcessFunctionBase.applySchemaChange(UpdatedDataFieldsProcessFunctionBase.java:125) ~[?:?]
at org.apache.paimon.flink.sink.cdc.UpdatedDataFieldsProcessFunction.processElement(UpdatedDataFieldsProcessFunction.java:60) ~[?:?]
at org.apache.paimon.flink.sink.cdc.UpdatedDataFieldsProcessFunction.processElement(UpdatedDataFieldsProcessFunction.java:39) ~[?:?]
at org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:237) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:146) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:550) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) ~[flink-dist-1.17.0-ctrip-RELEASE.jar:1.17.0-ctrip-RELEASE]
at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_121]
```
## fix
I find in org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils#fromDebeziumType, if dbzType is float DataType will convert to DataTypes.STRING()
```
private static DataType fromDebeziumType(String dbzType) {
switch (dbzType) {
case "int8":
return DataTypes.TINYINT();
case "int16":
return DataTypes.SMALLINT();
case "int32":
return DataTypes.INT();
case "int64":
return DataTypes.BIGINT();
case "float32":
case "float64":
return DataTypes.FLOAT();
case "double":
return DataTypes.DOUBLE();
case "boolean":
return DataTypes.BOOLEAN();
case "bytes":
return DataTypes.BYTES();
case "string":
default:
return DataTypes.STRING();
}
}
```
so just add "float" situation as following :
```
private static DataType fromDebeziumType(String dbzType) {
switch (dbzType) {
case "int8":
return DataTypes.TINYINT();
case "int16":
return DataTypes.SMALLINT();
case "int32":
return DataTypes.INT();
case "int64":
return DataTypes.BIGINT();
case "float":
case "float32":
case "float64":
return DataTypes.FLOAT();
case "double":
return DataTypes.DOUBLE();
case "boolean":
return DataTypes.BOOLEAN();
case "bytes":
return DataTypes.BYTES();
case "string":
default:
return DataTypes.STRING();
}
}
```
### Anything else?
_No response_
### Are you willing to submit a PR?
- [X] I'm willing to submit a PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug] fix `float` type convert error in DebeziumSchemaUtils#fromDebeziumType [paimon]
Posted by "chunji96 (via GitHub)" <gi...@apache.org>.
chunji96 commented on issue #3846:
URL: https://github.com/apache/paimon/issues/3846#issuecomment-2258560406
merged
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug] fix `float` type convert error in DebeziumSchemaUtils#fromDebeziumType [paimon]
Posted by "chunji96 (via GitHub)" <gi...@apache.org>.
chunji96 closed issue #3846: [Bug] fix `float` type convert error in DebeziumSchemaUtils#fromDebeziumType
URL: https://github.com/apache/paimon/issues/3846
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org