refactor

Author: func25

lib/logstorage/block_search.go

@@ -66,8 +66,7 @@ var blockSearchWorkBatchPool sync.Pool
 
 func (bswb *blockSearchWorkBatch) appendBlockSearchWork(p *part, so *searchOptions, bh *blockHeader) bool {
 	bsws := bswb.bsws
-	var dm *deleteMarker
-	dm = p.marker.delete.Load()
+	dm := p.deleteMarker.Load()
 
 	bsws = append(bsws, blockSearchWork{
 		p:  p,

lib/logstorage/block_stream_merger.go

@@ -93,9 +93,6 @@ type blockStreamMerger struct {
 	//
 	// It is used for limiting the number of columns written per block
 	uniqueFields int
-
-	// Accumulated delete marker for blocks we keep and mark during merge.
-	dmNew *deleteMarker
 }
 
 func (bsm *blockStreamMerger) reset() {
@@ -110,7 +107,6 @@ func (bsm *blockStreamMerger) reset() {
 	bsm.streamID.reset()
 	bsm.resetRows()
 	bsm.uniqueFields = 0
-	bsm.dmNew = nil
 }
 
 func (bsm *blockStreamMerger) resetRows() {
@@ -344,13 +340,8 @@ func (h *blockStreamReadersHeap) Pop() any {
 //
 // In both cases the original reader is advanced to the next block or popped when exhausted.
 func (bsm *blockStreamMerger) processDeleteMarker(bsr *blockStreamReader, blockID uint64) bool {
-
-	miAgg := bsr.marker
-	if miAgg == nil {
-		return false
-	}
-	dm := miAgg.delete.Load()
-	if dm == nil {
+	dm := bsr.deleteMarker
+	if len(dm.blockIDs) == 0 {
 		return false
 	}
 
@@ -381,62 +372,8 @@ func (bsm *blockStreamMerger) processDeleteMarker(bsr *blockStreamReader, blockI
 	// Obtain the new blockID assigned by the writer.
 	newID := bsm.bsw.LastBlockID()
 
-	// Accumulate delete-marker pointing to the new blockID.
-	if bsm.dmNew == nil {
-		bsm.dmNew = &deleteMarker{}
-	}
-	bsm.dmNew.AddBlock(newID, bm)
-
-	// Uncomment the call below to enable the legacy re-queue logic instead of
-	// keeping the original compressed block.
-	// bsm.requeuePartialDeleteBlock(bsr, bm, rowsTotal)
-
-	return true // block handled; caller will advance reader
-}
-
-// requeuePartialDeleteBlock recreates a pruned block as an in-memory reader and
-// pushes it back into the merge heap so ordering is preserved.
-func (bsm *blockStreamMerger) requeuePartialDeleteBlock(bsr *blockStreamReader, bm boolRLE, rowsTotal int) {
-	if bsm.sbu == nil {
-		bsm.sbu = getStringsBlockUnmarshaler()
-	}
-	if bsm.vd == nil {
-		bsm.vd = getValuesDecoder()
-	}
-
-	// Unmarshal original rows.
-	var rs rows
-	if err := bsr.blockData.unmarshalRows(&rs, bsm.sbu, bsm.vd); err != nil {
-		logger.Panicf("BUG: cannot unmarshal rows for partial delete pruning: %s", err)
-	}
-
-	// Collect rows that remain after applying delete marker.
-	keptTS := make([]int64, 0, rowsTotal)
-	keptRows := make([][]Field, 0, rowsTotal)
-	bm.ForEachZeroBit(rowsTotal, func(i int) {
-		keptTS = append(keptTS, rs.timestamps[i])
-		keptRows = append(keptRows, rs.rows[i])
-	})
-
-	if len(keptTS) == 0 {
-		logger.Panicf("BUG: length of rows after partial delete marker pruning is 0")
-	}
-
-	// Build a tiny in-memory part with the kept rows.
-	lr := getLogRows()
-	for i, ts := range keptTS {
-		lr.mustAddRow(bsr.blockData.streamID, ts, keptRows[i])
-	}
-	mp := getInmemoryPart()
-	mp.mustInitFromRows(lr)
-	putLogRows(lr)
-
-	bsrNew := getBlockStreamReader()
-	bsrNew.MustInitFromInmemoryPart(mp)
-	if !bsrNew.NextBlock() {
-		logger.Panicf("BUG: pruned in-memory reader has no blocks to merge")
-	}
+	// Accumulate delete-marker directly in writer
+	bsm.bsw.dm.AddBlock(newID, bm)
 
-	heap.Push(&bsm.readersHeap, bsrNew)
-	bsm.bsrs = append(bsm.bsrs, bsrNew)
+	return true
 }

lib/logstorage/block_stream_reader.go

@@ -65,9 +65,6 @@ type streamReaders struct {
 	columnsHeaderReader      readerWithStats
 	timestampsReader         readerWithStats
 
-	// Marker reader for tracking bytes read during streaming
-	markerDatReader readerWithStats
-
 	messageBloomValuesReader bloomValuesReader
 	oldBloomValuesReader     bloomValuesReader
 	bloomValuesShards        []bloomValuesReader
@@ -119,8 +116,6 @@ func (sr *streamReaders) reset() {
 	sr.columnsHeaderReader.reset()
 	sr.timestampsReader.reset()
 
-	sr.markerDatReader.reset()
-
 	sr.messageBloomValuesReader.reset()
 	sr.oldBloomValuesReader.reset()
 	for i := range sr.bloomValuesShards {
@@ -133,9 +128,7 @@ func (sr *streamReaders) reset() {
 }
 
 func (sr *streamReaders) init(partFormatVersion uint, columnNamesReader, columnIdxsReader, metaindexReader, indexReader,
-	columnsHeaderIndexReader, columnsHeaderReader, timestampsReader filestream.ReadCloser,
-	markerDatReader filestream.ReadCloser,
-	messageBloomValuesReader, oldBloomValuesReader bloomValuesStreamReader, bloomValuesShards []bloomValuesStreamReader,
+	columnsHeaderIndexReader, columnsHeaderReader, timestampsReader filestream.ReadCloser, messageBloomValuesReader, oldBloomValuesReader bloomValuesStreamReader, bloomValuesShards []bloomValuesStreamReader,
 ) {
 	sr.partFormatVersion = partFormatVersion
 
@@ -147,8 +140,6 @@ func (sr *streamReaders) init(partFormatVersion uint, columnNamesReader, columnI
 	sr.columnsHeaderReader.init(columnsHeaderReader)
 	sr.timestampsReader.init(timestampsReader)
 
-	sr.markerDatReader.init(markerDatReader)
-
 	sr.messageBloomValuesReader.init(messageBloomValuesReader)
 	sr.oldBloomValuesReader.init(oldBloomValuesReader)
 	sr.bloomValuesShards = slicesutil.SetLength(sr.bloomValuesShards, len(bloomValuesShards))
@@ -195,7 +186,6 @@ func (sr *streamReaders) MustClose() {
 	sr.columnsHeaderIndexReader.MustClose()
 	sr.columnsHeaderReader.MustClose()
 	sr.timestampsReader.MustClose()
-	sr.markerDatReader.MustClose()
 	sr.messageBloomValuesReader.MustClose()
 	sr.oldBloomValuesReader.MustClose()
 	for i := range sr.bloomValuesShards {
@@ -238,8 +228,8 @@ type blockStreamReader struct {
 	// ph is the header for the part
 	ph partHeader
 
-	// marker aggregates marker data for the source part (delete, ttl, etc.).
-	marker *marker
+	// deleteMarker holds delete marker data for the given part.
+	deleteMarker deleteMarker
 
 	// streamReaders contains data readers in stream mode
 	streamReaders streamReaders
@@ -301,7 +291,7 @@ func (bsr *blockStreamReader) reset() {
 	bsr.globalRowsCount = 0
 	bsr.globalBlocksCount = 0
 
-	bsr.marker = nil
+	bsr.deleteMarker = deleteMarker{}
 
 	bsr.sidLast.reset()
 	bsr.minTimestampLast = 0
@@ -320,7 +310,7 @@ func (bsr *blockStreamReader) MustInitFromInmemoryPart(mp *inmemoryPart) {
 	bsr.ph = mp.ph
 
 	// propagate delete-marker data
-	bsr.marker = nil
+	bsr.deleteMarker = deleteMarker{}
 
 	// Initialize streamReaders
 	columnNamesReader := mp.columnNames.NewReader()
@@ -351,11 +341,15 @@ func (bsr *blockStreamReader) MustInitFromInmemoryPart(mp *inmemoryPart) {
 
 	bsr.streamReaders.init(bsr.ph.FormatVersion, columnNamesReader, columnIdxsReader, metaindexReader, indexReader,
 		columnsHeaderIndexReader, columnsHeaderReader, timestampsReader,
-		nil, // no marker data reader for in-memory parts
 		messageBloomValuesReader, oldBloomValuesReader, bloomValuesShards)
 
 	// Read metaindex data
 	bsr.indexBlockHeaders = mustReadIndexBlockHeaders(bsr.indexBlockHeaders[:0], &bsr.streamReaders.metaindexReader)
+
+	// Link deleteMarker directly from in-memory part
+	if len(mp.deleteMarker.blockIDs) > 0 {
+		bsr.deleteMarker = mp.deleteMarker
+	}
 }
 
 // MustInitFromFilePart initializes bsr from file part at the given path.
@@ -432,15 +426,14 @@ func (bsr *blockStreamReader) MustInitFromFilePart(path string) {
 	// Initialize streamReaders
 	bsr.streamReaders.init(bsr.ph.FormatVersion, columnNamesReader, columnIdxsReader, metaindexReader, indexReader,
 		columnsHeaderIndexReader, columnsHeaderReader, timestampsReader,
-		markerDatReader,
 		messageBloomValuesReader, oldBloomValuesReader, bloomValuesShards)
 
 	// Read metaindex data
 	bsr.indexBlockHeaders = mustReadIndexBlockHeaders(bsr.indexBlockHeaders[:0], &bsr.streamReaders.metaindexReader)
 
 	// Read marker index if available
 	if markerDatReader != nil {
-		bsr.marker = mustReadMarkerData(&bsr.streamReaders.markerDatReader, bsr.ph.BlocksCount)
+		bsr.deleteMarker = mustReadDeleteMarkerData(markerDatReader)
 	}
 }
 

lib/logstorage/block_stream_writer.go

@@ -204,6 +204,15 @@ type blockStreamWriter struct {
 	// streamWriters contains writer for block data
 	streamWriters streamWriters
 
+	// partPath is non-empty when writing a file-based part.
+	partPath string
+
+	// mp points to the destination in-memory part when writing an in-memory part.
+	mp *inmemoryPart
+
+	// dm accumulates per-block delete markers generated while writing.
+	dm deleteMarker
+
 	// sidLast is the streamID for the last written block
 	sidLast streamID
 
@@ -276,12 +285,18 @@ func (bsw *blockStreamWriter) reset() {
 	}
 
 	bsw.indexBlockHeader.reset()
+
+	bsw.partPath = ""
+	bsw.mp = nil
+	bsw.dm = deleteMarker{}
 }
 
 // MustInitForInmemoryPart initializes bsw from mp
 func (bsw *blockStreamWriter) MustInitForInmemoryPart(mp *inmemoryPart) {
 	bsw.reset()
 
+	bsw.mp = mp
+
 	messageBloomValues := mp.messageBloomValues.NewStreamWriter()
 	createBloomValuesWriter := func(_ uint64) bloomValuesStreamWriter {
 		return mp.fieldBloomValues.NewStreamWriter()
@@ -296,6 +311,8 @@ func (bsw *blockStreamWriter) MustInitForInmemoryPart(mp *inmemoryPart) {
 func (bsw *blockStreamWriter) MustInitForFilePart(path string, nocache bool) {
 	bsw.reset()
 
+	bsw.partPath = path
+
 	fs.MustMkdirFailIfExist(path)
 
 	columnNamesPath := filepath.Join(path, columnNamesFilename)
@@ -472,6 +489,18 @@ func (bsw *blockStreamWriter) Finalize(ph *partHeader) {
 	ph.CompressedSizeBytes = bsw.streamWriters.totalBytesWritten()
 
 	bsw.streamWriters.MustClose()
+
+	// Persist accumulated delete markers if any
+	if len(bsw.dm.blockIDs) > 0 {
+		if bsw.mp != nil {
+			bsw.mp.deleteMarker = bsw.dm
+		} else if bsw.partPath != "" {
+			datBuf := bsw.dm.Marshal(nil)
+			datPath := filepath.Join(bsw.partPath, rowMarkerDatFilename)
+			fs.MustWriteSync(datPath, datBuf)
+		}
+	}
+
 	bsw.reset()
 }
 

lib/logstorage/datadb.go

@@ -507,7 +507,7 @@ func (ddb *datadb) mustMergeParts(pws []*partWrapper, isFinal bool) {
 	var totalRows int64
 	for i := range pws {
 		pw := pws[i]
-		delete := pw.p.marker.delete.Load()
+		delete := pw.p.deleteMarker.Load()
 		if delete != nil {
 			deleteMarkerBlocks += int64(len(delete.blockIDs))
 			for i := range delete.blockIDs {
@@ -549,7 +549,6 @@ func (ddb *datadb) mustMergeParts(pws []*partWrapper, isFinal bool) {
 	if isFinal && len(pws) == 1 && pws[0].mp != nil {
 		// Fast path: flush a single in-memory part to disk.
 		mp := pws[0].mp
-		// Persist applied sequence from the source part so the newly flushed part is immediately up-to-date.
 		mp.MustStoreToDisk(dstPartPath)
 		pwNew := ddb.openCreatedPart(&mp.ph, pws, nil, dstPartPath)
 		ddb.swapSrcWithDstParts(pws, pwNew, dstPartType)
@@ -597,6 +596,7 @@ func (ddb *datadb) mustMergeParts(pws []*partWrapper, isFinal bool) {
 		mpNew.ph = ph
 	} else {
 		ph.mustWriteMetadata(dstPartPath)
+
 		// Make sure the created part directory contents is synced and visible in case of unclean shutdown.
 		fs.MustSyncPathAndParentDir(dstPartPath)
 	}

lib/logstorage/filenames.go

@@ -16,7 +16,7 @@ const (
 	messageBloomFilename       = "message_bloom.bin"
 
 	// Filename for per-row marker data (e.g. delete markers).
-	rowMarkerDatFilename = "row_marker.bin"
+	rowMarkerDatFilename = "deleted_rows.bin"
 
 	// Filename for async tasks storage at partition level.
 	asyncTasksFilename = "async_tasks.json"

lib/logstorage/inmemory_part.go

@@ -24,6 +24,8 @@ type inmemoryPart struct {
 
 	messageBloomValues bloomValuesBuffer
 	fieldBloomValues   bloomValuesBuffer
+
+	deleteMarker deleteMarker
 }
 
 type bloomValuesBuffer struct {
@@ -64,6 +66,9 @@ func (mp *inmemoryPart) reset() {
 
 	mp.messageBloomValues.reset()
 	mp.fieldBloomValues.reset()
+
+	// Drop any attached deleteMarker.
+	mp.deleteMarker = deleteMarker{}
 }
 
 // mustInitFromRows initializes mp from lr.
@@ -130,6 +135,13 @@ func (mp *inmemoryPart) MustStoreToDisk(path string) {
 	fs.MustWriteStreamSync(messageBloomFilterPath, &mp.messageBloomValues.bloom)
 	fs.MustWriteStreamSync(messageValuesPath, &mp.messageBloomValues.values)
 
+	// Persist delete-marker data if present.
+	if len(mp.deleteMarker.blockIDs) > 0 {
+		datBuf := mp.deleteMarker.Marshal(nil)
+		datPath := filepath.Join(path, rowMarkerDatFilename)
+		fs.MustWriteSync(datPath, datBuf)
+	}
+
 	bloomPath := getBloomFilePath(path, 0)
 	fs.MustWriteStreamSync(bloomPath, &mp.fieldBloomValues.bloom)