Skip to content

Commit

Permalink
Merge pull request #168 from ijsong/change-trim
Browse files Browse the repository at this point in the history 
feat(storage): change the trim not to remove the commit context
  • Loading branch information
@ijsong
ijsong authored Oct 5, 2022
2 parents 3059482 + b80964b commit 0b64855
Show file tree
Hide file tree
Showing 2 changed files with 169 additions and 118 deletions.
43 changes: 15 additions & 28 deletions internal/storage/storage.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -212,8 +212,10 @@ func (s *Storage) NextCommitContextOf(cc CommitContext) (next CommitContext, err
return decodeCommitContextKey(it.Key()), nil
}

// Trim deletes log entries whose GLSNs are less than or equal to the argument glsn.
// It returns the ErrNoCommitContext if there are no logs to delete.
// Trim deletes log entries whose GLSNs are less than or equal to the argument
// glsn. Internally, it removes records for both data and commits but does not
// remove the commit context.
// It returns the ErrNoLogEntry if there are no logs to delete.
func (s *Storage) Trim(glsn types.GLSN) error {
lem, err := s.findLTE(glsn)
if err != nil {
Expand All @@ -227,50 +229,35 @@ func (s *Storage) Trim(glsn types.GLSN) error {
_ = batch.Close()
}()

// commit context
cc, err := s.CommitContextOf(trimGLSN)
if err != nil {
return err
}
cckBegin := make([]byte, commitContextKeyLength)
cckBegin = encodeCommitContextKeyInternal(CommitContext{}, cckBegin)
cckEnd := make([]byte, commitContextKeyLength)
cckEnd = encodeCommitContextKeyInternal(cc, cckEnd)
if cc.CommittedGLSNEnd-1 == trimGLSN {
// cc is deletable.
cc.Version++
cckEnd = encodeCommitContextKeyInternal(cc, cckEnd)
}
if err := batch.DeleteRange(cckBegin, cckEnd, nil); err != nil {
return err
}

// commit
ckBegin := make([]byte, commitKeyLength)
ckBegin = encodeCommitKeyInternal(types.MinGLSN, ckBegin)
ckEnd := make([]byte, commitKeyLength)
ckEnd = encodeCommitKeyInternal(trimGLSN+1, ckEnd)
if err := batch.DeleteRange(ckBegin, ckEnd, nil); err != nil {
return err
}
_ = batch.DeleteRange(ckBegin, ckEnd, nil)

// data
dkBegin := make([]byte, dataKeyLength)
dkBegin = encodeDataKeyInternal(types.MinLLSN, dkBegin)
dkEnd := make([]byte, dataKeyLength)
dkEnd = encodeDataKeyInternal(trimLLSN+1, dkEnd)
if err := batch.DeleteRange(dkBegin, dkEnd, nil); err != nil {
return err
}
_ = batch.DeleteRange(dkBegin, dkEnd, nil)

return batch.Commit(s.writeOpts)
}

func (s *Storage) findLTE(glsn types.GLSN) (lem varlogpb.LogEntryMeta, err error) {
ck := make([]byte, commitKeyLength)
var upper []byte
if glsn < types.MaxGLSN {
upper = make([]byte, commitKeyLength)
upper = encodeCommitKeyInternal(glsn+1, upper)
} else {
upper = []byte{commitKeySentinelPrefix}
}

it := s.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{commitKeyPrefix},
UpperBound: encodeCommitKeyInternal(glsn+1, ck),
UpperBound: upper,
})
defer func() {
_ = it.Close()
Expand Down
244 changes: 154 additions & 90 deletions internal/storage/storage_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -718,99 +718,163 @@ func TestStorage_CommitContextOf(t *testing.T) {
})
}

