From 5660e57a4147f281f1c3a8403d3c3d5247e04d0d Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Thu, 24 Oct 2024 11:27:47 +0200 Subject: [PATCH] ethdb: add DeleteRange feature --- core/rawdb/table.go | 17 +++++ core/rawdb/table_test.go | 4 + core/state/pruner/bloom.go | 2 + eth/protocols/snap/gentrie_test.go | 4 + ethdb/database.go | 3 + ethdb/dbtest/testsuite.go | 113 +++++++++++++++++++++++++++++ ethdb/delete_range.go | 77 ++++++++++++++++++++ ethdb/leveldb/leveldb.go | 28 +++++-- ethdb/memorydb/memorydb.go | 11 +++ ethdb/pebble/pebble.go | 20 +++++ ethdb/remotedb/remotedb.go | 4 + internal/ethapi/api.go | 4 + trie/sync_test.go | 4 + trie/trie_test.go | 2 + trie/trienode/proof.go | 8 ++ triedb/hashdb/database.go | 4 + 16 files changed, 298 insertions(+), 7 deletions(-) create mode 100644 ethdb/delete_range.go diff --git a/core/rawdb/table.go b/core/rawdb/table.go index bc1d354d10..47bd6ca8ab 100644 --- a/core/rawdb/table.go +++ b/core/rawdb/table.go @@ -129,6 +129,11 @@ func (t *table) Delete(key []byte) error { return t.db.Delete(append([]byte(t.prefix), key...)) } +// DeleteRange removes all keys in the range [start,end) from the database. +func (t *table) DeleteRange(start, end []byte) error { + return t.db.DeleteRange(append([]byte(t.prefix), start...), append([]byte(t.prefix), end...)) +} + // NewIterator creates a binary-alphabetical iterator over a subset // of database content with a particular key prefix, starting at a particular // initial key (or after, if it does not exist). @@ -211,6 +216,12 @@ func (b *tableBatch) Delete(key []byte) error { return b.batch.Delete(append([]byte(b.prefix), key...)) } +// DeleteRange inserts the removal all of the keys in the range [start,end) into +// the batch for later committing. +func (b *tableBatch) DeleteRange(start, end []byte) error { + return b.batch.DeleteRange(append([]byte(b.prefix), start...), append([]byte(b.prefix), end...)) +} + // ValueSize retrieves the amount of data queued up for writing. func (b *tableBatch) ValueSize() int { return b.batch.ValueSize() @@ -245,6 +256,12 @@ func (r *tableReplayer) Delete(key []byte) error { return r.w.Delete(trimmed) } +// DeleteRange implements the interface KeyValueWriter. +func (r *tableReplayer) DeleteRange(start, end []byte) error { + trimmedStart, trimmedEnd := start[len(r.prefix):], end[len(r.prefix):] + return r.w.DeleteRange(trimmedStart, trimmedEnd) +} + // Replay replays the batch contents. func (b *tableBatch) Replay(w ethdb.KeyValueWriter) error { return b.batch.Replay(&tableReplayer{w: w, prefix: b.prefix}) diff --git a/core/rawdb/table_test.go b/core/rawdb/table_test.go index aa6adf3e72..333d7960e7 100644 --- a/core/rawdb/table_test.go +++ b/core/rawdb/table_test.go @@ -41,6 +41,10 @@ func (r *testReplayer) Delete(key []byte) error { return nil } +func (r *testReplayer) DeleteRange(start, end []byte) error { + panic("not implemented") +} + func testTableDatabase(t *testing.T, prefix string) { db := NewTable(NewMemoryDatabase(), prefix) diff --git a/core/state/pruner/bloom.go b/core/state/pruner/bloom.go index dad2b5b2a8..df1881b45f 100644 --- a/core/state/pruner/bloom.go +++ b/core/state/pruner/bloom.go @@ -116,6 +116,8 @@ func (bloom *stateBloom) Put(key []byte, value []byte) error { // Delete removes the key from the key-value data store. func (bloom *stateBloom) Delete(key []byte) error { panic("not supported") } +func (bloom *stateBloom) DeleteRange(start, end []byte) error { panic("not supported") } + // Contain is the wrapper of the underlying contains function which // reports whether the key is contained. // - If it says yes, the key may be contained diff --git a/eth/protocols/snap/gentrie_test.go b/eth/protocols/snap/gentrie_test.go index 2da4f3c866..8c9174e32b 100644 --- a/eth/protocols/snap/gentrie_test.go +++ b/eth/protocols/snap/gentrie_test.go @@ -97,6 +97,10 @@ func (r *replayer) Delete(key []byte) error { return nil } +func (r *replayer) DeleteRange(start, end []byte) error { + panic("not implemented") +} + func byteToHex(str []byte) []byte { l := len(str) * 2 var nibbles = make([]byte, l) diff --git a/ethdb/database.go b/ethdb/database.go index c6e76fd2fe..cf5df79c67 100644 --- a/ethdb/database.go +++ b/ethdb/database.go @@ -35,6 +35,9 @@ type KeyValueWriter interface { // Delete removes the key from the key-value data store. Delete(key []byte) error + + // DeleteRange removes all keys in the range [start,end) from the key-value store. + DeleteRange(start, end []byte) error } // KeyValueStater wraps the Stat method of a backing data store. diff --git a/ethdb/dbtest/testsuite.go b/ethdb/dbtest/testsuite.go index 1af55a0e38..8f3a896e9e 100644 --- a/ethdb/dbtest/testsuite.go +++ b/ethdb/dbtest/testsuite.go @@ -21,6 +21,7 @@ import ( "crypto/rand" "slices" "sort" + "strconv" "testing" "github.com/ethereum/go-ethereum/ethdb" @@ -343,6 +344,95 @@ func TestDatabaseSuite(t *testing.T, New func() ethdb.KeyValueStore) { t.Fatalf("expected error on batch.Write after Close") } }) + + t.Run("DeleteRange", func(t *testing.T) { + db := New() + defer db.Close() + + test := func(addToBatch, deleteFromBatch bool) { + var batch ethdb.Batch + addRange := func(start, stop int) { + if addToBatch { + batch = db.NewBatch() + } + for i := start; i <= stop; i++ { + if addToBatch { + batch.Put([]byte(strconv.Itoa(i)), nil) + } else { + db.Put([]byte(strconv.Itoa(i)), nil) + } + } + if addToBatch && !deleteFromBatch { + batch.Write() + batch = nil + } + } + + deleteRange := func(start, end []byte) { + if deleteFromBatch { + if batch == nil { + batch = db.NewBatch() + } + batch.DeleteRange(start, end) + batch.Write() + batch = nil + } else { + db.DeleteRange(start, end) + } + } + + checkRange := func(start, stop int, exp bool) { + for i := start; i <= stop; i++ { + has, _ := db.Has([]byte(strconv.Itoa(i))) + if has && !exp { + t.Fatalf("unexpected key %d", i) + } + if !has && exp { + t.Fatalf("missing expected key %d", i) + } + } + } + + addRange(1, 9) + deleteRange([]byte("9"), []byte("1")) + checkRange(1, 9, true) + deleteRange([]byte("5"), []byte("5")) + checkRange(1, 9, true) + deleteRange([]byte("5"), []byte("50")) + checkRange(1, 4, true) + checkRange(5, 5, false) + checkRange(6, 9, true) + deleteRange([]byte(""), []byte("a")) + checkRange(1, 9, false) + + addRange(1, 999) + deleteRange([]byte("12345"), []byte("54321")) + checkRange(1, 1, true) + checkRange(2, 5, false) + checkRange(6, 12, true) + checkRange(13, 54, false) + checkRange(55, 123, true) + checkRange(124, 543, false) + checkRange(544, 999, true) + + addRange(1, 999) + deleteRange([]byte("3"), []byte("7")) + checkRange(1, 2, true) + checkRange(3, 6, false) + checkRange(7, 29, true) + checkRange(30, 69, false) + checkRange(70, 299, true) + checkRange(300, 699, false) + checkRange(700, 999, true) + + deleteRange([]byte(""), []byte("a")) + checkRange(1, 999, false) + } + + test(false, false) + test(false, true) + test(true, true) + }) } // BenchDatabaseSuite runs a suite of benchmarks against a KeyValueStore database @@ -438,6 +528,29 @@ func BenchDatabaseSuite(b *testing.B, New func() ethdb.KeyValueStore) { benchBatchWrite(b, keys, vals) }) }) + b.Run("DeleteRange", func(b *testing.B) { + benchDeleteRange := func(b *testing.B, count int) { + db := New() + defer db.Close() + + for i := 0; i < count; i++ { + db.Put([]byte(strconv.Itoa(i)), nil) + } + b.ResetTimer() + b.ReportAllocs() + + db.DeleteRange([]byte("0"), []byte("999999999")) + } + b.Run("DeleteRange100", func(b *testing.B) { + benchDeleteRange(b, 100) + }) + b.Run("DeleteRange1k", func(b *testing.B) { + benchDeleteRange(b, 1000) + }) + b.Run("DeleteRange10k", func(b *testing.B) { + benchDeleteRange(b, 10000) + }) + }) } func iterateKeys(it ethdb.Iterator) []string { diff --git a/ethdb/delete_range.go b/ethdb/delete_range.go new file mode 100644 index 0000000000..dad5d2f8ef --- /dev/null +++ b/ethdb/delete_range.go @@ -0,0 +1,77 @@ +// Copyright 2024 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package ethdb defines the interfaces for an Ethereum data store. +package ethdb + +import ( + "bytes" + "errors" +) + +var ErrTooManyKeys = errors.New("too many keys in deleted range") + +// DeleteRangeWithIterator is a fallback method for deleting a key range from a +// database that does not natively support range deletion. +// Note that the number of deleted keys is limited in order to avoid blocking for +// a very long time. ErrTooManyKeys is returned if the range has only been +// partially deleted. In this case the caller can repeat the call until it +// finally succeeds. +func DeleteRangeWithIterator(db KeyValueStore, start, end []byte) error { + batch := db.NewBatch() + if err := deleteRangeWithIterator(batch, db, start, end); err != nil { + return err + } + return batch.Write() +} + +// DeleteRangeFromBatch is a fallback method for deleting a key range in a batch +// from a database that does not natively support range deletion. +func DeleteRangeFromBatch(target Batch, source Iteratee, start, end []byte) error { + deleteRangeWithIterator(target, source, start, end) + var keys [][]byte + writer := HookedBatch{ + Batch: target, + OnPut: func(key []byte, value []byte) { + if bytes.Compare(start, key) <= 0 && bytes.Compare(end, key) > 0 { + keys = append(keys, key) + } + }, + } + target.Replay(writer) + for _, key := range keys { + if err := target.Delete(key); err != nil { + return err + } + } + return nil +} + +func deleteRangeWithIterator(target KeyValueWriter, source Iteratee, start, end []byte) error { + it := source.NewIterator(nil, start) + var count int + for it.Next() && bytes.Compare(end, it.Key()) > 0 { + count++ + if count > 10000 { // should not block for more than a second + return ErrTooManyKeys + } + if err := target.Delete(it.Key()); err != nil { + return err + } + } + it.Release() + return nil +} diff --git a/ethdb/leveldb/leveldb.go b/ethdb/leveldb/leveldb.go index 24925a4f04..19840e21b3 100644 --- a/ethdb/leveldb/leveldb.go +++ b/ethdb/leveldb/leveldb.go @@ -206,20 +206,27 @@ func (db *Database) Delete(key []byte) error { return db.db.Delete(key, nil) } +// DeleteRange removes all keys in the range [start,end) from the key-value store. +func (db *Database) DeleteRange(start, end []byte) error { + return ethdb.DeleteRangeWithIterator(db, start, end) +} + // NewBatch creates a write-only key-value store that buffers changes to its host // database until a final write is called. func (db *Database) NewBatch() ethdb.Batch { return &batch{ - db: db.db, - b: new(leveldb.Batch), + db: db.db, + wrappedDb: db, + b: new(leveldb.Batch), } } // NewBatchWithSize creates a write-only database batch with pre-allocated buffer. func (db *Database) NewBatchWithSize(size int) ethdb.Batch { return &batch{ - db: db.db, - b: leveldb.MakeBatch(size), + db: db.db, + wrappedDb: db, + b: leveldb.MakeBatch(size), } } @@ -413,9 +420,10 @@ func (db *Database) meter(refresh time.Duration, namespace string) { // batch is a write-only leveldb batch that commits changes to its host database // when Write is called. A batch cannot be used concurrently. type batch struct { - db *leveldb.DB - b *leveldb.Batch - size int + db *leveldb.DB + wrappedDb *Database + b *leveldb.Batch + size int } // Put inserts the given value into the batch for later committing. @@ -432,6 +440,12 @@ func (b *batch) Delete(key []byte) error { return nil } +// DeleteRange inserts the removal all of the keys in the range [start,end) into +// the batch for later committing. +func (b *batch) DeleteRange(start, end []byte) error { + return ethdb.DeleteRangeFromBatch(b, b.wrappedDb, start, end) +} + // ValueSize retrieves the amount of data queued up for writing. func (b *batch) ValueSize() int { return b.size diff --git a/ethdb/memorydb/memorydb.go b/ethdb/memorydb/memorydb.go index 532e0dfe3f..64f7fcd4ba 100644 --- a/ethdb/memorydb/memorydb.go +++ b/ethdb/memorydb/memorydb.go @@ -121,6 +121,11 @@ func (db *Database) Delete(key []byte) error { return nil } +// DeleteRange removes all keys in the range [start,end) from the key-value store. +func (db *Database) DeleteRange(start, end []byte) error { + return ethdb.DeleteRangeWithIterator(db, start, end) +} + // NewBatch creates a write-only key-value store that buffers changes to its host // database until a final write is called. func (db *Database) NewBatch() ethdb.Batch { @@ -223,6 +228,12 @@ func (b *batch) Delete(key []byte) error { return nil } +// DeleteRange inserts the removal all of the keys in the range [start,end) into +// the batch for later committing. +func (b *batch) DeleteRange(start, end []byte) error { + return ethdb.DeleteRangeFromBatch(b, b.db, start, end) +} + // ValueSize retrieves the amount of data queued up for writing. func (b *batch) ValueSize() int { return b.size diff --git a/ethdb/pebble/pebble.go b/ethdb/pebble/pebble.go index e2ba9b8c7b..2e45628267 100644 --- a/ethdb/pebble/pebble.go +++ b/ethdb/pebble/pebble.go @@ -338,6 +338,16 @@ func (d *Database) Delete(key []byte) error { return d.db.Delete(key, nil) } +// DeleteRange removes all keys in the range [start,end) from the key-value store. +func (d *Database) DeleteRange(start, end []byte) error { + d.quitLock.RLock() + defer d.quitLock.RUnlock() + if d.closed { + return pebble.ErrClosed + } + return d.db.DeleteRange(start, end, nil) +} + // NewBatch creates a write-only key-value store that buffers changes to its host // database until a final write is called. func (d *Database) NewBatch() ethdb.Batch { @@ -539,6 +549,16 @@ func (b *batch) Delete(key []byte) error { return nil } +// DeleteRange inserts the removal all of the keys in the range [start,end) into +// the batch for later committing. +func (b *batch) DeleteRange(start, end []byte) error { + if err := b.b.DeleteRange(start, end, nil); err != nil { + return err + } + b.size += len(start) + len(end) + return nil +} + // ValueSize retrieves the amount of data queued up for writing. func (b *batch) ValueSize() int { return b.size diff --git a/ethdb/remotedb/remotedb.go b/ethdb/remotedb/remotedb.go index d0f018cb01..247a4392db 100644 --- a/ethdb/remotedb/remotedb.go +++ b/ethdb/remotedb/remotedb.go @@ -94,6 +94,10 @@ func (db *Database) Delete(key []byte) error { panic("not supported") } +func (db *Database) DeleteRange(start, end []byte) error { + panic("not supported") +} + func (db *Database) ModifyAncients(f func(ethdb.AncientWriteOp) error) (int64, error) { panic("not supported") } diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index 10d79c85ae..3a590809e3 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -704,6 +704,10 @@ func (n *proofList) Delete(key []byte) error { panic("not supported") } +func (n *proofList) DeleteRange(start, end []byte) error { + panic("not supported") +} + // GetProof returns the Merkle-proof for a given account and optionally some storage keys. func (api *BlockChainAPI) GetProof(ctx context.Context, address common.Address, storageKeys []string, blockNrOrHash rpc.BlockNumberOrHash) (*AccountResult, error) { var ( diff --git a/trie/sync_test.go b/trie/sync_test.go index 2ff02576d4..787a02b5fa 100644 --- a/trie/sync_test.go +++ b/trie/sync_test.go @@ -918,6 +918,10 @@ func (w *hookWriter) Delete(key []byte) error { return w.db.Delete(key) } +func (w *hookWriter) DeleteRange(start, end []byte) error { + panic("not supported") +} + func testSyncAbort(t *testing.T, scheme string) { var ( srcDisk = rawdb.NewMemoryDatabase() diff --git a/trie/trie_test.go b/trie/trie_test.go index 9b2530bdd4..acfbf9db0a 100644 --- a/trie/trie_test.go +++ b/trie/trie_test.go @@ -819,6 +819,7 @@ type spongeDb struct { func (s *spongeDb) Has(key []byte) (bool, error) { panic("implement me") } func (s *spongeDb) Get(key []byte) ([]byte, error) { return nil, errors.New("no such elem") } func (s *spongeDb) Delete(key []byte) error { panic("implement me") } +func (s *spongeDb) DeleteRange(start, end []byte) error { panic("implement me") } func (s *spongeDb) NewBatch() ethdb.Batch { return &spongeBatch{s} } func (s *spongeDb) NewBatchWithSize(size int) ethdb.Batch { return &spongeBatch{s} } func (s *spongeDb) Stat() (string, error) { panic("implement me") } @@ -867,6 +868,7 @@ func (b *spongeBatch) Put(key, value []byte) error { return nil } func (b *spongeBatch) Delete(key []byte) error { panic("implement me") } +func (b *spongeBatch) DeleteRange(start, end []byte) error { panic("implement me") } func (b *spongeBatch) ValueSize() int { return 100 } func (b *spongeBatch) Write() error { return nil } func (b *spongeBatch) Reset() {} diff --git a/trie/trienode/proof.go b/trie/trienode/proof.go index d3075ecccf..112e74a764 100644 --- a/trie/trienode/proof.go +++ b/trie/trienode/proof.go @@ -69,6 +69,10 @@ func (db *ProofSet) Delete(key []byte) error { return nil } +func (db *ProofSet) DeleteRange(start, end []byte) error { + panic("not supported") +} + // Get returns a stored node func (db *ProofSet) Get(key []byte) ([]byte, error) { db.lock.RLock() @@ -152,6 +156,10 @@ func (n *ProofList) Delete(key []byte) error { panic("not supported") } +func (n *ProofList) DeleteRange(start, end []byte) error { + panic("not supported") +} + // DataSize returns the aggregated data size of nodes in the list func (n ProofList) DataSize() int { var size int diff --git a/triedb/hashdb/database.go b/triedb/hashdb/database.go index 5de7805c31..2768f01d05 100644 --- a/triedb/hashdb/database.go +++ b/triedb/hashdb/database.go @@ -532,6 +532,10 @@ func (c *cleaner) Delete(key []byte) error { panic("not implemented") } +func (c *cleaner) DeleteRange(start, end []byte) error { + panic("not supported") +} + // Initialized returns an indicator if state data is already initialized // in hash-based scheme by checking the presence of genesis state. func (db *Database) Initialized(genesisRoot common.Hash) bool {