enhance: add create segment message, enable empty segment flush #37407

chyezh · 2024-11-04T06:52:15Z

add redo interceptor to implement append context refresh. (make new timetick)
add create segment handler for flusher.
make empty segment flushable and directly change it into dropped.
add create segment message into wal when creating new growing segment.
make the insert operation into following seq: createSegment -> insert -> insert -> flushSegment.
make manual flush into following seq: flushTs -> flushsegment -> flushsegment -> manualflush.

sre-ci-robot · 2024-11-04T06:52:26Z

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by: chyezh
To complete the pull request process, please assign yanliang567 after the PR has been reviewed.
You can assign the PR to them by writing /assign @yanliang567 in a comment when ready.

The full list of commands accepted by this bot can be found here.

Needs approval from an approver in each of these files:

OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

mergify · 2024-11-04T07:05:20Z

@chyezh go-sdk check failed, comment rerun go-sdk can trigger the job again.

mergify · 2024-11-04T07:14:07Z

@chyezh E2e jenkins job failed, comment /run-cpu-e2e can trigger the job again.

chyezh · 2024-11-04T07:26:53Z

internal/datacoord/meta.go

@@ -1013,6 +1013,24 @@ func UpdateIsImporting(segmentID int64, isImporting bool) UpdateOperator {
 	}
 }

+// UpdateAsDroppedIfEmptyWhenFlushing updates segment state to Dropped if segment is empty and in Flushing state
+// It's used to make a empty flushing segment to be dropped directly.
+func UpdateAsDroppedIfEmptyWhenFlushing(segmentID int64) UpdateOperator {


Used to set a empty segment into dropped when savebinlogpath

chyezh · 2024-11-04T07:27:39Z

internal/flushcommon/metacache/actions.go

@@ -147,6 +155,14 @@ func UpdateNumOfRows(numOfRows int64) SegmentAction {
 	}
 }

