optimize: optimize parsePkValues to reduce nested loops (#1018)

* optimize: optimize parsePkValues to reduce nested loops

* optimize parsePkValues primary key extraction

* fix

Author: everfid-ever

pkg/datasource/sql/undo/executor/executor.go

@@ -168,21 +168,28 @@ func (b *BaseExecutor) queryCurrentRecords(ctx context.Context, conn *sql.Conn)
 }
 
 func (b *BaseExecutor) parsePkValues(rows []types.RowImage, pkNameList []string) map[string][]types.ColumnImage {
+	if len(rows) == 0 {
+		return make(map[string][]types.ColumnImage)
+	}
+
+	pkLookup := make(map[string]string, len(pkNameList))
+	for _, pk := range pkNameList {
+		pkLookup[strings.ToLower(pk)] = pk
+	}
+
 	pkValues := make(map[string][]types.ColumnImage)
-	// todo optimize 3 fors
+
 	for _, row := range rows {
 		for _, column := range row.Columns {
-			for _, pk := range pkNameList {
-				if strings.EqualFold(pk, column.ColumnName) {
-					values := pkValues[strings.ToUpper(pk)]
-					if values == nil {
-						values = make([]types.ColumnImage, 0)
-					}
-					values = append(values, column)
-					pkValues[pk] = values
+			columnNameLower := strings.ToLower(column.ColumnName)
+			if originalPk, exists := pkLookup[columnNameLower]; exists {
+				if pkValues[originalPk] == nil {
+					pkValues[originalPk] = make([]types.ColumnImage, 0, len(rows))
 				}
+				pkValues[originalPk] = append(pkValues[originalPk], column)
 			}
 		}
 	}
+
 	return pkValues
 }