WIP: ADR 073: Built-in Indexer Implementation

.gitignore

@@ -63,3 +63,4 @@ debug_container.log
 *.synctex.gz
 /x/genutil/config/priv_validator_key.json
 /x/genutil/data/priv_validator_state.json
+.env.local

baseapp/streaming.go

@@ -7,6 +7,8 @@ import (
 
 	"github.com/spf13/cast"
 
+	"cosmossdk.io/schema/decoding"
+	"cosmossdk.io/schema/indexing"
 	"cosmossdk.io/store/streaming"
 	storetypes "cosmossdk.io/store/types"
 
@@ -22,6 +24,33 @@ const (
 	StreamingABCIStopNodeOnErrTomlKey = "stop-node-on-err"
 )
 
+func (app *BaseApp) EnableIndexer(indexerOpts interface{}, keys map[string]*storetypes.KVStoreKey, appModules map[string]any) error {
+	optsMap, ok := indexerOpts.(map[string]interface{})
+	if !ok {
+		return fmt.Errorf("invalid indexer options type %T, expected a map", indexerOpts)
+	}
+
+	listener, err := indexing.Start(indexing.Options{
+		Options:    optsMap,
+		Resolver:   decoding.ModuleSetDecoderResolver(appModules),
+		SyncSource: nil,
+		Logger:     app.logger.With("module", "indexer"),
+	})
+	if err != nil {
+		return err
+	}
+
+	exposedKeys := exposeStoreKeysSorted([]string{"*"}, keys)
+	app.cms.AddListeners(exposedKeys)
+
+	app.streamingManager = storetypes.StreamingManager{
+		ABCIListeners: []storetypes.ABCIListener{storetypes.FromSchemaListener(listener)},
+		StopNodeOnErr: true,
+	}
+
+	return nil
+}
+
 // RegisterStreamingServices registers streaming services with the BaseApp.
 func (app *BaseApp) RegisterStreamingServices(appOpts servertypes.AppOptions, keys map[string]*storetypes.KVStoreKey) error {
 	// register streaming services

codec/collections.go

@@ -8,6 +8,9 @@
 	gogotypes "github.com/cosmos/gogoproto/types"
 	"google.golang.org/protobuf/encoding/protojson"
 	protov2 "google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protoreflect"
+
+	"cosmossdk.io/schema"
 
 	"cosmossdk.io/collections"
 	collcodec "cosmossdk.io/collections/codec"
@@ -91,6 +94,16 @@
 	return "github.com/cosmos/gogoproto/" + c.messageName
 }
 
+func (c collValue[T, PT]) SchemaColumns() []schema.Field {
+	var pt PT
+	msgName := proto.MessageName(pt)
+	desc, err := proto.HybridResolver.FindDescriptorByName(protoreflect.FullName(msgName))
+	if err != nil {
+		panic(fmt.Errorf("could not find descriptor for %s: %w", msgName, err))
+	}
+	return protoCols(desc.(protoreflect.MessageDescriptor))
+}
+
 type protoMessageV2[T any] interface {
 	*T
 	protov2.Message
@@ -136,6 +149,11 @@
 	return "google.golang.org/protobuf/" + c.messageName
 }
 
+func (c collValue2[T, PT]) SchemaColumns() []schema.Field {
+	var pt PT
+	return protoCols(pt.ProtoReflect().Descriptor())
+}
+
 // CollInterfaceValue instantiates a new collections.ValueCodec for a generic
 // interface value. The codec must be able to marshal and unmarshal the
 // interface.
@@ -179,3 +197,70 @@
 	var t T
 	return fmt.Sprintf("%T", t)
 }
