diff --git a/cmd/juno/dbcmd.go b/cmd/juno/dbcmd.go index 2f7fb27a73..2bfd05d2ec 100644 --- a/cmd/juno/dbcmd.go +++ b/cmd/juno/dbcmd.go @@ -5,6 +5,9 @@ import ( "errors" "fmt" "os" + "runtime" + "strconv" + "time" "github.com/NethermindEth/juno/blockchain" "github.com/NethermindEth/juno/blockchain/networks" @@ -22,6 +25,7 @@ import ( const ( dbRevertToBlockF = "to-block" + dbCompactForceF = "force" ) type DBInfo struct { @@ -44,7 +48,7 @@ func DBCmd(defaultDBPath string) *cobra.Command { dbCmd.PersistentFlags().String(dbPathF, defaultDBPath, dbPathUsage) dbCmd.PersistentFlags().Bool(newStateF, defaultNewState, newStateUsage) - dbCmd.AddCommand(DBInfoCmd(), DBSizeCmd(), DBRevertCmd()) + dbCmd.AddCommand(DBInfoCmd(), DBSizeCmd(), DBRevertCmd(), DBCompactCmd()) return dbCmd } @@ -78,6 +82,72 @@ func DBRevertCmd() *cobra.Command { return cmd } +func DBCompactCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "compact", + Short: "Compact the database so current table options apply to existing data", + Long: `This subcommand compacts the full key range with the current table options ` + + `(bloom filters, compression). Without --force, data already settled in the ` + + `bottom level — most of an aged database — is not rewritten; pass --force to ` + + `rewrite every sstable. It may take a long time and needs enough free disk ` + + `space to hold rewritten tables while old ones are dropped.`, + RunE: dbCompact, + } + cmd.Flags().String(dbCompressionF, "", + dbCompressionUsage+" Required: rewritten data is stored with it.") + cmd.Flags().Bool(dbCompactForceF, false, + "Rewrite every sstable even if the database is already fully compacted") + cmd.Flags().String(dbCompactionConcurrencyF, strconv.Itoa(runtime.GOMAXPROCS(0)), + "Number of concurrent compactions; the database is held exclusively, so default to all cores") + + return cmd +} + +func dbCompact(cmd *cobra.Command, args []string) error { + dbPath, err := cmd.Flags().GetString(dbPathF) + if err != nil { + return err + } + + compression, err := cmd.Flags().GetString(dbCompressionF) + if err != nil { + return err + } + if compression == "" { + return fmt.Errorf("--%s is required: compaction stores rewritten data with it", dbCompressionF) + } + + force, err := cmd.Flags().GetBool(dbCompactForceF) + if err != nil { + return err + } + + concurrency, err := cmd.Flags().GetString(dbCompactionConcurrencyF) + if err != nil { + return err + } + + database, err := openDB( + dbPath, + pebblev2.WithCompression(compression), + pebblev2.WithCompactionConcurrency(concurrency), + pebblev2.WithOfflineCompaction(), + ) + if err != nil { + return err + } + defer database.Close() + + fmt.Fprintln(cmd.OutOrStdout(), "Compacting the whole database, this may take a while") + start := time.Now() + if err := database.(*pebblev2.DB).CompactAll(cmd.Context(), force); err != nil { + return fmt.Errorf("compacting database: %w", err) + } + fmt.Fprintf(cmd.OutOrStdout(), "Compaction finished in %s\n", time.Since(start).Round(time.Second)) + + return nil +} + func dbInfo(cmd *cobra.Command, args []string) error { dbPath, err := cmd.Flags().GetString(dbPathF) if err != nil { @@ -355,13 +425,13 @@ func getNetwork( return "unknown" } -func openDB(path string) (db.KeyValueStore, error) { +func openDB(path string, options ...pebblev2.Option) (db.KeyValueStore, error) { _, err := os.Stat(path) if os.IsNotExist(err) { return nil, errors.New("database path does not exist") } - database, err := pebblev2.New(path) + database, err := pebblev2.New(path, append(options, pebblev2.WithBloomFilter())...) if err != nil { return nil, fmt.Errorf("failed to open db: %w", err) } diff --git a/db/pebblev2/db.go b/db/pebblev2/db.go index 01a952f573..b65d869162 100644 --- a/db/pebblev2/db.go +++ b/db/pebblev2/db.go @@ -1,14 +1,19 @@ package pebblev2 import ( + "bytes" "context" "errors" + "fmt" + "slices" "sync" "time" "github.com/NethermindEth/juno/db" "github.com/NethermindEth/juno/db/dbutils" "github.com/cockroachdb/pebble/v2" + "github.com/cockroachdb/pebble/v2/sstable" + "golang.org/x/sync/errgroup" ) var ( @@ -25,6 +30,16 @@ type DB struct { writeOpt *pebble.WriteOptions listener db.EventListener closeLock *sync.RWMutex // Ensures that the database is closed correctly + // compactionConcurrency mirrors the options' upper compaction concurrency + // bound, which pebble does not expose back; CompactAll fans out this many + // concurrent manual compactions. + compactionConcurrency int + // tableFilterPolicy and tableCompression mirror the options' bottom-level + // table settings; forced compaction skips tables whose properties already + // match both, making it resumable. Empty when not configured, which + // disables skipping. + tableFilterPolicy string + tableCompression string } // New opens a new database at the given path with default options @@ -43,17 +58,35 @@ func New(path string, options ...Option) (db.KeyValueStore, error) { } } + bottomLevel := &opts.Levels[len(opts.Levels)-1] + var tableFilterPolicy, tableCompression string + if bottomLevel.FilterPolicy != nil { + tableFilterPolicy = bottomLevel.FilterPolicy.Name() + } + if bottomLevel.Compression != nil { + tableCompression = bottomLevel.Compression().Name + } + pDB, err := pebble.Open(path, &opts) if err != nil { return nil, err } + compactionConcurrency := 1 // pebble's default range is (1, 1) + if opts.CompactionConcurrencyRange != nil { + _, upper := opts.CompactionConcurrencyRange() + compactionConcurrency = max(upper, 1) + } + return &DB{ - db: pDB, - path: path, - closeLock: new(sync.RWMutex), - listener: &db.SelectiveListener{}, - writeOpt: &pebble.WriteOptions{Sync: true}, // TODO: can we use non-sync writes for performance? + db: pDB, + path: path, + closeLock: new(sync.RWMutex), + listener: &db.SelectiveListener{}, + writeOpt: &pebble.WriteOptions{Sync: true}, // TODO: can we use non-sync writes for performance? + compactionConcurrency: compactionConcurrency, + tableFilterPolicy: tableFilterPolicy, + tableCompression: tableCompression, }, nil } @@ -234,6 +267,279 @@ func (d *DB) NewSnapshot() db.Snapshot { return NewSnapshot(d.db, d.listener) } +const ( + // maxForceCompactPasses bounds the forced-rewrite loop. A stale table + // escaping a pass via a move compaction descends at least one level, so + // passes beyond the level count mean no progress is possible. + maxForceCompactPasses = 8 + + // forceCompactChunkTables is the target number of stale tables per forced + // compaction chunk. Chunks cover disjoint key ranges and compact + // independently, letting pebble run them concurrently — a single + // full-range manual compaction has one contiguous in-use range and runs + // on one goroutine no matter the compaction concurrency. + forceCompactChunkTables = 16 +) + +// forceChunk is one independently-compactable slice of the forced rewrite. +type forceChunk struct { + start, end []byte + // bottomSmallest holds the smallest key of every stale bottom-level table + // in range; each gets a rewrite marker planted at it. + bottomSmallest [][]byte +} + +// CompactAll rewrites every sstable by compacting the full key range, +// materializing the current table options (bloom filters, compression) +// across existing data. Manual compaction rewrites bottom-level tables only +// as outputs of a compaction from the level above, so on an already-compacted +// database it picks nothing and rewrites nothing. force overwrites every +// stale bottom-level table's smallest key with its current value first, +// making each table an output of a real compaction, and repeats until no +// table predating the call remains: single-file compactions with no overlap +// below are "moves" that relink the old file a level down without rewriting +// it, so one pass is not always enough. Tables whose properties already match +// the configured filter policy and compression are skipped, so an interrupted +// forced compaction resumes instead of starting over. Force mode expects a +// database opened with WithOfflineCompaction — automatic compactions may +// otherwise merge the per-chunk marker sstables and serialize the chunk +// compactions. +func (d *DB) CompactAll(ctx context.Context, force bool) error { + d.closeLock.RLock() + defer d.closeLock.RUnlock() + + if d.closed { + return pebble.ErrClosed + } + + if !force { + return d.db.Compact(ctx, nil, bytes.Repeat([]byte{0xff}, 8), true) + } + + // Rewritten tables always receive a fresh file number, moved tables keep + // theirs: tables numbered at or below the baseline still need a rewrite. + baseline, err := d.maxTableNum() + if err != nil { + return err + } + + for range maxForceCompactPasses { + chunks, err := d.staleChunks(baseline) + if err != nil { + return err + } + if len(chunks) == 0 { + return nil + } + + if err := d.compactChunks(ctx, chunks); err != nil { + return err + } + } + + stale, err := d.staleChunks(baseline) + if err != nil { + return err + } + if len(stale) == 0 { + return nil + } + return fmt.Errorf("stale sstables remain after %d passes", maxForceCompactPasses) +} + +// compactChunks plants each chunk's markers and compacts its range, fanning +// the chunks out so pebble can run their compactions concurrently. Planting +// and flushing are serialized so every chunk's markers land in own sstables +// bounded to its range; a carrier spanning several chunks would make their +// compactions conflict on it and serialize. Chunks launch in stride order — +// key-space neighbors can share a straddling upper-level table, and pebble +// stalls the whole manual-compaction queue while its head conflicts with a +// running compaction, so the concurrently-active set is kept spread out. +func (d *DB) compactChunks(ctx context.Context, chunks []forceChunk) error { + var plantMu sync.Mutex + group, ctx := errgroup.WithContext(ctx) + group.SetLimit(d.compactionConcurrency) + + stride := max(1, (len(chunks)+d.compactionConcurrency-1)/d.compactionConcurrency) + ordered := make([]forceChunk, 0, len(chunks)) + for offset := range stride { + for i := offset; i < len(chunks); i += stride { + ordered = append(ordered, chunks[i]) + } + } + + for _, chunk := range ordered { + group.Go(func() error { + plantMu.Lock() + err := d.plantRewriteMarkers(chunk.bottomSmallest) + if err == nil && len(chunk.bottomSmallest) > 0 { + err = d.db.Flush() + } + plantMu.Unlock() + if err != nil { + return fmt.Errorf("planting rewrite markers: %w", err) + } + + return d.db.Compact(ctx, chunk.start, chunk.end, false) + }) + } + + return group.Wait() +} + +func (d *DB) maxTableNum() (pebble.TableNum, error) { + tables, err := d.db.SSTables() + if err != nil { + return 0, err + } + + var maxNum pebble.TableNum + for _, level := range tables { + for i := range level { + maxNum = max(maxNum, level[i].FileNum) + } + } + return maxNum, nil +} + +// tableUpToDate reports whether a table's properties already match the +// configured filter policy and compression, so a forced compaction can skip +// rewriting it. Skipping is disabled when either expectation is unknown. +func (d *DB) tableUpToDate(props *sstable.Properties) bool { + return d.tableFilterPolicy != "" && d.tableCompression != "" && props != nil && + props.FilterPolicyName == d.tableFilterPolicy && + props.CompressionName == d.tableCompression +} + +// keyRange is one table's inclusive user-key span. +type keyRange struct { + start, end []byte +} + +// staleTables returns the ranges of every stale table — numbered at or below +// baseline and not matching the configured table options — sorted by start +// key and split into bottom-level and upper-level tables. +func (d *DB) staleTables(baseline pebble.TableNum) (bottom, upper []keyRange, err error) { + tables, err := d.db.SSTables(pebble.WithProperties()) + if err != nil { + return nil, nil, err + } + + bottomLevel := len(tables) - 1 + for level, files := range tables { + for i := range files { + if files[i].FileNum > baseline || d.tableUpToDate(files[i].Properties) { + continue + } + stale := keyRange{ + start: bytes.Clone(files[i].Smallest.UserKey), + end: bytes.Clone(files[i].Largest.UserKey), + } + if level == bottomLevel { + bottom = append(bottom, stale) + } else { + upper = append(upper, stale) + } + } + } + byStart := func(a, b keyRange) int { return bytes.Compare(a.start, b.start) } + slices.SortFunc(bottom, byStart) + slices.SortFunc(upper, byStart) + return bottom, upper, nil +} + +// gapChunks returns extra chunks for the stale upper-level tables overlapping +// no chunk. Tables that overlap one need no extra chunk: the first chunk +// compaction to reach an upper level consumes every overlapping table there +// whole. The rest sit in a single inter-chunk gap each; tables in the same +// gap merge into one chunk — two chunks compacting overlapping ranges would +// conflict in pebble and stall the whole manual queue. +func gapChunks(chunks []forceChunk, upper []keyRange) []forceChunk { + overlapsChunk := func(t keyRange) bool { + next, _ := slices.BinarySearchFunc(chunks, t, func(c forceChunk, t keyRange) int { + return bytes.Compare(c.start, t.start) + }) + // Candidates: the chunk starting at or after t, and the one before it. + return next < len(chunks) && bytes.Compare(chunks[next].start, t.end) <= 0 || + next > 0 && bytes.Compare(t.start, chunks[next-1].end) <= 0 + } + + var gaps []forceChunk + for _, table := range upper { + if overlapsChunk(table) { + continue + } + if last := len(gaps) - 1; last >= 0 && bytes.Compare(table.start, gaps[last].end) <= 0 { + if bytes.Compare(table.end, gaps[last].end) > 0 { + gaps[last].end = table.end + } + continue + } + gaps = append(gaps, forceChunk{start: table.start, end: table.end}) + } + return gaps +} + +// staleChunks groups the stale bottom-level tables into chunks of +// forceCompactChunkTables tables. Bottom-level ranges are disjoint, so the +// chunks are too, and pebble can compact them concurrently. Stale upper-level +// tables ride along with the chunks they overlap, or get gap chunks of their +// own. +func (d *DB) staleChunks(baseline pebble.TableNum) ([]forceChunk, error) { + bottom, upper, err := d.staleTables(baseline) + if err != nil { + return nil, err + } + + var chunks []forceChunk + for start := 0; start < len(bottom); start += forceCompactChunkTables { + group := bottom[start:min(start+forceCompactChunkTables, len(bottom))] + chunk := forceChunk{start: group[0].start, end: group[len(group)-1].end} + for _, table := range group { + chunk.bottomSmallest = append(chunk.bottomSmallest, table.start) + } + chunks = append(chunks, chunk) + } + chunks = append(chunks, gapChunks(chunks, upper)...) + + for i := range chunks { + // Compact requires start < end; extend a single-key chunk minimally. + if bytes.Equal(chunks[i].start, chunks[i].end) { + chunks[i].end = append(bytes.Clone(chunks[i].end), 0) + } + } + return chunks, nil +} + +// plantRewriteMarkers overwrites each given bottom-level table's smallest key +// with its current merged value, or re-deletes it when a newer tombstone +// shadows it. Either marker changes no data, and its flushed sstable overlaps +// the table's range, forcing a following manual compaction to rewrite it. +func (d *DB) plantRewriteMarkers(smallestKeys [][]byte) error { + for _, smallest := range smallestKeys { + value, closer, err := d.db.Get(smallest) + if errors.Is(err, pebble.ErrNotFound) { + if err := d.db.Delete(smallest, pebble.NoSync); err != nil { + return err + } + continue + } + if err != nil { + return err + } + + value = bytes.Clone(value) // closing the getter invalidates the value + if err := closer.Close(); err != nil { + return err + } + if err := d.db.Set(smallest, value, pebble.NoSync); err != nil { + return err + } + } + + return nil +} + type Item struct { Count uint Size db.DataSize diff --git a/db/pebblev2/db_test.go b/db/pebblev2/db_test.go index 3ff5f8e27e..ac4881d1aa 100644 --- a/db/pebblev2/db_test.go +++ b/db/pebblev2/db_test.go @@ -1,7 +1,10 @@ package pebblev2 import ( + "bytes" "context" + "fmt" + "slices" "testing" "github.com/NethermindEth/juno/db" @@ -28,6 +31,249 @@ func newPebbleMem(t *testing.T) *DB { return db.(*DB) } +func TestCompactAll(t *testing.T) { + testDB := newPebbleMem(t) + + keys := [][]byte{{0}, {1, 2, 3}, {0xfe}, {0xff, 0xff}} + for _, key := range keys { + require.NoError(t, testDB.Put(key, key)) + } + + require.NoError(t, testDB.CompactAll(t.Context(), false)) + + for _, key := range keys { + require.NoError(t, testDB.Get(key, func(value []byte) error { + assert.Equal(t, key, value) + return nil + })) + } +} + +func TestCompactAllForce(t *testing.T) { + dir := t.TempDir() + keys := [][]byte{{0}, {1, 2, 3}, {0xfe}, {0xff, 0xff}} + + // Fully compact without a filter policy: the flat bottom level has no + // filter blocks. + testDB, err := New(dir) + require.NoError(t, err) + for _, key := range keys { + require.NoError(t, testDB.Put(key, key)) + } + require.NoError(t, testDB.(*DB).CompactAll(t.Context(), false)) + require.Empty(t, bottomFilterPolicies(t, testDB.(*DB))) + require.NoError(t, testDB.Close()) + + // A plain compaction of the already-flat database is a no-op, force + // rewrites it with the now-configured filter policy. + testDB, err = New(dir, WithBloomFilter()) + require.NoError(t, err) + pDB := testDB.(*DB) + + require.NoError(t, pDB.CompactAll(t.Context(), false)) + require.Empty(t, bottomFilterPolicies(t, pDB)) + + require.NoError(t, pDB.CompactAll(t.Context(), true)) + policies := bottomFilterPolicies(t, pDB) + require.NotEmpty(t, policies) + for _, policy := range policies { + assert.Equal(t, "rocksdb.BuiltinBloomFilter", policy) + } + + for _, key := range keys { + require.NoError(t, pDB.Get(key, func(value []byte) error { + assert.Equal(t, key, value) + return nil + })) + } + require.NoError(t, pDB.Close()) +} + +func TestCompactAllForceRewritesMovedTables(t *testing.T) { + dir := t.TempDir() + + // A flat filter-less bottom level in one key range, plus a filter-less + // upper-level table in a disjoint range: with no bottom-level overlap the + // latter enters the bottom level through a move compaction, which relinks + // the file without rewriting it. + testDB, err := New(dir) + require.NoError(t, err) + pDB := testDB.(*DB) + require.NoError(t, pDB.Put([]byte{1, 1}, []byte{1})) + require.NoError(t, pDB.Put([]byte{1, 2}, []byte{1})) + require.NoError(t, pDB.CompactAll(t.Context(), false)) + require.NoError(t, pDB.Put([]byte{2, 2}, []byte{2})) + require.NoError(t, pDB.Put([]byte{2, 3}, []byte{2})) + require.NoError(t, pDB.db.Flush()) + require.NoError(t, pDB.Close()) + + testDB, err = New(dir, WithBloomFilter()) + require.NoError(t, err) + pDB = testDB.(*DB) + + require.NoError(t, pDB.CompactAll(t.Context(), true)) + + tables, err := pDB.db.SSTables(pebble.WithProperties()) + require.NoError(t, err) + for _, level := range tables { + for i := range level { + assert.Equal(t, "rocksdb.BuiltinBloomFilter", level[i].Properties.FilterPolicyName, + "table %s has no filter", level[i].FileNum) + } + } + + for _, key := range [][]byte{{1, 1}, {1, 2}, {2, 2}, {2, 3}} { + require.NoError(t, pDB.Get(key, func(value []byte) error { + assert.Equal(t, key[:1], value) + return nil + })) + } + require.NoError(t, pDB.Close()) +} + +func TestCompactAllForceSingleKeyTable(t *testing.T) { + dir := t.TempDir() + + // A flat filter-less bottom level holding one sstable with a single key: + // a foreign tombstone has no room inside its range, only the same-key + // rewrite marker can force it. + testDB, err := New(dir) + require.NoError(t, err) + require.NoError(t, testDB.Put([]byte{7}, []byte{7})) + require.NoError(t, testDB.(*DB).CompactAll(t.Context(), false)) + require.NoError(t, testDB.Close()) + + testDB, err = New(dir, WithBloomFilter()) + require.NoError(t, err) + pDB := testDB.(*DB) + + require.NoError(t, pDB.CompactAll(t.Context(), true)) + + policies := bottomFilterPolicies(t, pDB) + require.NotEmpty(t, policies) + for _, policy := range policies { + assert.Equal(t, "rocksdb.BuiltinBloomFilter", policy) + } + + require.NoError(t, pDB.Get([]byte{7}, func(value []byte) error { + assert.Equal(t, []byte{7}, value) + return nil + })) + require.NoError(t, pDB.Close()) +} + +func TestStaleChunksMultiLevel(t *testing.T) { + dir := t.TempDir() + small := func(opts *pebble.Options) error { + opts.MemTableSize = 1 << 20 + for i := range opts.TargetFileSizes { + opts.TargetFileSizes[i] = 256 << 10 + } + return nil + } + + // Fill with automatic compaction running so tables settle into several + // levels, like on a long-running node. Wide upper-level tables straddle + // the bottom-level ones and must not chain the chunks together. + testDB, err := New(dir, small) + require.NoError(t, err) + pDB := testDB.(*DB) + for round := range 3 { + for done := 0; done < 80000; done += 10000 { + batch := pDB.NewBatch() + for i := done; i < done+10000; i++ { + key := fmt.Appendf(nil, "key-%03d-%06d", (i*7+round)%977, i) + require.NoError(t, batch.Put(key, bytes.Repeat([]byte{byte(i)}, 512))) + } + require.NoError(t, batch.Write()) + } + } + require.NoError(t, pDB.Close()) + + testDB, err = New(dir, small, WithBloomFilter(), WithOfflineCompaction()) + require.NoError(t, err) + pDB = testDB.(*DB) + defer pDB.Close() + + tables, err := pDB.db.SSTables() + require.NoError(t, err) + var levels int + for _, level := range tables { + if len(level) > 0 { + levels++ + } + } + require.Greater(t, levels, 1, "fill did not span multiple levels") + + baseline, err := pDB.maxTableNum() + require.NoError(t, err) + chunks, err := pDB.staleChunks(baseline) + require.NoError(t, err) + require.Greater(t, len(chunks), 1, "multi-level database must split into several chunks") + + slices.SortFunc(chunks, func(a, b forceChunk) int { return bytes.Compare(a.start, b.start) }) + for i := range chunks { + require.Negative(t, bytes.Compare(chunks[i].start, chunks[i].end)) + if i > 0 { + require.Negative(t, bytes.Compare(chunks[i-1].end, chunks[i].start), + "chunks %d and %d overlap", i-1, i) + } + } +} + +func TestCompactAllForceSkipsUpToDateTables(t *testing.T) { + dir := t.TempDir() + + testDB, err := New(dir, WithCompression("zstd")) + require.NoError(t, err) + for i := range byte(8) { + require.NoError(t, testDB.Put([]byte{i}, []byte{i})) + } + require.NoError(t, testDB.(*DB).CompactAll(t.Context(), false)) + require.NoError(t, testDB.Close()) + + testDB, err = New(dir, WithCompression("zstd"), WithBloomFilter(), WithOfflineCompaction()) + require.NoError(t, err) + pDB := testDB.(*DB) + defer pDB.Close() + + tableNums := func() map[pebble.TableNum]bool { + tables, err := pDB.db.SSTables() + require.NoError(t, err) + nums := make(map[pebble.TableNum]bool) + for _, level := range tables { + for i := range level { + nums[level[i].FileNum] = true + } + } + return nums + } + + require.NoError(t, pDB.CompactAll(t.Context(), true)) + require.NotEmpty(t, bottomFilterPolicies(t, pDB)) + + before := tableNums() + require.NoError(t, pDB.CompactAll(t.Context(), true)) + assert.Equal(t, before, tableNums(), "second forced compaction must rewrite nothing") +} + +// bottomFilterPolicies returns the filter policy name of every non-empty +// bottom-level sstable. +func bottomFilterPolicies(t *testing.T, pDB *DB) []string { + t.Helper() + tables, err := pDB.db.SSTables(pebble.WithProperties()) + require.NoError(t, err) + + var policies []string + bottom := tables[len(tables)-1] + for i := range bottom { + if name := bottom[i].Properties.FilterPolicyName; name != "" { + policies = append(policies, name) + } + } + return policies +} + func TestCalculatePrefixSize(t *testing.T) { t.Run("empty db", func(t *testing.T) { testDB := newPebbleMem(t) diff --git a/db/pebblev2/option.go b/db/pebblev2/option.go index b0130bc557..73959f414e 100644 --- a/db/pebblev2/option.go +++ b/db/pebblev2/option.go @@ -2,12 +2,14 @@ package pebblev2 import ( "fmt" + "math" "runtime" "strconv" "strings" "github.com/NethermindEth/juno/db" "github.com/cockroachdb/pebble/v2" + "github.com/cockroachdb/pebble/v2/bloom" "github.com/cockroachdb/pebble/v2/sstable/block" ) @@ -15,6 +17,10 @@ const ( // minCache is the minimum amount of memory in megabytes to allocate to pebble // read and write caching. This is also pebble's default value. minCacheSizeMB = 8 + + // bloomFilterBitsPerKey gives a ~1% false-positive rate, the industry + // standard trade-off between filter size and skipped reads. + bloomFilterBitsPerKey = 10 ) type Option = func(*pebble.Options) error @@ -27,6 +33,30 @@ func WithCacheSize(cacheSizeMB uint) Option { } } +// WithBloomFilter enables per-sstable bloom filters so point reads skip +// tables that cannot contain the key. Filters are built as sstables are +// written; existing tables gain them through compaction. +func WithBloomFilter() Option { + return func(opts *pebble.Options) error { + for i := range opts.Levels { + opts.Levels[i].FilterPolicy = bloom.FilterPolicy(bloomFilterBitsPerKey) + } + return nil + } +} + +// WithOfflineCompaction tunes an exclusively-held database for bulk manual +// compaction: automatic compactions stay off so they cannot merge the forced +// rewrite's per-chunk marker sstables into one (serializing the chunks), +// and L0 never stalls writes. +func WithOfflineCompaction() Option { + return func(opts *pebble.Options) error { + opts.DisableAutomaticCompactions = true + opts.L0StopWritesThreshold = math.MaxInt32 + return nil + } +} + func WithMaxOpenFiles(maxOpenFiles int) Option { return func(opts *pebble.Options) error { opts.MaxOpenFiles = maxOpenFiles diff --git a/node/node.go b/node/node.go index f5d1d8625a..35a1ecf020 100644 --- a/node/node.go +++ b/node/node.go @@ -224,6 +224,7 @@ func New(cfg *Config, version string, logLevel *log.Level) (*Node, error) { pebblev2.WithMemtableSize(cfg.DBMemtableSize), pebblev2.WithMemtableCount(cfg.DBMemtableCount), pebblev2.WithCompression(cfg.DBCompression), + pebblev2.WithBloomFilter(), ) }