Skip to content

Commit 888a58b

Browse files
lance6716lance6716
authored
lightning: update dependency pebbledb to v1.1.0 (#52400) (#55127)
close #52402
1 parent 3c8f260 commit 888a58b

File tree

13 files changed

+588
-1004
lines changed

13 files changed

+588
-1004
lines changed

‎DEPS.bzl

Lines changed: 444 additions & 808 deletions
Large diffs are not rendered by default.

‎go.mod

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,8 @@ require (
2525
github.com/cheggaaa/pb/v3 v3.0.8
2626
github.com/cheynewallace/tabby v1.1.1
2727
github.com/cloudfoundry/gosigar v1.3.6
28-
github.com/cockroachdb/errors v1.8.1
29-
github.com/cockroachdb/pebble v0.0.0-20220415182917-06c9d3be25b3
28+
github.com/cockroachdb/errors v1.11.1
29+
github.com/cockroachdb/pebble v1.1.0
3030
github.com/coocood/freecache v1.2.1
3131
github.com/coreos/go-semver v0.3.1
3232
github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548
@@ -68,7 +68,7 @@ require (
6868
github.com/johannesboyne/gofakes3 v0.0.0-20230506070712-04da935ef877
6969
github.com/joho/sqltocsv v0.0.0-20210428211105-a6d6801d59df
7070
github.com/kisielk/errcheck v1.7.0
71-
github.com/klauspost/compress v1.17.4
71+
github.com/klauspost/compress v1.17.7
7272
github.com/ks3sdklib/aws-sdk-go v1.2.9
7373
github.com/kyoh86/exportloopref v0.1.11
7474
github.com/lestrrat-go/jwx/v2 v2.0.21
@@ -90,7 +90,7 @@ require (
9090
github.com/pingcap/tidb/pkg/parser v0.0.0-20211011031125-9b13dc409c5e
9191
github.com/pingcap/tipb v0.0.0-20240318032315-55a7867ddd50
9292
github.com/prometheus/client_golang v1.19.0
93-
github.com/prometheus/client_model v0.6.0
93+
github.com/prometheus/client_model v0.6.1
9494
github.com/prometheus/common v0.52.2
9595
github.com/prometheus/prometheus v0.50.1
9696
github.com/robfig/cron/v3 v3.0.1
@@ -127,7 +127,7 @@ require (
127127
go.uber.org/mock v0.4.0
128128
go.uber.org/multierr v1.11.0
129129
go.uber.org/zap v1.27.0
130-
golang.org/x/exp v0.0.0-20240205201215-2c58cdc269a3
130+
golang.org/x/exp v0.0.0-20240404231335-c0f41cb1a7a0
131131
golang.org/x/net v0.24.0
132132
golang.org/x/oauth2 v0.18.0
133133
golang.org/x/sync v0.7.0
@@ -145,6 +145,11 @@ require (
145145
sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67
146146
)
147147

148+
require (
149+
github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect
150+
github.com/getsentry/sentry-go v0.27.0 // indirect
151+
)
152+
148153
require (
149154
cloud.google.com/go v0.112.0 // indirect
150155
cloud.google.com/go/compute v1.24.0 // indirect
@@ -154,8 +159,7 @@ require (
154159
github.com/Azure/azure-sdk-for-go/sdk/internal v1.5.1 // indirect
155160
github.com/Azure/go-ntlmssp v0.0.0-20221128193559-754e69321358 // indirect
156161
github.com/AzureAD/microsoft-authentication-library-for-go v1.2.1 // indirect
157-
github.com/DataDog/zstd v1.4.5 // indirect
158-
github.com/HdrHistogram/hdrhistogram-go v1.1.2 // indirect
162+
github.com/DataDog/zstd v1.5.5 // indirect
159163
github.com/Masterminds/goutils v1.1.1 // indirect
160164
github.com/Masterminds/semver/v3 v3.1.1 // indirect
161165
github.com/Masterminds/sprig/v3 v3.2.2 // indirect
@@ -165,11 +169,10 @@ require (
165169
github.com/beorn7/perks v1.0.1 // indirect
166170
github.com/bmatcuk/doublestar/v2 v2.0.4 // indirect
167171
github.com/cenkalti/backoff/v4 v4.2.1 // indirect
168-
github.com/cespare/xxhash/v2 v2.2.0 // indirect
172+
github.com/cespare/xxhash/v2 v2.3.0 // indirect
169173
github.com/chavacava/garif v0.1.0 // indirect
170-
github.com/cockroachdb/logtags v0.0.0-20190617123548-eb05cc24525f // indirect
171-
github.com/cockroachdb/redact v1.0.8 // indirect
172-
github.com/cockroachdb/sentry-go v0.6.1-cockroachdb.2 // indirect
174+
github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect
175+
github.com/cockroachdb/redact v1.1.5 // indirect
173176
github.com/coocood/bbloom v0.0.0-20190830030839-58deb6228d64 // indirect
174177
github.com/coocood/rtutil v0.0.0-20190304133409-c84515f646f2 // indirect
175178
github.com/coreos/go-systemd/v22 v22.5.0 // indirect
@@ -255,7 +258,7 @@ require (
255258
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
256259
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
257260
github.com/rivo/uniseg v0.4.6 // indirect
258-
github.com/rogpeppe/go-internal v1.11.0 // indirect
261+
github.com/rogpeppe/go-internal v1.12.0 // indirect
259262
github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 // indirect
260263
github.com/segmentio/asm v1.2.0 // indirect
261264
github.com/shabbyrobe/gocovmerge v0.0.0-20190829150210-3e036491d500 // indirect

‎go.sum

Lines changed: 28 additions & 150 deletions
Large diffs are not rendered by default.

‎pkg/lightning/backend/external/engine_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ func testNewIter(
6363
}
6464

6565
func checkDupDB(t *testing.T, db *pebble.DB, expectedKeys, expectedValues [][]byte) {
66-
iter := db.NewIter(nil)
66+
iter, _ := db.NewIter(nil)
6767
var (
6868
gotKeys, gotValues [][]byte
6969
)

‎pkg/lightning/backend/local/BUILD.bazel

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ go_library(
6464
"//pkg/util/mathutil",
6565
"//pkg/util/ranger",
6666
"@com_github_cockroachdb_pebble//:pebble",
67+
"@com_github_cockroachdb_pebble//objstorage/objstorageprovider",
6768
"@com_github_cockroachdb_pebble//sstable",
6869
"@com_github_cockroachdb_pebble//vfs",
6970
"@com_github_coreos_go_semver//semver",
@@ -157,7 +158,9 @@ go_test(
157158
"//pkg/util/hack",
158159
"//pkg/util/mock",
159160
"@com_github_cockroachdb_pebble//:pebble",
161+
"@com_github_cockroachdb_pebble//objstorage/objstorageprovider",
160162
"@com_github_cockroachdb_pebble//sstable",
163+
"@com_github_cockroachdb_pebble//vfs",
161164
"@com_github_coreos_go_semver//semver",
162165
"@com_github_data_dog_go_sqlmock//:go-sqlmock",
163166
"@com_github_docker_go_units//:go-units",

‎pkg/lightning/backend/local/engine.go

Lines changed: 38 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,9 @@ import (
2929
"time"
3030

3131
"github.com/cockroachdb/pebble"
32+
"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
3233
"github.com/cockroachdb/pebble/sstable"
34+
"github.com/cockroachdb/pebble/vfs"
3335
"github.com/google/btree"
3436
"github.com/google/uuid"
3537
"github.com/pingcap/errors"
@@ -982,7 +984,12 @@ func (e *Engine) newKVIter(ctx context.Context, opts *pebble.IterOptions, buf *m
982984
opts = &newOpts
983985
}
984986
if !e.duplicateDetection {
985-
return &pebbleIter{Iterator: e.getDB().NewIter(opts), buf: buf}
987+
iter, err := e.getDB().NewIter(opts)
988+
if err != nil {
989+
e.logger.Panic("fail to create iterator")
990+
return nil
991+
}
992+
return &pebbleIter{Iterator: iter, buf: buf}
986993
}
987994
logger := log.FromContext(ctx).With(
988995
zap.String("table", common.UniqueTable(e.tableInfo.DB, e.tableInfo.Name)),
@@ -1362,15 +1369,21 @@ var errorUnorderedSSTInsertion = errors.New("inserting KVs into SST without orde
13621369
type sstWriter struct {
13631370
*sstMeta
13641371
writer *sstable.Writer
1372+
1373+
// To dedup keys before write them into the SST file.
1374+
// NOTE: keys should be sorted and deduped when construct one SST file.
1375+
lastKey []byte
1376+
13651377
logger log.Logger
13661378
}
13671379

13681380
func newSSTWriter(path string, blockSize int) (*sstable.Writer, error) {
1369-
f, err := os.Create(path)
1381+
f, err := vfs.Default.Create(path)
13701382
if err != nil {
13711383
return nil, errors.Trace(err)
13721384
}
1373-
writer := sstable.NewWriter(f, sstable.WriterOptions{
1385+
writable := objstorageprovider.NewFileWritable(f)
1386+
writer := sstable.NewWriter(writable, sstable.WriterOptions{
13741387
TablePropertyCollectors: []func() pebble.TablePropertyCollector{
13751388
newRangePropertiesCollector,
13761389
},
@@ -1393,9 +1406,8 @@ func (sw *sstWriter) writeKVs(kvs []common.KvPair) error {
13931406
internalKey := sstable.InternalKey{
13941407
Trailer: uint64(sstable.InternalKeyKindSet),
13951408
}
1396-
var lastKey []byte
13971409
for _, p := range kvs {
1398-
if bytes.Equal(p.Key, lastKey) {
1410+
if sw.lastKey != nil && bytes.Equal(p.Key, sw.lastKey) {
13991411
sw.logger.Warn("duplicated key found, skip write", logutil.Key("key", p.Key))
14001412
continue
14011413
}
@@ -1404,10 +1416,10 @@ func (sw *sstWriter) writeKVs(kvs []common.KvPair) error {
14041416
return errors.Trace(err)
14051417
}
14061418
sw.totalSize += int64(len(p.Key)) + int64(len(p.Val))
1407-
lastKey = p.Key
1419+
sw.lastKey = p.Key
14081420
}
14091421
sw.totalCount += int64(len(kvs))
1410-
sw.maxKey = append(sw.maxKey[:0], lastKey...)
1422+
sw.maxKey = append(sw.maxKey[:0], sw.lastKey...)
14111423
return nil
14121424
}
14131425

@@ -1486,9 +1498,16 @@ func (h *sstIterHeap) Next() ([]byte, []byte, error) {
14861498
}
14871499

14881500
var k *pebble.InternalKey
1489-
k, iter.val = iter.iter.Next()
1501+
var v pebble.LazyValue
1502+
k, v = iter.iter.Next()
1503+
14901504
if k != nil {
1505+
vBytes, _, err := v.Value(nil)
1506+
if err != nil {
1507+
return nil, nil, errors.Trace(err)
1508+
}
14911509
iter.key = k.UserKey
1510+
iter.val = vBytes
14921511
iter.valid = true
14931512
heap.Fix(h, 0)
14941513
} else {
@@ -1528,11 +1547,15 @@ func (i dbSSTIngester) mergeSSTs(metas []*sstMeta, dir string, blockSize int) (*
15281547
}
15291548

15301549
for _, p := range metas {
1531-
f, err := os.Open(p.path)
1550+
f, err := vfs.Default.Open(p.path)
1551+
if err != nil {
1552+
return nil, errors.Trace(err)
1553+
}
1554+
readable, err := sstable.NewSimpleReadable(f)
15321555
if err != nil {
15331556
return nil, errors.Trace(err)
15341557
}
1535-
reader, err := sstable.NewReader(f, sstable.ReaderOptions{})
1558+
reader, err := sstable.NewReader(readable, sstable.ReaderOptions{})
15361559
if err != nil {
15371560
return nil, errors.Trace(err)
15381561
}
@@ -1544,14 +1567,18 @@ func (i dbSSTIngester) mergeSSTs(metas []*sstMeta, dir string, blockSize int) (*
15441567
if key == nil {
15451568
continue
15461569
}
1570+
valBytes, _, err := val.Value(nil)
1571+
if err != nil {
1572+
return nil, errors.Trace(err)
1573+
}
15471574
if iter.Error() != nil {
15481575
return nil, errors.Trace(iter.Error())
15491576
}
15501577
mergeIter.iters = append(mergeIter.iters, &sstIter{
15511578
name: p.path,
15521579
iter: iter,
15531580
key: key.UserKey,
1554-
val: val,
1581+
val: valBytes,
15551582
reader: reader,
15561583
valid: true,
15571584
})

‎pkg/lightning/backend/local/engine_mgr.go

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -194,10 +194,10 @@ func (em *engineManager) flushAllEngines(parentCtx context.Context) (err error)
194194

195195
func (em *engineManager) openEngineDB(engineUUID uuid.UUID, readOnly bool) (*pebble.DB, error) {
196196
opt := &pebble.Options{
197-
MemTableSize: em.MemTableSize,
197+
MemTableSize: uint64(em.MemTableSize),
198198
// the default threshold value may cause write stall.
199199
MemTableStopWritesThreshold: 8,
200-
MaxConcurrentCompactions: 16,
200+
MaxConcurrentCompactions: func() int { return 16 },
201201
// set threshold to half of the max open files to avoid trigger compaction
202202
L0CompactionThreshold: math.MaxInt32,
203203
L0StopWritesThreshold: math.MaxInt32,
@@ -525,7 +525,10 @@ func (em *engineManager) close() {
525525

526526
if em.duplicateDB != nil {
527527
// Check if there are duplicates that are not collected.
528-
iter := em.duplicateDB.NewIter(&pebble.IterOptions{})
528+
iter, err := em.duplicateDB.NewIter(&pebble.IterOptions{})
529+
if err != nil {
530+
em.logger.Panic("fail to create iterator")
531+
}
529532
hasDuplicates := iter.First()
530533
allIsWell := true
531534
if err := iter.Error(); err != nil {

‎pkg/lightning/backend/local/engine_mgr_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ func TestEngineManager(t *testing.T) {
8686
require.NoError(t, em.closeEngine(ctx, &backend.EngineConfig{}, engine1ID))
8787
require.Equal(t, 0, int(em.getImportedKVCount(engine1ID)))
8888
// close non-existent engine
89-
require.ErrorContains(t, em.closeEngine(ctx, &backend.EngineConfig{}, uuid.New()), "no such file or directory")
89+
require.ErrorContains(t, em.closeEngine(ctx, &backend.EngineConfig{}, uuid.New()), "does not exist")
9090

9191
// reset non-existent engine should work
9292
require.NoError(t, em.resetEngine(ctx, uuid.New(), false))

‎pkg/lightning/backend/local/engine_test.go

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,9 @@ import (
2525
"testing"
2626

2727
"github.com/cockroachdb/pebble"
28+
"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
2829
"github.com/cockroachdb/pebble/sstable"
30+
"github.com/cockroachdb/pebble/vfs"
2931
"github.com/google/uuid"
3032
"github.com/pingcap/tidb/br/pkg/membuf"
3133
"github.com/pingcap/tidb/pkg/lightning/backend"
@@ -47,7 +49,7 @@ func makePebbleDB(t *testing.T, opt *pebble.Options) (*pebble.DB, string) {
4749
func TestGetEngineSizeWhenImport(t *testing.T) {
4850
opt := &pebble.Options{
4951
MemTableSize: 1024 * 1024,
50-
MaxConcurrentCompactions: 16,
52+
MaxConcurrentCompactions: func() int { return 16 },
5153
L0CompactionThreshold: math.MaxInt32, // set to max try to disable compaction
5254
L0StopWritesThreshold: math.MaxInt32, // set to max try to disable compaction
5355
DisableWAL: true,
@@ -85,7 +87,7 @@ func TestGetEngineSizeWhenImport(t *testing.T) {
8587
func TestIngestSSTWithClosedEngine(t *testing.T) {
8688
opt := &pebble.Options{
8789
MemTableSize: 1024 * 1024,
88-
MaxConcurrentCompactions: 16,
90+
MaxConcurrentCompactions: func() int { return 16 },
8991
L0CompactionThreshold: math.MaxInt32, // set to max try to disable compaction
9092
L0StopWritesThreshold: math.MaxInt32, // set to max try to disable compaction
9193
DisableWAL: true,
@@ -107,9 +109,10 @@ func TestIngestSSTWithClosedEngine(t *testing.T) {
107109
f.db.Store(db)
108110
f.sstIngester = dbSSTIngester{e: f}
109111
sstPath := path.Join(tmpPath, uuid.New().String()+".sst")
110-
file, err := os.Create(sstPath)
112+
file, err := vfs.Default.Create(sstPath)
111113
require.NoError(t, err)
112-
w := sstable.NewWriter(file, sstable.WriterOptions{})
114+
writable := objstorageprovider.NewFileWritable(file)
115+
w := sstable.NewWriter(writable, sstable.WriterOptions{})
113116
for i := 0; i < 10; i++ {
114117
require.NoError(t, w.Add(sstable.InternalKey{
115118
Trailer: uint64(sstable.InternalKeyKindSet),

‎pkg/lightning/backend/local/iterator.go

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -164,9 +164,13 @@ func newDupDetectIter(
164164
}
165165

166166
detector := common.NewDupDetector(keyAdapter, dupDB.NewBatch(), logger, dupOpt)
167+
iter, err := db.NewIter(newOpts)
168+
if err != nil {
169+
panic("fail to create iterator")
170+
}
167171
return &dupDetectIter{
168172
keyAdapter: keyAdapter,
169-
iter: db.NewIter(newOpts),
173+
iter: iter,
170174
buf: buf,
171175
dupDetector: detector,
172176
logger: logger,
@@ -256,8 +260,12 @@ func newDupDBIter(dupDB *pebble.DB, keyAdapter common.KeyAdapter, opts *pebble.I
256260
if len(opts.UpperBound) > 0 {
257261
newOpts.UpperBound = keyAdapter.Encode(nil, opts.UpperBound, common.MinRowID)
258262
}
263+
iter, err := dupDB.NewIter(newOpts)
264+
if err != nil {
265+
panic("fail to create iterator")
266+
}
259267
return &dupDBIter{
260-
iter: dupDB.NewIter(newOpts),
268+
iter: iter,
261269
keyAdapter: keyAdapter,
262270
}
263271
}

0 commit comments

Comments
 (0)