cerc-io/go-ethereum
12
0
Fork 0
Code Issues 9 Pull Requests Packages Projects Releases 66 Wiki Activity

Fix storage_cids

Browse Source
This commit is contained in:
telackey
2023-03-07 19:02:58 -06:00
parent 3604d98891
commit 9b535e2901
2 changed files with 20 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
statediff/indexer/database/sql/lazy_tx.go
+7 -7
View File
@@ -41,20 +41,20 @@ func (tx *DelayedTx) QueryRow(ctx context.Context, sql string, args ...interface
return tx.db.QueryRow(ctx, sql, args...)
}
func (tx *DelayedTx) findPrevCopyFrom(tableName []string, columnNames []string, limit int) *copyFrom {
func (tx *DelayedTx) findPrevCopyFrom(tableName []string, columnNames []string, limit int) (*copyFrom, int) {
for pos, count := len(tx.cache)-1, 0; pos >= 0 && count < limit; pos, count = pos-1, count+1 {
prevCopy, ok := tx.cache[pos].(*copyFrom)
if ok && prevCopy.matches(tableName, columnNames) {
return prevCopy
return prevCopy, count
}
}
return nil
return nil, -1
}
func (tx *DelayedTx) CopyFrom(ctx context.Context, tableName []string, columnNames []string, rows [][]interface{}) (int64, error) {
if prevCopy := tx.findPrevCopyFrom(tableName, columnNames, copyFromCheckLimit); nil != prevCopy {
log.Trace("statediff lazy_tx : Appending rows to COPY", "table", tableName,
"current", len(prevCopy.rows), "append", len(rows))
if prevCopy, distance := tx.findPrevCopyFrom(tableName, columnNames, copyFromCheckLimit); nil != prevCopy {
log.Trace("statediff lazy_tx : Appending to COPY", "table", tableName,
"current", len(prevCopy.rows), "new", len(rows), "distance", distance)
prevCopy.appendRows(rows)
} else {
tx.cache = append(tx.cache, &copyFrom{tableName, columnNames, rows})
@@ -84,9 +84,9 @@ func (tx *DelayedTx) Commit(ctx context.Context) error {
for _, item := range tx.cache {
switch item := item.(type) {
case *copyFrom:
log.Trace("statediff lazy_tx : COPY", "table", item.tableName, "rows", len(item.rows))
_, err := base.CopyFrom(ctx, item.tableName, item.columnNames, item.rows)
if err != nil {
log.Error("COPY error", "table", item.tableName, "err", err)
return err
}
case cachedStmt:
statediff/indexer/database/sql/writer.go
+13 -5
View File
@@ -150,14 +150,18 @@ func (w *Writer) upsertStateCID(tx Tx, stateNode models.StateNodeModel) error {
}
if w.db.UseCopyFrom() {
var row []interface{}
blockNum, _ := strconv.ParseInt(stateNode.BlockNumber, 10, 64)
blockNum, err := strconv.ParseInt(stateNode.BlockNumber, 10, 64)
if err != nil {
return insertError{"eth.state_cids", err, "COPY", stateNode}
}
row = append(row, blockNum, stateNode.HeaderID, stateKey, stateNode.CID,
stateNode.Path, stateNode.NodeType, true, stateNode.MhKey)
var rows [][]interface{}
rows = append(rows, row)
_, err := tx.CopyFrom(w.db.Context(), w.db.StateTableName(), w.db.StateColumnNames(), rows)
_, err = tx.CopyFrom(w.db.Context(), w.db.StateTableName(), w.db.StateColumnNames(), rows)
if err != nil {
return insertError{"eth.state_cids", err, "COPY", stateNode}
}
@@ -197,16 +201,20 @@ func (w *Writer) upsertStorageCID(tx Tx, storageCID models.StorageNodeModel) err
}
if w.db.UseCopyFrom() {
var row []interface{}
blockNum, _ := strconv.ParseInt(storageCID.BlockNumber, 10, 64)
blockNum, err := strconv.ParseInt(storageCID.BlockNumber, 10, 64)
if err != nil {
return insertError{"eth.storage_cids", err, "COPY", storageCID}
}
row = append(row, blockNum, storageCID.HeaderID, storageCID.StatePath, storageKey, storageCID.CID,
storageCID.Path, storageCID.NodeType, true, storageCID.MhKey)
var rows [][]interface{}
rows = append(rows, row)
_, err := tx.CopyFrom(w.db.Context(), w.db.StateTableName(), w.db.StateColumnNames(), rows)
_, err = tx.CopyFrom(w.db.Context(), w.db.StorageTableName(), w.db.StorageColumnNames(), rows)
if err != nil {
return insertError{"eth.state_cids", err, "COPY", storageCID}
return insertError{"eth.storage_cids", err, "COPY", storageCID}
}
} else {
_, err := tx.Exec(w.db.Context(), w.db.InsertStorageStm(),