func TestStorage_Trim(t *testing.T) {
testStorage(t, func(t testing.TB, stg *Storage) {
err := stg.Trim(1)
assert.ErrorIs(t, err, ErrNoLogEntry)

// CC : +-1-+
// LLSN: 1 2 3
// GLSN: 1 2 3
wb := stg.NewWriteBatch()
assert.NoError(t, wb.Set(1, nil))
assert.NoError(t, wb.Set(2, nil))
assert.NoError(t, wb.Set(3, nil))
assert.NoError(t, wb.Apply())
assert.NoError(t, wb.Close())
func TestStorage_TrimWhenNoLogEntry(t *testing.T) {
tcs := []struct {
name string
testf func(t testing.TB, stg *Storage)
}{
{
name: "MinGLSN",
testf: func(t testing.TB, stg *Storage) {
err := stg.Trim(types.MinGLSN)
assert.ErrorIs(t, err, ErrNoLogEntry)
},
},
{
name: "MaxGLSN",
testf: func(t testing.TB, stg *Storage) {
err := stg.Trim(types.MaxGLSN)
assert.ErrorIs(t, err, ErrNoLogEntry)
},
},
{
name: "InvalidGLSN",
testf: func(t testing.TB, stg *Storage) {
err := stg.Trim(types.InvalidGLSN)
assert.ErrorIs(t, err, ErrNoLogEntry)
},
},
}

cb, err := stg.NewCommitBatch(CommitContext{
Version: 1,
HighWatermark: 5,
CommittedGLSNBegin: 1,
CommittedGLSNEnd: 4,
CommittedLLSNBegin: 1,
for _, tc := range tcs {
t.Run(tc.name, func(t *testing.T) {
testStorage(t, func(t testing.TB, stg *Storage) {
tc.testf(t, stg)
})
})
assert.NoError(t, err)
assert.NoError(t, cb.Set(1, 1))
assert.NoError(t, cb.Set(2, 2))
assert.NoError(t, cb.Set(3, 3))
assert.NoError(t, cb.Apply())
assert.NoError(t, cb.Close())
}
}

// CC : +-1-+
// LLSN: _ 2 3
// GLSN: _ 2 3
err = stg.Trim(1)
assert.NoError(t, err)
it := stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{dataKeyPrefix},
UpperBound: []byte{dataKeySentinelPrefix},
})
assert.True(t, it.First())
assert.Equal(t, types.LLSN(2), decodeDataKey(it.Key()))
assert.True(t, it.Next())
assert.Equal(t, types.LLSN(3), decodeDataKey(it.Key()))
assert.NoError(t, it.Close())
it = stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{commitKeyPrefix},
UpperBound: []byte{commitKeySentinelPrefix},
})
assert.True(t, it.First())
assert.Equal(t, types.GLSN(2), decodeCommitKey(it.Key()))
assert.True(t, it.Next())
assert.Equal(t, types.GLSN(3), decodeCommitKey(it.Key()))
assert.NoError(t, it.Close())
it = stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{commitContextKeyPrefix},
UpperBound: []byte{commitContextKeySentinelPrefix},
})
assert.True(t, it.First())
assert.Equal(t, CommitContext{
Version: 1,
HighWatermark: 5,
CommittedGLSNBegin: 1,
CommittedGLSNEnd: 4,
CommittedLLSNBegin: 1,
}, decodeCommitContextKey(it.Key()))
assert.NoError(t, it.Close())
func TestStorage_Trim(t *testing.T) {
expectedCC := CommitContext{
Version: 1,
HighWatermark: 3,
CommittedGLSNBegin: 1,
CommittedGLSNEnd: 4,
CommittedLLSNBegin: 1,
}

// CC : +-_-+
// LLSN: _ _ _
// GLSN: _ _ _
err = stg.Trim(3)
assert.NoError(t, err)
it = stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{dataKeyPrefix},
UpperBound: []byte{dataKeySentinelPrefix},
})
assert.False(t, it.First())
assert.NoError(t, it.Close())
it = stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{commitKeyPrefix},
UpperBound: []byte{commitKeySentinelPrefix},
})
assert.False(t, it.First())
assert.NoError(t, it.Close())
it = stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{commitContextKeyPrefix},
UpperBound: []byte{commitContextKeySentinelPrefix},
})
assert.False(t, it.First())
assert.NoError(t, it.Close())
tcs := []struct {
name string
testf func(t testing.TB, stg *Storage)
}{
{
name: "MinGLSN",
testf: func(t testing.TB, stg *Storage) {
err := stg.Trim(types.MinGLSN)
assert.NoError(t, err)

it := stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{dataKeyPrefix},
UpperBound: []byte{dataKeySentinelPrefix},
})
assert.True(t, it.First())
assert.Equal(t, types.LLSN(2), decodeDataKey(it.Key()))
assert.True(t, it.Next())
assert.Equal(t, types.LLSN(3), decodeDataKey(it.Key()))
assert.NoError(t, it.Close())

it = stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{commitKeyPrefix},
UpperBound: []byte{commitKeySentinelPrefix},
})
assert.True(t, it.First())
assert.Equal(t, types.GLSN(2), decodeCommitKey(it.Key()))
assert.True(t, it.Next())
assert.Equal(t, types.GLSN(3), decodeCommitKey(it.Key()))
assert.NoError(t, it.Close())

