Repository: numaproj/numaflow
Pull request #546 (2023-02-14 21:34:44)

chore: rename WAL function names
We were abusing the word `entry`. We need to differentiate between WAL header v/s ReadMessage header and its body.


Patch:
diff --git a/pkg/reduce/pbq/store/wal/bootstrap.go b/pkg/reduce/pbq/store/wal/bootstrap.go
index d4c2cea8..62d4ce07 100644
--- a/pkg/reduce/pbq/store/wal/bootstrap.go
+++ b/pkg/reduce/pbq/store/wal/bootstrap.go
@@ -44,7 +44,7 @@ func (w *WAL) IsCorrupted() bool {
 	return w.corrupted
 }
 
-func (w *WAL) readHeader() (*partition.ID, error) {
+func (w *WAL) readWALHeader() (*partition.ID, error) {
 	if w.openMode == os.O_WRONLY {
 		return nil, fmt.Errorf("opened using O_WRONLY")
 	}
@@ -53,7 +53,7 @@ func (w *WAL) readHeader() (*partition.ID, error) {
 		return nil, fmt.Errorf("header has already been read, current readoffset is at %d", w.rOffset)
 	}
 
-	id, err := decodeHeader(w.fp)
+	id, err := decodeWALHeader(w.fp)
 	if err != nil {
 		return nil, err
 	}
@@ -74,17 +74,17 @@ func (w *WAL) isEnd() bool {
 	return w.rOffset >= w.readUpTo
 }
 
-// decodeHeader decodes the header which is encoded by encodeHeader.
-func decodeHeader(buf io.Reader) (*partition.ID, error) {
+// decodeWALHeader decodes the header which is encoded by encodeWALHeader.
+func decodeWALHeader(buf io.Reader) (*partition.ID, error) {
 	var err error
 	// read the fixed vals
-	var hp = new(headerPreamble)
+	var hp = new(walHeaderPreamble)
 	err = binary.Read(buf, binary.LittleEndian, hp)
 	if err != nil {
 		return nil, err
 	}
 	// read the variadic key
-	var key = make([]rune, hp.KLen)
+	var key = make([]rune, hp.SLen)
 	err = binary.Read(buf, binary.LittleEndian, key)
 	if err != nil {
 		return nil, err
@@ -111,7 +111,7 @@ func (w *WAL) Read(size int64) ([]*isb.ReadMessage, bool, error) {
 	// we will assign size with the number of messages in the store
 	start := w.rOffset
 	for size > w.rOffset-start && !w.isEnd() {
-		message, sizeRead, err := decodeEntry(w.fp)
+		message, sizeRead, err := decodeReadMessage(w.fp)
 		if err != nil {
 			if err == errChecksumMismatch {
 				w.corrupted = true
@@ -133,14 +133,14 @@ func (w *WAL) Read(size int64) ([]*isb.ReadMessage, bool, error) {
 	return messages, false, nil
 }
 
-// decodeEntry decodes the header which is encoded by encodeEntry.
-func decodeEntry(buf io.Reader) (*isb.ReadMessage, int64, error) {
-	entryHeader, err := decodeEntryHeader(buf)
+// decodeReadMessage decodes the header which is encoded by encodeReadMessage.
+func decodeReadMessage(buf io.Reader) (*isb.ReadMessage, int64, error) {
+	entryHeader, err := decodeReadMessageHeader(buf)
 	if err != nil {
 		return nil, 0, err
 	}
 
-	entryBody, err := decodeEntryBody(buf, entryHeader)
+	entryBody, err := decodeReadMessageBody(buf, entryHeader)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -153,10 +153,10 @@ func decodeEntry(buf io.Reader) (*isb.ReadMessage, int64, error) {
 	}, size, nil
 }
 
-// decodeEntryHeader decodes the header which is encoded by encodeEntryHeader.
-func decodeEntryHeader(buf io.Reader) (*entryHeaderPreamble, error) {
+// decodeReadMessageHeader decodes the header which is encoded by encodeReadMessageHeader.
+func decodeReadMessageHeader(buf io.Reader) (*readMessageHeaderPreamble, error) {
 	// read the fixed vals
-	var entryHeader = new(entryHeaderPreamble)
+	var entryHeader = new(readMessageHeaderPreamble)
 	err := binary.Read(buf, binary.LittleEndian, entryHeader)
 	if err != nil {
 		return nil, err
@@ -164,9 +164,9 @@ func decodeEntryHeader(buf io.Reader) (*entryHeaderPreamble, error) {
 	return entryHeader, nil
 }
 
-// decodeEntryBody decodes the header which is encoded by encodeEntryBody.
+// decodeReadMessageBody decodes the header which is encoded by encodeReadMessageBody.
 // Returns errChecksumMismatch to indicate if corrupted entry is found.
-func decodeEntryBody(buf io.Reader, entryHeader *entryHeaderPreamble) (*isb.Message, error) {
+func decodeReadMessageBody(buf io.Reader, entryHeader *readMessageHeaderPreamble) (*isb.Message, error) {
 	var err error
 
 	body := make([]byte, entryHeader.MessageLen)
diff --git a/pkg/reduce/pbq/store/wal/segment.go b/pkg/reduce/pbq/store/wal/segment.go
index 44cb577d..b05aa029 100644
--- a/pkg/reduce/pbq/store/wal/segment.go
+++ b/pkg/reduce/pbq/store/wal/segment.go
@@ -74,19 +74,19 @@ type WAL struct {
 	numOfUnsyncedMsgs int64
 }
 
-// writeHeader writes the header to the file
-func (w *WAL) writeHeader() (err error) {
+// writeWALHeader writes the WAL header to the file.
+func (w *WAL) writeWALHeader() (err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
 				metrics.LabelPipeline:           w.walStores.pipelineName,
 				metrics.LabelVertex:             w.walStores.vertexName,
 				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-				labelErrorKind:                  "writeHeader",
+				labelErrorKind:                  "writeWALHeader",
 			}).Inc()
 		}
 	}()
-	header, err := w.encodeHeader(w.partitionID)
+	header, err := w.encodeWALHeader(w.partitionID)
 	if err != nil {
 		return err
 	}
@@ -100,44 +100,45 @@ func (w *WAL) writeHeader() (err error) {
 	return err
 }
 
-// headerPreamble is the header preamble (excludes variadic key)
-type headerPreamble struct {
+// walHeaderPreamble is the header preamble (excludes variadic key)
+type walHeaderPreamble struct {
 	S    int64
 	E    int64
-	KLen int16
+	SLen int16
 }
 
-// entryHeaderPreamble is the header for each WAL entry
-type entryHeaderPreamble struct {
+// readMessageHeaderPreamble is the header for each WAL entry
+type readMessageHeaderPreamble struct {
 	WaterMark  int64
 	Offset     int64
 	MessageLen int64
 	Checksum   uint32
 }
 
-// encodeHeader builds the header. The header is of the following format.
+// encodeWALHeader builds the WAL header. WAL header is per WAL and has information to build the WAL partition.
+// The header is of the following format.
 //
-//	+--------------------+------------------+-----------------+------------+
-//	| start time (int64) | end time (int64) | key-len (int16) | key []rune |
-//	+--------------------+------------------+-----------------+------------+
+//	+--------------------+------------------+------------------+-------------+
+//	| start time (int64) | end time (int64) | slot-len (int16) | slot []rune |
+//	+--------------------+------------------+------------------+-------------+
 //
 // We require the key-len because key is variadic.
-func (w *WAL) encodeHeader(id *partition.ID) (buf *bytes.Buffer, err error) {
+func (w *WAL) encodeWALHeader(id *partition.ID) (buf *bytes.Buffer, err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
 				metrics.LabelPipeline:           w.walStores.pipelineName,
 				metrics.LabelVertex:             w.walStores.vertexName,
 				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-				labelErrorKind:                  "encodeHeader",
+				labelErrorKind:                  "encodeWALHeader",
 			}).Inc()
 		}
 	}()
 	buf = new(bytes.Buffer)
-	hp := headerPreamble{
+	hp := walHeaderPreamble{
 		S:    id.Start.UnixMilli(),
 		E:    id.End.UnixMilli(),
-		KLen: int16(len(id.Slot)),
+		SLen: int16(len(id.Slot)),
 	}
 
 	// write the fixed values
@@ -152,26 +153,26 @@ func (w *WAL) encodeHeader(id *partition.ID) (buf *bytes.Buffer, err error) {
 	return buf, err
 }
 
-func (w *WAL) encodeEntry(message *isb.ReadMessage) (buf *bytes.Buffer, err error) {
+func (w *WAL) encodeReadMessage(message *isb.ReadMessage) (buf *bytes.Buffer, err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
 				metrics.LabelPipeline:           w.walStores.pipelineName,
 				metrics.LabelVertex:             w.walStores.vertexName,
 				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-				labelErrorKind:                  "encodeEntry",
+				labelErrorKind:                  "encodeReadMessage",
 			}).Inc()
 		}
 	}()
 	buf = new(bytes.Buffer)
-	body, err := w.encodeEntryBody(message)
+	body, err := w.encodeReadMessageBody(message)
 	if err != nil {
 		return nil, err
 	}
 	checksum := calculateChecksum(body.Bytes())
 
 	// Writes the message header
-	header, err := w.encodeEntryHeader(message, int64(body.Len()), checksum)
+	header, err := w.encodeReadMessageHeader(message, int64(body.Len()), checksum)
 	if err != nil {
 		return nil, err
 	}
@@ -194,7 +195,7 @@ func (w *WAL) encodeEntry(message *isb.ReadMessage) (buf *bytes.Buffer, err erro
 	return buf, nil
 }
 
-func (w *WAL) encodeEntryHeader(message *isb.ReadMessage, messageLen int64, checksum uint32) (*bytes.Buffer, error) {
+func (w *WAL) encodeReadMessageHeader(message *isb.ReadMessage, messageLen int64, checksum uint32) (*bytes.Buffer, error) {
 	watermark := message.Watermark.UnixMilli()
 
 	offset, err := message.ReadOffset.Sequence()
@@ -202,7 +203,7 @@ func (w *WAL) encodeEntryHeader(message *isb.ReadMessage, messageLen int64, chec
 		return nil, err
 	}
 
-	entryHeader := entryHeaderPreamble{
+	msgHeader := readMessageHeaderPreamble{
 		WaterMark:  watermark,
 		Offset:     offset,
 		MessageLen: messageLen,
@@ -211,20 +212,20 @@ func (w *WAL) encodeEntryHeader(message *isb.ReadMessage, messageLen int64, chec
 
 	// write the fixed values
 	buf := new(bytes.Buffer)
-	err = binary.Write(buf, binary.LittleEndian, entryHeader)
+	err = binary.Write(buf, binary.LittleEndian, msgHeader)
 	if err != nil {
 		walErrors.With(map[string]string{
 			metrics.LabelPipeline:           w.walStores.pipelineName,
 			metrics.LabelVertex:             w.walStores.vertexName,
 			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-			labelErrorKind:                  "encodeEntryHeader",
+			labelErrorKind:                  "encodeReadMessageHeader",
 		}).Inc()
 		return nil, err
 	}
 	return buf, nil
 }
 
-func (w *WAL) encodeEntryBody(message *isb.ReadMessage) (*bytes.Buffer, error) {
+func (w *WAL) encodeReadMessageBody(message *isb.ReadMessage) (*bytes.Buffer, error) {
 	m := new(bytes.Buffer)
 	enc := gob.NewEncoder(m)
 	err := enc.Encode(message.Message)
@@ -233,9 +234,9 @@ func (w *WAL) encodeEntryBody(message *isb.ReadMessage) (*bytes.Buffer, error) {
 			metrics.LabelPipeline:           w.walStores.pipelineName,
 			metrics.LabelVertex:             w.walStores.vertexName,
 			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-			labelErrorKind:                  "encodeEntryBody",
+			labelErrorKind:                  "encodeReadMessageBody",
 		}).Inc()
-		return nil, fmt.Errorf("entry body encountered encode err: %w", err)
+		return nil, fmt.Errorf("encodeReadMessageBody encountered encode err: %w", err)
 	}
 	return m, nil
 }
@@ -251,7 +252,7 @@ func calculateChecksum(data []byte) uint32 {
 //	| watermark (int64) | offset (int64) | msg-len (int64) | CRC (unit32) | message []byte |
 //	+-------------------+----------------+-----------------+--------------+----------------+
 //
-// CRC will be used for detecting entry corruptions.
+// CRC will be used for detecting ReadMessage corruptions.
 func (w *WAL) Write(message *isb.ReadMessage) (err error) {
 	defer func() {
 		if err != nil {
@@ -264,7 +265,7 @@ func (w *WAL) Write(message *isb.ReadMessage) (err error) {
 		}
 	}()
 	encodeStart := time.Now()
-	entry, err := w.encodeEntry(message)
+	entry, err := w.encodeReadMessage(message)
 	entryEncodeLatency.With(map[string]string{
 		metrics.LabelPipeline:           w.walStores.pipelineName,
 		metrics.LabelVertex:             w.walStores.vertexName,
diff --git a/pkg/reduce/pbq/store/wal/stores.go b/pkg/reduce/pbq/store/wal/stores.go
index 8e3aab79..0aa8892e 100644
--- a/pkg/reduce/pbq/store/wal/stores.go
+++ b/pkg/reduce/pbq/store/wal/stores.go
@@ -120,7 +120,7 @@ func (ws *walStores) openOrCreateWAL(id *partition.ID) (*WAL, error) {
 			numOfUnsyncedMsgs: 0,
 		}
 
-		err = wal.writeHeader()
+		err = wal.writeWALHeader()
 		if err != nil {
 			return nil, err
 		}
@@ -146,7 +146,7 @@ func (ws *walStores) openOrCreateWAL(id *partition.ID) (*WAL, error) {
 			walStores:         ws,
 			numOfUnsyncedMsgs: 0,
 		}
-		readPartition, err := wal.readHeader()
+		readPartition, err := wal.readWALHeader()
 		if err != nil {
 			return nil, err
 		}
@@ -204,7 +204,7 @@ func (ws *walStores) openWAL(filePath string) (*WAL, error) {
 		walStores: ws,
 	}
 
-	w.partitionID, err = w.readHeader()
+	w.partitionID, err = w.readWALHeader()
 
 	return w, err
 }


Test patch:
diff --git a/pkg/reduce/pbq/store/wal/segment_test.go b/pkg/reduce/pbq/store/wal/segment_test.go
index 9bc7c694..2c7c386f 100644
--- a/pkg/reduce/pbq/store/wal/segment_test.go
+++ b/pkg/reduce/pbq/store/wal/segment_test.go
@@ -57,7 +57,7 @@ func Test_writeReadHeader(t *testing.T) {
 	fName := wal.fp.Name()
 	assert.NoError(t, err)
 	// read will fail because the file was opened only in write only mode
-	_, err = wal.readHeader()
+	_, err = wal.readWALHeader()
 	assert.Error(t, err)
 	err = wal.Close()
 	fmt.Println(fName)
@@ -66,7 +66,7 @@ func Test_writeReadHeader(t *testing.T) {
 	openWAL, err := wal.walStores.openWAL(fName)
 	assert.NoError(t, err)
 	// we have already read the header in OpenWAL
-	_, err = openWAL.readHeader()
+	_, err = openWAL.readWALHeader()
 	assert.Error(t, err)
 
 	// compare the original ID with read ID
@@ -109,13 +109,13 @@ func Test_encodeDecodeHeader(t *testing.T) {
 			wal, err := stores.CreateStore(context.Background(), *tt.id)
 			assert.NoError(t, err)
 			newWal := wal.(*WAL)
-			got, err := newWal.encodeHeader(tt.id)
-			if !tt.wantErr(t, err, fmt.Sprintf("encodeHeader(%v)", tt.id)) {
+			got, err := newWal.encodeWALHeader(tt.id)
+			if !tt.wantErr(t, err, fmt.Sprintf("encodeWALHeader(%v)", tt.id)) {
 				return
 			}
-			result, err := decodeHeader(got)
+			result, err := decodeWALHeader(got)
 			assert.NoError(t, err)
-			assert.Equalf(t, tt.id, result, "encodeHeader(%v)", tt.id)
+			assert.Equalf(t, tt.id, result, "encodeWALHeader(%v)", tt.id)
 			err = newWal.Close()
 			assert.NoError(t, err)
 		})
@@ -148,7 +148,7 @@ func Test_writeReadEntry(t *testing.T) {
 	assert.NoError(t, err)
 	newWal := store.(*WAL)
 	// we have already read the header in OpenWAL
-	_, err = newWal.readHeader()
+	_, err = newWal.readWALHeader()
 	assert.Error(t, err)
 
 	actualMessages, finished, err := newWal.Read(10000)
@@ -165,7 +165,7 @@ func Test_writeReadEntry(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeEntry(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)
@@ -210,20 +210,20 @@ func Test_encodeDecodeEntry(t *testing.T) {
 			assert.NoError(t, err)
 			newWal := wal.(*WAL)
 
-			got, err := newWal.encodeEntry(tt.message)
-			if !tt.wantErr(t, err, fmt.Sprintf("encodeEntry(%v)", tt.message)) {
+			got, err := newWal.encodeReadMessage(tt.message)
+			if !tt.wantErr(t, err, fmt.Sprintf("encodeReadMessage(%v)", tt.message)) {
 				return
 			}
 
-			result, _, err := decodeEntry(bytes.NewReader(got.Bytes()))
+			result, _, err := decodeReadMessage(bytes.NewReader(got.Bytes()))
 			assert.NoError(t, err)
-			assert.Equalf(t, tt.message.Message, result.Message, "encodeEntry(%v)", tt.message.Message)
+			assert.Equalf(t, tt.message.Message, result.Message, "encodeReadMessage(%v)", tt.message.Message)
 			expectedOffset, err := tt.message.ReadOffset.Sequence()
 			assert.NoError(t, err)
 			actualOffset, err := result.ReadOffset.Sequence()
 			assert.NoError(t, err)
-			assert.Equalf(t, expectedOffset, actualOffset, "encodeEntry(%v)", tt.message.ReadOffset)
-			assert.Equalf(t, tt.message.Watermark, result.Watermark, "encodeEntry(%v)", tt.message.Watermark)
+			assert.Equalf(t, expectedOffset, actualOffset, "encodeReadMessage(%v)", tt.message.ReadOffset)
+			assert.Equalf(t, tt.message.Watermark, result.Watermark, "encodeReadMessage(%v)", tt.message.Watermark)
 			err = newWal.Close()
 			assert.NoError(t, err)
 		})
@@ -269,7 +269,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	assert.NoError(t, err)
 	newWal := store.(*WAL)
 	// we have already read the header in OpenWAL
-	_, err = newWal.readHeader()
+	_, err = newWal.readWALHeader()
 	assert.Error(t, err)
 
 	actualMessages, finished, err := newWal.Read(10000)
@@ -286,7 +286,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeEntry(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)
@@ -335,7 +335,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	assert.NoError(t, err)
 	newWal := store.(*WAL)
 	// we have already read the header in OpenWAL
-	_, err = newWal.readHeader()
+	_, err = newWal.readWALHeader()
 	assert.Error(t, err)
 
 	actualMessages, finished, err := newWal.Read(10000)
@@ -352,7 +352,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeEntry(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)