From 08bc2f87ab2b0fed0a9b28c4b39d105ba60ab46c Mon Sep 17 00:00:00 2001 From: prathamesh0 Date: Mon, 9 May 2022 14:50:14 +0530 Subject: [PATCH] Add block number while making inserts --- postgres/batch.go | 13 +++++++++---- postgres/batch_test.go | 5 +++++ postgres/database.go | 19 ++++++++----------- postgres/database_test.go | 23 ++++++++++++++--------- 4 files changed, 36 insertions(+), 24 deletions(-) diff --git a/postgres/batch.go b/postgres/batch.go index 1ceeaf8..0099567 100644 --- a/postgres/batch.go +++ b/postgres/batch.go @@ -17,6 +17,8 @@ package pgipfsethdb import ( + "math/big" + "github.com/ethereum/go-ethereum/ethdb" "github.com/jmoiron/sqlx" ) @@ -28,13 +30,16 @@ type Batch struct { db *sqlx.DB tx *sqlx.Tx valueSize int + + blockNumber *big.Int } // NewBatch returns a ethdb.Batch interface for PG-IPFS -func NewBatch(db *sqlx.DB, tx *sqlx.Tx) ethdb.Batch { +func NewBatch(db *sqlx.DB, tx *sqlx.Tx, blockNumber *big.Int) ethdb.Batch { b := &Batch{ - db: db, - tx: tx, + db: db, + tx: tx, + blockNumber: blockNumber, } if tx == nil { b.Reset() @@ -50,7 +55,7 @@ func (b *Batch) Put(key []byte, value []byte) (err error) { if err != nil { return err } - if _, err = b.tx.Exec(putPgStr, mhKey, value); err != nil { + if _, err = b.tx.Exec(putPgStr, mhKey, value, b.blockNumber.Uint64()); err != nil { return err } b.valueSize += len(value) diff --git a/postgres/batch_test.go b/postgres/batch_test.go index 199a183..ddb8328 100644 --- a/postgres/batch_test.go +++ b/postgres/batch_test.go @@ -49,6 +49,11 @@ var _ = Describe("Batch", func() { } database = pgipfsethdb.NewDatabase(db, cacheConfig) + + databaseWithBlock, ok := database.(*pgipfsethdb.Database) + Expect(ok).To(BeTrue()) + (*databaseWithBlock).BlockNumber = testBlockNumber + batch = database.NewBatch() }) AfterEach(func() { diff --git a/postgres/database.go b/postgres/database.go index 47334bc..e408c32 100644 --- a/postgres/database.go +++ b/postgres/database.go @@ -20,6 +20,7 @@ import ( "context" "errors" "fmt" + "math/big" "strconv" "strings" "time" @@ -32,15 +33,9 @@ import ( var errNotSupported = errors.New("this operation is not supported") var ( - hasPgStr = "SELECT exists(select 1 from public.blocks WHERE key = $1)" - - // TODO Fix the get query to accomodate block_number field - // Using LIMIT 1 for now - getPgStr = "SELECT data FROM public.blocks WHERE key = $1 LIMIT 1" - - // TODO Fix the put query to accomodate block_number field - // block_number has been put as 1 always for now. - putPgStr = "INSERT INTO public.blocks (key, data, block_number) VALUES ($1, $2, 1) ON CONFLICT DO NOTHING" + hasPgStr = "SELECT exists(select 1 from public.blocks WHERE key = $1)" + getPgStr = "SELECT data FROM public.blocks WHERE key = $1 LIMIT 1" + putPgStr = "INSERT INTO public.blocks (key, data, block_number) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING" deletePgStr = "DELETE FROM public.blocks WHERE key = $1" dbSizePgStr = "SELECT pg_database_size(current_database())" ) @@ -51,6 +46,8 @@ var _ ethdb.Database = &Database{} type Database struct { db *sqlx.DB cache *groupcache.Group + + BlockNumber *big.Int } func (d *Database) ModifyAncients(f func(ethdb.AncientWriteOp) error) (int64, error) { @@ -142,7 +139,7 @@ func (d *Database) Put(key []byte, value []byte) error { if err != nil { return err } - _, err = d.db.Exec(putPgStr, mhKey, value) + _, err = d.db.Exec(putPgStr, mhKey, value, d.BlockNumber.Uint64()) return err } @@ -251,7 +248,7 @@ func (d *Database) Compact(start []byte, limit []byte) error { // NewBatch creates a write-only database that buffers changes to its host db // until a final write is called func (d *Database) NewBatch() ethdb.Batch { - return NewBatch(d.db, nil) + return NewBatch(d.db, nil, d.BlockNumber) } // NewBatchWithSize satisfies the ethdb.Batcher interface. diff --git a/postgres/database_test.go b/postgres/database_test.go index eab71a7..2a9751d 100644 --- a/postgres/database_test.go +++ b/postgres/database_test.go @@ -32,13 +32,14 @@ import ( ) var ( - database ethdb.Database - db *sqlx.DB - err error - testHeader = types.Header{Number: big.NewInt(1337)} - testValue, _ = rlp.EncodeToBytes(testHeader) - testEthKey = testHeader.Hash().Bytes() - testMhKey, _ = pgipfsethdb.MultihashKeyFromKeccak256(testEthKey) + database ethdb.Database + db *sqlx.DB + err error + testBlockNumber = big.NewInt(1337) + testHeader = types.Header{Number: testBlockNumber} + testValue, _ = rlp.EncodeToBytes(testHeader) + testEthKey = testHeader.Hash().Bytes() + testMhKey, _ = pgipfsethdb.MultihashKeyFromKeccak256(testEthKey) ) var _ = Describe("Database", func() { @@ -53,6 +54,10 @@ var _ = Describe("Database", func() { } database = pgipfsethdb.NewDatabase(db, cacheConfig) + + databaseWithBlock, ok := database.(*pgipfsethdb.Database) + Expect(ok).To(BeTrue()) + (*databaseWithBlock).BlockNumber = testBlockNumber }) AfterEach(func() { groupcache.DeregisterGroup("db") @@ -67,7 +72,7 @@ var _ = Describe("Database", func() { Expect(has).ToNot(BeTrue()) }) It("returns true if a key-pair exists in the db", func() { - _, err = db.Exec("INSERT into public.blocks (key, data, block_number) VALUES ($1, $2, 1)", testMhKey, testValue) + _, err = db.Exec("INSERT into public.blocks (key, data, block_number) VALUES ($1, $2, $3)", testMhKey, testValue, testBlockNumber.Uint64()) Expect(err).ToNot(HaveOccurred()) has, err := database.Has(testEthKey) Expect(err).ToNot(HaveOccurred()) @@ -82,7 +87,7 @@ var _ = Describe("Database", func() { Expect(err.Error()).To(ContainSubstring("sql: no rows in result set")) }) It("returns the value associated with the key, if the pair exists", func() { - _, err = db.Exec("INSERT into public.blocks (key, data, block_number) VALUES ($1, $2, 1)", testMhKey, testValue) + _, err = db.Exec("INSERT into public.blocks (key, data, block_number) VALUES ($1, $2, $3)", testMhKey, testValue, testBlockNumber.Uint64()) Expect(err).ToNot(HaveOccurred()) val, err := database.Get(testEthKey) Expect(err).ToNot(HaveOccurred())