Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Flink]supported schemaEvolution when restarting the paimon cdc job #3311

Closed
wants to merge 13 commits into from
Closed

[Flink]supported schemaEvolution when restarting the paimon cdc job #3311

Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
ed79d13
[Flink]Fixed UpdatedDataFieldsProcessFunction.canConvert parameter of…
MOBIN-F May 9, 2024
a071a8f
[cdc]schemaCompatible failed after changing the length of the data type
MOBIN-F May 11, 2024
9c988f6
[cdc]fixed
MOBIN-F May 11, 2024
3262643
fix paimon-pg-cdc varchar/char always convert string type
May 17, 2024
0f32e5a
Supported schemaEvolution when restarting the paimon cdc job
May 17, 2024
ce114cf
Merge branch 'master' into schemaCompatible-Bug
MOBIN-F May 17, 2024
55627fd
optimized code
May 17, 2024
c4e5e70
Merge branch 'schemaCompatible-Bug' of https://github.com/MOBIN-F/pai…
May 17, 2024
488e841
resolve conflict
May 18, 2024
fe5803b
add pg-test
May 18, 2024
2addf77
fixed
May 18, 2024
3d20632
update test cast,support ColumnComment change
May 18, 2024
c936ab7
fixed
May 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 4 additions & 13 deletions ...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,24 +225,15 @@ private DataType extractFieldType(
case "boolean":
return DataTypes.BOOLEAN();
case "string":
int newLength = afterData.get(field.field()).asText().length();
if (paimonField == null) {
return DataTypes.VARCHAR(afterData.get(field.field()).asText().length());
return DataTypes.VARCHAR(newLength);
} else if (paimonField.type() instanceof VarCharType) {
int oldLength = ((VarCharType) paimonField.type()).getLength();
int newLength = afterData.get(field.field()).asText().length();
if (oldLength < newLength) {
return DataTypes.VARCHAR(newLength);
} else {
return DataTypes.VARCHAR(oldLength);
}
return DataTypes.VARCHAR(Math.max(oldLength, newLength));
} else if (paimonField.type() instanceof CharType) {
int oldLength = ((CharType) paimonField.type()).getLength();
int newLength = afterData.get(field.field()).asText().length();
if (oldLength < newLength) {
return DataTypes.CHAR(newLength);
} else {
return DataTypes.CHAR(oldLength);
}
return DataTypes.CHAR(Math.max(oldLength, newLength));
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in case "string", We cannot convert the varchar(n)/char(n) type into the string type, which will cause schemaCompatible to fail if the field changes when restarting the paimon-cdc job, because the string length of paimon is integer.max is always greater than n.
At the same time, when pg table performs column type alter(example: n changes from 10 to 20), because pg-debezium does not have ddl event, we can only get the length value n from the record

}
return DataTypes.STRING();
case "bytes":
Expand Down