+
+func protoCols(desc protoreflect.MessageDescriptor) []schema.Field {
+	nFields := desc.Fields()
+	cols := make([]schema.Field, 0, nFields.Len())
+	for i := 0; i < nFields.Len(); i++ {
+		f := nFields.Get(i)
+		cols = append(cols, protoCol(f))
+	}
+	return cols
+}
+
+func protoCol(f protoreflect.FieldDescriptor) schema.Field {
+	col := schema.Field{Name: string(f.Name())}
+
+	if f.IsMap() || f.IsList() {
+		col.Kind = schema.JSONKind
+		col.Nullable = true
+	} else {
+		switch f.Kind() {
+		case protoreflect.BoolKind:
+			col.Kind = schema.BoolKind
+		case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind:
+			col.Kind = schema.Int32Kind
+		case protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind:
+			col.Kind = schema.Int64Kind
+		case protoreflect.Uint32Kind, protoreflect.Fixed32Kind:
+			col.Kind = schema.Int64Kind
+		case protoreflect.Uint64Kind, protoreflect.Fixed64Kind:
+			col.Kind = schema.IntegerStringKind
+		case protoreflect.FloatKind:
+			col.Kind = schema.Float32Kind
+		case protoreflect.DoubleKind:
+			col.Kind = schema.Float64Kind
+		case protoreflect.StringKind:
+			col.Kind = schema.StringKind
+		case protoreflect.BytesKind:
+			col.Kind = schema.BytesKind
+		case protoreflect.EnumKind:
+			col.Kind = schema.EnumKind
+			enumDesc := f.Enum()
+			var vals []string
+			n := enumDesc.Values().Len()
+			for i := 0; i < n; i++ {
+				vals = append(vals, string(enumDesc.Values().Get(i).Name()))
+			}
+			col.EnumDefinition = schema.EnumDefinition{
+				Name:   string(enumDesc.Name()),
+				Values: vals,
+			}
+		case protoreflect.MessageKind:
+			col.Nullable = true
+			fullName := f.Message().FullName()
+			if fullName == "google.protobuf.Timestamp" {
+				col.Kind = schema.TimeKind
+			} else if fullName == "google.protobuf.Duration" {
+				col.Kind = schema.DurationKind
+			} else {
+				col.Kind = schema.JSONKind
+			}
+		}
+		if f.HasPresence() {
+			col.Nullable = true
+		}
+	}
+
+	return col
+}

collections/codec/alternative_value_test.go

