[merge after release] dupsort of plain state (#913)

* dupsort of plain state

* rebase master

common/dbutils/bucket.go

@@ -4,21 +4,39 @@ import (
 	"sort"
 	"strings"
 
-	"github.com/ledgerwatch/turbo-geth/common/debug"
 	"github.com/ledgerwatch/turbo-geth/metrics"
 )
 
 // Buckets
 var (
 	// "Plain State". The same as CurrentStateBucket, but the keys arent' hashed.
 
-	// Contains Accounts:
-	//   key - address (unhashed)
-	//   value - account encoded for storage
-	// Contains Storage:
-	//   key - address (unhashed) + incarnation + storage key (unhashed)
-	//   value - storage value(common.hash)
-	PlainStateBucket = "PLAIN-CST"
+	/*
+		Logical layout:
+			Contains Accounts:
+			  key - address (unhashed)
+			  value - account encoded for storage
+			Contains Storage:
+			  key - address (unhashed) + incarnation + storage key (unhashed)
+			  value - storage value(common.hash)
+
+		Physical layout:
+			PlainStateBucket and CurrentStateBucket utilises DupSort feature of LMDB (store multiple values inside 1 key).
+		-------------------------------------------------------------
+			   key              |            value
+		-------------------------------------------------------------
+		[acc_hash]              | [acc_value]
+		[acc_hash]+[inc]        | [storage1_hash]+[storage1_value]
+								| [storage2_hash]+[storage2_value] // this value has no own key. it's 2nd value of [acc_hash]+[inc] key.
+								| [storage3_hash]+[storage3_value]
+								| ...
+		[acc_hash]+[old_inc]    | [storage1_hash]+[storage1_value]
+								| ...
+		[acc2_hash]             | [acc2_value]
+								...
+	*/
+	PlainStateBucket     = "PLAIN-CST2"
+	PlainStateBucketOld1 = "PLAIN-CST"
 
 	// "Plain State"
 	//key - address+incarnation
@@ -204,13 +222,14 @@ var DeprecatedBuckets = []string{
 	SyncStageProgressOld1,
 	SyncStageUnwindOld1,
 	CurrentStateBucketOld1,
+	PlainStateBucketOld1,
 }
 
 var BucketsCfg = map[string]*BucketConfigItem{}
 
 type BucketConfigItem struct {
 	ID         int
-	IsDupsort  bool
+	IsDupSort  bool
 	DupToLen   int
 	DupFromLen int
 }
@@ -232,7 +251,7 @@ var dupSortConfig = []dupSortConfigEntry{
 	},
 	{
 		Bucket:    PlainStateBucket,
-		IsDupSort: debug.IsPlainStateDupsortEnabled(),
+		IsDupSort: true,
 		ToLen:     28,
 		FromLen:   60,
 	},
@@ -262,7 +281,7 @@ func createBucketConfig(id int, name string) *BucketConfigItem {
 
 		cfg.DupFromLen = dupCfg.FromLen
 		cfg.DupToLen = dupCfg.ToLen
-		cfg.IsDupsort = dupCfg.IsDupSort
+		cfg.IsDupSort = dupCfg.IsDupSort
 	}
 
 	return cfg

common/debug/experiments.go

@@ -68,15 +68,3 @@ func TestDB() string {
 	})
 	return testDB
 }
-
-var (
-	dupsortPlain    bool
-	getDupsortPlain sync.Once
-)
-
-func IsPlainStateDupsortEnabled() bool {
-	getDupsortPlain.Do(func() {
-		_, dupsortPlain = os.LookupEnv("DUPSORT_PLAIN")
-	})
-	return dupsortPlain
-}

ethdb/kv_abstract_test.go

