Skip to content

Commit

Permalink
Fix flush panic after compaction (milvus-io#18677)
Browse files Browse the repository at this point in the history 
See also: milvus-io#18565

Signed-off-by: yangxuan <xuan.yang@zilliz.com>

Signed-off-by: yangxuan <xuan.yang@zilliz.com>
  • Loading branch information
@XuanYang-cn
XuanYang-cn authored Aug 17, 2022
1 parent 0db281c commit 16dcd97
Show file tree
Hide file tree
Showing 3 changed files with 227 additions and 73 deletions.
151 changes: 83 additions & 68 deletions internal/datanode/flow_graph_delete_node.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -105,71 +105,6 @@ func (dn *deleteNode) Close() {
log.Info("Flowgraph Delete Node closing")
}

func (dn *deleteNode) bufferDeleteMsg(msg *msgstream.DeleteMsg, tr TimeRange) ([]UniqueID, error) {
log.Debug("bufferDeleteMsg", zap.Any("primary keys", msg.PrimaryKeys), zap.String("vChannelName", dn.channelName))

// Update delBuf for merged segments
compactedTo2From := dn.replica.listCompactedSegmentIDs()
for compactedTo, compactedFrom := range compactedTo2From {
compactToDelBuff := newDelDataBuf()
for _, segID := range compactedFrom {
value, loaded := dn.delBuf.LoadAndDelete(segID)
if loaded {
compactToDelBuff.updateFromBuf(value.(*DelDataBuf))
}
}
dn.delBuf.Store(compactedTo, compactToDelBuff)
dn.replica.removeSegments(compactedFrom...)
log.Debug("update delBuf for merged segments",
zap.Int64("compactedTo segmentID", compactedTo),
zap.Int64s("compactedFrom segmentIDs", compactedFrom),
)
}

primaryKeys := storage.ParseIDs2PrimaryKeys(msg.PrimaryKeys)
segIDToPks, segIDToTss := dn.filterSegmentByPK(msg.PartitionID, primaryKeys, msg.Timestamps)

segIDs := make([]UniqueID, 0, len(segIDToPks))
for segID, pks := range segIDToPks {
segIDs = append(segIDs, segID)

rows := len(pks)
tss, ok := segIDToTss[segID]
if !ok || rows != len(tss) {
return nil, fmt.Errorf("primary keys and timestamp's element num mis-match, segmentID = %d", segID)
}

var delDataBuf *DelDataBuf
value, ok := dn.delBuf.Load(segID)
if ok {
delDataBuf = value.(*DelDataBuf)
} else {
delDataBuf = newDelDataBuf()
}
delData := delDataBuf.delData

for i := 0; i < rows; i++ {
delData.Pks = append(delData.Pks, pks[i])
delData.Tss = append(delData.Tss, tss[i])
// this log will influence the data node performance as it may be too many,
// only use it when we focus on delete operations
//log.Debug("delete",
// zap.Any("primary key", pks[i]),
// zap.Uint64("ts", tss[i]),
// zap.Int64("segmentID", segID),
// zap.String("vChannelName", dn.channelName))
}

// store
delDataBuf.updateSize(int64(rows))
metrics.DataNodeConsumeMsgRowsCount.WithLabelValues(fmt.Sprint(Params.DataNodeCfg.GetNodeID()), metrics.DeleteLabel).Add(float64(rows))
delDataBuf.updateTimeRange(tr)
dn.delBuf.Store(segID, delDataBuf)
}

return segIDs, nil
}

func (dn *deleteNode) showDelBuf(segIDs []UniqueID, ts Timestamp) {
for _, segID := range segIDs {
if v, ok := dn.delBuf.Load(segID); ok {
Expand Down Expand Up @@ -209,6 +144,12 @@ func (dn *deleteNode) Operate(in []Msg) []Msg {
msg.SetTraceCtx(ctx)
}

// update compacted segment before operation
if len(fgMsg.deleteMessages) > 0 || len(fgMsg.segmentsToFlush) > 0 {
dn.updateCompactedSegments()
}

// process delete messages
var segIDs []UniqueID
for i, msg := range fgMsg.deleteMessages {
traceID, _, _ := trace.InfoFromSpan(spans[i])
Expand All @@ -224,12 +165,12 @@ func (dn *deleteNode) Operate(in []Msg) []Msg {
segIDs = append(segIDs, tmpSegIDs...)
}

// show changed segment's status in dn.delBuf of a certain ts
// display changed segment's status in dn.delBuf of a certain ts
if len(fgMsg.deleteMessages) != 0 {
dn.showDelBuf(segIDs, fgMsg.timeRange.timestampMax)
}

// handle flush
// process flush messages
if len(fgMsg.segmentsToFlush) > 0 {
log.Debug("DeleteNode receives flush message",
zap.Int64s("segIDs", fgMsg.segmentsToFlush),
Expand All @@ -254,7 +195,7 @@ func (dn *deleteNode) Operate(in []Msg) []Msg {
}
}

// drop collection signal, delete node shall notify flush manager all data are cleared and send signal to DataSyncService cleaner
// process drop collection message, delete node shall notify flush manager all data are cleared and send signal to DataSyncService cleaner
if fgMsg.dropCollection {
dn.flushManager.notifyAllFlushed()
log.Debug("DeleteNode notifies BackgroundGC to release vchannel", zap.String("vChannelName", dn.channelName))
Expand All @@ -267,6 +208,80 @@ func (dn *deleteNode) Operate(in []Msg) []Msg {
return nil
}

// update delBuf for compacted segments
func (dn *deleteNode) updateCompactedSegments() {
compactedTo2From := dn.replica.listCompactedSegmentIDs()
for compactedTo, compactedFrom := range compactedTo2From {
var compactToDelBuff *DelDataBuf
delBuf, loaded := dn.delBuf.Load(compactedTo)
if !loaded {
compactToDelBuff = newDelDataBuf()
} else {
compactToDelBuff = delBuf.(*DelDataBuf)
}

for _, segID := range compactedFrom {
if value, loaded := dn.delBuf.LoadAndDelete(segID); loaded {
compactToDelBuff.updateFromBuf(value.(*DelDataBuf))
dn.delBuf.Store(compactedTo, compactToDelBuff)

}
}
log.Debug("update delBuf for compacted segments",
zap.Int64("compactedTo segmentID", compactedTo),
zap.Int64s("compactedFrom segmentIDs", compactedFrom),
)
dn.replica.removeSegments(compactedFrom...)
}
}

func (dn *deleteNode) bufferDeleteMsg(msg *msgstream.DeleteMsg, tr TimeRange) ([]UniqueID, error) {
log.Debug("bufferDeleteMsg", zap.Any("primary keys", msg.PrimaryKeys), zap.String("vChannelName", dn.channelName))

primaryKeys := storage.ParseIDs2PrimaryKeys(msg.PrimaryKeys)
segIDToPks, segIDToTss := dn.filterSegmentByPK(msg.PartitionID, primaryKeys, msg.Timestamps)

segIDs := make([]UniqueID, 0, len(segIDToPks))
for segID, pks := range segIDToPks {
segIDs = append(segIDs, segID)

rows := len(pks)
tss, ok := segIDToTss[segID]
if !ok || rows != len(tss) {
return nil, fmt.Errorf("primary keys and timestamp's element num mis-match, segmentID = %d", segID)
}

var delDataBuf *DelDataBuf
value, ok := dn.delBuf.Load(segID)
if ok {
delDataBuf = value.(*DelDataBuf)
} else {
delDataBuf = newDelDataBuf()
}
delData := delDataBuf.delData

for i := 0; i < rows; i++ {
delData.Pks = append(delData.Pks, pks[i])
delData.Tss = append(delData.Tss, tss[i])
// this log will influence the data node performance as it may be too many,
// only use it when we focus on delete operations
//log.Debug("delete",
// zap.Any("primary key", pks[i]),
// zap.Uint64("ts", tss[i]),
// zap.Int64("segmentID", segID),
// zap.String("vChannelName", dn.channelName))
}

// store
delDataBuf.updateSize(int64(rows))
metrics.DataNodeConsumeMsgRowsCount.WithLabelValues(fmt.Sprint(Params.DataNodeCfg.GetNodeID()), metrics.DeleteLabel).Add(float64(rows))
delDataBuf.updateTimeRange(tr)
dn.delBuf.Store(segID, delDataBuf)
}

return segIDs, nil
}

// filterSegmentByPK returns the bloom filter check result.
// If the key may exists in the segment, returns it in map.
// If the key not exists in the segment, the segment is filter out.
Expand Down
135 changes: 135 additions & 0 deletions internal/datanode/flow_graph_delete_node_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -358,6 +358,7 @@ func TestFlowGraphDeleteNode_Operate(t *testing.T) {
// send again shall trigger empty buffer flush
delNode.Operate([]flowgraph.Msg{fgMsg})
})

t.Run("Test deleteNode Operate valid with dropCollection", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
Expand Down Expand Up @@ -426,6 +427,56 @@ func TestFlowGraphDeleteNode_Operate(t *testing.T) {
delNode.Operate([]flowgraph.Msg{fgMsg})
})
})

t.Run("Test issue#18565", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

chanName := "datanode-test-FlowGraphDeletenode-issue18565"
testPath := "/test/datanode/root/meta"
assert.NoError(t, clearEtcd(testPath))
Params.EtcdCfg.MetaRootPath = testPath
Params.DataNodeCfg.DeleteBinlogRootPath = testPath

replica := &SegmentReplica{
newSegments: make(map[UniqueID]*Segment),
normalSegments: make(map[UniqueID]*Segment),
flushedSegments: make(map[UniqueID]*Segment),
compactedSegments: make(map[UniqueID]*Segment),
}

c := &nodeConfig{
replica: replica,
allocator: NewAllocatorFactory(),
vChannelName: chanName,
}
delNode, err := newDeleteNode(ctx, fm, make(chan string, 1), c)
assert.Nil(t, err)

compactedSegment := UniqueID(10020987)
replica.compactedSegments[compactedSegment] = &Segment{
segmentID: compactedSegment,
compactedTo: 100,
}

msg := genFlowGraphDeleteMsg(int64Pks, chanName)
msg.deleteMessages = []*msgstream.DeleteMsg{}
msg.segmentsToFlush = []UniqueID{compactedSegment}

delNode.delBuf.Store(compactedSegment, &DelDataBuf{delData: &DeleteData{}})
delNode.flushManager = NewRendezvousFlushManager(&allocator{}, cm, replica, func(*segmentFlushPack) {}, emptyFlushAndDropFunc)

var fgMsg flowgraph.Msg = &msg
flowGraphRetryOpt = retry.Attempts(1)
assert.NotPanics(t, func() {
delNode.Operate([]flowgraph.Msg{fgMsg})
})

_, ok := delNode.delBuf.Load(100)
assert.False(t, ok)
_, ok = delNode.delBuf.Load(compactedSegment)
assert.False(t, ok)
})
}

func TestFlowGraphDeleteNode_showDelBuf(t *testing.T) {
Expand Down Expand Up @@ -468,3 +519,87 @@ func TestFlowGraphDeleteNode_showDelBuf(t *testing.T) {

delNode.showDelBuf([]UniqueID{111, 112, 113}, 100)
}

func TestFlowGraphDeleteNode_updateCompactedSegments(t *testing.T) {
cm := storage.NewLocalChunkManager(storage.RootPath(deleteNodeTestDir))
defer cm.RemoveWithPrefix("")

fm := NewRendezvousFlushManager(NewAllocatorFactory(), cm, &mockReplica{}, func(*segmentFlushPack) {}, emptyFlushAndDropFunc)

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

chanName := "datanode-test-FlowGraphDeletenode-showDelBuf"
testPath := "/test/datanode/root/meta"
assert.NoError(t, clearEtcd(testPath))
Params.EtcdCfg.MetaRootPath = testPath
Params.DataNodeCfg.DeleteBinlogRootPath = testPath

replica := SegmentReplica{
newSegments: make(map[UniqueID]*Segment),
normalSegments: make(map[UniqueID]*Segment),
flushedSegments: make(map[UniqueID]*Segment),
compactedSegments: make(map[UniqueID]*Segment),
}

c := &nodeConfig{
replica: &replica,
allocator: NewAllocatorFactory(),
vChannelName: chanName,
}
delNode, err := newDeleteNode(ctx, fm, make(chan string, 1), c)
require.NoError(t, err)

tests := []struct {
description string
segIDsInBuffer []UniqueID
compactedToIDs []UniqueID
compactedFromIDs []UniqueID

expectedSegsRemain []UniqueID
}{
{"zero segments",
[]UniqueID{}, []UniqueID{}, []UniqueID{}, []UniqueID{}},
{"segment no compaction",
[]UniqueID{100, 101}, []UniqueID{}, []UniqueID{}, []UniqueID{100, 101}},
{"segment compacted not in buffer",
[]UniqueID{100, 101}, []UniqueID{200}, []UniqueID{103}, []UniqueID{100, 101}},
{"segment compacted in buffer one",
[]UniqueID{100, 101}, []UniqueID{201}, []UniqueID{100}, []UniqueID{101, 201}},
{"segment compacted in buffer all-1",
[]UniqueID{100, 101}, []UniqueID{201, 201}, []UniqueID{100, 101}, []UniqueID{201}},
{"segment compacted in buffer all-2",
[]UniqueID{100, 101}, []UniqueID{201, 202}, []UniqueID{100, 101}, []UniqueID{201, 202}},
}

for _, test := range tests {
t.Run(test.description, func(t *testing.T) {
for _, seg := range test.segIDsInBuffer {
delBuf := newDelDataBuf()
delNode.delBuf.Store(seg, delBuf)
}

for i, seg := range test.compactedFromIDs {
replica.compactedSegments[seg] = &Segment{
segmentID: seg,
compactedTo: test.compactedToIDs[i],
}
}

delNode.updateCompactedSegments()

for _, remain := range test.expectedSegsRemain {
_, ok := delNode.delBuf.Load(remain)
assert.True(t, ok)
}

var count int
delNode.delBuf.Range(func(key, value interface{}) bool {
count++
return true
})

assert.Equal(t, len(test.expectedSegsRemain), count)
})
}
}
14 changes: 9 additions & 5 deletions internal/datanode/segment_replica.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,12 +41,16 @@ const (
maxBloomFalsePositive float64 = 0.005
)

type primaryKey = storage.PrimaryKey
type int64PrimaryKey = storage.Int64PrimaryKey
type varCharPrimaryKey = storage.VarCharPrimaryKey
type (
primaryKey = storage.PrimaryKey
int64PrimaryKey = storage.Int64PrimaryKey
varCharPrimaryKey = storage.VarCharPrimaryKey
)

var newInt64PrimaryKey = storage.NewInt64PrimaryKey
var newVarCharPrimaryKey = storage.NewVarCharPrimaryKey
var (
newInt64PrimaryKey = storage.NewInt64PrimaryKey
newVarCharPrimaryKey = storage.NewVarCharPrimaryKey
)

// Replica is DataNode unique replication
type Replica interface {
Expand Down

0 comments on commit 16dcd97

Please sign in to comment.