feat: Charset parameter for Mysql connection #1027

sigalikanevsky · 2025-04-27T14:20:38Z

No description provided.

…ests Removed FlushBinlog which was only used for testing

…deleted_while_reading Fix/eitam/add skip when table is deleted while reading

Feature/eitam/removed spamming log

…_decode Added support to convert byte to string dynamically

…on_sp Remove unneeded log on SP

…_to_date Change mysql_date default zero based value to nil

…ault nil

…_to_date Change mysql_MYSQL_TYPE_TIMESTAMP2 and MYSQL_TYPE_TIMESTAMP to be def…

…_to_date Change decodeDatetime2 default to zero

Change decodeTimestamp2 to return nil in case sec = 0

Add limit to max host name length

…p_log Changed SP is missing to debug level

lance6716

Please fix git conflict

lance6716 · 2025-04-28T01:11:22Z

README.md

@@ -132,7 +132,7 @@ type MyEventHandler struct {
 }

 func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
-	log.Infof("%s %v\n", e.Action, e.Rows)
+	log.Debugf("%s %v\n", e.Action, e.Rows)


the log level should be kept, other developers may rely on this behaviour.
Also for other files.

lance6716 · 2025-04-28T01:12:23Z

canal/canal.go

@@ -65,6 +65,10 @@ func NewCanal(cfg *Config) (*Canal, error) {

 	c.ctx, c.cancel = context.WithCancel(context.Background())

+	if cfg.WaitTimeBetweenConnectionSeconds > 0 {


it should be <= 0 to set the default value?

lance6716 · 2025-04-28T01:15:21Z

canal/sync.go

@@ -270,38 +279,10 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 		return errors.Errorf("%s not supported now", e.Header.EventType)
 	}
 	events := newRowsEvent(t, action, ev.Rows, e.Header)
+	events.Header.Gtid = c.SyncedGTIDSet()


Can you explain why this is changed?

lance6716 · 2025-04-28T01:17:46Z

replication/binlogsyncer.go

@@ -697,7 +711,8 @@ func (b *BinlogSyncer) onStream(s *BinlogStreamer) {
 							return
 						}

-						log.Errorf("retry sync err: %v, wait 1s and retry again", err)
+						log.Errorf("retry sync err: %v, wait %s s and retry again", err, b.cfg.WaitTimeBetweenConnectionSeconds)
+						time.Sleep(b.cfg.WaitTimeBetweenConnectionSeconds)


the sleep is at line 705. Don't need to add extra time.Sleep.

lance6716 · 2025-04-28T01:18:25Z

replication/binlogsyncer.go

@@ -719,7 +734,11 @@ func (b *BinlogSyncer) onStream(s *BinlogStreamer) {

 		switch data[0] {
 		case OK_HEADER:
-			if err = b.parseEvent(s, data); err != nil {
+			if err = b.parseEvent(b.nextPos.Name, s, data); err != nil {
+				// if the error is errMissingTableMapEvent skip


data may be lost if we don't return error

lance6716 · 2025-04-28T01:21:05Z

replication/row_event.go

 	case MYSQL_TYPE_STRING:
-		v, n = decodeString(data, length)
+		if !utf8.Valid(data) && e.Table.isLatin {
+			v, n = decodeStringLatin1(data, length)


The decoding and converting should be done at upper layer, not here. replication package aims to provide a golang representation of binlog.

dveeden

Please reference an issue or describe the changes in the description of the PR

dveeden · 2025-04-28T05:21:51Z

replication/row_event.go

@@ -1182,6 +1217,29 @@ func decodeString(data []byte, length int) (v string, n int) {
 	return
 }

+func decodeStringLatin1(data []byte, length int) (v string, n int) {
+	// Define the Latin1 decoder
+	decoder := charmap.ISO8859_1.NewDecoder()


Note that latin1 in MySQL is cp1252 and not ISO 8859-1:

"MySQL's latin1 is the same as the Windows cp1252 character set. This means it is the same as the official ISO 8859-1 or IANA (Internet Assigned Numbers Authority) latin1, except that IANA latin1 treats the code points between 0x80 and 0x9f as “undefined,” whereas cp1252, and therefore MySQL's latin1, assign characters for those positions. For example, 0x80 is the Euro sign. For the “undefined” entries in cp1252, MySQL translates 0x81 to Unicode 0x0081, 0x8d to 0x008d, 0x8f to 0x008f, 0x90 to 0x0090, and 0x9d to 0x009d. "

source: https://dev.mysql.com/doc/refman/8.4/en/charset-we-sets.html

Suggested change

decoder := charmap.ISO8859_1.NewDecoder()

decoder := charmap.Windows1252.NewDecoder()

dveeden · 2025-04-28T05:26:11Z

replication/row_event.go

@@ -1182,6 +1217,29 @@ func decodeString(data []byte, length int) (v string, n int) {
 	return
 }

+func decodeStringLatin1(data []byte, length int) (v string, n int) {


Might be good to make things a bit more generic, so it could work with multiple charsets and not just latin1.

However I think a vast majority of MySQL installations use either latin1 or utf8/utf8mb4. Maybe some ascii, but that's a subset of both latin1 and utf8/utf8mb4...

dveeden · 2025-05-06T05:23:23Z

@sigalikanevsky Could you fix the conflicts by merging the master branch into your branch (or rebasing on top of the master branch)?

ccoVeille · 2025-05-06T21:08:57Z

canal/sync.go

-func (c *Canal) FlushBinlog() error {
-	_, err := c.Execute("FLUSH BINARY LOGS")
-	return errors.Trace(err)
-}
-
-func (c *Canal) WaitUntilPos(pos mysql.Position, timeout time.Duration) error {
-	timer := time.NewTimer(timeout)
-	for {
-		select {
-		case <-timer.C:
-			return errors.Errorf("wait position %v too long > %s", pos, timeout)
-		default:
-			err := c.FlushBinlog()
-			if err != nil {
-				return errors.Trace(err)
-			}
-			curPos := c.master.Position()
-			if curPos.Compare(pos) >= 0 {
-				return nil
-			} else {
-				log.Debugf("master pos is %v, wait catching %v", curPos, pos)
-				time.Sleep(100 * time.Millisecond)
-			}
-		}
-	}
-
-	return nil
-}


Removing this break the API of the package

I don't see why they were moved as functions accepting Canal

ccoVeille · 2025-05-06T21:09:50Z

canal/sync.go

@@ -142,7 +143,15 @@ func (c *Canal) runSyncBinlog() error {
 		case *replication.QueryEvent:
 			stmts, _, err := c.parser.Parse(string(e.Query), "", "")
 			if err != nil {
-				log.Errorf("parse query(%s) err %v, will skip this event", e.Query, err)
+				msg := err.Error()
+				if strings.Contains(strings.ToLower(msg), strings.ToLower("procedure")) {


Parsing error message like this is a bad practice.

Don't you have an error code should could catch?

ccoVeille · 2025-05-06T21:13:03Z

canal/sync.go

-		log.Infof("start sync binlog at binlog file %v", pos)
+		log.Debugf("start sync binlog at binlog file %v", pos)


You might think there is a need to change the log lever for these information messages.

But it should be done in another PR, not this one.

It brings confusion, especially because there are no links between these changes and the topic of the PR

ccoVeille · 2025-05-06T21:13:25Z

canal/sync.go

@@ -270,38 +279,10 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 		return errors.Errorf("%s not supported now", e.Header.EventType)
 	}
 	events := newRowsEvent(t, action, ev.Rows, e.Header)
+	events.Header.Gtid = c.SyncedGTIDSet()


ccoVeille · 2025-05-06T21:19:10Z

replication/binlogsyncer.go

+	// MaxWaitTimeBetween Check connection
+	WaitTimeBetweenConnectionSeconds time.Duration


Suggested change

// MaxWaitTimeBetween Check connection

WaitTimeBetweenConnectionSeconds time.Duration

// WaitTimeBetweenConnectionSeconds is the delay between two connections when an connection fails

WaitTimeBetweenConnectionSeconds time.Duration

Also, it's a time.Duration, so for naming you should avoid mentioning seconds

Suggested change

// MaxWaitTimeBetween Check connection

WaitTimeBetweenConnectionSeconds time.Duration

// DelayBetweenConnection is the delay between two connections when an connection fails

DelayBetweenConnection time.Duration

ccoVeille · 2025-05-06T21:21:10Z

replication/binlogsyncer.go

-			if err = b.parseEvent(s, data); err != nil {
+			if err = b.parseEvent(b.nextPos.Name, s, data); err != nil {
+				// if the error is errMissingTableMapEvent skip
+				if strings.Contains(strings.ToLower(errors.Cause(err).Error()), errMissingTableMapEvent.Error()) {


This way to test the error is absolutely discouraged.

Using errors.Is maybe?

eitamring and others added 30 commits January 6, 2022 15:02

Added waitTimeBetweenConnection check

ef55c4c

Added safe conn close

6855741

Added file name to each event

e6565cd

Added gtid to each event

cb4f084

Removed FlushBinlog which was only used for testing

8d7d13c

Fixed testing

f81397d

Merge pull request #1 from RiveryIO/feature/eitam/removed_flush_for_t…

f77308e

…ests Removed FlushBinlog which was only used for testing

Added delete on table when deleted

824d650

probably grammar

61a148b

Merge pull request #2 from RiveryIO/fix/eitam/add_skip_when_table_is_…

8cc642f

…deleted_while_reading Fix/eitam/add skip when table is deleted while reading

Removed spamming log

05b87bb

Merge pull request #3 from RiveryIO/feature/eitam/removed_spaming_log

2025086

Feature/eitam/removed spamming log

Added covert to string interface if needed

fa2e8fa

Merge pull request #4 from RiveryIO/feature/eitam/support_base64_byte…

e95e5cd

…_decode Added support to convert byte to string dynamically

Remove unneeded log on SP

fdd11a6

Merge pull request #5 from RiveryIO/feautre/eitam/remove_uneeded_log_…

d947c9e

…on_sp Remove unneeded log on SP

Change mysql_date default zero based value to nil

73477d3

Merge pull request #6 from RiveryIO/feature/eitam/add_nil_based_value…

eb2ea8e

…_to_date Change mysql_date default zero based value to nil

Change mysql_MYSQL_TYPE_TIMESTAMP2 and MYSQL_TYPE_TIMESTAMP to be def…

007d419

…ault nil

Merge pull request #7 from RiveryIO/feature/eitam/add_nil_based_value…

c6c9248

…_to_date Change mysql_MYSQL_TYPE_TIMESTAMP2 and MYSQL_TYPE_TIMESTAMP to be def…

Change decodeDatetime2 default to zero

d15a7ce

Merge pull request #8 from RiveryIO/feature/eitam/add_nil_based_value…

4298599

…_to_date Change decodeDatetime2 default to zero

Change decodeTimestamp2 to return nil in case sec = 0

1f7d701

Merge pull request #9 from RiveryIO/fix/eitam/decode_timestamp

98108dd

Change decodeTimestamp2 to return nil in case sec = 0

Add limit to max host name length

411b75d

Merge pull request #10 from RiveryIO/feature/eitam/limit_max_host_name

9bd5965

Add limit to max host name length

Change log debug level

c349655

Merge pull request #12 from RiveryIO/feature/eitam/change_log_debug_s…

e7d51c3

…p_log Changed SP is missing to debug level

add latin support

c3593a7

add latin support

fd2c582

sigalikanevsky added 6 commits April 27, 2025 12:43

add latin support

3d213ae

add latin support

b34a98d

add latin support

7222b87

add latin support

b4851e0

move to function

b939f18

move to function

88f0609

lance6716 requested changes Apr 28, 2025

View reviewed changes

dveeden requested changes Apr 28, 2025

View reviewed changes

dveeden reviewed Apr 28, 2025

View reviewed changes

sigalikanevsky added 2 commits April 28, 2025 11:50

fixes and tests

0799a57

fixes

a2053cd

ccoVeille reviewed May 6, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Charset parameter for Mysql connection #1027

feat: Charset parameter for Mysql connection #1027

sigalikanevsky commented Apr 27, 2025

lance6716 left a comment

lance6716 Apr 28, 2025

lance6716 Apr 28, 2025

lance6716 Apr 28, 2025

ccoVeille May 6, 2025

lance6716 Apr 28, 2025

lance6716 Apr 28, 2025

lance6716 Apr 28, 2025

dveeden left a comment

dveeden Apr 28, 2025

dveeden Apr 28, 2025

dveeden commented May 6, 2025 •

edited

Loading

ccoVeille May 6, 2025

ccoVeille May 6, 2025

ccoVeille May 6, 2025

ccoVeille May 6, 2025

ccoVeille May 6, 2025

ccoVeille May 6, 2025

		@@ -65,6 +65,10 @@ func NewCanal(cfg Config) (Canal, error) {

		c.ctx, c.cancel = context.WithCancel(context.Background())

		if cfg.WaitTimeBetweenConnectionSeconds > 0 {

	decoder := charmap.ISO8859_1.NewDecoder()
	decoder := charmap.Windows1252.NewDecoder()

		log.Infof("start sync binlog at binlog file %v", pos)
		log.Debugf("start sync binlog at binlog file %v", pos)

		// MaxWaitTimeBetween Check connection
		WaitTimeBetweenConnectionSeconds time.Duration

feat: Charset parameter for Mysql connection #1027

Are you sure you want to change the base?

feat: Charset parameter for Mysql connection #1027

Conversation

sigalikanevsky commented Apr 27, 2025

lance6716 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dveeden left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dveeden commented May 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dveeden commented May 6, 2025 •

edited

Loading