From 8646dddb3fd0f9f1558d625634efff916c23b76f Mon Sep 17 00:00:00 2001 From: Yue Yuan Date: Tue, 4 Jul 2023 14:44:23 -0400 Subject: [PATCH 1/3] log --- pkg/clone/transactionwriter.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pkg/clone/transactionwriter.go b/pkg/clone/transactionwriter.go index 9a13cd2..d9b3150 100644 --- a/pkg/clone/transactionwriter.go +++ b/pkg/clone/transactionwriter.go @@ -617,6 +617,7 @@ func (m *Mutation) replace(ctx context.Context, tx DBWriter) error { valueStrings := make([]string, 0, len(m.Rows)) valueArgs := make([]interface{}, 0, len(m.Rows)*len(tableSchema.Columns)) for _, row := range m.Rows { + fmt.Printf("row = %v\n", row) if len(row) != len(m.Table.Columns) { panic(fmt.Sprintf("row column count %d doesn't match the cached table schema columns: %v (%v), "+ "there may have been a schema change and you will most likely need to restart replication by deleting"+ From b760d14b2c5ab5ab0a39c135b09c24ffd208f2ae Mon Sep 17 00:00:00 2001 From: Yue Yuan Date: Tue, 4 Jul 2023 14:46:42 -0400 Subject: [PATCH 2/3] comment --- pkg/clone/transactionstream.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/clone/transactionstream.go b/pkg/clone/transactionstream.go index 931c656..fc9e7ee 100644 --- a/pkg/clone/transactionstream.go +++ b/pkg/clone/transactionstream.go @@ -127,7 +127,7 @@ func (s *TransactionStream) Run(ctx context.Context, b backoff.BackOff, output c continue } currentTransaction.Mutations = append(currentTransaction.Mutations, s.toMutation(e, event)) - case *replication.XIDEvent: + case *replication.XIDEvent: // the last event in a transaction gset := event.GSet currentTransaction.FinalPosition = Position{ File: nextPos.Name, From 3709e744361b29bc24cf8d14695bb142bed3be7c Mon Sep 17 00:00:00 2001 From: Yue Yuan Date: Tue, 4 Jul 2023 16:41:25 -0400 Subject: [PATCH 3/3] logging --- pkg/clone/transactionstream.go | 1 + pkg/clone/transactionwriter.go | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/clone/transactionstream.go b/pkg/clone/transactionstream.go index fc9e7ee..fbc9337 100644 --- a/pkg/clone/transactionstream.go +++ b/pkg/clone/transactionstream.go @@ -156,6 +156,7 @@ func (s *TransactionStream) Run(ctx context.Context, b backoff.BackOff, output c func (s *TransactionStream) toMutation(e *replication.BinlogEvent, event *replication.RowsEvent) Mutation { mutationType := toMutationType(e.Header.EventType) + fmt.Printf("mutationType = %s, RowsEvent = %v\n", mutationType, event) switch mutationType { case Update: if len(event.Rows)%2 != 0 { diff --git a/pkg/clone/transactionwriter.go b/pkg/clone/transactionwriter.go index d9b3150..e267036 100644 --- a/pkg/clone/transactionwriter.go +++ b/pkg/clone/transactionwriter.go @@ -617,7 +617,7 @@ func (m *Mutation) replace(ctx context.Context, tx DBWriter) error { valueStrings := make([]string, 0, len(m.Rows)) valueArgs := make([]interface{}, 0, len(m.Rows)*len(tableSchema.Columns)) for _, row := range m.Rows { - fmt.Printf("row = %v\n", row) + fmt.Printf("tableName = %s, row = %v\n", tableName, row) if len(row) != len(m.Table.Columns) { panic(fmt.Sprintf("row column count %d doesn't match the cached table schema columns: %v (%v), "+ "there may have been a schema change and you will most likely need to restart replication by deleting"+