Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix get snapshot panic for GC #20924

Merged
merged 6 commits into from
Dec 25, 2024
Merged

Fix get snapshot panic for GC #20924

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
6bb7891
GetTombstonesByBlockId skip tombstone
LeftHandCold Dec 24, 2024
2d2e762
Fix bug
LeftHandCold Dec 24, 2024
ba6216f
Fix reader
LeftHandCold Dec 24, 2024
ebf1a7d
Merge branch 'main' into dev-skip-tombstone
LeftHandCold Dec 24, 2024
4d3fcdd
remove log & add ut
LeftHandCold Dec 25, 2024
efe9fd6
Merge branch '2.0-dev' into dev-skip-tombstone
LeftHandCold Dec 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
151 changes: 151 additions & 0 deletions pkg/vm/engine/tae/db/test/db_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4348,6 +4348,157 @@ func TestBlockRead(t *testing.T) {
)
}

func TestBlockRead2(t *testing.T) {
blockio.RunPipelineTest(
func() {
defer testutils.AfterTest(t)()
ctx := context.Background()

opts := config.WithLongScanAndCKPOpts(nil)
tae := testutil.NewTestEngine(ctx, ModuleName, t, opts)
tsAlloc := types.NewTsAlloctor(opts.Clock)
defer tae.Close()
schema := catalog.MockSchemaAll(2, 1)
schema.Extra.BlockMaxRows = 20
schema.Extra.ObjectMaxBlocks = 2
tae.BindSchema(schema)
bat := catalog.MockBatch(schema, 40)

tae.CreateRelAndAppend(bat, true)
tae.CompactBlocks(false)

_, rel := tae.GetRelation()
blkEntry := testutil.GetOneObject(rel).GetMeta().(*catalog.ObjectEntry)

beforeDel := tsAlloc.Alloc()
txn1, rel := tae.GetRelation()
for i := 0; i < blkEntry.GetNonAppendableBlockCnt(); i++ {
commonID := blkEntry.AsCommonID()
commonID.SetBlockOffset(uint16(i))
assert.NoError(t, rel.RangeDelete(commonID, 0, 12, handle.DT_Normal))
}
assert.NoError(t, txn1.Commit(context.Background()))

afterFirstDel := tsAlloc.Alloc()
txn2, rel := tae.GetRelation()
for i := 0; i < blkEntry.GetNonAppendableBlockCnt(); i++ {
commonID := blkEntry.AsCommonID()
commonID.SetBlockOffset(uint16(i))
assert.NoError(t, rel.RangeDelete(commonID, 13, 15, handle.DT_Normal))
}
assert.NoError(t, txn2.Commit(context.Background()))

afterSecondDel := tsAlloc.Alloc()

tae.CompactBlocks(false)
_, rel = tae.GetRelation()
tombstoneObjectEntry := testutil.GetOneTombstoneMeta(rel)
objectEntry := testutil.GetOneBlockMeta(rel)
objStats := tombstoneObjectEntry.ObjectMVCCNode
ds := logtail.NewSnapshotDataSource(ctx, tae.DB.Runtime.Fs.Service, beforeDel, []objectio.ObjectStats{objStats.ObjectStats})
bid, _ := blkEntry.ID(), blkEntry.ID()

info := &objectio.BlockInfo{
BlockID: *objectio.NewBlockidWithObjectID(bid, 0),
}
metaloc := objectEntry.ObjectLocation()
metaloc.SetRows(schema.Extra.BlockMaxRows)
info.SetMetaLocation(metaloc)

columns := make([]string, 0)
colIdxs := make([]uint16, 0)
colTyps := make([]types.Type, 0)
defs := schema.ColDefs[:]
rand.Shuffle(len(defs), func(i, j int) { defs[i], defs[j] = defs[j], defs[i] })
for _, col := range defs {
columns = append(columns, col.Name)
colIdxs = append(colIdxs, uint16(col.Idx))
colTyps = append(colTyps, col.Type)
}
t.Log("read columns: ", columns)
fs := tae.DB.Runtime.Fs.Service
pool, err := mpool.NewMPool("test", 0, mpool.NoFixed)
assert.NoError(t, err)
infos := make([]*objectio.BlockInfo, 0)
infos = append(infos, info)
err = blockio.Prefetch("", fs, infos[0].MetaLocation())
assert.NoError(t, err)
buildBatch := func(typs []types.Type) *batch.Batch {
bat := batch.NewWithSize(len(typs))
//bat.Attrs = append(bat.Attrs, cols...)

for i := 0; i < len(typs); i++ {
bat.Vecs[i] = vector.NewVec(typs[i])
}
return bat
}
for i := 0; i < 2; i++ {
info.MetaLocation().SetID(uint16(i))
b1 := buildBatch(colTyps)
phyAddrColumnPos := -1
cacheVectors := containers.NewVectors(len(colIdxs) + 1)
ds.SetTS(beforeDel)
err = blockio.BlockDataReadInner(
context.Background(), info, ds, colIdxs, colTyps, phyAddrColumnPos,
beforeDel, nil, fileservice.Policy(0), b1, cacheVectors, pool, fs,
)
assert.NoError(t, err)
assert.Equal(t, len(columns), len(b1.Vecs))
assert.Equal(t, 20, b1.Vecs[0].Length())

ds.SetTS(afterFirstDel)

b2 := buildBatch(colTyps)
logutil.Infof("meta location: %v", info.MetaLocation().String())
err = blockio.BlockDataReadInner(
context.Background(), info, ds, colIdxs, colTyps, phyAddrColumnPos,
afterFirstDel, nil, fileservice.Policy(0), b2, cacheVectors, pool, fs,
)
assert.NoError(t, err)
assert.Equal(t, 7, b2.Vecs[0].Length())

ds.SetTS(afterSecondDel)

b3 := buildBatch(colTyps)
err = blockio.BlockDataReadInner(
context.Background(), info, ds, colIdxs, colTyps, phyAddrColumnPos,
afterSecondDel, nil, fileservice.Policy(0), b3, cacheVectors, pool, fs,
)
assert.NoError(t, err)
assert.Equal(t, len(columns), len(b2.Vecs))
assert.Equal(t, 4, b3.Vecs[0].Length())
// read rowid column only
b4 := buildBatch([]types.Type{types.T_Rowid.ToType()})
err = blockio.BlockDataReadInner(
context.Background(), info,
ds,
[]uint16{2},
[]types.Type{types.T_Rowid.ToType()},
0,
afterSecondDel, nil, fileservice.Policy(0), b4, cacheVectors, pool, fs,
)
assert.NoError(t, err)
assert.Equal(t, 1, len(b4.Vecs))
assert.Equal(t, 4, b4.Vecs[0].Length())

// read rowid column only
//info.Appendable = false
b5 := buildBatch([]types.Type{types.T_Rowid.ToType()})
err = blockio.BlockDataReadInner(
context.Background(), info,
ds, []uint16{2},
[]types.Type{types.T_Rowid.ToType()},
0,
afterSecondDel, nil, fileservice.Policy(0), b5, cacheVectors, pool, fs,
)
assert.NoError(t, err)
assert.Equal(t, 1, len(b5.Vecs))
assert.Equal(t, 4, b5.Vecs[0].Length())
}
},
)
}

