forked from Restream/reindexer
/
clickhouse.go
104 lines (89 loc) · 2.43 KB
/
clickhouse.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
package repo
import (
"log"
"math/rand"
"strconv"
"github.com/jmoiron/sqlx"
_ "github.com/kshvakov/clickhouse"
)
type ClickHouseRepo struct {
db *sqlx.DB
}
func (repo *ClickHouseRepo) Init() bool {
var err error
if repo.db, err = sqlx.Connect("clickhouse", "tcp://127.0.0.1:9000?debug=false"); err != nil {
panic(err)
}
repo.db.SetMaxOpenConns(8)
repo.db.SetMaxIdleConns(8)
return true
}
func (repo *ClickHouseRepo) Seed(itemsInDataSet int) bool {
log.Printf("Seeding data to Clickhouse")
if _, err := repo.db.Exec("DROP TABLE IF EXISTS items;"); err != nil {
panic(err)
}
sqlStmt := `
CREATE TABLE IF NOT EXISTS items (
EventDate Date,
id UInt32,
name String,
year UInt32,
description String
) engine=Memory;
`
if _, err := repo.db.Exec(sqlStmt); err != nil {
panic(err)
}
tx, err := repo.db.Begin()
if err != nil {
panic(err)
}
stmt, err := tx.Prepare("INSERT INTO items(id, name, year, description) VALUES (?,?,?,?)")
if err != nil {
panic(err)
}
defer stmt.Close()
for i := 0; i < itemsInDataSet; i++ {
it := newItem(i)
if _, err = stmt.Exec(i, it.Name, it.Year, it.Description); err != nil {
panic(err)
}
}
tx.Commit()
return true
}
func (repo *ClickHouseRepo) QueryFullText(textQuery func() string, N int, limit int) (ret []*Item) {
return nil
}
func (repo *ClickHouseRepo) QueryByID(N int, onlyQuery bool) (it *Item) {
stmt, err := repo.db.Preparex("select id,name,year,description from items where id = ?")
if err != nil {
log.Fatal(err)
}
return sqlFetchOne(stmt, N, onlyQuery, rand.Int()%itemsInDataSet)
}
func (repo *ClickHouseRepo) Query2Cond(N int, onlyQuery bool, limit int) (ret []*Item) {
stmt, err := repo.db.Preparex("select id,name,year,description from items where year > ? and name = ? limit " + strconv.Itoa(limit))
if err != nil {
log.Fatal(err)
}
return sqlFetchAll(stmt, N, onlyQuery, limit, 2010, randString())
}
func (repo *ClickHouseRepo) Query1Cond(N int, onlyQuery bool, limit int) (ret []*Item) {
stmt, err := repo.db.Preparex("select id,name,year,description from items where year > ? limit " + strconv.Itoa(limit))
if err != nil {
log.Fatal(err)
}
return sqlFetchAll(stmt, N, onlyQuery, limit, 2010)
}
func (repo *ClickHouseRepo) Update(N int) {
stmt, err := repo.db.Preparex("UPDATE items SET name=?,year=?,description=? WHERE id=?")
if err != nil {
log.Fatal(err)
}
sqlUpdate(stmt, 1)
}
func init() {
registerRepo("clickhouse", &ClickHouseRepo{})
}