Fix dd node filter bug (#5736)

Signed-off-by: sunby <bingyi.sun@zilliz.com>
2021-06-11 11:37:24 +08:00 · 2021-06-11 11:37:24 +08:00 · 8a0335be44
parent 33a43f48c8
commit 8a0335be44
1 changed files with 2 additions and 3 deletions
--- a/internal/datanode/flow_graph_dd_node.go
+++ b/internal/datanode/flow_graph_dd_node.go
@ -90,7 +90,6 @@ func (ddn *ddNode) Operate(in []flowgraph.Msg) []flowgraph.Msg {
 					continue
 				}
 			}
-
 			iMsg.insertMessages = append(iMsg.insertMessages, msg.(*msgstream.InsertMsg))
 		}
 	}
@ -110,10 +109,10 @@ func (ddn *ddNode) filterFlushedSegmentInsertMessages(msg *msgstream.InsertMsg)

 	ddn.mu.Lock()
 	if si, ok := ddn.seg2SegInfo[msg.GetSegmentID()]; ok {
-		if msg.EndTs() > si.GetDmlPosition().GetTimestamp() {
-			delete(ddn.seg2SegInfo, msg.GetSegmentID())
+		if msg.EndTs() <= si.GetDmlPosition().GetTimestamp() {
 			return true
 		}
+		delete(ddn.seg2SegInfo, msg.GetSegmentID())
 	}

 	ddn.mu.Unlock()