mirror/dolt
1
0
Fork 0
mirror of https://github.com/dolthub/dolt.git synced 2026-01-24 03:09:22 -06:00
Code Issues Packages Projects Releases 123 Wiki Activity

Add child refs to memtable only when a chunk was added.

Browse Source 
Fixes https://github.com/dolthub/dolt/issues/5301
This commit is contained in:
Pavel Safronov
2023-02-07 17:03:59 -08:00
parent 826f5787d0
commit c0cd58c192
2 changed files with 21 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
go/store/nbs/mem_table.go
View File

@@ -33,6 +33,14 @@ import (
"github.com/dolthub/dolt/go/store/hash"
)
type addChunkResult int
const (
chunkExists addChunkResult = iota
chunkAdded
chunkNotAdded
)
func WriteChunks(chunks []chunks.Chunk) (string, []byte, error) {
var size uint64
for _, chunk := range chunks {
@@ -46,7 +54,8 @@ func WriteChunks(chunks []chunks.Chunk) (string, []byte, error) {
func writeChunksToMT(mt *memTable, chunks []chunks.Chunk) (string, []byte, error) {
for _, chunk := range chunks {
if !mt.addChunk(addr(chunk.Hash()), chunk.Data()) {
res := mt.addChunk(addr(chunk.Hash()), chunk.Data())
if res == chunkNotAdded {
return "", nil, errors.New("didn't create this memory table with enough space to add all the chunks")
}
}
@@ -78,17 +87,19 @@ func newMemTable(memTableSize uint64) *memTable {
return &memTable{chunks: map[addr][]byte{}, maxData: memTableSize}
}
func (mt *memTable) addChunk(h addr, data []byte) bool {
func (mt *memTable) addChunk(h addr, data []byte) addChunkResult {
if len(data) == 0 {
panic("NBS blocks cannot be zero length")
}
if _, ok := mt.chunks[h]; ok {
return true
return chunkExists
}
dataLen := uint64(len(data))
if mt.totalData+dataLen > mt.maxData {
return false
return chunkNotAdded
}
mt.totalData += dataLen
mt.chunks[h] = data
mt.order = append(mt.order, hasRecord{
@@ -97,7 +108,7 @@ func (mt *memTable) addChunk(h addr, data []byte) bool {
len(mt.order),
false,
})
return true
return chunkAdded
}
func (mt *memTable) addChildRefs(addrs hash.HashSet) {

10
go/store/nbs/store.go
View File

@@ -623,8 +623,8 @@ func (nbs *NomsBlockStore) addChunk(ctx context.Context, ch chunks.Chunk, addrs
}
a := addr(ch.Hash())
ok := nbs.mt.addChunk(a, ch.Data())
if !ok {
addChunkRes := nbs.mt.addChunk(a, ch.Data())
if addChunkRes == chunkNotAdded {
ts, err := nbs.tables.append(ctx, nbs.mt, checker, nbs.stats)
if err != nil {
if errors.Is(err, ErrDanglingRef) {
@@ -634,12 +634,12 @@ func (nbs *NomsBlockStore) addChunk(ctx context.Context, ch chunks.Chunk, addrs
}
nbs.tables = ts
nbs.mt = newMemTable(nbs.mtSize)
ok = nbs.mt.addChunk(a, ch.Data())
addChunkRes = nbs.mt.addChunk(a, ch.Data())
}
if ok {
if addChunkRes == chunkAdded {
nbs.mt.addChildRefs(addrs)
}
return ok, nil
return true, nil
}
// refCheck checks that no dangling references are being committed.