func TestCompactDeltaBlk(t *testing.T) {
defer testutils.AfterTest(t)()
testutils.EnsureNoLeak(t)
Expand Down
73 changes: 46 additions & 27 deletions pkg/vm/engine/tae/logtail/backup.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,16 @@ func GetTombstonesByBlockId(
}

for idx := 0; idx < int(obj.BlkCnt()); idx++ {
if idx >= len(oData.data) {
logutil.Warn("GetTombstonesByBlockId skip tombstone",
zap.Int("idx", idx),
zap.Int("len", len(oData.data)),
zap.Int("blkcnt ", int(obj.BlkCnt())),
zap.Uint64("tid", oData.tid),
zap.String("name", obj.ObjectName().String()),
zap.String("stats", obj.String()))
return true, nil
}
rowids := vector.MustFixedColWithTypeCheck[types.Rowid](oData.data[idx].Vecs[0])
start, end := blockio.FindStartEndOfBlockFromSortedRowids(rowids, bid)
if start == end {
Expand Down Expand Up @@ -212,38 +222,47 @@ func (d *BackupDeltaLocDataSource) GetTombstones(
}
}
name := tombstone.ObjectName()
logutil.Infof("[GetSnapshot] tombstone object %v", name.String())
bat, _, err := blockio.LoadOneBlock(ctx, d.fs, tombstone.ObjectLocation(), objectio.SchemaData)
if err != nil {
return false, err
}
if !tombstone.GetCNCreated() {
deleteRow := make([]int64, 0)
for v := 0; v < bat.Vecs[0].Length(); v++ {
var commitTs types.TS
err = commitTs.Unmarshal(bat.Vecs[len(bat.Vecs)-1].GetRawBytesAt(v))
if err != nil {
return false, err
logutil.Infof("[GetSnapshot] tombstone object: %v, block count: %d", name.String(), tombstone.BlkCnt())
for id := uint32(0); id < tombstone.BlkCnt(); id++ {
location := tombstone.ObjectLocation()
location.SetID(uint16(id))
bat, _, err := blockio.LoadOneBlock(ctx, d.fs, location, objectio.SchemaData)
if err != nil {
return false, err
}
if !tombstone.GetCNCreated() {
deleteRow := make([]int64, 0)
for v := 0; v < bat.Vecs[0].Length(); v++ {
var commitTs types.TS
err = commitTs.Unmarshal(bat.Vecs[len(bat.Vecs)-1].GetRawBytesAt(v))
if err != nil {
return false, err
}
if commitTs.GT(&d.ts) {

logutil.Debug("[GetSnapshot]",
zap.Int("row", v),
zap.String("commitTs", commitTs.ToString()),
zap.String("location", location.String()))
} else {
deleteRow = append(deleteRow, int64(v))
}
}
if commitTs.GT(&d.ts) {
logutil.Debugf("delete row %v, commitTs %v, location %v",
v, commitTs.ToString(), name.String())
} else {
deleteRow = append(deleteRow, int64(v))
if len(deleteRow) != bat.Vecs[0].Length() {
bat.Shrink(deleteRow, false)
}
}
if len(deleteRow) != bat.Vecs[0].Length() {
bat.Shrink(deleteRow, false)
if id == 0 {
d.ds[name.String()] = &objData{
stats: &tombstone,
dataType: objectio.SchemaData,
sortKey: uint16(math.MaxUint16),
data: make([]*batch.Batch, 0),
appendable: true,
}
}
d.ds[name.String()].data = append(d.ds[name.String()].data, bat)
}
d.ds[name.String()] = &objData{
stats: &tombstone,
dataType: objectio.SchemaData,
sortKey: uint16(math.MaxUint16),
data: make([]*batch.Batch, 0),
appendable: true,
}
d.ds[name.String()].data = append(d.ds[name.String()].data, bat)
return true, nil
},
d.tombstones,
Expand Down
Loading