+func SetStartPositionIfNil(startPos *msgpb.MsgPosition) SegmentAction {


Setup startPosition when first insert message comes, applied to writebuffer

chyezh · 2024-11-04T07:28:30Z

internal/flushcommon/pipeline/flow_graph_dd_node.go

@@ -236,6 +236,19 @@ func (ddn *ddNode) Operate(in []Msg) []Msg {
 				WithLabelValues(fmt.Sprint(paramtable.GetNodeID()), metrics.DeleteLabel).
 				Add(float64(dmsg.GetNumRows()))
 			fgMsg.DeleteMessages = append(fgMsg.DeleteMessages, dmsg)
+		case commonpb.MsgType_CreateSegment:
+			createSegment := msg.(*adaptor.CreateSegmentMessageBody)


create new segment when create segment incoming.
TODO: Move the create new segment operation on datacoord at interceptor to here.

chyezh · 2024-11-04T07:29:40Z

internal/flushcommon/syncmgr/storage_serializer.go

@@ -257,7 +258,11 @@ func (s *storageV1Serializer) serializeMergedPkStats(pack *SyncPack) (*storage.B
 			BF:      pks.PkFilter,
 			PkType:  int64(s.pkField.GetDataType()),
 		}
-	}), segment.NumOfRows())
+	})
+	if len(stats) == 0 {


allow to merge a empty stats if the segment is empty.

mergify · 2024-11-04T08:15:08Z

@chyezh E2e jenkins job failed, comment /run-cpu-e2e can trigger the job again.

mergify · 2024-11-04T08:46:14Z

@chyezh go-sdk check failed, comment rerun go-sdk can trigger the job again.

mergify · 2024-11-04T08:46:15Z

@chyezh cpp-unit-test check failed, comment rerun cpp-unit-test can trigger the job again.

mergify · 2024-11-04T08:57:46Z

@chyezh E2e jenkins job failed, comment /run-cpu-e2e can trigger the job again.

bigsheeper · 2024-11-04T12:33:07Z

internal/flushcommon/pipeline/flow_graph_dd_node.go

+			logger := log.With(
+				zap.String("vchannel", ddn.Name()),
+				zap.Int32("msgType", int32(msg.Type())),
+				zap.Uint64("timetick", createSegment.CreateSegmentMessage.TimeTick()),


better to log segmentID as well

fix it soon

bigsheeper · 2024-11-04T12:47:02Z

internal/streamingnode/server/flusher/flushmsg_handler.go

@@ -16,9 +16,15 @@

 package flusher

-import "github.com/milvus-io/milvus/pkg/streaming/util/message"
+import (


rename the file?

bigsheeper · 2024-11-04T13:01:30Z

internal/streamingnode/server/wal/interceptors/segment/manager/partition_manager.go

 	m.mu.Lock()
 	defer m.mu.Unlock()
+	// no-op if the incoming time tick is less than the fenced time tick.
+	if timeTick <= m.fencedAssignTimeTick {


When would this happen?

Concurrent maunalflush incoming, The older manualflush will be ignored.

If manualflush trigger a flushsegment operation, it will redo the operation to generate a new timetick to keep the message sequence：flushTs -> flushsegment -> flushsegment -> manualflush. So the redo operation will be rejected here. In previous implementation, the sequence will be flushTs -> manualflush -> flushsegment -> flushsegment, it's wierd.

mergify · 2024-11-04T14:29:55Z

@chyezh E2e jenkins job failed, comment /run-cpu-e2e can trigger the job again.

codecov · 2024-11-04T15:04:07Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 67.31%. Comparing base (f813fb4) to head (1936a4d).
Report is 7 commits behind head on master.

Additional details and impacted files

@@           Coverage Diff           @@
##           master   #37407   +/-   ##
=======================================
  Coverage   67.31%   67.31%           
=======================================
  Files         290      290           
  Lines       25377    25377           
=======================================
  Hits        17082    17082           
  Misses       8295     8295

Components	Coverage Δ
Client	`∅ <ø> (∅)`
Core	`67.31% <ø> (ø)`
Go	`∅ <ø> (∅)`

mergify · 2024-11-05T09:24:08Z

@chyezh E2e jenkins job failed, comment /run-cpu-e2e can trigger the job again.

chyezh · 2024-11-05T15:10:54Z

rerun ut

mergify · 2024-11-06T01:39:30Z

@chyezh E2e jenkins job failed, comment /run-cpu-e2e can trigger the job again.

chyezh · 2024-11-06T03:33:32Z

/run-cpu-e2e

- add redo interceptor to implement append context refresh. (make new timetick) - add create segment handler for flusher. - make empty segment flushable and directly change it into dropped. - add create segment message into wal when creating new growing segment. Signed-off-by: chyezh <[email protected]>

mergify · 2024-11-07T04:11:05Z

@chyezh E2e jenkins job failed, comment /run-cpu-e2e can trigger the job again.

chyezh · 2024-11-07T04:50:33Z

/run-cpu-e2e

sre-ci-robot requested review from congqixia and czs007 November 4, 2024 06:52

sre-ci-robot added size/XXL Denotes a PR that changes 1000+ lines. area/dependency Pull requests that update a dependency file labels Nov 4, 2024

mergify bot added dco-passed DCO check passed. kind/enhancement Issues or changes related to enhancement labels Nov 4, 2024

chyezh force-pushed the fix_streaming_flush_empty_segment branch from 39b8028 to cb5a5e9 Compare November 4, 2024 07:25

chyezh commented Nov 4, 2024

View reviewed changes

chyezh force-pushed the fix_streaming_flush_empty_segment branch from cb5a5e9 to 6ab0ba1 Compare November 4, 2024 08:16

bigsheeper reviewed Nov 4, 2024

View reviewed changes

chyezh force-pushed the fix_streaming_flush_empty_segment branch 2 times, most recently from 0807515 to 9a09f1a Compare November 4, 2024 13:50

chyezh force-pushed the fix_streaming_flush_empty_segment branch from 9a09f1a to a2fbfc4 Compare November 5, 2024 08:35

chyezh requested a review from bigsheeper November 5, 2024 08:36

chyezh force-pushed the fix_streaming_flush_empty_segment branch from a2fbfc4 to 32bed9b Compare November 5, 2024 12:50

mergify bot added ci-passed and removed ci-passed labels Nov 5, 2024

mergify bot added the ci-passed label Nov 6, 2024

chyezh force-pushed the fix_streaming_flush_empty_segment branch from 32bed9b to 1936a4d Compare November 7, 2024 03:02

mergify bot removed the ci-passed label Nov 7, 2024

mergify bot added the ci-passed label Nov 7, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

enhance: add create segment message, enable empty segment flush #37407

enhance: add create segment message, enable empty segment flush #37407

chyezh commented Nov 4, 2024 •

edited

Loading

sre-ci-robot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

chyezh Nov 4, 2024

chyezh Nov 4, 2024

chyezh Nov 4, 2024

chyezh Nov 4, 2024

mergify bot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

bigsheeper Nov 4, 2024

chyezh Nov 5, 2024

bigsheeper Nov 4, 2024

bigsheeper Nov 4, 2024

chyezh Nov 5, 2024

mergify bot commented Nov 4, 2024

codecov bot commented Nov 4, 2024 •

edited

Loading

mergify bot commented Nov 5, 2024

chyezh commented Nov 5, 2024

mergify bot commented Nov 6, 2024

chyezh commented Nov 6, 2024

mergify bot commented Nov 7, 2024

chyezh commented Nov 7, 2024

enhance: add create segment message, enable empty segment flush #37407

Are you sure you want to change the base?

enhance: add create segment message, enable empty segment flush #37407

Conversation

chyezh commented Nov 4, 2024 • edited Loading

sre-ci-robot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mergify bot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

mergify bot commented Nov 4, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mergify bot commented Nov 4, 2024

codecov bot commented Nov 4, 2024 • edited Loading

Codecov Report

mergify bot commented Nov 5, 2024

chyezh commented Nov 5, 2024

mergify bot commented Nov 6, 2024

chyezh commented Nov 6, 2024

mergify bot commented Nov 7, 2024

chyezh commented Nov 7, 2024

chyezh commented Nov 4, 2024 •

edited

Loading

codecov bot commented Nov 4, 2024 •

edited

Loading