/
event.go
179 lines (160 loc) · 4.84 KB
/
event.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
// Copyright 2022 PingCAP, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.
package txn
import (
"encoding/binary"
"hash/fnv"
"sort"
"strings"
"time"
"github.com/pingcap/log"
"github.com/pingcap/tidb/pkg/parser/mysql"
"github.com/pingcap/tiflow/cdc/model"
"github.com/pingcap/tiflow/cdc/sink/dmlsink"
"go.uber.org/zap"
)
type txnEvent struct {
*dmlsink.TxnCallbackableEvent
start time.Time
conflictResolved time.Time
}
func newTxnEvent(event *dmlsink.TxnCallbackableEvent) *txnEvent {
return &txnEvent{TxnCallbackableEvent: event, start: time.Now()}
}
func (e *txnEvent) OnConflictResolved() {
e.conflictResolved = time.Now()
}
// GenSortedDedupKeysHash implements causality.txnEvent interface.
func (e *txnEvent) GenSortedDedupKeysHash(numSlots uint64) []uint64 {
hashes := genTxnKeys(e.TxnCallbackableEvent.Event)
// Sort and dedup hashes.
// Sort hashes by `hash % numSlots` to avoid deadlock, and then dedup
// hashes, so the same txn will not check confict with the same hash twice to
// prevent potential cyclic self dependency in the causality dependency
// graph.
return sortAndDedupHashes(hashes, numSlots)
}
func sortAndDedupHashes(hashes []uint64, numSlots uint64) []uint64 {
if len(hashes) == 0 {
return nil
}
// Sort hashes by `hash % numSlots` to avoid deadlock.
sort.Slice(hashes, func(i, j int) bool { return hashes[i]%numSlots < hashes[j]%numSlots })
// Dedup hashes
last := hashes[0]
j := 1
for i, hash := range hashes {
if i == 0 {
// skip first one, start checking duplication from 2nd one
continue
}
if hash == last {
continue
}
last = hash
hashes[j] = hash
j++
}
hashes = hashes[:j]
return hashes
}
// genTxnKeys returns hash keys for `txn`.
func genTxnKeys(txn *model.SingleTableTxn) []uint64 {
if len(txn.Rows) == 0 {
return nil
}
hashRes := make(map[uint64]struct{}, len(txn.Rows))
hasher := fnv.New32a()
for _, row := range txn.Rows {
for _, key := range genRowKeys(row) {
if n, err := hasher.Write(key); n != len(key) || err != nil {
log.Panic("transaction key hash fail")
}
hashRes[uint64(hasher.Sum32())] = struct{}{}
hasher.Reset()
}
}
keys := make([]uint64, 0, len(hashRes))
for key := range hashRes {
keys = append(keys, key)
}
return keys
}
func genRowKeys(row *model.RowChangedEvent) [][]byte {
var keys [][]byte
if len(row.Columns) != 0 {
for iIdx, idxCol := range row.TableInfo.IndexColumnsOffset {
key := genKeyList(row.GetColumns(), iIdx, idxCol, row.PhysicalTableID)
if len(key) == 0 {
continue
}
keys = append(keys, key)
}
}
if len(row.PreColumns) != 0 {
for iIdx, idxCol := range row.TableInfo.IndexColumnsOffset {
key := genKeyList(row.GetPreColumns(), iIdx, idxCol, row.PhysicalTableID)
if len(key) == 0 {
continue
}
keys = append(keys, key)
}
}
if len(keys) == 0 {
// use table ID as key if no key generated (no PK/UK),
// no concurrence for rows in the same table.
log.Debug("Use table id as the key", zap.Int64("tableID", row.PhysicalTableID))
tableKey := make([]byte, 8)
binary.BigEndian.PutUint64(tableKey, uint64(row.PhysicalTableID))
keys = [][]byte{tableKey}
}
return keys
}
func genKeyList(
columns []*model.Column, iIdx int, colIdx []int, tableID int64,
) []byte {
var key []byte
for _, i := range colIdx {
// if a column value is null, we can ignore this index
// If the index contain generated column, we can't use this key to detect conflict with other DML,
// Because such as insert can't specify the generated value.
if columns[i] == nil || columns[i].Value == nil || columns[i].Flag.IsGeneratedColumn() {
return nil
}
val := model.ColumnValueString(columns[i].Value)
if columnNeeds2LowerCase(columns[i].Type, columns[i].Collation) {
val = strings.ToLower(val)
}
key = append(key, []byte(val)...)
key = append(key, 0)
}
if len(key) == 0 {
return nil
}
tableKey := make([]byte, 16)
binary.BigEndian.PutUint64(tableKey[:8], uint64(iIdx))
binary.BigEndian.PutUint64(tableKey[8:], uint64(tableID))
key = append(key, tableKey...)
return key
}
func columnNeeds2LowerCase(mysqlType byte, collation string) bool {
switch mysqlType {
case mysql.TypeVarchar, mysql.TypeString, mysql.TypeVarString, mysql.TypeTinyBlob,
mysql.TypeMediumBlob, mysql.TypeBlob, mysql.TypeLongBlob:
return collationNeeds2LowerCase(collation)
}
return false
}
func collationNeeds2LowerCase(collation string) bool {
return strings.HasSuffix(collation, "_ci")
}