fix(embedded/tbtree): add marker entry to store timestamp when tree is empty

Signed-off-by: Stefano Scafiti <[email protected]>

Author: ostafen

embedded/tbtree/snapshot.go

@@ -505,11 +505,23 @@ func (l *leafNode) writeTo(nw, hw io.Writer, writeOpts *WriteOpts, buf []byte) (
 	buf[bi] = LeafNodeType
 	bi++
 
-	binary.BigEndian.PutUint16(buf[bi:], uint16(len(l.values)))
-	bi += 2
+	if len(l.values) == 0 && l._ts > 0 {
+		// NOTE: we store a marker entry to remember the highest timestamp seen
 
-	accH := int64(0)
+		binary.BigEndian.PutUint16(buf[bi:], 1)
+		bi += 2
 
+		binary.BigEndian.PutUint16(buf[bi:], 0)
+		bi += 2
+
+		binary.BigEndian.PutUint64(buf[bi:], l._ts)
+		bi += 8
+	} else {
+		binary.BigEndian.PutUint16(buf[bi:], uint16(len(l.values)))
+		bi += 2
+	}
+
+	accH := int64(0)
 	for _, v := range l.values {
 		timedValue := v.timedValues[0]
 

embedded/tbtree/tbtree.go

@@ -825,7 +825,6 @@ func (t *TBtree) readNodeFrom(r *appendable.Reader) (node, error) {
 		n.off = off
 		return n, nil
 	}
-
 	return nil, ErrReadingFileContent
 }
 
@@ -914,6 +913,16 @@ func (t *TBtree) readLeafNodeFrom(r *appendable.Reader) (*leafNode, error) {
 			return nil, err
 		}
 
+		if ksize == 0 {
+			ts, err := r.ReadUint64()
+			if err != nil {
+				return nil, err
+			}
+
+			l._ts = ts
+			return l, nil
+		}
+
 		key := make([]byte, ksize)
 		_, err = r.Read(key)
 		if err != nil {
@@ -959,7 +968,6 @@ func (t *TBtree) readLeafNodeFrom(r *appendable.Reader) (*leafNode, error) {
 			l._ts = ts
 		}
 	}
-
 	return l, nil
 }
 
@@ -1607,7 +1615,6 @@ func (t *TBtree) IncreaseTs(ts uint64) error {
 		_, _, err := t.flushTree(t.cleanupPercentage, false, false, "increaseTs")
 		return err
 	}
-
 	return nil
 }
 

embedded/tbtree/tbtree_test.go

@@ -1314,6 +1314,30 @@ func TestTBTreeIncreaseTs(t *testing.T) {
 	require.ErrorIs(t, err, ErrAlreadyClosed)
 }
 
+func TestTBTreeFlushEmptyTree(t *testing.T) {
+	dir := t.TempDir()
+	t.Cleanup(func() {
+		os.RemoveAll(dir)
+	})
+
+	tbtree, err := Open(dir, DefaultOptions())
+	require.NoError(t, err)
+
+	err = tbtree.IncreaseTs(100)
+	require.NoError(t, err)
+
+	_, _, err = tbtree.Flush()
+	require.NoError(t, err)
+
+	err = tbtree.Close()
+	require.NoError(t, err)
+
+	tbtree, err = Open(dir, DefaultOptions())
+	require.NoError(t, err)
+
+	require.Equal(t, tbtree.Ts(), uint64(100))
+}
+
 func BenchmarkRandomInsertion(b *testing.B) {
 	seed := rand.NewSource(time.Now().UnixNano())
 	rnd := rand.New(seed)