enhance: [2.4] release the record in delete codec and add some log for compaction (#34506)

/kind improvement
- pr: #34454

Signed-off-by: SimFG <bang.fu@zilliz.com>

internal/datanode/compaction/mix_compactor.go

@@ -124,6 +124,7 @@ func (t *mixCompactionTask) merge(
 		syncBatchCount    int   // binlog batch count
 		remainingRowCount int64 // the number of remaining entities
 		expiredRowCount   int64 // the number of expired entities
+		deletedRowCount   int64 = 0
 		unflushedRowCount int64 = 0
 
 		// All binlog meta of a segment
@@ -177,6 +178,7 @@ func (t *mixCompactionTask) merge(
 			}
 			v := iter.Value()
 			if isValueDeleted(v) {
+				deletedRowCount++
 				continue
 			}
 
@@ -257,6 +259,7 @@ func (t *mixCompactionTask) merge(
 
 	log.Info("compact merge end",
 		zap.Int64("remaining row count", remainingRowCount),
+		zap.Int64("deleted row count", deletedRowCount),
 		zap.Int64("expired entities", expiredRowCount),
 		zap.Int("binlog batch count", syncBatchCount),
 		zap.Duration("download binlogs elapse", downloadTimeCost),

internal/storage/data_codec.go

@@ -892,7 +892,7 @@ func (deleteCodec *DeleteCodec) Deserialize(blobs []*Blob) (partitionID UniqueID
 
 		deleteLog := &DeleteLog{}
 
-		for rr.Next() {
+		handleRecord := func() error {
 			rec := rr.Record()
 			defer rec.Release()
 			column := rec.Column(0)
@@ -905,6 +905,14 @@ func (deleteCodec *DeleteCodec) Deserialize(blobs []*Blob) (partitionID UniqueID
 
 				result.Append(deleteLog.Pk, deleteLog.Ts)
 			}
+			return nil
+		}
+
+		for rr.Next() {
+			err := handleRecord()
+			if err != nil {
+				return err
+			}
 		}
 		return nil
 	}