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

maxwell: fix some bug and refine output #1129

Merged
merged 21 commits into from
Dec 7, 2020
Merged

maxwell: fix some bug and refine output #1129

Changes from 1 commit
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
ccdd5b1
Fix base64 data output, fix maxwell output tso to unix timestamp, fix…
shldreams Nov 26, 2020
a24018e
Merge branch 'master' into fix_base64_decode_to_maxwell
shldreams Nov 26, 2020
ae6243e
fix test case
shldreams Nov 26, 2020
f263a27
Merge branch 'fix_base64_decode_to_maxwell' of https://github.com/shl…
shldreams Nov 26, 2020
db2e93a
fix
shldreams Nov 26, 2020
88391b5
fix
shldreams Nov 26, 2020
2186d78
fix
shldreams Nov 26, 2020
02085bd
fix varchar output
shldreams Nov 30, 2020
7057d48
Merge branch 'master' into fix_base64_decode_to_maxwell
shldreams Nov 30, 2020
d1b466a
remove decode code
shldreams Dec 3, 2020
aa48b93
fix
shldreams Dec 3, 2020
9e635ab
Merge branch 'master' into fix_base64_decode_to_maxwell
shldreams Dec 3, 2020
aa6b231
fix
shldreams Dec 3, 2020
d7163d8
fix
shldreams Dec 3, 2020
07db747
Merge branch 'fix_base64_decode_to_maxwell' of https://github.com/shl…
shldreams Dec 3, 2020
453127f
Merge branch 'master' into fix_base64_decode_to_maxwell
shldreams Dec 3, 2020
3d35b59
Merge branch 'master' into fix_base64_decode_to_maxwell
shldreams Dec 3, 2020
531dea8
fix code
shldreams Dec 3, 2020
2fb4342
Merge branch 'fix_base64_decode_to_maxwell' of https://github.com/shl…
shldreams Dec 3, 2020
2e35901
Merge branch 'master' into fix_base64_decode_to_maxwell
shldreams Dec 3, 2020
51ad98d
Merge branch 'master' into fix_base64_decode_to_maxwell
shldreams Dec 4, 2020
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
Prev Previous commit
Next Next commit
fix varchar output
  • Loading branch information
@shldreams
shldreams committed Nov 30, 2020
commit 02085bd9651b4aaca0b0aa95395a6a4f32a42d0b
69 changes: 29 additions & 40 deletions cdc/sink/codec/maxwell.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ import (
"bytes"
"encoding/binary"
"encoding/json"
"strconv"

"github.com/pingcap/errors"
model2 "github.com/pingcap/parser/model"
Expand Down Expand Up @@ -107,16 +106,13 @@ func rowEventToMaxwellMessage(e *model.RowChangedEvent) (*mqMessageKey, *maxwell
value.Type = "insert"
for _, v := range e.Columns {
switch v.Type {
case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
if v.Value != nil {
str := string(v.Value.([]byte))
if v.Flag.IsBinary() {
str = strconv.Quote(str)
str = str[1 : len(str)-1]
}
value.Data[v.Name] = str
} else {
case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeBlob:
if v.Value == nil {
value.Data[v.Name] = nil
} else if v.Flag.IsBinary() {
value.Data[v.Name] = v.Value
} else {
value.Data[v.Name] = string(v.Value.([]byte))
}
default:
value.Data[v.Name] = v.Value
Expand All @@ -126,16 +122,13 @@ func rowEventToMaxwellMessage(e *model.RowChangedEvent) (*mqMessageKey, *maxwell
value.Type = "delete"
for _, v := range e.PreColumns {
switch v.Type {
case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
if v.Value != nil {
str := string(v.Value.([]byte))
if v.Flag.IsBinary() {
str = strconv.Quote(str)
str = str[1 : len(str)-1]
}
value.Old[v.Name] = str
} else {
case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeBlob:
if v.Value == nil {
value.Old[v.Name] = nil
} else if v.Flag.IsBinary() {
value.Old[v.Name] = v.Value
} else {
value.Old[v.Name] = string(v.Value.([]byte))
}
default:
value.Old[v.Name] = v.Value
Expand All @@ -145,35 +138,31 @@ func rowEventToMaxwellMessage(e *model.RowChangedEvent) (*mqMessageKey, *maxwell
value.Type = "update"
for _, v := range e.Columns {
switch v.Type {
case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
if v.Value != nil {
str := string(v.Value.([]byte))
if v.Flag.IsBinary() {
str = strconv.Quote(str)
str = str[1 : len(str)-1]
}
value.Data[v.Name] = str
} else {
case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeBlob:
if v.Value == nil {
value.Data[v.Name] = nil
} else if v.Flag.IsBinary() {
value.Data[v.Name] = v.Value
} else {
value.Data[v.Name] = string(v.Value.([]byte))
}
default:
value.Data[v.Name] = v.Value
}
}
for _, v := range e.PreColumns {
switch v.Type {
case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
if v.Value != nil {
str := string(v.Value.([]byte))
if v.Flag.IsBinary() {
str = strconv.Quote(str)
str = str[1 : len(str)-1]
case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeBlob:
if v.Value == nil {
if value.Data[v.Name] != nil {
value.Old[v.Name] = nil
} else {
continue
}
if value.Data[v.Name] != str {
value.Old[v.Name] = str
}
} else if value.Data[v.Name] != nil && v.Value == nil {
value.Old[v.Name] = nil
} else if v.Flag.IsBinary() {
value.Old[v.Name] = v.Value
} else {
value.Old[v.Name] = string(v.Value.([]byte))
}
default:
if value.Data[v.Name] != v.Value {
Expand Down Expand Up @@ -204,7 +193,7 @@ func (d *MaxwellEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEven
d.valueBuf.Write(value)

d.batchSize++
return EncoderNoOperation, nil
return EncoderNeedAsyncWrite, nil
}

// SetParams is no-op for Maxwell for now
Expand Down