mirror of
https://bitbucket.org/s_l_teichmann/mtsatellite
synced 2024-12-24 01:00:18 +01:00
Interleaver: Commit blocks in batches of 128 to speed up build a new SQLiet databases.
This commit is contained in:
parent
758ef7afc5
commit
d51ecc6eef
@ -23,6 +23,8 @@ const (
|
|||||||
|
|
||||||
var ErrDatabaseNotExists = errors.New("Database does not exists.")
|
var ErrDatabaseNotExists = errors.New("Database does not exists.")
|
||||||
|
|
||||||
|
const blocksPerTx = 128
|
||||||
|
|
||||||
type (
|
type (
|
||||||
SQLiteBlockProducer struct {
|
SQLiteBlockProducer struct {
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
@ -33,6 +35,8 @@ type (
|
|||||||
SQLiteBlockConsumer struct {
|
SQLiteBlockConsumer struct {
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
insertStmt *sql.Stmt
|
insertStmt *sql.Stmt
|
||||||
|
tx *sql.Tx
|
||||||
|
txCounter int
|
||||||
joiner common.KeyJoiner
|
joiner common.KeyJoiner
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
@ -58,6 +62,11 @@ func NewSQLiteBlockConsumer(
|
|||||||
db.Close()
|
db.Close()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
} else {
|
||||||
|
if _, err = db.Exec(deleteSql); err != nil {
|
||||||
|
db.Close()
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
var insertStmt *sql.Stmt
|
var insertStmt *sql.Stmt
|
||||||
@ -66,21 +75,49 @@ func NewSQLiteBlockConsumer(
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var tx *sql.Tx
|
||||||
|
if tx, err = db.Begin(); err != nil {
|
||||||
|
insertStmt.Close()
|
||||||
|
db.Close()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
sbc = &SQLiteBlockConsumer{
|
sbc = &SQLiteBlockConsumer{
|
||||||
db: db,
|
db: db,
|
||||||
insertStmt: insertStmt,
|
insertStmt: insertStmt,
|
||||||
|
tx: tx,
|
||||||
joiner: joiner}
|
joiner: joiner}
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (sbc *SQLiteBlockConsumer) Close() error {
|
func (sbc *SQLiteBlockConsumer) Close() error {
|
||||||
|
sbc.tx.Commit()
|
||||||
sbc.insertStmt.Close()
|
sbc.insertStmt.Close()
|
||||||
return sbc.db.Close()
|
return sbc.db.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (sbc *SQLiteBlockConsumer) getTx() (tx *sql.Tx, err error) {
|
||||||
|
if sbc.txCounter >= blocksPerTx {
|
||||||
|
sbc.txCounter = 0
|
||||||
|
if err = sbc.tx.Commit(); err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if sbc.tx, err = sbc.db.Begin(); err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sbc.txCounter++
|
||||||
|
tx = sbc.tx
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
func (sbc *SQLiteBlockConsumer) Consume(block *Block) (err error) {
|
func (sbc *SQLiteBlockConsumer) Consume(block *Block) (err error) {
|
||||||
_, err = sbc.insertStmt.Exec(sbc.joiner(block.Coord), block.Data)
|
var tx *sql.Tx
|
||||||
|
if tx, err = sbc.getTx(); err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
_, err = tx.Stmt(sbc.insertStmt).Exec(sbc.joiner(block.Coord), block.Data)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user