cassandra/exec.go at main · core-go/cassandra · GitHub

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
package cassandra

import (
	"context"
	"github.com/apache/cassandra-gocql-driver"
)

func Exec(ses *gocql.Session, query string, values ...interface{}) error {
	q := ses.Query(query, values...)
	return q.Exec()
}
func ExecuteAll(ctx context.Context, ses *gocql.Session, stmts ...Statement) (int64, error) {
	return ExecuteAllWithSize(ctx, ses, 5, stmts...)
}
func ExecuteAllWithSize(ctx context.Context, ses *gocql.Session, size int, stmts ...Statement) (int64, error) {
	if stmts == nil || len(stmts) == 0 {
		return 0, nil
	}
	batch := ses.NewBatch(gocql.UnloggedBatch).WithContext(ctx)
	l := len(stmts)
	for i := 0; i < l; i++ {
		var args []interface{}
		args = stmts[i].Params
		batch.Entries = append(batch.Entries, gocql.BatchEntry{
			Stmt:       stmts[i].Query,
			Args:       args,
			Idempotent: true,
		})
		if i%size == 0 || i == l-1 {
			err := ses.ExecuteBatch(batch)
			if err != nil {
				return int64(i + 1), err
			}
			batch = ses.NewBatch(gocql.UnloggedBatch).WithContext(ctx)
		}
	}
	return int64(l), nil
}

func Insert(ses *gocql.Session, table string, model interface{}, options ...*Schema) error {
	return InsertWithVersion(ses, table, model, -1, options...)
}
func InsertWithVersion(ses *gocql.Session, table string, model interface{}, versionIndex int, options ...*Schema) error {
	query, values := BuildToInsertWithVersion(table, model, versionIndex, false, options...)
	return Exec(ses, query, values...)
}
func Update(ses *gocql.Session, table string, model interface{}, options ...*Schema) error {
	return UpdateWithVersion(ses, table, model, -1, options...)
}
func UpdateWithVersion(ses *gocql.Session, table string, model interface{}, versionIndex int, options ...*Schema) error {
	query, values := BuildToUpdateWithVersion(table, model, versionIndex, options...)
	return Exec(ses, query, values...)
}
func Save(ses *gocql.Session, table string, model interface{}, options ...*Schema) error {
	query, values := BuildToSave(table, model, options...)
	return Exec(ses, query, values...)
}

func InsertBatchWithSizeAndVersion(ctx context.Context, ses *gocql.Session, size int, table string, models interface{}, versionIndex int, options ...*Schema) (int64, error) {
	s, err := BuildToInsertBatchWithVersion(table, models, versionIndex, false, options...)
	if err != nil {
		return -1, err
	}
	return ExecuteAllWithSize(ctx, ses, size, s...)
}
func InsertBatchWithVersion(ctx context.Context, ses *gocql.Session, table string, models interface{}, versionIndex int, options ...*Schema) (int64, error) {
	return InsertBatchWithSizeAndVersion(ctx, ses, 5, table, models, versionIndex, options...)
}
func InsertBatch(ctx context.Context, ses *gocql.Session, table string, models interface{}, options ...*Schema) (int64, error) {
	return InsertBatchWithSizeAndVersion(ctx, ses, 5, table, models, -1, options...)
}
func UpdateBatchWithSizeAndVersion(ctx context.Context, ses *gocql.Session, size int, table string, models interface{}, versionIndex int, options ...*Schema) (int64, error) {
	s, err := BuildToUpdateBatchWithVersion(table, models, versionIndex, options...)
	if err != nil {
		return -1, err
	}
	return ExecuteAllWithSize(ctx, ses, size, s...)
}
func UpdateBatchWithVersion(ctx context.Context, ses *gocql.Session, table string, models interface{}, versionIndex int, options ...*Schema) (int64, error) {
	return UpdateBatchWithSizeAndVersion(ctx, ses, 5, table, models, versionIndex, options...)
}
func UpdateBatch(ctx context.Context, ses *gocql.Session, table string, models interface{}, options ...*Schema) (int64, error) {
	return UpdateBatchWithSizeAndVersion(ctx, ses, 5, table, models, -1, options...)
}
func SaveBatchWithSize(ctx context.Context, ses *gocql.Session, size int, table string, models interface{}, options ...*Schema) (int64, error) {
	s, err := BuildToInsertBatchWithVersion(table, models, -1, true, options...)
	if err != nil {
		return -1, err
	}
	return ExecuteAllWithSize(ctx, ses, size, s...)
}
func SaveBatch(ctx context.Context, ses *gocql.Session, table string, models interface{}, options ...*Schema) (int64, error) {
	return SaveBatchWithSize(ctx, ses, 5, table, models, options...)
}