Use VersionedCollapsingMergeTree and insert based deletes (#149)

### TL;DR
Migrated database tables from ReplacingMergeTree to VersionedCollapsingMergeTree engine and updated column names for better consistency.

### What changed?
- Replaced lightweight deletes with inserts for blocks, logs, transactions and traces
- Changed database engine from ReplacingMergeTree to VersionedCollapsingMergeTree
- Added `sign` and `insert_timestamp` columns for versioning support
- Renamed block table columns for consistency:
  - `number` -> `block_number`
  - `timestamp` -> `block_timestamp`
- Updated block timestamp type from UInt64 to DateTime
- Optimized index granularity settings
- Adjusted table ordering to use block_timestamp
- Removed nullable constraints from certain columns
- Updated corresponding Go structs to match new schema
- Added option to force consistent data (uses FINAL with clickhouse)

### How to test?
1. Drop existing tables
2. Run new table creation scripts
3. Verify data insertion works with new schema
4. Test data retrieval using updated column names
5. Verify versioning works by inserting duplicate records
6. Confirm proper handling of deleted records

### Why make this change?
- Better support for data versioning and deletion tracking
- Improved data consistency with standardized column naming
- Enhanced query performance through optimized indexing
- More efficient handling of data updates and deletions
- Better type handling for timestamp-related operations

Author: iuwqyir

api/api.go

@@ -44,6 +44,8 @@ type QueryParams struct {
 
 	// @Description Flag to enable abi decoding of tx data
 	Decode bool `schema:"decode"`
+	// @Description Flag to force consistent data at the expense of query speed
+	ForceConsistentData bool `schema:"force_consistent_data"`
 }
 
 // Meta represents metadata for a query response

configs/config.go

@@ -1,7 +1,9 @@
 package config
 
 import (
+	"encoding/json"
 	"fmt"
+	"os"
 	"strings"
 
 	"github.com/rs/zerolog/log"
@@ -63,15 +65,23 @@ type StorageConnectionConfig struct {
 	Redis      *RedisConfig      `mapstructure:"redis"`
 }
 
+type TableConfig struct {
+	DefaultSelectFields []string `mapstructure:"defaultSelectFields"`
+	TableName           string   `mapstructure:"tableName"`
+}
+
+type TableOverrideConfig map[string]TableConfig
+
 type ClickhouseConfig struct {
-	Host             string `mapstructure:"host"`
-	Port             int    `mapstructure:"port"`
-	Username         string `mapstructure:"username"`
-	Password         string `mapstructure:"password"`
-	Database         string `mapstructure:"database"`
-	DisableTLS       bool   `mapstructure:"disableTLS"`
-	AsyncInsert      bool   `mapstructure:"asyncInsert"`
-	MaxRowsPerInsert int    `mapstructure:"maxRowsPerInsert"`
+	Host             string                         `mapstructure:"host"`
+	Port             int                            `mapstructure:"port"`
+	Username         string                         `mapstructure:"username"`
+	Password         string                         `mapstructure:"password"`
+	Database         string                         `mapstructure:"database"`
+	DisableTLS       bool                           `mapstructure:"disableTLS"`
+	AsyncInsert      bool                           `mapstructure:"asyncInsert"`
+	MaxRowsPerInsert int                            `mapstructure:"maxRowsPerInsert"`
+	ChainBasedConfig map[string]TableOverrideConfig `mapstructure:"chainBasedConfig"`
 }
 
 type MemoryConfig struct {
@@ -160,5 +170,48 @@ func LoadConfig(cfgFile string) error {
 		return fmt.Errorf("error unmarshalling config: %v", err)
 	}
 
+	err = setCustomJSONConfigs()
+	if err != nil {
+		return fmt.Errorf("error setting custom JSON configs: %v", err)
+	}
+
+	// Add debug logging
+	if clickhouse := Cfg.Storage.Main.Clickhouse; clickhouse != nil {
+		log.Debug().
+			Interface("chainConfig", clickhouse.ChainBasedConfig).
+			Msgf("Loaded chain config %v", clickhouse.ChainBasedConfig)
+	}
+
+	return nil
+}
+
+func setCustomJSONConfigs() error {
+	if chainConfigJSON := os.Getenv("STORAGE_MAIN_CLICKHOUSE_CHAINBASEDCONFIG"); chainConfigJSON != "" {
+		var mainChainConfig map[string]TableOverrideConfig
+		if err := json.Unmarshal([]byte(chainConfigJSON), &mainChainConfig); err != nil {
+			return fmt.Errorf("error parsing main chainBasedConfig JSON: %v", err)
+		}
+		if Cfg.Storage.Main.Clickhouse != nil {
+			Cfg.Storage.Main.Clickhouse.ChainBasedConfig = mainChainConfig
+		}
+	}
+	if chainConfigJSON := os.Getenv("STORAGE_STAGING_CLICKHOUSE_CHAINBASEDCONFIG"); chainConfigJSON != "" {
+		var stagingChainConfig map[string]TableOverrideConfig
+		if err := json.Unmarshal([]byte(chainConfigJSON), &stagingChainConfig); err != nil {
+			return fmt.Errorf("error parsing staging chainBasedConfig JSON: %v", err)
+		}
+		if Cfg.Storage.Staging.Clickhouse != nil {
+			Cfg.Storage.Staging.Clickhouse.ChainBasedConfig = stagingChainConfig
+		}
+	}
+	if chainConfigJSON := os.Getenv("STORAGE_ORCHESTRATOR_CLICKHOUSE_CHAINBASEDCONFIG"); chainConfigJSON != "" {
+		var orchestratorChainConfig map[string]TableOverrideConfig
+		if err := json.Unmarshal([]byte(chainConfigJSON), &orchestratorChainConfig); err != nil {
+			return fmt.Errorf("error parsing orchestrator chainBasedConfig JSON: %v", err)
+		}
+		if Cfg.Storage.Main.Clickhouse != nil {
+			Cfg.Storage.Main.Clickhouse.ChainBasedConfig = orchestratorChainConfig
+		}
+	}
 	return nil
 }

internal/common/block.go

@@ -2,31 +2,34 @@ package common
 
 import (
 	"math/big"
+	"time"
 )
 
 type Block struct {
-	ChainId          *big.Int `json:"chain_id" ch:"chain_id"`
-	Number           *big.Int `json:"number" ch:"number"`
-	Hash             string   `json:"hash" ch:"hash"`
-	ParentHash       string   `json:"parent_hash" ch:"parent_hash"`
-	Timestamp        uint64   `json:"timestamp" ch:"timestamp"`
-	Nonce            string   `json:"nonce" ch:"nonce"`
-	Sha3Uncles       string   `json:"sha3_uncles" ch:"sha3_uncles"`
-	MixHash          string   `json:"mix_hash" ch:"mix_hash"`
-	Miner            string   `json:"miner" ch:"miner"`
-	StateRoot        string   `json:"state_root" ch:"state_root"`
-	TransactionsRoot string   `json:"transactions_root" ch:"transactions_root"`
-	ReceiptsRoot     string   `json:"receipts_root" ch:"receipts_root"`
-	LogsBloom        string   `json:"logs_bloom" ch:"logs_bloom"`
-	Size             uint64   `json:"size" ch:"size"`
-	ExtraData        string   `json:"extra_data" ch:"extra_data"`
-	Difficulty       *big.Int `json:"difficulty" ch:"difficulty"`
-	TotalDifficulty  *big.Int `json:"total_difficulty" ch:"total_difficulty"`
-	TransactionCount uint64   `json:"transaction_count" ch:"transaction_count"`
-	GasLimit         *big.Int `json:"gas_limit" ch:"gas_limit"`
-	GasUsed          *big.Int `json:"gas_used" ch:"gas_used"`
-	WithdrawalsRoot  string   `json:"withdrawals_root" ch:"withdrawals_root"`
-	BaseFeePerGas    uint64   `json:"base_fee_per_gas" ch:"base_fee_per_gas"`
+	ChainId          *big.Int  `json:"chain_id" ch:"chain_id"`
+	Number           *big.Int  `json:"block_number" ch:"block_number"`
+	Hash             string    `json:"hash" ch:"hash"`
+	ParentHash       string    `json:"parent_hash" ch:"parent_hash"`
+	Timestamp        time.Time `json:"block_timestamp" ch:"block_timestamp"`
+	Nonce            string    `json:"nonce" ch:"nonce"`
+	Sha3Uncles       string    `json:"sha3_uncles" ch:"sha3_uncles"`
+	MixHash          string    `json:"mix_hash" ch:"mix_hash"`
+	Miner            string    `json:"miner" ch:"miner"`
+	StateRoot        string    `json:"state_root" ch:"state_root"`
+	TransactionsRoot string    `json:"transactions_root" ch:"transactions_root"`
+	ReceiptsRoot     string    `json:"receipts_root" ch:"receipts_root"`
+	LogsBloom        string    `json:"logs_bloom" ch:"logs_bloom"`
+	Size             uint64    `json:"size" ch:"size"`
+	ExtraData        string    `json:"extra_data" ch:"extra_data"`
+	Difficulty       *big.Int  `json:"difficulty" ch:"difficulty"`
+	TotalDifficulty  *big.Int  `json:"total_difficulty" ch:"total_difficulty"`
+	TransactionCount uint64    `json:"transaction_count" ch:"transaction_count"`
+	GasLimit         *big.Int  `json:"gas_limit" ch:"gas_limit"`
+	GasUsed          *big.Int  `json:"gas_used" ch:"gas_used"`
+	WithdrawalsRoot  string    `json:"withdrawals_root" ch:"withdrawals_root"`
+	BaseFeePerGas    uint64    `json:"base_fee_per_gas" ch:"base_fee_per_gas"`
+	Sign             int8      `json:"sign" ch:"sign"`
+	InsertTimestamp  time.Time `json:"insert_timestamp" ch:"insert_timestamp"`
 }
 
 type BlockData struct {

internal/common/log.go

@@ -5,26 +5,29 @@ import (
 	"fmt"
 	"math/big"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	gethCommon "github.com/ethereum/go-ethereum/common"
 	"github.com/rs/zerolog/log"
 )
 
 type Log struct {
-	ChainId          *big.Int `json:"chain_id" ch:"chain_id" swaggertype:"string"`
-	BlockNumber      *big.Int `json:"block_number" ch:"block_number" swaggertype:"string"`
-	BlockHash        string   `json:"block_hash" ch:"block_hash"`
-	BlockTimestamp   uint64   `json:"block_timestamp" ch:"block_timestamp"`
-	TransactionHash  string   `json:"transaction_hash" ch:"transaction_hash"`
-	TransactionIndex uint64   `json:"transaction_index" ch:"transaction_index"`
-	LogIndex         uint64   `json:"log_index" ch:"log_index"`
-	Address          string   `json:"address" ch:"address"`
-	Data             string   `json:"data" ch:"data"`
-	Topic0           string   `json:"topic_0" ch:"topic_0"`
-	Topic1           string   `json:"topic_1" ch:"topic_1"`
-	Topic2           string   `json:"topic_2" ch:"topic_2"`
-	Topic3           string   `json:"topic_3" ch:"topic_3"`
+	ChainId          *big.Int  `json:"chain_id" ch:"chain_id" swaggertype:"string"`
+	BlockNumber      *big.Int  `json:"block_number" ch:"block_number" swaggertype:"string"`
+	BlockHash        string    `json:"block_hash" ch:"block_hash"`
+	BlockTimestamp   time.Time `json:"block_timestamp" ch:"block_timestamp"`
+	TransactionHash  string    `json:"transaction_hash" ch:"transaction_hash"`
+	TransactionIndex uint64    `json:"transaction_index" ch:"transaction_index"`
+	LogIndex         uint64    `json:"log_index" ch:"log_index"`
+	Address          string    `json:"address" ch:"address"`
+	Data             string    `json:"data" ch:"data"`
+	Topic0           string    `json:"topic_0" ch:"topic_0"`
+	Topic1           string    `json:"topic_1" ch:"topic_1"`
+	Topic2           string    `json:"topic_2" ch:"topic_2"`
+	Topic3           string    `json:"topic_3" ch:"topic_3"`
+	Sign             int8      `json:"sign" ch:"sign"`
+	InsertTimestamp  time.Time `json:"insert_timestamp" ch:"insert_timestamp"`
 }
 
 func (l *Log) GetTopic(index int) (string, error) {

internal/common/trace.go

@@ -2,30 +2,33 @@ package common
 
 import (
 	"math/big"
+	"time"
 )
 
 type Trace struct {
-	ChainID          *big.Int `json:"chain_id" ch:"chain_id"`
-	BlockNumber      *big.Int `json:"block_number" ch:"block_number"`
-	BlockHash        string   `json:"block_hash" ch:"block_hash"`
-	BlockTimestamp   uint64   `json:"block_timestamp" ch:"block_timestamp"`
-	TransactionHash  string   `json:"transaction_hash" ch:"transaction_hash"`
-	TransactionIndex uint64   `json:"transaction_index" ch:"transaction_index"`
-	Subtraces        int64    `json:"subtraces" ch:"subtraces"`
-	TraceAddress     []uint64 `json:"trace_address" ch:"trace_address"`
-	TraceType        string   `json:"trace_type" ch:"type"`
-	CallType         string   `json:"call_type" ch:"call_type"`
-	Error            string   `json:"error" ch:"error"`
-	FromAddress      string   `json:"from_address" ch:"from_address"`
-	ToAddress        string   `json:"to_address" ch:"to_address"`
-	Gas              *big.Int `json:"gas" ch:"gas"`
-	GasUsed          *big.Int `json:"gas_used" ch:"gas_used"`
-	Input            string   `json:"input" ch:"input"`
-	Output           string   `json:"output" ch:"output"`
-	Value            *big.Int `json:"value" ch:"value"`
-	Author           string   `json:"author" ch:"author"`
-	RewardType       string   `json:"reward_type" ch:"reward_type"`
-	RefundAddress    string   `json:"refund_address" ch:"refund_address"`
+	ChainID          *big.Int  `json:"chain_id" ch:"chain_id"`
+	BlockNumber      *big.Int  `json:"block_number" ch:"block_number"`
+	BlockHash        string    `json:"block_hash" ch:"block_hash"`
+	BlockTimestamp   time.Time `json:"block_timestamp" ch:"block_timestamp"`
+	TransactionHash  string    `json:"transaction_hash" ch:"transaction_hash"`
+	TransactionIndex uint64    `json:"transaction_index" ch:"transaction_index"`
+	Subtraces        int64     `json:"subtraces" ch:"subtraces"`
+	TraceAddress     []uint64  `json:"trace_address" ch:"trace_address"`
+	TraceType        string    `json:"trace_type" ch:"type"`
+	CallType         string    `json:"call_type" ch:"call_type"`
+	Error            string    `json:"error" ch:"error"`
+	FromAddress      string    `json:"from_address" ch:"from_address"`
+	ToAddress        string    `json:"to_address" ch:"to_address"`
+	Gas              *big.Int  `json:"gas" ch:"gas"`
+	GasUsed          *big.Int  `json:"gas_used" ch:"gas_used"`
+	Input            string    `json:"input" ch:"input"`
+	Output           string    `json:"output" ch:"output"`
+	Value            *big.Int  `json:"value" ch:"value"`
+	Author           string    `json:"author" ch:"author"`
+	RewardType       string    `json:"reward_type" ch:"reward_type"`
+	RefundAddress    string    `json:"refund_address" ch:"refund_address"`
+	Sign             int8      `json:"sign" ch:"sign"`
+	InsertTimestamp  time.Time `json:"insert_timestamp" ch:"insert_timestamp"`
 }
 
 type RawTraces = []map[string]interface{}

internal/common/transaction.go

@@ -5,41 +5,44 @@ import (
 	"math/big"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/rs/zerolog/log"
 )
 
 type Transaction struct {
-	ChainId              *big.Int `json:"chain_id" ch:"chain_id" swaggertype:"string"`
-	Hash                 string   `json:"hash" ch:"hash"`
-	Nonce                uint64   `json:"nonce" ch:"nonce"`
-	BlockHash            string   `json:"block_hash" ch:"block_hash"`
-	BlockNumber          *big.Int `json:"block_number" ch:"block_number" swaggertype:"string"`
-	BlockTimestamp       uint64   `json:"block_timestamp" ch:"block_timestamp"`
-	TransactionIndex     uint64   `json:"transaction_index" ch:"transaction_index"`
-	FromAddress          string   `json:"from_address" ch:"from_address"`
-	ToAddress            string   `json:"to_address" ch:"to_address"`
-	Value                *big.Int `json:"value" ch:"value" swaggertype:"string"`
-	Gas                  uint64   `json:"gas" ch:"gas"`
-	GasPrice             *big.Int `json:"gas_price" ch:"gas_price" swaggertype:"string"`
-	Data                 string   `json:"data" ch:"data"`
-	FunctionSelector     string   `json:"function_selector" ch:"function_selector"`
-	MaxFeePerGas         *big.Int `json:"max_fee_per_gas" ch:"max_fee_per_gas" swaggertype:"string"`
-	MaxPriorityFeePerGas *big.Int `json:"max_priority_fee_per_gas" ch:"max_priority_fee_per_gas" swaggertype:"string"`
-	TransactionType      uint8    `json:"transaction_type" ch:"transaction_type"`
-	R                    *big.Int `json:"r" ch:"r" swaggertype:"string"`
-	S                    *big.Int `json:"s" ch:"s" swaggertype:"string"`
-	V                    *big.Int `json:"v" ch:"v" swaggertype:"string"`
-	AccessListJson       *string  `json:"access_list_json" ch:"access_list"`
-	ContractAddress      *string  `json:"contract_address" ch:"contract_address"`
-	GasUsed              *uint64  `json:"gas_used" ch:"gas_used"`
-	CumulativeGasUsed    *uint64  `json:"cumulative_gas_used" ch:"cumulative_gas_used"`
-	EffectiveGasPrice    *big.Int `json:"effective_gas_price" ch:"effective_gas_price" swaggertype:"string"`
-	BlobGasUsed          *uint64  `json:"blob_gas_used" ch:"blob_gas_used"`
-	BlobGasPrice         *big.Int `json:"blob_gas_price" ch:"blob_gas_price" swaggertype:"string"`
-	LogsBloom            *string  `json:"logs_bloom" ch:"logs_bloom"`
-	Status               *uint64  `json:"status" ch:"status"`
+	ChainId              *big.Int  `json:"chain_id" ch:"chain_id" swaggertype:"string"`
+	Hash                 string    `json:"hash" ch:"hash"`
+	Nonce                uint64    `json:"nonce" ch:"nonce"`
+	BlockHash            string    `json:"block_hash" ch:"block_hash"`
+	BlockNumber          *big.Int  `json:"block_number" ch:"block_number" swaggertype:"string"`
+	BlockTimestamp       time.Time `json:"block_timestamp" ch:"block_timestamp"`
+	TransactionIndex     uint64    `json:"transaction_index" ch:"transaction_index"`
+	FromAddress          string    `json:"from_address" ch:"from_address"`
+	ToAddress            string    `json:"to_address" ch:"to_address"`
+	Value                *big.Int  `json:"value" ch:"value" swaggertype:"string"`
+	Gas                  uint64    `json:"gas" ch:"gas"`
+	GasPrice             *big.Int  `json:"gas_price" ch:"gas_price" swaggertype:"string"`
+	Data                 string    `json:"data" ch:"data"`
+	FunctionSelector     string    `json:"function_selector" ch:"function_selector"`
+	MaxFeePerGas         *big.Int  `json:"max_fee_per_gas" ch:"max_fee_per_gas" swaggertype:"string"`
+	MaxPriorityFeePerGas *big.Int  `json:"max_priority_fee_per_gas" ch:"max_priority_fee_per_gas" swaggertype:"string"`
+	TransactionType      uint8     `json:"transaction_type" ch:"transaction_type"`
+	R                    *big.Int  `json:"r" ch:"r" swaggertype:"string"`
+	S                    *big.Int  `json:"s" ch:"s" swaggertype:"string"`
+	V                    *big.Int  `json:"v" ch:"v" swaggertype:"string"`
+	AccessListJson       *string   `json:"access_list_json" ch:"access_list"`
+	ContractAddress      *string   `json:"contract_address" ch:"contract_address"`
+	GasUsed              *uint64   `json:"gas_used" ch:"gas_used"`
+	CumulativeGasUsed    *uint64   `json:"cumulative_gas_used" ch:"cumulative_gas_used"`
+	EffectiveGasPrice    *big.Int  `json:"effective_gas_price" ch:"effective_gas_price" swaggertype:"string"`
+	BlobGasUsed          *uint64   `json:"blob_gas_used" ch:"blob_gas_used"`
+	BlobGasPrice         *big.Int  `json:"blob_gas_price" ch:"blob_gas_price" swaggertype:"string"`
+	LogsBloom            *string   `json:"logs_bloom" ch:"logs_bloom"`
+	Status               *uint64   `json:"status" ch:"status"`
+	Sign                 int8      `json:"sign" ch:"sign"`
+	InsertTimestamp      time.Time `json:"insert_timestamp" ch:"insert_timestamp"`
 }
 
 type DecodedTransactionData struct {

internal/common/utils.go

@@ -163,20 +163,21 @@ func isType(word string) bool {
 var allowedFunctions = map[string]struct{}{
 	"sum":                  {},
 	"count":                {},
-	"reinterpretAsUInt256": {},
+	"reinterpretasuint256": {},
 	"reverse":              {},
 	"unhex":                {},
 	"substring":            {},
 	"length":               {},
-	"toUInt256":            {},
+	"touint256":            {},
 	"if":                   {},
-	"toStartOfMonth":       {},
-	"toStartOfDay":         {},
-	"toStartOfHour":        {},
-	"toStartOfMinute":      {},
-	"toDate":               {},
-	"toDateTime":           {},
+	"tostartofmonth":       {},
+	"tostartofday":         {},
+	"tostartofhour":        {},
+	"tostartofminute":      {},
+	"todate":               {},
+	"todatetime":           {},
 	"concat":               {},
+	"in":                   {},
 }
 
 var disallowedPatterns = []string{
@@ -207,7 +208,7 @@ func ValidateQuery(query string) error {
 	matches := functionPattern.FindAllStringSubmatch(query, -1)
 	for _, match := range matches {
 		funcName := match[1]
-		if _, ok := allowedFunctions[funcName]; !ok {
+		if _, ok := allowedFunctions[strings.ToLower(funcName)]; !ok {
 			return fmt.Errorf("function '%s' is not allowed", funcName)
 		}
 	}