puller(ticdc): refactor ddl puller remove verbose code and reduce logs

cdc/entry/mounter.go

@@ -314,22 +314,21 @@ func ParseDDLJob(tblInfo *model.TableInfo, rawKV *model.RawKVEntry, id int64) (*
 
 // parseJob unmarshal the job from "v".
 func parseJob(v []byte, startTs, CRTs uint64) (*timodel.Job, error) {
-	job := &timodel.Job{}
-	err := json.Unmarshal(v, job)
+	var job timodel.Job
+	err := json.Unmarshal(v, &job)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	log.Debug("get new DDL job", zap.String("detail", job.String()))
 	if !job.IsDone() {
 		return nil, nil
 	}
 	// FinishedTS is only set when the job is synced,
 	// but we can use the entry's ts here
 	job.StartTS = startTs
-	// Since ddl in stateDone is not contain the FinishedTS,
+	// Since ddl in stateDone doesn't contain the FinishedTS,
 	// we need to set it as the txn's commit ts.
 	job.BinlogInfo.FinishedTS = CRTs
-	return job, nil
+	return &job, nil
 }
 
 func datum2Column(

cdc/entry/schema_storage.go

@@ -74,8 +74,8 @@ func NewSchemaStorage(
 ) (SchemaStorage, error) {
 	var (
 		snap    *schema.Snapshot
-		err     error
 		version int64
+		err     error
 	)
 	if meta == nil {
 		snap = schema.NewEmptySnapshot(forceReplicate)
@@ -89,19 +89,14 @@ func NewSchemaStorage(
 			return nil, errors.Trace(err)
 		}
 	}
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-
-	schema := &schemaStorageImpl{
+	return &schemaStorageImpl{
 		snaps:          []*schema.Snapshot{snap},
 		resolvedTs:     startTs,
 		forceReplicate: forceReplicate,
 		id:             id,
 		schemaVersion:  version,
 		role:           role,
-	}
-	return schema, nil
+	}, nil
 }
 
 // getSnapshot returns the snapshot which currentTs is less than(but most close to)
@@ -189,7 +184,7 @@ func (s *schemaStorageImpl) HandleDDLJob(job *timodel.Job) error {
 		// We use schemaVersion to check if an already-executed DDL job is processed for a second time.
 		// Unexecuted DDL jobs should have largest schemaVersions.
 		if job.BinlogInfo.FinishedTS <= lastSnap.CurrentTs() || job.BinlogInfo.SchemaVersion <= s.schemaVersion {
-			log.Info("ignore foregone DDL",
+			log.Info("schemaStorage: ignore foregone DDL",
 				zap.String("namespace", s.id.Namespace),
 				zap.String("changefeed", s.id.ID),
 				zap.String("DDL", job.Query),
@@ -205,26 +200,29 @@ func (s *schemaStorageImpl) HandleDDLJob(job *timodel.Job) error {
 		snap = schema.NewEmptySnapshot(s.forceReplicate)
 	}
 	if err := snap.HandleDDL(job); err != nil {
-		log.Error("handle DDL failed",
+		log.Error("schemaStorage: update snapshot by the DDL job failed",
 			zap.String("namespace", s.id.Namespace),
 			zap.String("changefeed", s.id.ID),
-			zap.String("DDL", job.Query),
-			zap.Stringer("job", job), zap.Error(err),
-			zap.Uint64("finishTs", job.BinlogInfo.FinishedTS),
-			zap.String("role", s.role.String()))
+			zap.String("schema", job.SchemaName),
+			zap.String("table", job.TableName),
+			zap.String("query", job.Query),
+			zap.Uint64("finishedTs", job.BinlogInfo.FinishedTS),
+			zap.String("role", s.role.String()),
+			zap.Error(err))
 		return errors.Trace(err)
 	}
-	log.Info("handle DDL",
-		zap.String("namespace", s.id.Namespace),
-		zap.String("changefeed", s.id.ID),
-		zap.String("DDL", job.Query),
-		zap.Stringer("job", job),
-		zap.Uint64("finishTs", job.BinlogInfo.FinishedTS),
-		zap.String("role", s.role.String()))
-
 	s.snaps = append(s.snaps, snap)
 	s.schemaVersion = job.BinlogInfo.SchemaVersion
 	s.AdvanceResolvedTs(job.BinlogInfo.FinishedTS)
+	log.Info("schemaStorage: update snapshot by the DDL job",
+		zap.String("namespace", s.id.Namespace),
+		zap.String("changefeed", s.id.ID),
+		zap.String("schema", job.SchemaName),
+		zap.String("table", job.TableName),
+		zap.String("query", job.Query),
+		zap.Uint64("finishedTs", job.BinlogInfo.FinishedTS),
+		zap.Uint64("schemaVersion", uint64(s.schemaVersion)),
+		zap.String("role", s.role.String()))
 	return nil
 }
 

cdc/entry/schema_storage_test.go

@@ -881,10 +881,10 @@ func TestSchemaStorage(t *testing.T) {
 		require.Nil(t, err)
 
 		schemaStorage, err := NewSchemaStorage(nil, 0, false, model.DefaultChangeFeedID("dummy"), util.RoleTester, f)
-		require.Nil(t, err)
+		require.NoError(t, err)
 		for _, job := range jobs {
 			err := schemaStorage.HandleDDLJob(job)
-			require.Nil(t, err)
+			require.NoError(t, err)
 		}
 
 		for _, job := range jobs {

cdc/model/owner.go

@@ -74,7 +74,6 @@ type DDLJobEntry struct {
 	Job    *timodel.Job
 	OpType OpType
 	CRTs   uint64
-	Err    error
 }
 
 // TaskPosition records the process information of a capture

cdc/processor/processor.go

@@ -711,21 +711,16 @@ func (p *processor) initDDLHandler(ctx context.Context) error {
 	}
 
 	meta := kv.GetSnapshotMeta(p.upstream.KVStorage, ddlStartTs)
-	f, err := filter.NewFilter(p.latestInfo.Config, "")
-	if err != nil {
-		return errors.Trace(err)
-	}
 	schemaStorage, err := entry.NewSchemaStorage(meta, ddlStartTs,
-		forceReplicate, p.changefeedID, util.RoleProcessor, f)
+		forceReplicate, p.changefeedID, util.RoleProcessor, p.filter)
 	if err != nil {
 		return errors.Trace(err)
 	}
 
 	serverCfg := config.GetGlobalServerConfig()
-
 	changefeedID := model.DefaultChangeFeedID(p.changefeedID.ID + "_processor_ddl_puller")
 	ddlPuller := puller.NewDDLJobPuller(
-		ctx, p.upstream, ddlStartTs, serverCfg, changefeedID, schemaStorage, f,
+		ctx, p.upstream, ddlStartTs, serverCfg, changefeedID, schemaStorage, p.filter,
 	)
 	p.ddlHandler.r = &ddlHandler{puller: ddlPuller, schemaStorage: schemaStorage}
 	return nil
@@ -1007,10 +1002,6 @@ func (d *ddlHandler) Run(ctx context.Context, _ ...chan<- error) error {
 			if jobEntry.OpType == model.OpTypeResolved {
 				d.schemaStorage.AdvanceResolvedTs(jobEntry.CRTs)
 			}
-			err := jobEntry.Err
-			if err != nil {
-				return errors.Trace(err)
-			}
 		}
 	})
 	return g.Wait()