From a3f843bf39b323f9f28529e845455569c214f069 Mon Sep 17 00:00:00 2001
From: Ti Chi Robot <ti-community-prow-bot@tidb.io>
Date: Mon, 26 Jun 2023 15:55:05 +0800
Subject: [PATCH] drainer: fix ddl system table (#1251) (#1252)

close pingcap/tidb-binlog#1250
---
 drainer/schema.go | 12 ++++++++++++
 drainer/syncer.go |  4 ++--
 2 files changed, 14 insertions(+), 2 deletions(-)

diff --git a/drainer/schema.go b/drainer/schema.go
index d278905f4..d767b5ed0 100644
--- a/drainer/schema.go
+++ b/drainer/schema.go
@@ -18,6 +18,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	"go.uber.org/zap"
@@ -505,6 +506,9 @@ func (s *Schema) handleDDL(job *model.Job) (schemaName string, tableName string,
 			if table == nil {
 				return "", "", "", errors.NotValidf("job %d", job.ID)
 			}
+			if isDDLSystemTable(table.ID) {
+				continue
+			}
 
 			schema, ok := s.SchemaByID(job.SchemaID)
 			if !ok {
@@ -612,5 +616,13 @@ func addImplicitColumn(table *model.TableInfo) {
 // Now, it write DDL Binlog in the txn that the state of job is changed to *done* (before change to *synced*)
 // At state *done*, it will be always and only changed to *synced*.
 func skipJob(job *model.Job) bool {
+	if isDDLSystemTable(job.TableID) {
+		return true
+	}
 	return !job.IsSynced() && !job.IsDone()
 }
+
+// isDDLSystemTable checks if the tableID is a ddl system table ID.
+func isDDLSystemTable(tableID int64) bool {
+	return tableID > meta.MaxGlobalID
+}
diff --git a/drainer/syncer.go b/drainer/syncer.go
index 6f263f6e0..cb6df2f3a 100644
--- a/drainer/syncer.go
+++ b/drainer/syncer.go
@@ -433,7 +433,7 @@ ForLoop:
 		} else if jobID > 0 {
 			log.Debug("get ddl binlog job", zap.Stringer("job", b.job))
 
-			if skipUnsupportedDDLJob(b.job) {
+			if skipUnsupportedDDLJob(b.job) || isDDLSystemTable(b.job.TableID) {
 				log.Info("skip unsupported DDL job", zap.Stringer("job", b.job))
 				appendFakeBinlogIfNeeded(nil, commitTS)
 				continue
@@ -601,7 +601,7 @@ func skipDMLEvent(pv *pb.PrewriteValue, schema *Schema, filter *filter.Filter, b
 			return false, errors.Errorf("not found table id: %d", mutation.GetTableId())
 		}
 
-		if filter.SkipSchemaAndTable(schemaName, tableName) {
+		if filter.SkipSchemaAndTable(schemaName, tableName) || isDDLSystemTable(mutation.GetTableId()) {
 			log.Debug("skip dml", zap.String("schema", schemaName), zap.String("table", tableName))
 			continue
 		}