feat: add reading of transaction info file

Signed-off-by: Andres Taylor <andres@planetscale.com>

Author: systay

go/summarize/reading.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/vitessio/vt/go/keys"
 	"github.com/vitessio/vt/go/schema"
+	"github.com/vitessio/vt/go/transactions"
 )
 
 func readTracedFile(fileName string) traceSummary {
@@ -60,6 +61,27 @@ func readTracedFile(fileName string) traceSummary {
 	}
 }
 
+func readTransactionFile(fileName string) func(s *Summary) error {
+	c, err := os.ReadFile(fileName)
+	if err != nil {
+		exit("Error opening file: " + err.Error())
+	}
+
+	type txOutput struct {
+		FileType   string                   `json:"fileType"`
+		Signatures []transactions.Signature `json:"signatures"`
+	}
+
+	var to txOutput
+	err = json.Unmarshal(c, &to)
+	if err != nil {
+		exit("Error parsing json: " + err.Error())
+	}
+	return func(*Summary) error {
+		return nil
+	}
+}
+
 func readKeysFile(fileName string) func(s *Summary) error {
 	c, err := os.ReadFile(fileName)
 	if err != nil {

go/summarize/reading_test.go

@@ -0,0 +1,28 @@
+/*
+Copyright 2024 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package summarize
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestReadTransaction(t *testing.T) {
+	file := readTransactionFile("../testdata/small-slow-query-transactions.json")
+	assert.NotNil(t, file)
+}

go/summarize/summarize.go

@@ -61,12 +61,15 @@ func Run(files []string, hotMetric string) {
 	var workers []summaryWorker
 
 	for _, file := range files {
-		typ, _ := getFileType(file)
+		typ, err := getFileType(file)
+		if err != nil {
+			panic(err.Error())
+		}
 		switch typ {
 		case dbInfoFile:
 			workers = append(workers, readDBInfoFile(file))
 		case transactionFile:
-			fmt.Printf("transaction file: %s\n", file)
+			workers = append(workers, readTransactionFile(file))
 		case traceFile:
 			traces = append(traces, readTracedFile(file))
 		case keysFile:
@@ -76,16 +79,22 @@ func Run(files []string, hotMetric string) {
 		}
 	}
 
-	checkTraceConditions(traces, workers, hotMetric)
-
-	if len(traces) == 2 {
-		compareTraces(os.Stdout, terminalWidth(), highlightQuery, traces[0], traces[1])
+	traceCount := len(traces)
+	if traceCount <= 0 {
+		printSummary(hotMetric, workers)
 		return
-	} else if len(traces) == 1 {
+	}
+
+	checkTraceConditions(traces, workers, hotMetric)
+	switch traceCount {
+	case 1:
 		printTraceSummary(os.Stdout, terminalWidth(), highlightQuery, traces[0])
-		return
+	case 2:
+		compareTraces(os.Stdout, terminalWidth(), highlightQuery, traces[0], traces[1])
 	}
+}
 
+func printSummary(hotMetric string, workers []summaryWorker) {
 	s := NewSummary(hotMetric)
 	for _, worker := range workers {
 		err := worker(s)
@@ -134,9 +143,6 @@ func (s *Summary) AddTable(table *TableSummary) {
 }
 
 func checkTraceConditions(traces []traceSummary, workers []summaryWorker, hotMetric string) {
-	if len(traces) == 0 {
-		return
-	}
 	if len(workers) > 0 {
 		panic("Trace files cannot be mixed with other file types")
 	}

go/transactions/transaction_signature.go

@@ -28,20 +28,20 @@ import (
 )
 
 type (
-	TxSignature struct {
-		Count   int       `json:"count"`
-		Queries []TxQuery `json:"qqueries"`
+	Signature struct {
+		Count   int     `json:"count"`
+		Queries []Query `json:"qqueries"`
 	}
 
-	TxQuery struct {
+	Query struct {
 		Op             string          `json:"op"`
 		AffectedTable  string          `json:"affected_table"`
 		UpdatedColumns []string        `json:"updated_columns,omitempty"`
 		Predicates     []predicateInfo `json:"predicates,omitempty"`
 	}
 
 	txSignatureMap struct {
-		data map[uint64][]*TxSignature
+		data map[uint64][]*Signature
 	}
 
 	predicateInfo struct {
@@ -60,17 +60,17 @@ func (pi predicateInfo) String() string {
 	return fmt.Sprintf("%s.%s %s %s", pi.Table, pi.Col, pi.Op.ToString(), val)
 }
 
-func (tx *TxSignature) MarshalJSON() ([]byte, error) {
+func (tx *Signature) MarshalJSON() ([]byte, error) {
 	return json.Marshal(struct {
-		Count   int       `json:"count"`
-		Queries []TxQuery `json:"query-signatures"`
+		Count   int     `json:"count"`
+		Queries []Query `json:"query-signatures"`
 	}{
 		Count:   tx.Count,
 		Queries: tx.Queries,
 	})
 }
 
-func (tx *TxSignature) Hash64() uint64 {
+func (tx *Signature) Hash64() uint64 {
 	hasher := fnv.New64a()
 
 	for _, query := range tx.Queries {
@@ -80,7 +80,7 @@ func (tx *TxSignature) Hash64() uint64 {
 	return hasher.Sum64()
 }
 
-func (tx TxQuery) addToHash(hash hash.Hash64) {
+func (tx Query) addToHash(hash hash.Hash64) {
 	_, _ = hash.Write([]byte(tx.Op))
 	_, _ = hash.Write([]byte{0})
 	_, _ = hash.Write([]byte(tx.AffectedTable))
@@ -97,7 +97,7 @@ func (tx TxQuery) addToHash(hash hash.Hash64) {
 	}
 }
 
-func (tx TxQuery) Equals(other TxQuery) bool {
+func (tx Query) Equals(other Query) bool {
 	if tx.Op != other.Op {
 		return false
 	}
@@ -125,19 +125,19 @@ func (tx TxQuery) Equals(other TxQuery) bool {
 
 func newTxSignatureMap() *txSignatureMap {
 	return &txSignatureMap{
-		data: make(map[uint64][]*TxSignature),
+		data: make(map[uint64][]*Signature),
 	}
 }
 
-func (m *txSignatureMap) Add(tx *TxSignature) {
+func (m *txSignatureMap) Add(tx *Signature) {
 	hash := tx.Hash64()
 
 	bucket, exists := m.data[hash]
 
 	// Check if the hash already exists
 	if !exists {
 		tx.Count = 1
-		m.data[hash] = []*TxSignature{tx}
+		m.data[hash] = []*Signature{tx}
 		return
 	}
 
@@ -153,7 +153,7 @@ func (m *txSignatureMap) Add(tx *TxSignature) {
 	m.data[hash] = append(bucket, tx)
 }
 
-func (tx *TxSignature) Equals(other *TxSignature) bool {
+func (tx *Signature) Equals(other *Signature) bool {
 	if len(tx.Queries) != len(other.Queries) {
 		return false
 	}
@@ -167,7 +167,7 @@ func (tx *TxSignature) Equals(other *TxSignature) bool {
 }
 
 // CleanUp removes values that are only used once and replaces them with -1
-func (tx *TxSignature) CleanUp() *TxSignature {
+func (tx *Signature) CleanUp() *Signature {
 	usedValues := make(map[int]int)
 
 	// First let's count how many times each value is used
@@ -180,7 +180,7 @@ func (tx *TxSignature) CleanUp() *TxSignature {
 	// Now we replace values only used once with -1
 	newCount := 0
 	newValues := make(map[int]int)
-	newQueries := make([]TxQuery, 0, len(tx.Queries))
+	newQueries := make([]Query, 0, len(tx.Queries))
 	for _, query := range tx.Queries {
 		newPredicates := make([]predicateInfo, 0, len(query.Predicates))
 		for _, predicate := range query.Predicates {
@@ -198,23 +198,23 @@ func (tx *TxSignature) CleanUp() *TxSignature {
 			}
 			newPredicates = append(newPredicates, predicate)
 		}
-		newQueries = append(newQueries, TxQuery{
+		newQueries = append(newQueries, Query{
 			Op:             query.Op,
 			AffectedTable:  query.AffectedTable,
 			UpdatedColumns: query.UpdatedColumns,
 			Predicates:     newPredicates,
 		})
 	}
 
-	return &TxSignature{
+	return &Signature{
 		Queries: newQueries,
 		Count:   tx.Count,
 	}
 }
 
 func (m *txSignatureMap) MarshalJSON() ([]byte, error) {
 	// Collect all interesting TxSignatures into a slice
-	var signatures []*TxSignature
+	var signatures []*Signature
 	for _, bucket := range m.data {
 		for _, txSig := range bucket {
 			if txSig.Count > 1 {

go/transactions/transactions.go

@@ -204,7 +204,7 @@ func (s *state) consume(ch <-chan []sqlparser.Statement, wg *sync.WaitGroup) {
 	defer wg.Done()
 	for queries := range ch {
 		n := &normalizer{m: make(map[string]int)}
-		tx := &TxSignature{}
+		tx := &Signature{}
 		for _, query := range queries {
 			st, err := semantics.Analyze(query, "ks", s.si)
 			if err != nil {
@@ -222,7 +222,7 @@ func (s *state) consume(ch <-chan []sqlparser.Statement, wg *sync.WaitGroup) {
 	}
 }
 
-func (s *state) consumeUpdate(query *sqlparser.Update, st *semantics.SemTable, n *normalizer, tx *TxSignature) {
+func (s *state) consumeUpdate(query *sqlparser.Update, st *semantics.SemTable, n *normalizer, tx *Signature) {
 	// Find all predicates in the where clause that use a column and a literal
 	var predicates []predicateInfo
 	if query.Where != nil {
@@ -239,15 +239,15 @@ func (s *state) consumeUpdate(query *sqlparser.Update, st *semantics.SemTable, n
 		panic("multi-table updates not supported")
 	}
 
-	tx.Queries = append(tx.Queries, TxQuery{
+	tx.Queries = append(tx.Queries, Query{
 		Op:             "update",
 		AffectedTable:  sqlparser.String(query.TableExprs[0]),
 		UpdatedColumns: updatedColumns,
 		Predicates:     predicates,
 	})
 }
 
-func (s *state) consumeDelete(del *sqlparser.Delete, st *semantics.SemTable, n *normalizer, tx *TxSignature) {
+func (s *state) consumeDelete(del *sqlparser.Delete, st *semantics.SemTable, n *normalizer, tx *Signature) {
 	var predicates []predicateInfo
 	if del.Where != nil {
 		predicates = getPredicates(del.Where.Expr, st, n)
@@ -257,14 +257,14 @@ func (s *state) consumeDelete(del *sqlparser.Delete, st *semantics.SemTable, n *
 		panic("multi-table updates not supported")
 	}
 
-	tx.Queries = append(tx.Queries, TxQuery{
+	tx.Queries = append(tx.Queries, Query{
 		Op:            "delete",
 		AffectedTable: sqlparser.String(del.TableExprs[0]),
 		Predicates:    predicates,
 	})
 }
 
-func (s *state) addSignature(tx *TxSignature) {
+func (s *state) addSignature(tx *Signature) {
 	s.mu.Lock()
 	defer s.mu.Unlock()