codec(ticdc): fix some avro encoding bugs #4704
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -28,6 +28,7 @@ import ( | |
| "github.com/pingcap/log" | ||
| "github.com/pingcap/tidb/parser/mysql" | ||
| "github.com/pingcap/tidb/types" | ||
| "github.com/pingcap/tidb/util/rowcodec" | ||
| "github.com/pingcap/tiflow/cdc/model" | ||
| "github.com/pingcap/tiflow/pkg/config" | ||
| cerror "github.com/pingcap/tiflow/pkg/errors" | ||
|
|
@@ -90,7 +91,7 @@ func (a *AvroEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEvent) | |
| mqMessage := NewMQMessage(config.ProtocolAvro, nil, nil, e.CommitTs, model.MqMessageTypeRow, &e.Table.Schema, &e.Table.Table) | ||
|
|
||
| if !e.IsDelete() { | ||
| res, err := avroEncode(e.Table, a.valueSchemaManager, e.TableInfoVersion, e.Columns, e.ColInfos, a.tz) | ||
| if err != nil { | ||
| log.Warn("AppendRowChangedEvent: avro encoding failed", zap.String("table", e.Table.String())) | ||
| return errors.Annotate(err, "AppendRowChangedEvent could not encode to Avro") | ||
|
|
@@ -109,7 +110,7 @@ func (a *AvroEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEvent) | |
|
|
||
| pkeyCols := e.HandleKeyColumns() | ||
|
|
||
| res, err := avroEncode(e.Table, a.keySchemaManager, e.TableInfoVersion, pkeyCols, e.ColInfos, a.tz) | ||
| if err != nil { | ||
| log.Warn("AppendRowChangedEvent: avro encoding failed", zap.String("table", e.Table.String())) | ||
| return errors.Annotate(err, "AppendRowChangedEvent could not encode to Avro") | ||
|
|
@@ -158,7 +159,7 @@ func (a *AvroEventBatchEncoder) Size() int { | |
| return sum | ||
| } | ||
|
|
||
| func avroEncode(table *model.TableName, manager *AvroSchemaManager, tableVersion uint64, cols []*model.Column, colInfos []rowcodec.ColInfo, tz *time.Location) (*avroEncodeResult, error) { | ||
| schemaGen := func() (string, error) { | ||
| schema, err := ColumnInfoToAvroSchema(table.Table, cols) | ||
| if err != nil { | ||
|
|
@@ -173,7 +174,7 @@ func avroEncode(table *model.TableName, manager *AvroSchemaManager, tableVersion | |
| return nil, errors.Annotate(err, "AvroEventBatchEncoder: get-or-register failed") | ||
| } | ||
|
|
||
| native, err := rowToAvroNativeData(cols, colInfos, tz) | ||
| if err != nil { | ||
| return nil, errors.Annotate(err, "AvroEventBatchEncoder: converting to native failed") | ||
| } | ||
|
|
@@ -244,13 +245,13 @@ func ColumnInfoToAvroSchema(name string, columnInfo []*model.Column) (string, er | |
| return string(str), nil | ||
| } | ||
|
|
||
| func rowToAvroNativeData(cols []*model.Column, colInfos []rowcodec.ColInfo, tz *time.Location) (interface{}, error) { | ||
| ret := make(map[string]interface{}, len(cols)) | ||
| for i, col := range cols { | ||
| if col == nil { | ||
| continue | ||
| } | ||
| data, str, err := columnToAvroNativeData(col, colInfos[i].Ft, tz) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
@@ -293,7 +294,7 @@ func getAvroDataTypeFallback(v interface{}) (string, error) { | |
| var unsignedLongAvroType = avroLogicalType{ | ||
| Type: "bytes", | ||
| LogicalType: decimalType, | ||
| Precision: 64, // enough to hold all values and is the default precision of confluent schema registry | ||
| Scale: 0, | ||
| } | ||
|
|
||
|
|
@@ -305,6 +306,9 @@ func getAvroDataTypeFromColumn(col *model.Column) (interface{}, error) { | |
| case mysql.TypeDouble: | ||
| return "double", nil | ||
| case mysql.TypeVarchar, mysql.TypeString, mysql.TypeVarString: | ||
| if col.Flag.IsBinary() { | ||
|
There was a problem hiding this comment. Could you add some unit tests? There was a problem hiding this comment. In the future, maybe we need a flag to control the behavior? Like https://debezium.io/documentation/reference/connectors/mysql.html#mysql-property-binary-handling-mode Cc @leoppro |
||
| return "bytes", nil | ||
| } | ||
| return "string", nil | ||
| case mysql.TypeDate, mysql.TypeDatetime, mysql.TypeTimestamp: | ||
| return avroLogicalType{ | ||
|
|
@@ -317,9 +321,9 @@ func getAvroDataTypeFromColumn(col *model.Column) (interface{}, error) { | |
| LogicalType: timeMillis, | ||
| }, nil | ||
| case mysql.TypeEnum: | ||
| return "string", nil | ||
| case mysql.TypeSet: | ||
| return "string", nil | ||
| case mysql.TypeBit: | ||
| return unsignedLongAvroType, nil | ||
| case mysql.TypeNewDecimal: | ||
|
|
@@ -341,7 +345,10 @@ func getAvroDataTypeFromColumn(col *model.Column) (interface{}, error) { | |
| case mysql.TypeJSON: | ||
| return "string", nil | ||
| case mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeBlob: | ||
| if col.Flag.IsBinary() { | ||
| return "bytes", nil | ||
| } | ||
| return "string", nil | ||
| case mysql.TypeYear: | ||
| return "long", nil | ||
| default: | ||
|
|
@@ -355,7 +362,7 @@ var ( | |
| zeroDateStr = types.NewTime(types.ZeroCoreTime, mysql.TypeDate, 0).String() | ||
| ) | ||
|
|
||
| func columnToAvroNativeData(col *model.Column, ft *types.FieldType, tz *time.Location) (interface{}, string, error) { | ||
| if col.Value == nil { | ||
| return nil, "null", nil | ||
| } | ||
|
|
@@ -434,7 +441,7 @@ func columnToAvroNativeData(col *model.Column, tz *time.Location) (interface{}, | |
| d := types.NewDuration(hours, minutes, seconds, int(fracInt), fsp).Duration | ||
| const fullType = "int." + timeMillis | ||
| return d, string(fullType), nil | ||
| case mysql.TypeVarchar, mysql.TypeString, mysql.TypeVarString, mysql.TypeTinyBlob, mysql.TypeBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob: | ||
| if col.Flag.IsBinary() { | ||
| switch val := col.Value.(type) { | ||
| case string: | ||
|
|
@@ -459,9 +466,17 @@ func columnToAvroNativeData(col *model.Column, tz *time.Location) (interface{}, | |
| case mysql.TypeNewDecimal: | ||
| return col.Value.(string), "string", nil | ||
| case mysql.TypeEnum: | ||
| enumVar, err := types.ParseEnumValue(ft.Elems, col.Value.(uint64)) | ||
| if err != nil { | ||
| return nil, "", cerror.WrapError(cerror.ErrAvroEncodeFailed, err) | ||
| } | ||
| return enumVar.Name, "string", nil | ||
| case mysql.TypeSet: | ||
| setVar, err := types.ParseSetValue(ft.Elems, col.Value.(uint64)) | ||
| if err != nil { | ||
| return nil, "", cerror.WrapError(cerror.ErrAvroEncodeFailed, err) | ||
| } | ||
| return setVar.Name, "string", nil | ||
| case mysql.TypeBit: | ||
| return handleUnsignedInt64() | ||
| case mysql.TypeTiny, mysql.TypeShort, mysql.TypeInt24: | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please use
column_infos.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry I don't really get it. Do you mean change the
json:"column-infos"tojson:"column_infos"? But all other fields use hyphens instead of underscores.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
msg:"-"seems to prevent the information from being persisted for Redo. Could you verify the compatibility of this change with Redo?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Redo only applies to mysql sink currently so does none of avro business. So prevent
ColInfosfrom being persisted for Redo seems right.But I think the
ColInfosis important and maybe in future Redo could support avro? PersistColInfosseems bring no trouble. Upgrade is also safe since anilColInfosfield is safe for other codecs. But as @liuzix tells me, it will make the redo log large. So I'd like to leave it unpersist in this pull request.Could you take a look @ben1009 ?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
lgtm for redo part