Changes for using single-node v4 db schema

Author: prathamesh0

docker-compose.yml

@@ -5,9 +5,9 @@ services:
     restart: on-failure
     depends_on:
       - access-node
-    image: migrations-test
+    image: vulcanize/ipld-eth-db:v4.1.1-alpha
     environment:
-      DATABASE_USER: "postgres"
+      DATABASE_USER: "vdbm"
       DATABASE_NAME: "vulcanize_testing_v4"
       DATABASE_PASSWORD: "password"
       DATABASE_HOSTNAME: "access-node"

scripts/run_unit_test.sh

@@ -2,27 +2,24 @@
 
 set -e
 
-# Build database image
-docker build -t migrations-test https://github.com/vulcanize/ipld-eth-db.git#sharding
-
 mkdir -p out
 
-# Remove existing docker-multi-node directory
-rm -rf out/docker-multi-node/
+# Remove existing docker-tsdb directory
+rm -rf out/docker-tsdb/
 
-# Copy over files to setup multi-node database
-ID=$(docker create migrations-test)
-docker cp $ID:/app/docker-multi-node out/docker-multi-node/
+# Copy over files to setup TimescaleDB
+ID=$(docker create vulcanize/ipld-eth-db:v4.1.1-alpha)
+docker cp $ID:/app/docker-tsdb out/docker-tsdb/
 docker rm -v $ID
 
-# Spin up multi-node database
-docker-compose -f out/docker-multi-node/docker-compose.test.yml -f docker-compose.yml up ipld-eth-db
-sleep 20
+# Spin up TimescaleDB
+docker-compose -f out/docker-tsdb/docker-compose.test.yml -f docker-compose.yml up ipld-eth-db
+sleep 45
 
 # Run unit tests
 go clean -testcache
 make statedifftest
 
 # Clean up
-docker-compose -f out/docker-multi-node/docker-compose.test.yml -f docker-compose.yml down --remove-orphans --volumes
-rm -rf out/docker-multi-node/
+docker-compose -f out/docker-tsdb/docker-compose.test.yml -f docker-compose.yml down --remove-orphans --volumes
+rm -rf out/docker-tsdb/

statediff/indexer/database/file/indexer_test.go

