cdclog: fix the lost ddl files (#1094)

pkg/restore/log_client.go

@@ -192,7 +192,11 @@ func (l *LogClient) NeedRestoreDDL(fileName string) (bool, error) {
 	// maxUint64 - the first DDL event's commit ts as the file name to return the latest ddl file.
 	// see details at https://github.com/pingcap/ticdc/pull/826/files#diff-d2e98b3ed211b7b9bb7b6da63dd48758R81
 	ts = maxUint64 - ts
-	if l.maybeTSInRange(ts) {
+
+	// In cdc, we choose the first event as the file name of DDL file.
+	// so if the file ts is large than endTS, we can skip to execute it.
+	// FIXME find a unified logic to filter row changes files and ddl files.
+	if ts <= l.endTS {
 		return true, nil
 	}
 	log.Info("filter ddl file by ts", zap.String("name", fileName), zap.Uint64("ts", ts))

pkg/restore/log_client_test.go

@@ -84,7 +84,7 @@ func (s *testLogRestoreSuite) TestTsInRange(c *C) {
 	}
 
 	// format cdclog will collect, because file sink will generate cdclog for streaming write.
-	ddlFile := "ddl.1"
+	ddlFile := "ddl.18446744073709551615"
 	collected, err = s.client.NeedRestoreDDL(ddlFile)
 	c.Assert(err, IsNil)
 	c.Assert(collected, IsTrue)
@@ -95,4 +95,33 @@ func (s *testLogRestoreSuite) TestTsInRange(c *C) {
 		c.Assert(err, IsNil)
 		c.Assert(collected, IsFalse)
 	}
+
+	s.client.ResetTSRange(424839867765096449, 424839886560821249)
+	// ddl suffix records the first event's commit ts
+
+	// the file name include the end ts, collect it.(maxUint64 - 424839886560821249)
+	ddlFile = "ddl.18021904187148730366"
+	collected, err = s.client.NeedRestoreDDL(ddlFile)
+	c.Assert(err, IsNil)
+	c.Assert(collected, IsTrue)
+
+	// the file name include the start ts, collect it.(maxUint64 - 424839867765096449)
+	ddlFile = "ddl.18021904205944455166"
+	collected, err = s.client.NeedRestoreDDL(ddlFile)
+	c.Assert(err, IsNil)
+	c.Assert(collected, IsTrue)
+
+	// the file first event's ts is smaller than the start ts, collect it.
+	// because we only know this file's first event not in TSRange.
+	// FIXME find a unified logic for collection.
+	ddlFile = "ddl.18021904205944455167"
+	collected, err = s.client.NeedRestoreDDL(ddlFile)
+	c.Assert(err, IsNil)
+	c.Assert(collected, IsTrue)
+
+	// the file first event's ts is large than end ts, skip it.
+	ddlFile = "ddl.18021904187148730365"
+	collected, err = s.client.NeedRestoreDDL(ddlFile)
+	c.Assert(err, IsNil)
+	c.Assert(collected, IsFalse)
 }