Skip decoding of excluded table rows (#831)

* refactor: move decompression into DecodeData method

* feat: skip decoding of row events for tables we are not interested in

canal/canal.go

@@ -451,6 +451,19 @@ func (c *Canal) prepareSyncer() error {
 		Logger:                  c.cfg.Logger,
 		Dialer:                  c.cfg.Dialer,
 		Localhost:               c.cfg.Localhost,
+		RowsEventDecodeFunc: func(event *replication.RowsEvent, data []byte) error {
+			pos, err := event.DecodeHeader(data)
+			if err != nil {
+				return err
+			}
+
+			key := fmt.Sprintf("%s.%s", string(event.Table.Schema), string(event.Table.Table))
+			if !c.checkTableMatch(key) {
+				return nil
+			}
+
+			return event.DecodeData(pos, data)
+		},
 	}
 
 	if strings.Contains(c.cfg.Addr, "/") {

replication/row_event.go

@@ -1018,6 +1018,14 @@ func (e *RowsEvent) decodeExtraData(data []byte) (err2 error) {
 }
 
 func (e *RowsEvent) DecodeData(pos int, data []byte) (err2 error) {
+	if e.compressed {
+		data, err2 = DecompressMariadbData(data[pos:])
+		if err2 != nil {
+			//nolint:nakedret
+			return
+		}
+	}
+
 	// Rows_log_event::print_verbose()
 
 	var (
@@ -1073,13 +1081,6 @@ func (e *RowsEvent) Decode(data []byte) error {
 	if err != nil {
 		return err
 	}
-	if e.compressed {
-		uncompressedData, err := DecompressMariadbData(data[pos:])
-		if err != nil {
-			return err
-		}
-		return e.DecodeData(0, uncompressedData)
-	}
 	return e.DecodeData(pos, data)
 }