Fix measure merging issue (#549)

apache · Oct 7, 2024 · c92ee3a · c92ee3a
1 parent d14d599
commit c92ee3a
Show file tree

Hide file tree

Showing 5 changed files with 49 additions and 6 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -48,6 +48,7 @@ Release Notes.
 - Fix the bug that the etcd client doesn't reconnect when facing the context timeout in the startup phase.
 - Fix the bug that the long running query doesn't stop when the context is canceled.
 - Fix the bug that merge block with different tags or fields.
+- Fix the bug that the pending measure block is not released when a full block is merged.
 
 ### Documentation
 

diff --git a/banyand/measure/block.go b/banyand/measure/block.go
@@ -776,7 +776,7 @@ func (bi *blockPointer) updateMetadata() {
 }
 
 func (bi *blockPointer) copyFrom(src *blockPointer) {
-	bi.idx = 0
+	bi.reset()
 	bi.bm.copyFrom(&src.bm)
 	bi.appendAll(src)
 }
@@ -1006,7 +1006,7 @@ func (bi *blockPointer) isFull() bool {
 func (bi *blockPointer) reset() {
 	bi.idx = 0
 	bi.block.reset()
-	bi.bm = blockMetadata{}
+	bi.bm.reset()
 }
 
 func generateBlockPointer() *blockPointer {

diff --git a/banyand/measure/merger.go b/banyand/measure/merger.go
@@ -292,7 +292,6 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader) (*pa
 			(pendingBlock.isFull() && pendingBlock.bm.timestamps.max <= b.bm.timestamps.min) {
 			bw.mustWriteBlock(pendingBlock.bm.seriesID, &pendingBlock.block)
 			releaseDecoder()
-			pendingBlock.reset()
 			br.loadBlockData(getDecoder())
 			pendingBlock.copyFrom(b)
 			continue
@@ -316,6 +315,8 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader) (*pa
 
 		if len(tmpBlock.timestamps) <= maxBlockLength {
 			bw.mustWriteBlock(tmpBlock.bm.seriesID, &tmpBlock.block)
+			pendingBlock.reset()
+			pendingBlockIsEmpty = true
 			releaseDecoder()
 			continue
 		}

diff --git a/banyand/measure/merger_test.go b/banyand/measure/merger_test.go
@@ -291,14 +291,23 @@ func Test_mergeParts(t *testing.T) {
 				{seriesID: 1, count: 1265, uncompressedSizeBytes: 2130260},
 				{seriesID: 1, count: 1265, uncompressedSizeBytes: 2130260},
 				{seriesID: 1, count: 2470, uncompressedSizeBytes: 4159480},
-				{seriesID: 1, count: 2470, uncompressedSizeBytes: 4159480},
-				{seriesID: 1, count: 2470, uncompressedSizeBytes: 4159480},
-				{seriesID: 1, count: 2410, uncompressedSizeBytes: 4058440},
+				{seriesID: 1, count: 1265, uncompressedSizeBytes: 2130260},
+				{seriesID: 1, count: 1265, uncompressedSizeBytes: 2130260},
 				{seriesID: 1, count: 1205, uncompressedSizeBytes: 2029220},
 				{seriesID: 2, count: 2, uncompressedSizeBytes: 126},
 				{seriesID: 3, count: 2, uncompressedSizeBytes: 64},
 			},
 		},
+		{
+			name:    "Test with multiple parts with a large small quantity of different ts",
+			dpsList: []*dataPoints{generateHugeSmallDps(1, 5000, 1), generateHugeSmallDps(5001, 10000, 2)},
+			want: []blockMetadata{
+				{seriesID: 1, count: 8192, uncompressedSizeBytes: 262144},
+				{seriesID: 1, count: 1808, uncompressedSizeBytes: 57856},
+				{seriesID: 2, count: 2, uncompressedSizeBytes: 32},
+				{seriesID: 3, count: 2, uncompressedSizeBytes: 64},
+			},
+		},
 	}
 
 	for _, tt := range tests {

diff --git a/banyand/measure/tstable_test.go b/banyand/measure/tstable_test.go
@@ -582,3 +582,35 @@ func generateHugeDps(startTimestamp, endTimestamp, timestamp int64) *dataPoints
 	}}...)
 	return hugeDps
 }
+
+func generateHugeSmallDps(startTimestamp, endTimestamp, timestamp int64) *dataPoints {
+	hugeDps := &dataPoints{
+		seriesIDs:   []common.SeriesID{},
+		timestamps:  []int64{},
+		versions:    []int64{},
+		tagFamilies: [][]nameValues{},
+		fields:      []nameValues{},
+	}
+	now := time.Now().UnixNano()
+	for i := startTimestamp; i <= endTimestamp; i++ {
+		hugeDps.seriesIDs = append(hugeDps.seriesIDs, 1)
+		hugeDps.timestamps = append(hugeDps.timestamps, i)
+		hugeDps.versions = append(hugeDps.versions, now+i)
+		hugeDps.tagFamilies = append(hugeDps.tagFamilies, []nameValues{})
+		hugeDps.fields = append(hugeDps.fields, nameValues{
+			name: "skipped", values: []*nameValue{
+				{name: "intField", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(3330), valueArr: nil},
+			},
+		})
+	}
+	hugeDps.seriesIDs = append(hugeDps.seriesIDs, []common.SeriesID{2, 3}...)
+	hugeDps.timestamps = append(hugeDps.timestamps, []int64{timestamp, timestamp}...)
+	hugeDps.versions = append(hugeDps.versions, []int64{now + timestamp, now + timestamp}...)
+	hugeDps.tagFamilies = append(hugeDps.tagFamilies, [][]nameValues{{}, {}}...)
+	hugeDps.fields = append(hugeDps.fields, []nameValues{{}, {
+		name: "onlyFields", values: []*nameValue{
+			{name: "intField", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(4440), valueArr: nil},
+		},
+	}}...)
+	return hugeDps
+}