@@ -17,7 +17,7 @@ type altValue struct {
 func TestAltValueCodec(t *testing.T) {
 	// we assume we want to migrate the value from json(altValue) to just be
 	// the raw value uint64.
-	canonical := codec.KeyToValueCodec(codec.NewUint64Key[uint64]())
+	canonical := codec.KeyToValueCodec(codec.KeyCodec[uint64](codec.NewUint64Key[uint64]()))
 	alternative := func(v []byte) (uint64, error) {
 		var alt altValue
 		err := json.Unmarshal(v, &alt)

collections/codec/bool.go

@@ -6,9 +6,11 @@ import (
 	"strconv"
 )
 
-func NewBoolKey[T ~bool]() KeyCodec[T] { return boolKey[T]{} }
+func NewBoolKey[T ~bool]() NameableKeyCodec[T] { return boolKey[T]{} }
 
-type boolKey[T ~bool] struct{}
+type boolKey[T ~bool] struct {
+	name string
+}
 
 func (b boolKey[T]) Encode(buffer []byte, key T) (int, error) {
 	if key {
@@ -64,3 +66,12 @@ func (b boolKey[T]) DecodeNonTerminal(buffer []byte) (int, T, error) {
 func (b boolKey[T]) SizeNonTerminal(key T) int {
 	return b.Size(key)
 }
+
+func (b boolKey[T]) WithName(name string) NamedKeyCodec[T] {
+	b.name = name
+	return b
+}
+
+func (b boolKey[T]) Name() string {
+	return b.name
+}

collections/codec/bytes.go

@@ -10,9 +10,11 @@ import (
 // using the BytesKey KeyCodec.
 const MaxBytesKeyNonTerminalSize = math.MaxUint8
 
-func NewBytesKey[T ~[]byte]() KeyCodec[T] { return bytesKey[T]{} }
+func NewBytesKey[T ~[]byte]() NameableKeyCodec[T] { return bytesKey[T]{} }
 
-type bytesKey[T ~[]byte] struct{}
+type bytesKey[T ~[]byte] struct {
+	name string
+}
 
 func (b bytesKey[T]) Encode(buffer []byte, key T) (int, error) {
 	return copy(buffer, key), nil
@@ -77,3 +79,12 @@ func (bytesKey[T]) DecodeNonTerminal(buffer []byte) (int, T, error) {
 func (bytesKey[T]) SizeNonTerminal(key T) int {
 	return len(key) + 1
 }
+
+func (b bytesKey[T]) WithName(name string) NamedKeyCodec[T] {
+	b.name = name
+	return b
+}
+
+func (b bytesKey[T]) Name() string {
+	return b.name
+}

collections/codec/codec.go

@@ -3,6 +3,8 @@
 import (
 	"errors"
 	"fmt"
+
+	"cosmossdk.io/schema"
 )
 
 var ErrEncoding = errors.New("collections: encoding error")
@@ -74,6 +76,22 @@
 	ValueType() string
 }
 
+// IndexableCodec is an interface that all codec's should implement in order to properly support indexing.
+// It is not required by KeyCodec or ValueCodec in order to preserve backwards compatibility, but
+// a future version of collections may make it required and all codec's should aim to implement it.
+// If it is not implemented, fallback defaults will be used for indexing that may be sub-optimal.
+//
+// Implementations of IndexableCodec should test that they are conformant using the schema.ValidateWithKeyFields
+// and schema.ValidateWithValueFields depending on whether the codec is a KeyCodec or ValueCodec respectively.
+type IndexableCodec interface {
+	LogicalDecoder() LogicalDecoder
+}
+
+type LogicalDecoder struct {
+	Fields []schema.Field
+	Decode func([]byte) (any, error)
+}
+
 // NewUntypedValueCodec returns an UntypedValueCodec for the provided ValueCodec.
 func NewUntypedValueCodec[V any](v ValueCodec[V]) UntypedValueCodec {
 	typeName := fmt.Sprintf("%T", *new(V))

collections/codec/int.go

@@ -7,9 +7,11 @@ import (
 	"strconv"
 )
 
-func NewInt64Key[T ~int64]() KeyCodec[T] { return int64Key[T]{} }
+func NewInt64Key[T ~int64]() NameableKeyCodec[T] { return int64Key[T]{} }
 
-type int64Key[T ~int64] struct{}
+type int64Key[T ~int64] struct {
+	name string
+}
 
 func (i int64Key[T]) Encode(buffer []byte, key T) (int, error) {
 	binary.BigEndian.PutUint64(buffer, (uint64)(key))
@@ -64,11 +66,22 @@ func (i int64Key[T]) SizeNonTerminal(_ T) int {
 	return 8
 }
 
-func NewInt32Key[T ~int32]() KeyCodec[T] {
+func (i int64Key[T]) WithName(name string) NamedKeyCodec[T] {
+	i.name = name
+	return i
+}
+
+func (i int64Key[T]) Name() string {
+	return i.name
+}
+
+func NewInt32Key[T ~int32]() NameableKeyCodec[T] {
 	return int32Key[T]{}
 }
 
-type int32Key[T ~int32] struct{}
+type int32Key[T ~int32] struct {
+	name string
+}
 
 func (i int32Key[T]) Encode(buffer []byte, key T) (int, error) {
 	binary.BigEndian.PutUint32(buffer, (uint32)(key))
@@ -121,3 +134,12 @@ func (i int32Key[T]) DecodeNonTerminal(buffer []byte) (int, T, error) {
 func (i int32Key[T]) SizeNonTerminal(_ T) int {
 	return 4
 }
+
+func (i int32Key[T]) WithName(name string) NamedKeyCodec[T] {
+	i.name = name
+	return i
+}
+
+func (i int32Key[T]) Name() string {
+	return i.name
+}

collections/codec/naming.go

@@ -0,0 +1,35 @@
+package codec
+
+type HasName interface {
+	// Name returns the name of key in the schema if one is defined or the empty string.
+	// Multipart keys should separate names with commas, i.e. "name1,name2".
+	Name() string
+}
+
+// NameableKeyCodec is a KeyCodec that can be named.
+type NameableKeyCodec[T any] interface {
+	KeyCodec[T]
+
+	// WithName returns the KeyCodec with the provided name.
+	WithName(name string) NamedKeyCodec[T]
+}
+
+// NamedKeyCodec is a KeyCodec that has a name.
+type NamedKeyCodec[T any] interface {
+	KeyCodec[T]
+	HasName
+}
+
+// NameableValueCodec is a ValueCodec that can be named.
+type NameableValueCodec[T any] interface {
+	ValueCodec[T]
+
+	// WithName returns the ValueCodec with the provided name.
+	WithName(name string) NamedValueCodec[T]
+}
+
+// NamedValueCodec is a ValueCodec that has a name.
+type NamedValueCodec[T any] interface {
+	ValueCodec[T]
+	HasName
+}

collections/codec/string.go

@@ -6,14 +6,16 @@ import (
 	"fmt"
 )
 
-func NewStringKeyCodec[T ~string]() KeyCodec[T] { return stringKey[T]{} }
+func NewStringKeyCodec[T ~string]() NameableKeyCodec[T] { return stringKey[T]{} }
 
 const (
 	// StringDelimiter defines the delimiter of a string key when used in non-terminal encodings.
 	StringDelimiter uint8 = 0x0
 )
 
-type stringKey[T ~string] struct{}
+type stringKey[T ~string] struct {
+	name string
+}
 
 func (stringKey[T]) Encode(buffer []byte, key T) (int, error) {
 	return copy(buffer, key), nil
@@ -66,3 +68,12 @@ func (stringKey[T]) Stringify(key T) string {
 func (stringKey[T]) KeyType() string {
 	return "string"
 }
+
+func (s stringKey[T]) WithName(name string) NamedKeyCodec[T] {
+	s.name = name
+	return s
+}
+
+func (s stringKey[T]) Name() string {
+	return s.name
+}