use inserts instead of lightweight deletes to remove reorged data

Author: iuwqyir

internal/storage/clickhouse.go

@@ -64,9 +64,8 @@ func connectDB(cfg *config.ClickhouseConfig) (clickhouse.Conn, error) {
 		Settings: func() clickhouse.Settings {
 			if cfg.AsyncInsert {
 				return clickhouse.Settings{
-					"async_insert":             "1",
-					"wait_for_async_insert":    "1",
-					"lightweight_deletes_sync": "0",
+					"async_insert":          "1",
+					"wait_for_async_insert": "1",
 				}
 			}
 			return clickhouse.Settings{}
@@ -962,7 +961,7 @@ func (c *ClickHouseConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*
 
 	go func() {
 		defer wg.Done()
-		if err := c.deleteBlocksByNumbers(chainId, blockNumbers); err != nil {
+		if err := c.deleteBlocks(chainId, blockNumbers); err != nil {
 			deleteErrMutex.Lock()
 			deleteErr = fmt.Errorf("error deleting blocks: %v", err)
 			deleteErrMutex.Unlock()
@@ -971,7 +970,7 @@ func (c *ClickHouseConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*
 
 	go func() {
 		defer wg.Done()
-		if err := c.deleteLogsByNumbers(chainId, blockNumbers); err != nil {
+		if err := c.deleteLogs(chainId, blockNumbers); err != nil {
 			deleteErrMutex.Lock()
 			deleteErr = fmt.Errorf("error deleting logs: %v", err)
 			deleteErrMutex.Unlock()
@@ -980,7 +979,7 @@ func (c *ClickHouseConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*
 
 	go func() {
 		defer wg.Done()
-		if err := c.deleteTransactionsByNumbers(chainId, blockNumbers); err != nil {
+		if err := c.deleteTransactions(chainId, blockNumbers); err != nil {
 			deleteErrMutex.Lock()
 			deleteErr = fmt.Errorf("error deleting transactions: %v", err)
 			deleteErrMutex.Unlock()
@@ -989,7 +988,7 @@ func (c *ClickHouseConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*
 
 	go func() {
 		defer wg.Done()
-		if err := c.deleteTracesByNumbers(chainId, blockNumbers); err != nil {
+		if err := c.deleteTraces(chainId, blockNumbers); err != nil {
 			deleteErrMutex.Lock()
 			deleteErr = fmt.Errorf("error deleting traces: %v", err)
 			deleteErrMutex.Unlock()
@@ -1004,21 +1003,33 @@ func (c *ClickHouseConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*
 	return nil
 }
 
-func (c *ClickHouseConnector) deleteBlocksByNumbers(chainId *big.Int, blockNumbers []*big.Int) error {
-	query := fmt.Sprintf("DELETE FROM %s.blocks WHERE _partition_value.1 = ? AND chain_id = ? AND number IN (?)", c.cfg.Database)
+func (c *ClickHouseConnector) deleteBlocks(chainId *big.Int, blockNumbers []*big.Int) error {
+	query := fmt.Sprintf("INSERT INTO %s.blocks (chain_id, number, is_deleted)", c.cfg.Database)
+
+	batch, err := c.conn.PrepareBatch(context.Background(), query)
+	if err != nil {
+		return err
+	}
 
 	blockNumbersStr := make([]string, len(blockNumbers))
 	for i, bn := range blockNumbers {
 		blockNumbersStr[i] = bn.String()
+		err := batch.Append(
+			chainId,
+			bn,
+			1,
+		)
+		if err != nil {
+			return err
+		}
 	}
-	err := c.conn.Exec(context.Background(), query, chainId, chainId, blockNumbersStr)
-	if err != nil {
-		return fmt.Errorf("error deleting blocks: %w", err)
+	if batch.Rows() < 1 {
+		return nil // No blocks to delete
 	}
-	return nil
+	return batch.Send()
 }
 
-func (c *ClickHouseConnector) deleteLogsByNumbers(chainId *big.Int, blockNumbers []*big.Int) error {
+func (c *ClickHouseConnector) deleteLogs(chainId *big.Int, blockNumbers []*big.Int) error {
 	blockNumbersStr := make([]string, len(blockNumbers))
 	for i, bn := range blockNumbers {
 		blockNumbersStr[i] = bn.String()
@@ -1031,34 +1042,36 @@ func (c *ClickHouseConnector) deleteLogsByNumbers(chainId *big.Int, blockNumbers
 	}
 	defer rows.Close()
 
-	blockNumbersToDelete := common.NewSet[string]()
-	txHashesToDelete := common.NewSet[string]()
-	logIndexesToDelete := common.NewSet[uint64]()
+	query := fmt.Sprintf("INSERT INTO %s.logs (chain_id, block_number, transaction_hash, log_index, is_deleted)", c.cfg.Database)
+	batch, err := c.conn.PrepareBatch(context.Background(), query)
+	if err != nil {
+		return err
+	}
+
 	for rows.Next() {
 		var logToDelete common.Log
 		err := rows.ScanStruct(&logToDelete)
 		if err != nil {
 			return err
 		}
-		blockNumbersToDelete.Add(logToDelete.BlockNumber.String())
-		txHashesToDelete.Add(logToDelete.TransactionHash)
-		logIndexesToDelete.Add(logToDelete.LogIndex)
+		err = batch.Append(
+			chainId,
+			logToDelete.BlockNumber,
+			logToDelete.TransactionHash,
+			logToDelete.LogIndex,
+			1,
+		)
+		if err != nil {
+			return err
+		}
 	}
-
-	if txHashesToDelete.Size() == 0 {
+	if batch.Rows() < 1 {
 		return nil // No logs to delete
 	}
-
-	deleteQuery := fmt.Sprintf("DELETE FROM %s.logs WHERE _partition_value.1 = ? AND chain_id = ? AND block_number IN (?) AND transaction_hash IN (?) AND log_index IN (?)", c.cfg.Database)
-
-	err := c.conn.Exec(context.Background(), deleteQuery, chainId, chainId, blockNumbersToDelete.List(), txHashesToDelete.List(), logIndexesToDelete.List())
-	if err != nil {
-		return err
-	}
-	return nil
+	return batch.Send()
 }
 
-func (c *ClickHouseConnector) deleteTransactionsByNumbers(chainId *big.Int, blockNumbers []*big.Int) error {
+func (c *ClickHouseConnector) deleteTransactions(chainId *big.Int, blockNumbers []*big.Int) error {
 	blockNumbersStr := make([]string, len(blockNumbers))
 	for i, bn := range blockNumbers {
 		blockNumbersStr[i] = bn.String()
@@ -1071,68 +1084,74 @@ func (c *ClickHouseConnector) deleteTransactionsByNumbers(chainId *big.Int, bloc
 	}
 	defer rows.Close()
 
-	blockNumbersToDelete := common.NewSet[string]()
-	hashesToDelete := common.NewSet[string]()
+	query := fmt.Sprintf("INSERT INTO %s.transactions (chain_id, block_number, hash, is_deleted)", c.cfg.Database)
+	batch, err := c.conn.PrepareBatch(context.Background(), query)
+	if err != nil {
+		return err
+	}
 
 	for rows.Next() {
 		var txToDelete common.Transaction
 		err := rows.ScanStruct(&txToDelete)
 		if err != nil {
 			return err
 		}
-		blockNumbersToDelete.Add(txToDelete.BlockNumber.String())
-		hashesToDelete.Add(txToDelete.Hash)
+		err = batch.Append(
+			chainId,
+			txToDelete.BlockNumber,
+			txToDelete.Hash,
+			1,
+		)
+		if err != nil {
+			return err
+		}
 	}
-
-	if hashesToDelete.Size() == 0 {
+	if batch.Rows() < 1 {
 		return nil // No transactions to delete
 	}
-
-	deleteQuery := fmt.Sprintf("DELETE FROM %s.transactions WHERE _partition_value.1 = ? AND chain_id = ? AND block_number IN (?) AND hash IN (?)", c.cfg.Database)
-
-	err := c.conn.Exec(context.Background(), deleteQuery, chainId, chainId, blockNumbersToDelete.List(), hashesToDelete.List())
-	if err != nil {
-		return err
-	}
-	return nil
+	return batch.Send()
 }
 
-func (c *ClickHouseConnector) deleteTracesByNumbers(chainId *big.Int, blockNumbers []*big.Int) error {
+func (c *ClickHouseConnector) deleteTraces(chainId *big.Int, blockNumbers []*big.Int) error {
 	blockNumbersStr := make([]string, len(blockNumbers))
 	for i, bn := range blockNumbers {
 		blockNumbersStr[i] = bn.String()
 	}
-	getQuery := fmt.Sprintf("SELECT block_number, transaction_hash FROM %s.traces WHERE chain_id = %s AND block_number IN (?) AND is_deleted = 0", c.cfg.Database, chainId.String())
+	getQuery := fmt.Sprintf("SELECT block_number, transaction_hash, trace_address FROM %s.traces WHERE chain_id = %s AND block_number IN (?) AND is_deleted = 0", c.cfg.Database, chainId.String())
 
 	rows, getErr := c.conn.Query(context.Background(), getQuery, blockNumbersStr)
 	if getErr != nil {
 		return getErr
 	}
 	defer rows.Close()
 
-	blockNumbersToDelete := common.NewSet[string]()
-	txHashesToDelete := common.NewSet[string]()
+	query := fmt.Sprintf("INSERT INTO %s.traces (chain_id, block_number, transaction_hash, trace_address, is_deleted)", c.cfg.Database)
+	batch, err := c.conn.PrepareBatch(context.Background(), query)
+	if err != nil {
+		return err
+	}
+
 	for rows.Next() {
 		var traceToDelete common.Trace
 		err := rows.ScanStruct(&traceToDelete)
 		if err != nil {
 			return err
 		}
-		blockNumbersToDelete.Add(traceToDelete.BlockNumber.String())
-		txHashesToDelete.Add(traceToDelete.TransactionHash)
+		err = batch.Append(
+			chainId,
+			traceToDelete.BlockNumber,
+			traceToDelete.TransactionHash,
+			traceToDelete.TraceAddress,
+			1,
+		)
+		if err != nil {
+			return err
+		}
 	}
-
-	if txHashesToDelete.Size() == 0 {
+	if batch.Rows() < 1 {
 		return nil // No traces to delete
 	}
-
-	deleteQuery := fmt.Sprintf("DELETE FROM %s.traces WHERE _partition_value.1 = ? AND chain_id = ? AND block_number IN (?) AND transaction_hash IN (?)", c.cfg.Database)
-
-	err := c.conn.Exec(context.Background(), deleteQuery, chainId, chainId, blockNumbersToDelete.List(), txHashesToDelete.List())
-	if err != nil {
-		return err
-	}
-	return nil
+	return batch.Send()
 }
 
 // TODO make this atomic