[STO-310] [WIP] Parquet receiptdb option

app/app.go

@@ -107,7 +107,6 @@ import (
 	gigalib "github.com/sei-protocol/sei-chain/giga/executor/lib"
 	"github.com/sei-protocol/sei-chain/precompiles"
 	putils "github.com/sei-protocol/sei-chain/precompiles/utils"
-	"github.com/sei-protocol/sei-chain/sei-db/state_db/ss"
 	seidb "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/types"
 	"github.com/sei-protocol/sei-chain/sei-ibc-go/modules/apps/transfer"
 	ibctransferkeeper "github.com/sei-protocol/sei-chain/sei-ibc-go/modules/apps/transfer/keeper"
@@ -163,6 +162,7 @@ import (
 	// unnamed import of statik for openapi/swagger UI support
 	_ "github.com/sei-protocol/sei-chain/docs/swagger"
 	ssconfig "github.com/sei-protocol/sei-chain/sei-db/config"
+	receipt "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt"
 
 	gigaevmkeeper "github.com/sei-protocol/sei-chain/giga/deps/xevm/keeper"
 	gigaevmstate "github.com/sei-protocol/sei-chain/giga/deps/xevm/state"
@@ -392,7 +392,7 @@ type App struct {
 	genesisImportConfig genesistypes.GenesisImportConfig
 
 	stateStore   seidb.StateStore
-	receiptStore seidb.StateStore
+	receiptStore receipt.ReceiptStore
 
 	forkInitializer func(sdk.Context)
 
@@ -634,16 +634,28 @@ func New(
 		wasmOpts...,
 	)
 
-	receiptStorePath := filepath.Join(homePath, "data", "receipt.db")
-	ssConfig := ssconfig.DefaultStateStoreConfig()
-	ssConfig.KeepRecent = cast.ToInt(appOpts.Get(server.FlagMinRetainBlocks))
-	ssConfig.DBDirectory = receiptStorePath
-	ssConfig.KeepLastVersion = false
+	receiptConfig := ssconfig.DefaultReceiptStoreConfig()
+	receiptBackend := strings.TrimSpace(cast.ToString(appOpts.Get(FlagRSBackend)))
+	if receiptBackend != "" {
+		receiptConfig.Backend = receiptBackend
+	}
+	receiptConfig.DBDirectory = strings.TrimSpace(cast.ToString(appOpts.Get(FlagRSDirectory)))
+	receiptConfig.KeepRecent = cast.ToInt(appOpts.Get(server.FlagMinRetainBlocks))
+	if receiptConfig.DBDirectory == "" {
+		switch strings.ToLower(receiptConfig.Backend) {
+		case "parquet":
+			receiptConfig.DBDirectory = filepath.Join(homePath, "data", "receipt.parquet")
+		default:
+			receiptConfig.DBDirectory = filepath.Join(homePath, "data", "receipt.db")
+		}
+	}
+
 	if app.receiptStore == nil {
-		app.receiptStore, err = ss.NewStateStore(logger, receiptStorePath, ssConfig)
+		receiptStore, err := receipt.NewReceiptStore(logger, receiptConfig, keys[evmtypes.StoreKey])
 		if err != nil {
 			panic(fmt.Sprintf("error while creating receipt store: %s", err))
 		}
+		app.receiptStore = receiptStore
 	}
 	app.EvmKeeper = *evmkeeper.NewKeeper(keys[evmtypes.StoreKey],
 		tkeys[evmtypes.TransientStoreKey], app.GetSubspace(evmtypes.ModuleName), app.receiptStore, app.BankKeeper,

app/seidb.go

@@ -36,6 +36,10 @@ const (
 	FlagSSPruneInterval     = "state-store.ss-prune-interval"
 	FlagSSImportNumWorkers  = "state-store.ss-import-num-workers"
 
+	// Receipt store configs
+	FlagRSBackend   = "receipt-store.rs-backend"
+	FlagRSDirectory = "receipt-store.rs-db-directory"
+
 	// Other configs
 	FlagSnapshotInterval = "state-sync.snapshot-interval"
 	FlagMigrateIAVL      = "migrate-iavl"

app/test_helpers.go

@@ -14,10 +14,10 @@ import (
 	slashingtypes "github.com/cosmos/cosmos-sdk/x/slashing/types"
 	"github.com/cosmos/cosmos-sdk/x/staking"
 	ssconfig "github.com/sei-protocol/sei-chain/sei-db/config"
-	"github.com/sei-protocol/sei-chain/sei-db/state_db/ss"
-	seidbtypes "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/types"
+	receipt "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt"
 	"github.com/sei-protocol/sei-chain/sei-wasmd/x/wasm"
 	wasmkeeper "github.com/sei-protocol/sei-chain/sei-wasmd/x/wasm/keeper"
+	evmtypes "github.com/sei-protocol/sei-chain/x/evm/types"
 	"github.com/stretchr/testify/suite"
 	"github.com/tendermint/tendermint/config"
 
@@ -265,7 +265,7 @@ func (s *TestWrapper) EndBlock() {
 	legacyabci.EndBlock(s.Ctx, s.Ctx.BlockHeight(), 0, s.App.EndBlockKeepers)
 }
 
-func setupReceiptStore() (seidbtypes.StateStore, error) {
+func setupReceiptStore(storeKey sdk.StoreKey) (receipt.ReceiptStore, error) {
 	// Create a unique temporary directory per test process to avoid Pebble DB lock conflicts
 	baseDir := filepath.Join(os.TempDir(), "sei-testing")
 	if err := os.MkdirAll(baseDir, 0o750); err != nil {
@@ -276,11 +276,10 @@ func setupReceiptStore() (seidbtypes.StateStore, error) {
 		return nil, err
 	}
 
-	ssConfig := ssconfig.DefaultStateStoreConfig()
-	ssConfig.KeepRecent = 0 // No min retain blocks in test
-	ssConfig.DBDirectory = tempDir
-	ssConfig.KeepLastVersion = false
-	receiptStore, err := ss.NewStateStore(log.NewNopLogger(), tempDir, ssConfig)
+	receiptConfig := ssconfig.DefaultReceiptStoreConfig()
+	receiptConfig.KeepRecent = 0 // No min retain blocks in test
+	receiptConfig.DBDirectory = tempDir
+	receiptStore, err := receipt.NewReceiptStore(log.NewNopLogger(), receiptConfig, storeKey)
 	if err != nil {
 		return nil, err
 	}
@@ -297,7 +296,7 @@ func SetupWithAppOptsAndDefaultHome(isCheckTx bool, appOpts TestAppOpts, enableE
 
 	options := []AppOption{
 		func(app *App) {
-			receiptStore, err := setupReceiptStore()
+			receiptStore, err := setupReceiptStore(app.keys[evmtypes.StoreKey])
 			if err != nil {
 				panic(fmt.Sprintf("error while creating receipt store: %s", err))
 			}
@@ -370,7 +369,7 @@ func SetupWithDB(tb testing.TB, db dbm.DB, isCheckTx bool, enableEVMCustomPrecom
 
 	options := []AppOption{
 		func(app *App) {
-			receiptStore, err := setupReceiptStore()
+			receiptStore, err := setupReceiptStore(app.keys[evmtypes.StoreKey])
 			if err != nil {
 				panic(fmt.Sprintf("error while creating receipt store: %s", err))
 			}
@@ -436,7 +435,7 @@ func SetupWithSc(t *testing.T, isCheckTx bool, enableEVMCustomPrecompiles bool,
 
 	options := []AppOption{
 		func(app *App) {
-			receiptStore, err := setupReceiptStore()
+			receiptStore, err := setupReceiptStore(app.keys[evmtypes.StoreKey])
 			if err != nil {
 				panic(fmt.Sprintf("error while creating receipt store: %s", err))
 			}

cmd/seid/cmd/app_config.go

@@ -22,23 +22,25 @@ type WASMConfig struct {
 type CustomAppConfig struct {
 	srvconfig.Config
 
-	StateCommit     seidbconfig.StateCommitConfig `mapstructure:"state-commit"`
-	StateStore      seidbconfig.StateStoreConfig  `mapstructure:"state-store"`
-	WASM            WASMConfig                    `mapstructure:"wasm"`
-	EVM             evmrpcconfig.Config           `mapstructure:"evm"`
-	GigaExecutor    gigaconfig.Config             `mapstructure:"giga_executor"`
-	ETHReplay       replay.Config                 `mapstructure:"eth_replay"`
-	ETHBlockTest    blocktest.Config              `mapstructure:"eth_block_test"`
-	EvmQuery        querier.Config                `mapstructure:"evm_query"`
-	LightInvariance seiapp.LightInvarianceConfig  `mapstructure:"light_invariance"`
+	StateCommit     seidbconfig.StateCommitConfig  `mapstructure:"state-commit"`
+	StateStore      seidbconfig.StateStoreConfig   `mapstructure:"state-store"`
+	ReceiptStore    seidbconfig.ReceiptStoreConfig `mapstructure:"receipt-store"`
+	WASM            WASMConfig                     `mapstructure:"wasm"`
+	EVM             evmrpcconfig.Config            `mapstructure:"evm"`
+	GigaExecutor    gigaconfig.Config              `mapstructure:"giga_executor"`
+	ETHReplay       replay.Config                  `mapstructure:"eth_replay"`
+	ETHBlockTest    blocktest.Config               `mapstructure:"eth_block_test"`
+	EvmQuery        querier.Config                 `mapstructure:"evm_query"`
+	LightInvariance seiapp.LightInvarianceConfig   `mapstructure:"light_invariance"`
 }
 
 // NewCustomAppConfig creates a CustomAppConfig with the given base config and EVM config
 func NewCustomAppConfig(baseConfig *srvconfig.Config, evmConfig evmrpcconfig.Config) CustomAppConfig {
 	return CustomAppConfig{
-		Config:      *baseConfig,
-		StateCommit: seidbconfig.DefaultStateCommitConfig(),
-		StateStore:  seidbconfig.DefaultStateStoreConfig(),
+		Config:       *baseConfig,
+		StateCommit:  seidbconfig.DefaultStateCommitConfig(),
+		StateStore:   seidbconfig.DefaultStateStoreConfig(),
+		ReceiptStore: seidbconfig.DefaultReceiptStoreConfig(),
 		WASM: WASMConfig{
 			QueryGasLimit: 300000,
 			LruSize:       1,

cmd/seid/cmd/init_test.go

@@ -144,6 +144,7 @@ func TestInitModeConfiguration(t *testing.T) {
 
 			// Build custom template with all sections
 			customAppTemplate := srvconfig.ManualConfigTemplate + seidbconfig.StateCommitConfigTemplate + seidbconfig.StateStoreConfigTemplate +
+				seidbconfig.ReceiptStoreConfigTemplate +
 				srvconfig.AutoManagedConfigTemplate // Simplified - just need the pruning config
 
 			srvconfig.SetConfigTemplate(customAppTemplate)

cmd/seid/cmd/root.go

@@ -439,6 +439,7 @@ func initAppConfig() (string, interface{}) {
 	customAppTemplate := serverconfig.ManualConfigTemplate +
 		seidbconfig.StateCommitConfigTemplate +
 		seidbconfig.StateStoreConfigTemplate +
+		seidbconfig.ReceiptStoreConfigTemplate +
 		evmrpcconfig.ConfigTemplate +
 		gigaconfig.ConfigTemplate +
 		serverconfig.AutoManagedConfigTemplate + `

evmrpc/filter.go

@@ -783,7 +783,7 @@ func (f *LogFetcher) GetLogsByFilters(ctx context.Context, crit filters.FilterCr
 		localLogs := f.globalLogSlicePool.Get()
 
 		for _, block := range batch {
-			f.GetLogsForBlockPooled(block, crit, bloomIndexes, &localLogs)
+			f.GetLogsForBlockPooled(block, crit, &localLogs)
 		}
 
 		// Sort the local batch
@@ -907,68 +907,32 @@ func (f *LogFetcher) earliestHeight(ctx context.Context) (int64, error) {
 }
 
 // Pooled version that reuses slice allocation
-func (f *LogFetcher) GetLogsForBlockPooled(block *coretypes.ResultBlock, crit filters.FilterCriteria, filters [][]bloomIndexes, result *[]*ethtypes.Log) {
+func (f *LogFetcher) GetLogsForBlockPooled(block *coretypes.ResultBlock, crit filters.FilterCriteria, result *[]*ethtypes.Log) {
 	collector := &pooledCollector{logs: result}
-	f.collectLogs(block, crit, filters, collector, true) // Apply exact matching
-}
-
-func (f *LogFetcher) IsLogExactMatch(log *ethtypes.Log, crit filters.FilterCriteria) bool {
-	addrMatch := len(crit.Addresses) == 0
-	for _, addrFilter := range crit.Addresses {
-		if log.Address == addrFilter {
-			addrMatch = true
-			break
-		}
-	}
-	return addrMatch && matchTopics(crit.Topics, log.Topics)
+	f.collectLogs(block, crit, collector, true) // Apply exact matching
 }
 
 // Unified log collection logic
-func (f *LogFetcher) collectLogs(block *coretypes.ResultBlock, crit filters.FilterCriteria, filters [][]bloomIndexes, collector logCollector, applyExactMatch bool) {
+func (f *LogFetcher) collectLogs(block *coretypes.ResultBlock, crit filters.FilterCriteria, collector logCollector, applyExactMatch bool) {
 	ctx := f.ctxProvider(block.Block.Height)
-	totalLogs := uint(0)
-	evmTxIndex := 0
-
-	for _, hash := range getTxHashesFromBlock(f.ctxProvider, f.txConfigProvider, f.k, block, f.includeSyntheticReceipts, f.cacheCreationMutex, f.globalBlockCache) {
-		receipt, found := getOrSetCachedReceipt(f.cacheCreationMutex, f.globalBlockCache, ctx, f.k, block, hash.hash)
-		if !found {
-			ctx.Logger().Error(fmt.Sprintf("collectLogs: unable to find receipt for hash %s", hash.hash.Hex()))
-			continue
-		}
-
-		txLogs := keeper.GetLogsForTx(receipt, totalLogs)
+	store := f.k.ReceiptStore()
+	if store == nil {
+		return
+	}
 
-		if len(crit.Addresses) != 0 || len(crit.Topics) != 0 {
-			if len(receipt.LogsBloom) == 0 || MatchFilters(ethtypes.Bloom(receipt.LogsBloom), filters) {
-				if applyExactMatch {
-					for _, log := range txLogs {
-						log.TxIndex = uint(evmTxIndex)               //nolint:gosec
-						log.BlockNumber = uint64(block.Block.Height) //nolint:gosec
-						log.BlockHash = common.BytesToHash(block.BlockID.Hash)
-						if f.IsLogExactMatch(log, crit) {
-							collector.Append(log)
-						}
-					}
-				} else {
-					for _, log := range txLogs {
-						log.TxIndex = uint(evmTxIndex)               //nolint:gosec
-						log.BlockNumber = uint64(block.Block.Height) //nolint:gosec
-						log.BlockHash = common.BytesToHash(block.BlockID.Hash)
-						collector.Append(log)
-					}
-				}
-			}
-		} else {
-			for _, log := range txLogs {
-				log.TxIndex = uint(evmTxIndex)               //nolint:gosec
-				log.BlockNumber = uint64(block.Block.Height) //nolint:gosec
-				log.BlockHash = common.BytesToHash(block.BlockID.Hash)
-				collector.Append(log)
-			}
-		}
+	txHashes := getTxHashesFromBlock(f.ctxProvider, f.txConfigProvider, f.k, block, f.includeSyntheticReceipts, f.cacheCreationMutex, f.globalBlockCache)
+	hashes := make([]common.Hash, 0, len(txHashes))
+	for _, hash := range txHashes {
+		hashes = append(hashes, hash.hash)
+	}
 
-		totalLogs += uint(len(txLogs))
-		evmTxIndex++
+	logs, err := store.FilterLogs(ctx, block.Block.Height, common.BytesToHash(block.BlockID.Hash), hashes, crit, applyExactMatch)
+	if err != nil {
+		ctx.Logger().Error(fmt.Sprintf("collectLogs: %s", err.Error()))
+		return
+	}
+	for _, log := range logs {
+		collector.Append(log)
 	}
 }
 
@@ -1157,26 +1121,3 @@ func (f *LogFetcher) processBatch(ctx context.Context, start, end int64, crit fi
 		res <- block
 	}
 }
-
-func matchTopics(topics [][]common.Hash, eventTopics []common.Hash) bool {
-	for i, topicList := range topics {
-		if len(topicList) == 0 {
-			// anything matches for this position
-			continue
-		}
-		if i >= len(eventTopics) {
-			return false
-		}
-		matched := false
-		for _, topic := range topicList {
-			if topic == eventTopics[i] {
-				matched = true
-				break
-			}
-		}
-		if !matched {
-			return false
-		}
-	}
-	return true
-}

evmrpc/filter_test.go

@@ -6,9 +6,8 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	ethtypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/eth/filters"
 	testkeeper "github.com/sei-protocol/sei-chain/testutil/keeper"
-	"github.com/sei-protocol/sei-chain/x/evm/keeper"
 	evmtypes "github.com/sei-protocol/sei-chain/x/evm/types"
 	"github.com/stretchr/testify/require"
 )
@@ -580,28 +579,10 @@ func TestCollectLogsEvmTransactionIndex(t *testing.T) {
 		testkeeper.MustMockReceipt(t, k, ctx, txHash, receipt)
 	}
 
-	// Test the core logic that collectLogs implements
-	// This simulates what collectLogs does for each EVM transaction
-	var collectedLogs []*ethtypes.Log
-	evmTxIndex := 0
-	totalLogs := uint(0)
-
-	for _, txHash := range evmTxHashes {
-		receipt, err := k.GetReceipt(ctx, txHash)
-		require.NoError(t, err, "should be able to get receipt for tx %s", txHash.Hex())
-
-		// This simulates keeper.GetLogsForTx
-		logs := keeper.GetLogsForTx(receipt, totalLogs)
-
-		// This is the key part we're testing: setting the correct EVM transaction index
-		for _, log := range logs {
-			log.TxIndex = uint(evmTxIndex) // This should override receipt.TransactionIndex
-			collectedLogs = append(collectedLogs, log)
-		}
-
-		totalLogs += uint(len(receipt.Logs))
-		evmTxIndex++
-	}
+	store := k.ReceiptStore()
+	require.NotNil(t, store)
+	collectedLogs, err := store.FilterLogs(ctx, 2, common.HexToHash("0x2"), evmTxHashes, filters.FilterCriteria{}, true)
+	require.NoError(t, err)
 
 	// Verify that the transaction indices are set correctly
 	require.Equal(t, len(evmTxHashes), len(collectedLogs), "should have one log per transaction")

evmrpc/setup_test.go

@@ -577,7 +577,7 @@ func init() {
 		if err := store.SetLatestVersion(latest); err != nil {
 			panic(err)
 		}
-		_ = store.SetEarliestVersion(1, true)
+		_ = store.SetEarliestVersion(1)
 	}
 	ctxProvider := func(height int64) sdk.Context {
 		if height == MockHeight2 {
@@ -1100,7 +1100,7 @@ func setupLogs() {
 		if err := store.SetLatestVersion(MockHeight103); err != nil {
 			panic(err)
 		}
-		_ = store.SetEarliestVersion(1, true)
+		_ = store.SetEarliestVersion(1)
 	}
 
 }