cc, err := stg.ReadCommitContext()
assert.NoError(t, err)
assert.Equal(t, expectedCC, cc)
},
},
{
name: "LastCommittedGLSN",
testf: func(t testing.TB, stg *Storage) {
err := stg.Trim(3)
assert.NoError(t, err)
assert.NoError(t, err)

it := stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{dataKeyPrefix},
UpperBound: []byte{dataKeySentinelPrefix},
})
assert.False(t, it.First())
assert.NoError(t, it.Close())

it = stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{commitKeyPrefix},
UpperBound: []byte{commitKeySentinelPrefix},
})
assert.False(t, it.First())
assert.NoError(t, it.Close())

cc, err := stg.ReadCommitContext()
assert.NoError(t, err)
assert.Equal(t, expectedCC, cc)
},
},
{
name: "MaxGLSN",
testf: func(t testing.TB, stg *Storage) {
err := stg.Trim(types.MaxGLSN)
assert.NoError(t, err)

it := stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{dataKeyPrefix},
UpperBound: []byte{dataKeySentinelPrefix},
})
assert.False(t, it.First())
assert.NoError(t, it.Close())

it = stg.db.NewIter(&pebble.IterOptions{
LowerBound: []byte{commitKeyPrefix},
UpperBound: []byte{commitKeySentinelPrefix},
})
assert.False(t, it.First())
assert.NoError(t, it.Close())

cc, err := stg.ReadCommitContext()
assert.NoError(t, err)
assert.Equal(t, expectedCC, cc)
},
},
}

for glsn := types.MinGLSN; glsn < types.GLSN(5); glsn++ {
err = stg.Trim(glsn)
assert.ErrorIs(t, err, ErrNoLogEntry)
}
})
for _, tc := range tcs {
t.Run(tc.name, func(t *testing.T) {
testStorage(t, func(t testing.TB, stg *Storage) {
// CC : +---+
// LLSN: 1 2 3
// GLSN: 1 2 3
wb := stg.NewWriteBatch()
assert.NoError(t, wb.Set(1, nil))
assert.NoError(t, wb.Set(2, nil))
assert.NoError(t, wb.Set(3, nil))
assert.NoError(t, wb.Apply())
assert.NoError(t, wb.Close())

cb, err := stg.NewCommitBatch(expectedCC)
assert.NoError(t, err)
assert.NoError(t, cb.Set(1, 1))
assert.NoError(t, cb.Set(2, 2))
assert.NoError(t, cb.Set(3, 3))
assert.NoError(t, cb.Apply())
assert.NoError(t, cb.Close())

tc.testf(t, stg)
})
})
}
}

0 comments on commit 0b64855

Please sign in to comment.