From fc4e5d0a696f22dd8df72d7a506c913f8d84c6f4 Mon Sep 17 00:00:00 2001 From: lance6716 Date: Mon, 9 Sep 2024 11:39:51 +0800 Subject: [PATCH] drainer: if all DML is filtered, return `true` `ignore` (#1320) close pingcap/tidb-binlog#1317 --- drainer/syncer.go | 4 +++- drainer/syncer_test.go | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/drainer/syncer.go b/drainer/syncer.go index 6f263f6e0..8a08afb50 100644 --- a/drainer/syncer.go +++ b/drainer/syncer.go @@ -660,7 +660,9 @@ func skipDMLEvent(pv *pb.PrewriteValue, schema *Schema, filter *filter.Filter, b mutation.DeletedRows = mutation.DeletedRows[0:filteredDeleteIdx] } - muts = append(muts, mutation) + if len(mutation.Sequence) > 0 { + muts = append(muts, mutation) + } } pv.Mutations = muts diff --git a/drainer/syncer_test.go b/drainer/syncer_test.go index 5a4309387..2f5b7829e 100644 --- a/drainer/syncer_test.go +++ b/drainer/syncer_test.go @@ -54,7 +54,7 @@ func (s *syncerSuite) TestFilterTable(c *check.C) { // append one mutation that will not be dropped var keepID int64 = 2 schema.tableIDToName[keepID] = TableName{Schema: "keep", Table: "keep"} - pv.Mutations = append(pv.Mutations, pb.TableMutation{TableId: keepID}) + pv.Mutations = append(pv.Mutations, pb.TableMutation{TableId: keepID, Sequence: []pb.MutationType{pb.MutationType_Insert}}) ignore, err = skipDMLEvent(pv, schema, filter, nil) c.Assert(err, check.IsNil)