@@ -51,7 +51,7 @@ var (
 	sqlxdb    *sqlx.DB
 	err       error
 	ind       interfaces.StateDiffIndexer
-	ipfsPgGet = `SELECT COALESCE(data, '') as data FROM public.blocks
+	ipfsPgGet = `SELECT data FROM public.blocks
 					WHERE key = $1 AND block_number = $2`
 	tx1, tx2, tx3, tx4, tx5, rct1, rct2, rct3, rct4, rct5                          []byte
 	mockBlock                                                                      *types.Block
@@ -425,7 +425,7 @@ func TestFileIndexer(t *testing.T) {
 		}
 		for i := range rcts {
 			results := make([]logIPLD, 0)
-			pgStr = `SELECT log_cids.index, log_cids.address, COALESCE(log_cids.topic0, '') as topic0, COALESCE(log_cids.topic1, '') as topic1, data FROM eth.log_cids
+			pgStr = `SELECT log_cids.index, log_cids.address, log_cids.topic0, log_cids.topic1, data FROM eth.log_cids
     				INNER JOIN eth.receipt_cids ON (log_cids.rct_id = receipt_cids.tx_id)
 					INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key)
 					WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC`
@@ -480,7 +480,7 @@ func TestFileIndexer(t *testing.T) {
 
 		for idx, c := range rcts {
 			result := make([]models.IPLDModel, 0)
-			pgStr = `SELECT COALESCE(data, '') as data
+			pgStr = `SELECT data
 					FROM eth.receipt_cids
 					INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = public.blocks.key)
 					WHERE receipt_cids.leaf_cid = $1`
@@ -510,7 +510,7 @@ func TestFileIndexer(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			postStatePgStr := `SELECT COALESCE(post_state, '') as post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
+			postStatePgStr := `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 			switch c {
 			case rct1CID.String():
 				require.Equal(t, rctLeaf1, data)
@@ -564,7 +564,7 @@ func TestFileIndexer(t *testing.T) {
 
 		// check that state nodes were properly indexed and published
 		stateNodes := make([]models.StateNodeModel, 0)
-		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, COALESCE(state_cids.state_path, '') as state_path, state_cids.header_id
+		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
 				FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 				WHERE header_cids.block_number = $1 AND node_type != 3`
 		err = sqlxdb.Select(&stateNodes, pgStr, mocks.BlockNumber.Uint64())
@@ -584,7 +584,7 @@ func TestFileIndexer(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			pgStr = `SELECT cast(block_number AS TEXT), header_id, state_path, cast(balance AS TEXT), nonce, COALESCE(code_hash, '') as code_hash, storage_root from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
+			pgStr = `SELECT cast(block_number AS TEXT), header_id, state_path, cast(balance AS TEXT), nonce, code_hash, storage_root from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
 			var account models.StateAccountModel
 			err = sqlxdb.Get(&account, pgStr, stateNode.HeaderID, stateNode.Path)
 			if err != nil {
@@ -668,7 +668,7 @@ func TestFileIndexer(t *testing.T) {
 
 		// check that storage nodes were properly indexed
 		storageNodes := make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
+		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash
@@ -702,7 +702,7 @@ func TestFileIndexer(t *testing.T) {
 
 		// check that Removed storage nodes were properly indexed
 		storageNodes = make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
+		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash

statediff/indexer/database/sql/indexer_shared_test.go

@@ -22,7 +22,7 @@ var (
 	db        sql.Database
 	err       error
 	ind       interfaces.StateDiffIndexer
-	ipfsPgGet = `SELECT COALESCE(data, '') as data FROM public.blocks
+	ipfsPgGet = `SELECT data FROM public.blocks
 					WHERE key = $1 AND block_number = $2`
 	tx1, tx2, tx3, tx4, tx5, rct1, rct2, rct3, rct4, rct5                          []byte
 	mockBlock                                                                      *types.Block

statediff/indexer/database/sql/pgx_indexer_test.go

@@ -257,7 +257,7 @@ func TestPGXIndexer(t *testing.T) {
 				AND transaction_cids.header_id = header_cids.block_hash
 				AND header_cids.block_number = $1
 				ORDER BY transaction_cids.index`
-		logsPgStr := `SELECT log_cids.index, log_cids.address, COALESCE(log_cids.topic0, '') as topic0, COALESCE(log_cids.topic1, '') as topic1, data FROM eth.log_cids
+		logsPgStr := `SELECT log_cids.index, log_cids.address, log_cids.topic0, log_cids.topic1, data FROM eth.log_cids
     				INNER JOIN eth.receipt_cids ON (log_cids.rct_id = receipt_cids.tx_id)
 					INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key)
 					WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC`
@@ -328,7 +328,7 @@ func TestPGXIndexer(t *testing.T) {
 
 		for idx, c := range rcts {
 			result := make([]models.IPLDModel, 0)
-			pgStr = `SELECT COALESCE(data, '') as data
+			pgStr = `SELECT data
 					FROM eth.receipt_cids
 					INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = public.blocks.key)
 					WHERE receipt_cids.leaf_cid = $1`
@@ -359,7 +359,7 @@ func TestPGXIndexer(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			postStatePgStr := `SELECT COALESCE(post_state, '') as post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
+			postStatePgStr := `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 			switch c {
 			case rct1CID.String():
 				require.Equal(t, rctLeaf1, data)
@@ -412,7 +412,7 @@ func TestPGXIndexer(t *testing.T) {
 		defer checkTxClosure(t, 1, 0, 1)
 		// check that state nodes were properly indexed and published
 		stateNodes := make([]models.StateNodeModel, 0)
-		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, COALESCE(state_cids.state_path, '') as state_path, state_cids.header_id
+		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
 				FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 				WHERE header_cids.block_number = $1 AND node_type != 3`
 		err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
@@ -432,7 +432,7 @@ func TestPGXIndexer(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			pgStr = `SELECT cast(block_number AS TEXT), header_id, state_path, cast(balance AS TEXT), nonce, COALESCE(code_hash, '') as code_hash, storage_root from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
+			pgStr = `SELECT cast(block_number AS TEXT), header_id, state_path, cast(balance AS TEXT), nonce, code_hash, storage_root from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
 			var account models.StateAccountModel
 			err = db.Get(context.Background(), &account, pgStr, stateNode.HeaderID, stateNode.Path)
 			if err != nil {
@@ -472,7 +472,7 @@ func TestPGXIndexer(t *testing.T) {
 
 		// check that Removed state nodes were properly indexed and published
 		stateNodes = make([]models.StateNodeModel, 0)
-		pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, COALESCE(state_cids.state_path, '') as state_path, state_cids.header_id
+		pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
 				FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 				WHERE header_cids.block_number = $1 AND node_type = 3`
 		err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
@@ -515,7 +515,7 @@ func TestPGXIndexer(t *testing.T) {
 		defer checkTxClosure(t, 1, 0, 1)
 		// check that storage nodes were properly indexed
 		storageNodes := make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
+		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash
@@ -549,7 +549,7 @@ func TestPGXIndexer(t *testing.T) {
 
 		// check that Removed storage nodes were properly indexed
 		storageNodes = make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
+		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash

statediff/indexer/database/sql/postgres/config.go

@@ -50,7 +50,7 @@ var DefaultConfig = Config{
 	Hostname:     "localhost",
 	Port:         8066,
 	DatabaseName: "vulcanize_testing_v4",
-	Username:     "postgres",
+	Username:     "vdbm",
 	Password:     "password",
 }
 

statediff/indexer/database/sql/postgres/database.go

@@ -40,55 +40,55 @@ type DB struct {
 func (db *DB) InsertHeaderStm() string {
 	return `INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase)
 			VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
-			ON CONFLICT DO NOTHING`
+			ON CONFLICT (block_hash, block_number) DO UPDATE SET (parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase) = ($3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, eth.header_cids.times_validated + 1, $16)`
 }
 
 // InsertUncleStm satisfies the sql.Statements interface
 func (db *DB) InsertUncleStm() string {
 	return `INSERT INTO eth.uncle_cids (block_number, block_hash, header_id, parent_hash, cid, reward, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7)
-			ON CONFLICT DO NOTHING`
+			ON CONFLICT (block_hash, block_number) DO NOTHING`
 }
 
 // InsertTxStm satisfies the sql.Statements interface
 func (db *DB) InsertTxStm() string {
 	return `INSERT INTO eth.transaction_cids (block_number, header_id, tx_hash, cid, dst, src, index, mh_key, tx_data, tx_type, value) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-			ON CONFLICT DO NOTHING`
+			ON CONFLICT (tx_hash, block_number) DO NOTHING`
 }
 
 // InsertAccessListElementStm satisfies the sql.Statements interface
 func (db *DB) InsertAccessListElementStm() string {
 	return `INSERT INTO eth.access_list_elements (block_number, tx_id, index, address, storage_keys) VALUES ($1, $2, $3, $4, $5)
-			ON CONFLICT DO NOTHING`
+			ON CONFLICT (tx_id, index, block_number) DO NOTHING`
 }
 
 // InsertRctStm satisfies the sql.Statements interface
 func (db *DB) InsertRctStm() string {
 	return `INSERT INTO eth.receipt_cids (block_number, tx_id, leaf_cid, contract, contract_hash, leaf_mh_key, post_state, post_status, log_root) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-		  	ON CONFLICT DO NOTHING`
+			ON CONFLICT (tx_id, block_number) DO NOTHING`
 }
 
 // InsertLogStm satisfies the sql.Statements interface
 func (db *DB) InsertLogStm() string {
 	return `INSERT INTO eth.log_cids (block_number, leaf_cid, leaf_mh_key, rct_id, address, index, topic0, topic1, topic2, topic3, log_data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-			ON CONFLICT DO NOTHING`
+			ON CONFLICT (rct_id, index, block_number) DO NOTHING`
 }
 
 // InsertStateStm satisfies the sql.Statements interface
 func (db *DB) InsertStateStm() string {
 	return `INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
-			ON CONFLICT DO NOTHING`
+			ON CONFLICT (header_id, state_path, block_number) DO UPDATE SET (block_number, state_leaf_key, cid, node_type, diff, mh_key) = ($1, $3, $4, $6, $7, $8)`
 }
 
 // InsertAccountStm satisfies the sql.Statements interface
 func (db *DB) InsertAccountStm() string {
 	return `INSERT INTO eth.state_accounts (block_number, header_id, state_path, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5, $6, $7)
-		  	ON CONFLICT DO NOTHING`
+			ON CONFLICT (header_id, state_path, block_number) DO NOTHING`
 }
 
 // InsertStorageStm satisfies the sql.Statements interface
 func (db *DB) InsertStorageStm() string {
 	return `INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-		  	ON CONFLICT DO NOTHING`
+			ON CONFLICT (header_id, state_path, storage_path, block_number) DO UPDATE SET (block_number, storage_leaf_key, cid, node_type, diff, mh_key) = ($1, $4, $5, $7, $8, $9)`
 }
 
 // InsertIPLDStm satisfies the sql.Statements interface