lightning: pick the first file to check schema

br/pkg/lightning/restore/check_info.go

@@ -594,7 +594,11 @@ func (rc *Controller) SchemaIsValid(ctx context.Context, tableInfo *mydump.MDTab
 	// tidb_rowid have a default value.
 	defaultCols[model.ExtraHandleName.String()] = struct{}{}
 
-	for _, dataFile := range tableInfo.DataFiles {
+	// only check the first file of this table.
+	if len(tableInfo.DataFiles) > 0 {
+		dataFile := tableInfo.DataFiles[0]
+		log.L().Info("datafile to check", zap.String("db", tableInfo.DB),
+			zap.String("table", tableInfo.Name), zap.String("path", dataFile.FileMeta.Path))
 		// get columns name from data file.
 		dataFileMeta := dataFile.FileMeta
 
@@ -608,7 +612,7 @@ func (rc *Controller) SchemaIsValid(ctx context.Context, tableInfo *mydump.MDTab
 		}
 		if colsFromDataFile == nil && colCountFromDataFile == 0 {
 			log.L().Info("file contains no data, skip checking against schema validity", zap.String("path", dataFileMeta.Path))
-			continue
+			return msgs, nil
 		}
 
 		if colsFromDataFile == nil {
@@ -669,9 +673,6 @@ func (rc *Controller) SchemaIsValid(ctx context.Context, tableInfo *mydump.MDTab
 					tableInfo.DB, tableInfo.Name, col, col))
 			}
 		}
-		if len(msgs) > 0 {
-			return msgs, nil
-		}
 	}
 	return msgs, nil
 }

br/pkg/lightning/restore/restore_test.go

@@ -2205,6 +2205,65 @@ func (s *tableRestoreSuite) TestSchemaIsValid(c *C) {
 				},
 			},
 		},
+		// Case 4:
+		// table4 has two datafiles for table. we only check the first file.
+		// we expect the check success.
+		{
+			[]*config.IgnoreColumns{
+				{
+					DB:      "db1",
+					Table:   "table2",
+					Columns: []string{"cola"},
+				},
+			},
+			"",
+			0,
+			true,
+			map[string]*checkpoints.TidbDBInfo{
+				"db1": {
+					Name: "db1",
+					Tables: map[string]*checkpoints.TidbTableInfo{
+						"table2": {
+							ID:   1,
+							DB:   "db1",
+							Name: "table2",
+							Core: &model.TableInfo{
+								Columns: []*model.ColumnInfo{
+									{
+										// colB has the default value
+										Name:          model.NewCIStr("colB"),
+										DefaultIsExpr: true,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			&mydump.MDTableMeta{
+				DB:   "db1",
+				Name: "table2",
+				DataFiles: []mydump.FileInfo{
+					{
+						FileMeta: mydump.SourceFileMeta{
+							FileSize: 1 * units.TiB,
+							Path:     case2File,
+							Type:     mydump.SourceTypeCSV,
+						},
+					},
+					{
+						FileMeta: mydump.SourceFileMeta{
+							FileSize: 1 * units.TiB,
+							Path:     case2File,
+							// This type will make the check failed.
+							// but it's the second file for table.
+							// so it's unreachable so this case will success.
+							Type: mydump.SourceTypeIgnore,
+						},
+					},
+				},
+			},
+		},
 	}
 
 	for _, ca := range cases {