@@ -28,10 +28,10 @@ func TestManagedTx(t *testing.T) {
 	bucketID := 0
 	bucket1 := dbutils.Buckets[bucketID]
 	bucket2 := dbutils.Buckets[bucketID+1]
-	dbutils.BucketsCfg[bucket1].IsDupsort = true
+	dbutils.BucketsCfg[bucket1].IsDupSort = true
 	dbutils.BucketsCfg[bucket1].DupFromLen = 6
 	dbutils.BucketsCfg[bucket1].DupToLen = 4
-	dbutils.BucketsCfg[bucket2].IsDupsort = false
+	dbutils.BucketsCfg[bucket2].IsDupSort = false
 
 	writeDBs, readDBs, closeAll := setupDatabases()
 	defer closeAll()
@@ -110,7 +110,9 @@ func setupDatabases() (writeDBs []ethdb.KV, readDBs []ethdb.KV, close func()) {
 	return writeDBs, readDBs, func() {
 		grpcServer.Stop()
 
-		conn.Close()
+		if err := conn.Close(); err != nil {
+			panic(err)
+		}
 
 		for _, db := range readDBs {
 			db.Close()

ethdb/kv_lmdb.go

@@ -246,7 +246,7 @@ type lmdbTx struct {
 }
 
 type lmdbBucket struct {
-	isDupsort bool
+	isDupSort bool
 	dupFrom   int
 	dupTo     int
 	name      string
@@ -325,7 +325,7 @@ func (tx *lmdbTx) CreateBucket(name string) error {
 	if !tx.db.opts.readOnly {
 		flags |= lmdb.Create
 	}
-	if dbutils.BucketsCfg[name].IsDupsort {
+	if dbutils.BucketsCfg[name].IsDupSort {
 		flags |= lmdb.DupSort
 	}
 	dbi, err := tx.tx.OpenDBI(name, flags)
@@ -384,7 +384,7 @@ func (tx *lmdbTx) Bucket(name string) Bucket {
 		panic(fmt.Errorf("%w: %s", ErrUnknownBucket, name))
 	}
 
-	return &lmdbBucket{tx: tx, dbi: tx.db.buckets[name], isDupsort: cfg.IsDupsort, dupFrom: cfg.DupFromLen, dupTo: cfg.DupToLen, name: name}
+	return &lmdbBucket{tx: tx, dbi: tx.db.buckets[name], isDupSort: cfg.IsDupSort, dupFrom: cfg.DupFromLen, dupTo: cfg.DupToLen, name: name}
 }
 
 func (tx *lmdbTx) Commit(ctx context.Context) error {
@@ -458,7 +458,7 @@ func (c *LmdbCursor) NoValues() NoValuesCursor {
 func (tx *lmdbTx) Get(bucket string, key []byte) ([]byte, error) {
 	dbi := tx.db.buckets[bucket]
 	cfg := dbutils.BucketsCfg[bucket]
-	if cfg.IsDupsort {
+	if cfg.IsDupSort {
 		return tx.getDupSort(bucket, dbi, cfg, key)
 	}
 
@@ -503,7 +503,7 @@ func (tx *lmdbTx) getDupSort(bucket string, dbi lmdb.DBI, cfg *dbutils.BucketCon
 }
 
 func (b lmdbBucket) Get(key []byte) ([]byte, error) {
-	if b.isDupsort {
+	if b.isDupSort {
 		return b.getDupSort(key)
 	}
 
@@ -620,7 +620,7 @@ func (c *LmdbCursor) Last() ([]byte, []byte, error) {
 		return []byte{}, nil, err
 	}
 
-	if c.bucketCfg.IsDupsort {
+	if c.bucketCfg.IsDupSort {
 		if k == nil {
 			return k, v, nil
 		}
@@ -644,7 +644,7 @@ func (c *LmdbCursor) Seek(seek []byte) (k, v []byte, err error) {
 		}
 	}
 
-	if c.bucketCfg.IsDupsort {
+	if c.bucketCfg.IsDupSort {
 		return c.seekDupSort(seek)
 	}
 
@@ -734,7 +734,7 @@ func (c *LmdbCursor) Next() (k, v []byte, err error) {
 	default:
 	}
 
-	if c.bucketCfg.IsDupsort {
+	if c.bucketCfg.IsDupSort {
 		return c.nextDupSort()
 	}
 
@@ -791,7 +791,7 @@ func (c *LmdbCursor) Delete(key []byte) error {
 		}
 	}
 
-	if c.bucketCfg.IsDupsort {
+	if c.bucketCfg.IsDupSort {
 		return c.deleteDupSort(key)
 	}
 
@@ -854,7 +854,7 @@ func (c *LmdbCursor) Put(key []byte, value []byte) error {
 		}
 	}
 
-	if c.bucketCfg.IsDupsort {
+	if c.bucketCfg.IsDupSort {
 		return c.putDupSort(key, value)
 	}
 
@@ -986,7 +986,7 @@ func (c *LmdbCursor) Append(key []byte, value []byte) error {
 	}
 	b := c.bucketCfg
 	from, to := b.DupFromLen, b.DupToLen
-	if b.IsDupsort {
+	if b.IsDupSort {
 		if len(key) != from && len(key) >= to {
 			return fmt.Errorf("dupsort bucket: %s, can have keys of len==%d and len<%d. key: %x", c.bucketName, from, to, key)
 		}

migrations/dupsort_hash_state.go → migrations/dupsort_state.go

@@ -6,7 +6,7 @@ import (
 	"github.com/ledgerwatch/turbo-geth/ethdb"
 )
 
-var dupsortHashState = Migration{
+var dupSortHashState = Migration{
 	Name: "dupsort_hash_state",
 	Up: func(db ethdb.Database, datadir string, OnLoadCommit etl.LoadCommitHandler) error {
 		if exists, err := db.(ethdb.NonTransactional).BucketExists(dbutils.CurrentStateBucketOld1); err != nil {
@@ -40,3 +40,38 @@ var dupsortHashState = Migration{
 		return nil
 	},
 }
+
+var dupSortPlainState = Migration{
+	Name: "dupsort_plain_state",
+	Up: func(db ethdb.Database, datadir string, OnLoadCommit etl.LoadCommitHandler) error {
+		if exists, err := db.(ethdb.NonTransactional).BucketExists(dbutils.PlainStateBucketOld1); err != nil {
+			return err
+		} else if !exists {
+			return OnLoadCommit(db, nil, true)
+		}
+
+		if err := db.(ethdb.NonTransactional).ClearBuckets(dbutils.PlainStateBucket); err != nil {
+			return err
+		}
+		extractFunc := func(k []byte, v []byte, next etl.ExtractNextFunc) error {
+			return next(k, k, v)
+		}
+
+		if err := etl.Transform(
+			db,
+			dbutils.PlainStateBucketOld1,
+			dbutils.PlainStateBucket,
+			datadir,
+			extractFunc,
+			etl.IdentityLoadFunc,
+			etl.TransformArgs{OnLoadCommit: OnLoadCommit},
+		); err != nil {
+			return err
+		}
+
+		if err := db.(ethdb.NonTransactional).DropBuckets(dbutils.PlainStateBucketOld1); err != nil {
+			return err
+		}
+		return nil
+	},
+}

migrations/dupsort_hash_state_test.go → migrations/dupsort_state_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestDupsortHashState(t *testing.T) {
+func TestDupSortHashState(t *testing.T) {
 	require, db := require.New(t), ethdb.NewMemDatabase()
 
 	err := db.KV().Update(context.Background(), func(tx ethdb.Tx) error {
@@ -30,7 +30,7 @@ func TestDupsortHashState(t *testing.T) {
 	require.NoError(err)
 
 	migrator := NewMigrator()
-	migrator.Migrations = []Migration{dupsortHashState}
+	migrator.Migrations = []Migration{dupSortHashState}
 	err = migrator.Apply(db, "")
 	require.NoError(err)
 
@@ -73,3 +73,65 @@ func TestDupsortHashState(t *testing.T) {
 	require.Equal([]byte(storageKey)[keyLen:], v[:common.HashLength])
 	require.Equal([]byte{2}, v[common.HashLength:])
 }
+
+func TestDupSortPlainState(t *testing.T) {
+	require, db := require.New(t), ethdb.NewMemDatabase()
+
+	err := db.KV().Update(context.Background(), func(tx ethdb.Tx) error {
+		return tx.(ethdb.BucketMigrator).CreateBucket(dbutils.PlainStateBucketOld1)
+	})
+	require.NoError(err)
+
+	accKey := string(common.FromHex(fmt.Sprintf("%040x", 0)))
+	inc := string(common.FromHex("0000000000000001"))
+	storageKey := accKey + inc + string(common.FromHex(fmt.Sprintf("%064x", 0)))
+
+	err = db.Put(dbutils.PlainStateBucketOld1, []byte(accKey), []byte{1})
+	require.NoError(err)
+	err = db.Put(dbutils.PlainStateBucketOld1, []byte(storageKey), []byte{2})
+	require.NoError(err)
+
+	migrator := NewMigrator()
+	migrator.Migrations = []Migration{dupSortPlainState}
+	err = migrator.Apply(db, "")
+	require.NoError(err)
+
+	// test high-level data access didn't change
+	i := 0
+	err = db.Walk(dbutils.PlainStateBucket, nil, 0, func(k, v []byte) (bool, error) {
+		i++
+		return true, nil
+	})
+	require.NoError(err)
+	require.Equal(2, i)
+
+	v, err := db.Get(dbutils.PlainStateBucket, []byte(accKey))
+	require.NoError(err)
+	require.Equal([]byte{1}, v)
+
+	v, err = db.Get(dbutils.PlainStateBucket, []byte(storageKey))
+	require.NoError(err)
+	require.Equal([]byte{2}, v)
+
+	// test low-level data layout
+	rawKV := db.KV().(*ethdb.LmdbKV)
+	env := rawKV.Env()
+	allDBI := rawKV.AllDBI()
+
+	tx, err := env.BeginTxn(nil, lmdb.Readonly)
+	require.NoError(err)
+	c, err := tx.OpenCursor(allDBI[dbutils.PlainStateBucket])
+	require.NoError(err)
+
+	k, v, err := c.Get([]byte(accKey), nil, lmdb.Set)
+	require.NoError(err)
+	require.Equal([]byte(accKey), k)
+	require.Equal([]byte{1}, v)
+
+	keyLen := common.AddressLength + common.IncarnationLength
+	k, v, err = c.Get([]byte(storageKey)[:keyLen], []byte(storageKey)[keyLen:], lmdb.GetBothRange)
+	require.NoError(err)
+	require.Equal([]byte(storageKey)[:keyLen], k)
+	require.Equal([]byte(storageKey)[keyLen:], v[:common.HashLength])
+	require.Equal([]byte{2}, v[common.HashLength:])
+}

migrations/migrations.go

@@ -57,7 +57,8 @@ var migrations = []Migration{
 	unwindStagesToUseNamedKeys,
 	stagedsyncToUseStageBlockhashes,
 	unwindStagedsyncToUseStageBlockhashes,
-	dupsortHashState,
+	dupSortHashState,
+	dupSortPlainState,
 }
 
 type Migration struct {