Backport the psql indexer into v0.34.x (#6906)

This change backports the PostgreSQL indexing sink, addressing part of #6828.

Development on the main branch has diverged substantially since the v0.34.x
release. It includes package moves, breaking API and protobuf schema changes,
and new APIs, all of which together have a large footprint on the mapping
between the implementation at tip and the v0.34 release branch.

To avoid the need to retrofit all of those improvements, this change works by
injecting the new indexing sink into the existing (v0.34) indexing interfaces
by delegation. This means the backport does _not_ pull in all the newer APIs
for event handling, and thus has minimal impact on existing code written
against the v0.34 package structure.

This change includes the test for the `psql` implementation, and thus updates
some Go module dependencies. Because it does not interact with any other types,
however, I did not add any unit tests to other packages in this change.

Related changes:
 * Update module dependencies for psql backport.
 * Update test data to be type-compatible with the old protobuf types.
 * Add config settings for the PostgreSQL indexer.
 * Clean up some linter settings.
 * Hook up the psql indexer in the node main.

state/indexer/sink/psql/backport.go

@@ -0,0 +1,88 @@
+package psql
+
+// This file adds code to the psql package that is needed for integration with
+// v0.34, but which is not part of the original implementation.
+//
+// In v0.35, ADR 65 was implemented in which the TxIndexer and BlockIndexer
+// interfaces were merged into a hybrid EventSink interface. The Backport*
+// types defined here bridge the psql EventSink (which was built in terms of
+// the v0.35 interface) to the old interfaces.
+//
+// We took this narrower approach to backporting to avoid pulling in a much
+// wider-reaching set of changes in v0.35 that would have broken several of the
+// v0.34.x APIs. The result is sufficient to work with the node plumbing as it
+// exists in the v0.34 branch.
+
+import (
+	"context"
+	"errors"
+
+	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/libs/pubsub/query"
+	"github.com/tendermint/tendermint/state/txindex"
+	"github.com/tendermint/tendermint/types"
+)
+
+const (
+	eventTypeBeginBlock = "begin_block"
+	eventTypeEndBlock   = "end_block"
+)
+
+// TxIndexer returns a bridge from es to the Tendermint v0.34 transaction indexer.
+func (es *EventSink) TxIndexer() BackportTxIndexer {
+	return BackportTxIndexer{psql: es}
+}
+
+// BackportTxIndexer implements the txindex.TxIndexer interface by delegating
+// indexing operations to an underlying PostgreSQL event sink.
+type BackportTxIndexer struct{ psql *EventSink }
+
+// AddBatch indexes a batch of transactions in Postgres, as part of TxIndexer.
+func (b BackportTxIndexer) AddBatch(batch *txindex.Batch) error {
+	return b.psql.IndexTxEvents(batch.Ops)
+}
+
+// Index indexes a single transaction result in Postgres, as part of TxIndexer.
+func (b BackportTxIndexer) Index(txr *abci.TxResult) error {
+	return b.psql.IndexTxEvents([]*abci.TxResult{txr})
+}
+
+// Get is implemented to satisfy the TxIndexer interface, but is not supported
+// by the psql event sink and reports an error for all inputs.
+func (BackportTxIndexer) Get([]byte) (*abci.TxResult, error) {
+	return nil, errors.New("the TxIndexer.Get method is not supported")
+}
+
+// Search is implemented to satisfy the TxIndexer interface, but it is not
+// supported by the psql event sink and reports an error for all inputs.
+func (BackportTxIndexer) Search(context.Context, *query.Query) ([]*abci.TxResult, error) {
+	return nil, errors.New("the TxIndexer.Search method is not supported")
+}
+
+// BlockIndexer returns a bridge that implements the Tendermint v0.34 block
+// indexer interface, using the Postgres event sink as a backing store.
+func (es *EventSink) BlockIndexer() BackportBlockIndexer {
+	return BackportBlockIndexer{psql: es}
+}
+
+// BackportBlockIndexer implements the indexer.BlockIndexer interface by
+// delegating indexing operations to an underlying PostgreSQL event sink.
+type BackportBlockIndexer struct{ psql *EventSink }
+
+// Has is implemented to satisfy the BlockIndexer interface, but it is not
+// supported by the psql event sink and reports an error for all inputs.
+func (BackportBlockIndexer) Has(height int64) (bool, error) {
+	return false, errors.New("the BlockIndexer.Has method is not supported")
+}
+
+// Index indexes block begin and end events for the specified block.  It is
+// part of the BlockIndexer interface.
+func (b BackportBlockIndexer) Index(block types.EventDataNewBlockHeader) error {
+	return b.psql.IndexBlockEvents(block)
+}
+
+// Search is implemented to satisfy the BlockIndexer interface, but it is not
+// supported by the psql event sink and reports an error for all inputs.
+func (BackportBlockIndexer) Search(context.Context, *query.Query) ([]int64, error) {
+	return nil, errors.New("the BlockIndexer.Search method is not supported")
+}

state/indexer/sink/psql/backport_test.go

@@ -0,0 +1,11 @@
+package psql
+
+import (
+	"github.com/tendermint/tendermint/state/indexer"
+	"github.com/tendermint/tendermint/state/txindex"
+)
+
+var (
+	_ indexer.BlockIndexer = BackportBlockIndexer{}
+	_ txindex.TxIndexer    = BackportTxIndexer{}
+)

state/indexer/sink/psql/psql.go

@@ -0,0 +1,253 @@
+// Package psql implements an event sink backed by a PostgreSQL database.
+package psql
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/libs/pubsub/query"
+	"github.com/tendermint/tendermint/types"
+)
+
+const (
+	tableBlocks     = "blocks"
+	tableTxResults  = "tx_results"
+	tableEvents     = "events"
+	tableAttributes = "attributes"
+	driverName      = "postgres"
+)
+
+// EventSink is an indexer backend providing the tx/block index services.  This
+// implementation stores records in a PostgreSQL database using the schema
+// defined in state/indexer/sink/psql/schema.sql.
+type EventSink struct {
+	store   *sql.DB
+	chainID string
+}
+
+// NewEventSink constructs an event sink associated with the PostgreSQL
+// database specified by connStr. Events written to the sink are attributed to
+// the specified chainID.
+func NewEventSink(connStr, chainID string) (*EventSink, error) {
+	db, err := sql.Open(driverName, connStr)
+	if err != nil {
+		return nil, err
+	}
+
+	return &EventSink{
+		store:   db,
+		chainID: chainID,
+	}, nil
+}
+
+// DB returns the underlying Postgres connection used by the sink.
+// This is exported to support testing.
+func (es *EventSink) DB() *sql.DB { return es.store }
+
+// runInTransaction executes query in a fresh database transaction.
+// If query reports an error, the transaction is rolled back and the
+// error from query is reported to the caller.
+// Otherwise, the result of committing the transaction is returned.
+func runInTransaction(db *sql.DB, query func(*sql.Tx) error) error {
+	dbtx, err := db.Begin()
+	if err != nil {
+		return err
+	}
+	if err := query(dbtx); err != nil {
+		_ = dbtx.Rollback() // report the initial error, not the rollback
+		return err
+	}
+	return dbtx.Commit()
+}
+
+// queryWithID executes the specified SQL query with the given arguments,
+// expecting a single-row, single-column result containing an ID. If the query
+// succeeds, the ID from the result is returned.
+func queryWithID(tx *sql.Tx, query string, args ...interface{}) (uint32, error) {
+	var id uint32
+	if err := tx.QueryRow(query, args...).Scan(&id); err != nil {
+		return 0, err
+	}
+	return id, nil
+}
+
+// insertEvents inserts a slice of events and any indexed attributes of those
+// events into the database associated with dbtx.
+//
+// If txID > 0, the event is attributed to the Tendermint transaction with that
+// ID; otherwise it is recorded as a block event.
+func insertEvents(dbtx *sql.Tx, blockID, txID uint32, evts []abci.Event) error {
+	// Populate the transaction ID field iff one is defined (> 0).
+	var txIDArg interface{}
+	if txID > 0 {
+		txIDArg = txID
+	}
+
+	// Add each event to the events table, and retrieve its row ID to use when
+	// adding any attributes the event provides.
+	for _, evt := range evts {
+		// Skip events with an empty type.
+		if evt.Type == "" {
+			continue
+		}
+
+		eid, err := queryWithID(dbtx, `
+INSERT INTO `+tableEvents+` (block_id, tx_id, type) VALUES ($1, $2, $3)
+  RETURNING rowid;
+`, blockID, txIDArg, evt.Type)
+		if err != nil {
+			return err
+		}
+
+		// Add any attributes flagged for indexing.
+		for _, attr := range evt.Attributes {
+			if !attr.Index {
+				continue
+			}
+			compositeKey := evt.Type + "." + string(attr.Key)
+			if _, err := dbtx.Exec(`
+INSERT INTO `+tableAttributes+` (event_id, key, composite_key, value)
+  VALUES ($1, $2, $3, $4);
+`, eid, attr.Key, compositeKey, attr.Value); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+// makeIndexedEvent constructs an event from the specified composite key and
+// value. If the key has the form "type.name", the event will have a single
+// attribute with that name and the value; otherwise the event will have only
+// a type and no attributes.
+func makeIndexedEvent(compositeKey, value string) abci.Event {
+	i := strings.Index(compositeKey, ".")
+	if i < 0 {
+		return abci.Event{Type: compositeKey}
+	}
+	return abci.Event{Type: compositeKey[:i], Attributes: []abci.EventAttribute{
+		{Key: []byte(compositeKey[i+1:]), Value: []byte(value), Index: true},
+	}}
+}
+
+// IndexBlockEvents indexes the specified block header, part of the
+// indexer.EventSink interface.
+func (es *EventSink) IndexBlockEvents(h types.EventDataNewBlockHeader) error {
+	ts := time.Now().UTC()
+
+	return runInTransaction(es.store, func(dbtx *sql.Tx) error {
+		// Add the block to the blocks table and report back its row ID for use
+		// in indexing the events for the block.
+		blockID, err := queryWithID(dbtx, `
+INSERT INTO `+tableBlocks+` (height, chain_id, created_at)
+  VALUES ($1, $2, $3)
+  ON CONFLICT DO NOTHING
+  RETURNING rowid;
+`, h.Header.Height, es.chainID, ts)
+		if err == sql.ErrNoRows {
+			return nil // we already saw this block; quietly succeed
+		} else if err != nil {
+			return fmt.Errorf("indexing block header: %w", err)
+		}
+
+		// Insert the special block meta-event for height.
+		if err := insertEvents(dbtx, blockID, 0, []abci.Event{
+			makeIndexedEvent(types.BlockHeightKey, fmt.Sprint(h.Header.Height)),
+		}); err != nil {
+			return fmt.Errorf("block meta-events: %w", err)
+		}
+		// Insert all the block events. Order is important here,
+		if err := insertEvents(dbtx, blockID, 0, h.ResultBeginBlock.Events); err != nil {
+			return fmt.Errorf("begin-block events: %w", err)
+		}
+		if err := insertEvents(dbtx, blockID, 0, h.ResultEndBlock.Events); err != nil {
+			return fmt.Errorf("end-block events: %w", err)
+		}
+		return nil
+	})
+}
+
+func (es *EventSink) IndexTxEvents(txrs []*abci.TxResult) error {
+	ts := time.Now().UTC()
+
+	for _, txr := range txrs {
+		// Encode the result message in protobuf wire format for indexing.
+		resultData, err := proto.Marshal(txr)
+		if err != nil {
+			return fmt.Errorf("marshaling tx_result: %w", err)
+		}
+
+		// Index the hash of the underlying transaction as a hex string.
+		txHash := fmt.Sprintf("%X", types.Tx(txr.Tx).Hash())
+
+		if err := runInTransaction(es.store, func(dbtx *sql.Tx) error {
+			// Find the block associated with this transaction. The block header
+			// must have been indexed prior to the transactions belonging to it.
+			blockID, err := queryWithID(dbtx, `
+SELECT rowid FROM `+tableBlocks+` WHERE height = $1 AND chain_id = $2;
+`, txr.Height, es.chainID)
+			if err != nil {
+				return fmt.Errorf("finding block ID: %w", err)
+			}
+
+			// Insert a record for this tx_result and capture its ID for indexing events.
+			txID, err := queryWithID(dbtx, `
+INSERT INTO `+tableTxResults+` (block_id, index, created_at, tx_hash, tx_result)
+  VALUES ($1, $2, $3, $4, $5)
+  ON CONFLICT DO NOTHING
+  RETURNING rowid;
+`, blockID, txr.Index, ts, txHash, resultData)
+			if err == sql.ErrNoRows {
+				return nil // we already saw this transaction; quietly succeed
+			} else if err != nil {
+				return fmt.Errorf("indexing tx_result: %w", err)
+			}
+
+			// Insert the special transaction meta-events for hash and height.
+			if err := insertEvents(dbtx, blockID, txID, []abci.Event{
+				makeIndexedEvent(types.TxHashKey, txHash),
+				makeIndexedEvent(types.TxHeightKey, fmt.Sprint(txr.Height)),
+			}); err != nil {
+				return fmt.Errorf("indexing transaction meta-events: %w", err)
+			}
+			// Index any events packaged with the transaction.
+			if err := insertEvents(dbtx, blockID, txID, txr.Result.Events); err != nil {
+				return fmt.Errorf("indexing transaction events: %w", err)
+			}
+			return nil
+
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// SearchBlockEvents is not implemented by this sink, and reports an error for all queries.
+func (es *EventSink) SearchBlockEvents(ctx context.Context, q *query.Query) ([]int64, error) {
+	return nil, errors.New("block search is not supported via the postgres event sink")
+}
+
+// SearchTxEvents is not implemented by this sink, and reports an error for all queries.
+func (es *EventSink) SearchTxEvents(ctx context.Context, q *query.Query) ([]*abci.TxResult, error) {
+	return nil, errors.New("tx search is not supported via the postgres event sink")
+}
+
+// GetTxByHash is not implemented by this sink, and reports an error for all queries.
+func (es *EventSink) GetTxByHash(hash []byte) (*abci.TxResult, error) {
+	return nil, errors.New("getTxByHash is not supported via the postgres event sink")
+}
+
+// HasBlock is not implemented by this sink, and reports an error for all queries.
+func (es *EventSink) HasBlock(h int64) (bool, error) {
+	return false, errors.New("hasBlock is not supported via the postgres event sink")
+}
+
+// Stop closes the underlying PostgreSQL database.
+func (es *EventSink) Stop() error { return es.store.Close() }

state/indexer/sink/psql/psql_test.go

@@ -0,0 +1,341 @@
+package psql
+
+import (
+	"context"
+	"database/sql"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"os"
+	"os/signal"
+	"testing"
+	"time"
+
+	"github.com/adlio/schema"
+	"github.com/gogo/protobuf/proto"
+	"github.com/ory/dockertest"
+	"github.com/ory/dockertest/docker"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/types"
+
+	// Register the Postgres database driver.
+	_ "github.com/lib/pq"
+)
+
+var (
+	doPauseAtExit = flag.Bool("pause-at-exit", false,
+		"If true, pause the test until interrupted at shutdown, to allow debugging")
+
+	// A hook that test cases can call to obtain the shared database instance
+	// used for testing the sink. This is initialized in TestMain (see below).
+	testDB func() *sql.DB
+)
+
+const (
+	user     = "postgres"
+	password = "secret"
+	port     = "5432"
+	dsn      = "postgres://%s:%s@localhost:%s/%s?sslmode=disable"
+	dbName   = "postgres"
+	chainID  = "test-chainID"
+
+	viewBlockEvents = "block_events"
+	viewTxEvents    = "tx_events"
+)
+
+func TestMain(m *testing.M) {
+	flag.Parse()
+
+	// Set up docker and start a container running PostgreSQL.
+	pool, err := dockertest.NewPool(os.Getenv("DOCKER_URL"))
+	if err != nil {
+		log.Fatalf("Creating docker pool: %v", err)
+	}
+
+	resource, err := pool.RunWithOptions(&dockertest.RunOptions{
+		Repository: "postgres",
+		Tag:        "13",
+		Env: []string{
+			"POSTGRES_USER=" + user,
+			"POSTGRES_PASSWORD=" + password,
+			"POSTGRES_DB=" + dbName,
+			"listen_addresses = '*'",
+		},
+		ExposedPorts: []string{port},
+	}, func(config *docker.HostConfig) {
+		// set AutoRemove to true so that stopped container goes away by itself
+		config.AutoRemove = true
+		config.RestartPolicy = docker.RestartPolicy{
+			Name: "no",
+		}
+	})
+	if err != nil {
+		log.Fatalf("Starting docker pool: %v", err)
+	}
+
+	if *doPauseAtExit {
+		log.Print("Pause at exit is enabled, containers will not expire")
+	} else {
+		const expireSeconds = 60
+		_ = resource.Expire(expireSeconds)
+		log.Printf("Container expiration set to %d seconds", expireSeconds)
+	}
+
+	// Connect to the database, clear any leftover data, and install the
+	// indexing schema.
+	conn := fmt.Sprintf(dsn, user, password, resource.GetPort(port+"/tcp"), dbName)
+	var db *sql.DB
+
+	if err := pool.Retry(func() error {
+		sink, err := NewEventSink(conn, chainID)
+		if err != nil {
+			return err
+		}
+		db = sink.DB() // set global for test use
+		return db.Ping()
+	}); err != nil {
+		log.Fatalf("Connecting to database: %v", err)
+	}
+
+	if err := resetDatabase(db); err != nil {
+		log.Fatalf("Flushing database: %v", err)
+	}
+
+	sm, err := readSchema()
+	if err != nil {
+		log.Fatalf("Reading schema: %v", err)
+	} else if err := schema.NewMigrator().Apply(db, sm); err != nil {
+		log.Fatalf("Applying schema: %v", err)
+	}
+
+	// Set up the hook for tests to get the shared database handle.
+	testDB = func() *sql.DB { return db }
+
+	// Run the selected test cases.
+	code := m.Run()
+
+	// Clean up and shut down the database container.
+	if *doPauseAtExit {
+		log.Print("Testing complete, pausing for inspection. Send SIGINT to resume teardown")
+		waitForInterrupt()
+		log.Print("(resuming)")
+	}
+	log.Print("Shutting down database")
+	if err := pool.Purge(resource); err != nil {
+		log.Printf("WARNING: Purging pool failed: %v", err)
+	}
+	if err := db.Close(); err != nil {
+		log.Printf("WARNING: Closing database failed: %v", err)
+	}
+
+	os.Exit(code)
+}
+
+func TestIndexing(t *testing.T) {
+	t.Run("IndexBlockEvents", func(t *testing.T) {
+		indexer := &EventSink{store: testDB(), chainID: chainID}
+		require.NoError(t, indexer.IndexBlockEvents(newTestBlockHeader()))
+
+		verifyBlock(t, 1)
+		verifyBlock(t, 2)
+
+		verifyNotImplemented(t, "hasBlock", func() (bool, error) { return indexer.HasBlock(1) })
+		verifyNotImplemented(t, "hasBlock", func() (bool, error) { return indexer.HasBlock(2) })
+
+		verifyNotImplemented(t, "block search", func() (bool, error) {
+			v, err := indexer.SearchBlockEvents(context.Background(), nil)
+			return v != nil, err
+		})
+
+		require.NoError(t, verifyTimeStamp(tableBlocks))
+
+		// Attempting to reindex the same events should gracefully succeed.
+		require.NoError(t, indexer.IndexBlockEvents(newTestBlockHeader()))
+	})
+
+	t.Run("IndexTxEvents", func(t *testing.T) {
+		indexer := &EventSink{store: testDB(), chainID: chainID}
+
+		txResult := txResultWithEvents([]abci.Event{
+			makeIndexedEvent("account.number", "1"),
+			makeIndexedEvent("account.owner", "Ivan"),
+			makeIndexedEvent("account.owner", "Yulieta"),
+
+			{Type: "", Attributes: []abci.EventAttribute{
+				{
+					Key:   []byte("not_allowed"),
+					Value: []byte("Vlad"),
+					Index: true,
+				},
+			}},
+		})
+		require.NoError(t, indexer.IndexTxEvents([]*abci.TxResult{txResult}))
+
+		txr, err := loadTxResult(types.Tx(txResult.Tx).Hash())
+		require.NoError(t, err)
+		assert.Equal(t, txResult, txr)
+
+		require.NoError(t, verifyTimeStamp(tableTxResults))
+		require.NoError(t, verifyTimeStamp(viewTxEvents))
+
+		verifyNotImplemented(t, "getTxByHash", func() (bool, error) {
+			txr, err := indexer.GetTxByHash(types.Tx(txResult.Tx).Hash())
+			return txr != nil, err
+		})
+		verifyNotImplemented(t, "tx search", func() (bool, error) {
+			txr, err := indexer.SearchTxEvents(context.Background(), nil)
+			return txr != nil, err
+		})
+
+		// try to insert the duplicate tx events.
+		err = indexer.IndexTxEvents([]*abci.TxResult{txResult})
+		require.NoError(t, err)
+	})
+}
+
+func TestStop(t *testing.T) {
+	indexer := &EventSink{store: testDB()}
+	require.NoError(t, indexer.Stop())
+}
+
+// newTestBlockHeader constructs a fresh copy of a block header containing
+// known test values to exercise the indexer.
+func newTestBlockHeader() types.EventDataNewBlockHeader {
+	return types.EventDataNewBlockHeader{
+		Header: types.Header{Height: 1},
+		ResultBeginBlock: abci.ResponseBeginBlock{
+			Events: []abci.Event{
+				makeIndexedEvent("begin_event.proposer", "FCAA001"),
+				makeIndexedEvent("thingy.whatzit", "O.O"),
+			},
+		},
+		ResultEndBlock: abci.ResponseEndBlock{
+			Events: []abci.Event{
+				makeIndexedEvent("end_event.foo", "100"),
+				makeIndexedEvent("thingy.whatzit", "-.O"),
+			},
+		},
+	}
+}
+
+// readSchema loads the indexing database schema file
+func readSchema() ([]*schema.Migration, error) {
+	const filename = "schema.sql"
+	contents, err := ioutil.ReadFile(filename)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read sql file from '%s': %w", filename, err)
+	}
+
+	return []*schema.Migration{{
+		ID:     time.Now().Local().String() + " db schema",
+		Script: string(contents),
+	}}, nil
+}
+
+// resetDB drops all the data from the test database.
+func resetDatabase(db *sql.DB) error {
+	_, err := db.Exec(`DROP TABLE IF EXISTS blocks,tx_results,events,attributes CASCADE;`)
+	if err != nil {
+		return fmt.Errorf("dropping tables: %v", err)
+	}
+	_, err = db.Exec(`DROP VIEW IF EXISTS event_attributes,block_events,tx_events CASCADE;`)
+	if err != nil {
+		return fmt.Errorf("dropping views: %v", err)
+	}
+	return nil
+}
+
+// txResultWithEvents constructs a fresh transaction result with fixed values
+// for testing, that includes the specified events.
+func txResultWithEvents(events []abci.Event) *abci.TxResult {
+	return &abci.TxResult{
+		Height: 1,
+		Index:  0,
+		Tx:     types.Tx("HELLO WORLD"),
+		Result: abci.ResponseDeliverTx{
+			Data:   []byte{0},
+			Code:   abci.CodeTypeOK,
+			Log:    "",
+			Events: events,
+		},
+	}
+}
+
+func loadTxResult(hash []byte) (*abci.TxResult, error) {
+	hashString := fmt.Sprintf("%X", hash)
+	var resultData []byte
+	if err := testDB().QueryRow(`
+SELECT tx_result FROM `+tableTxResults+` WHERE tx_hash = $1;
+`, hashString).Scan(&resultData); err != nil {
+		return nil, fmt.Errorf("lookup transaction for hash %q failed: %v", hashString, err)
+	}
+
+	txr := new(abci.TxResult)
+	if err := proto.Unmarshal(resultData, txr); err != nil {
+		return nil, fmt.Errorf("unmarshaling txr: %v", err)
+	}
+
+	return txr, nil
+}
+
+func verifyTimeStamp(tableName string) error {
+	return testDB().QueryRow(fmt.Sprintf(`
+SELECT DISTINCT %[1]s.created_at
+  FROM %[1]s
+  WHERE %[1]s.created_at >= $1;
+`, tableName), time.Now().Add(-2*time.Second)).Err()
+}
+
+func verifyBlock(t *testing.T, height int64) {
+	// Check that the blocks table contains an entry for this height.
+	if err := testDB().QueryRow(`
+SELECT height FROM `+tableBlocks+` WHERE height = $1;
+`, height).Err(); err == sql.ErrNoRows {
+		t.Errorf("No block found for height=%d", height)
+	} else if err != nil {
+		t.Fatalf("Database query failed: %v", err)
+	}
+
+	// Verify the presence of begin_block and end_block events.
+	if err := testDB().QueryRow(`
+SELECT type, height, chain_id FROM `+viewBlockEvents+`
+  WHERE height = $1 AND type = $2 AND chain_id = $3;
+`, height, eventTypeBeginBlock, chainID).Err(); err == sql.ErrNoRows {
+		t.Errorf("No %q event found for height=%d", eventTypeBeginBlock, height)
+	} else if err != nil {
+		t.Fatalf("Database query failed: %v", err)
+	}
+
+	if err := testDB().QueryRow(`
+SELECT type, height, chain_id FROM `+viewBlockEvents+`
+  WHERE height = $1 AND type = $2 AND chain_id = $3;
+`, height, eventTypeEndBlock, chainID).Err(); err == sql.ErrNoRows {
+		t.Errorf("No %q event found for height=%d", eventTypeEndBlock, height)
+	} else if err != nil {
+		t.Fatalf("Database query failed: %v", err)
+	}
+}
+
+// verifyNotImplemented calls f and verifies that it returns both a
+// false-valued flag and a non-nil error whose string matching the expected
+// "not supported" message with label prefixed.
+func verifyNotImplemented(t *testing.T, label string, f func() (bool, error)) {
+	t.Helper()
+	t.Logf("Verifying that %q reports it is not implemented", label)
+
+	want := label + " is not supported via the postgres event sink"
+	ok, err := f()
+	assert.False(t, ok)
+	require.NotNil(t, err)
+	assert.Equal(t, want, err.Error())
+}
+
+// waitForInterrupt blocks until a SIGINT is received by the process.
+func waitForInterrupt() {
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, os.Interrupt)
+	<-ch
+}

state/indexer/sink/psql/schema.sql

@@ -0,0 +1,85 @@
+/*
+  This file defines the database schema for the PostgresQL ("psql") event sink
+  implementation in Tendermint. The operator must create a database and install
+  this schema before using the database to index events.
+ */
+
+-- The blocks table records metadata about each block.
+-- The block record does not include its events or transactions (see tx_results).
+CREATE TABLE blocks (
+  rowid      BIGSERIAL PRIMARY KEY,
+
+  height     BIGINT NOT NULL,
+  chain_id   VARCHAR NOT NULL,
+
+  -- When this block header was logged into the sink, in UTC.
+  created_at TIMESTAMPTZ NOT NULL,
+
+  UNIQUE (height, chain_id)
+);
+
+-- Index blocks by height and chain, since we need to resolve block IDs when
+-- indexing transaction records and transaction events.
+CREATE INDEX idx_blocks_height_chain ON blocks(height, chain_id);
+
+-- The tx_results table records metadata about transaction results.  Note that
+-- the events from a transaction are stored separately.
+CREATE TABLE tx_results (
+  rowid BIGSERIAL PRIMARY KEY,
+
+  -- The block to which this transaction belongs.
+  block_id BIGINT NOT NULL REFERENCES blocks(rowid),
+  -- The sequential index of the transaction within the block.
+  index INTEGER NOT NULL,
+  -- When this result record was logged into the sink, in UTC.
+  created_at TIMESTAMPTZ NOT NULL,
+  -- The hex-encoded hash of the transaction.
+  tx_hash VARCHAR NOT NULL,
+  -- The protobuf wire encoding of the TxResult message.
+  tx_result BYTEA NOT NULL,
+
+  UNIQUE (block_id, index)
+);
+
+-- The events table records events. All events (both block and transaction) are
+-- associated with a block ID; transaction events also have a transaction ID.
+CREATE TABLE events (
+  rowid BIGSERIAL PRIMARY KEY,
+
+  -- The block and transaction this event belongs to.
+  -- If tx_id is NULL, this is a block event.
+  block_id BIGINT NOT NULL REFERENCES blocks(rowid),
+  tx_id    BIGINT NULL REFERENCES tx_results(rowid),
+
+  -- The application-defined type label for the event.
+  type VARCHAR NOT NULL
+);
+
+-- The attributes table records event attributes.
+CREATE TABLE attributes (
+   event_id      BIGINT NOT NULL REFERENCES events(rowid),
+   key           VARCHAR NOT NULL, -- bare key
+   composite_key VARCHAR NOT NULL, -- composed type.key
+   value         VARCHAR NULL,
+
+   UNIQUE (event_id, key)
+);
+
+-- A joined view of events and their attributes. Events that do not have any
+-- attributes are represented as a single row with empty key and value fields.
+CREATE VIEW event_attributes AS
+  SELECT block_id, tx_id, type, key, composite_key, value
+  FROM events LEFT JOIN attributes ON (events.rowid = attributes.event_id);
+
+-- A joined view of all block events (those having tx_id NULL).
+CREATE VIEW block_events AS
+  SELECT blocks.rowid as block_id, height, chain_id, type, key, composite_key, value
+  FROM blocks JOIN event_attributes ON (blocks.rowid = event_attributes.block_id)
+  WHERE event_attributes.tx_id IS NULL;
+
+-- A joined view of all transaction events.
+CREATE VIEW tx_events AS
+  SELECT height, index, chain_id, type, key, composite_key, value, tx_results.created_at
+  FROM blocks JOIN tx_results ON (blocks.rowid = tx_results.block_id)
+  JOIN event_attributes ON (tx_results.rowid = event_attributes.tx_id)
+  WHERE event_attributes.tx_id IS NOT NULL;