sour-is/go-pkg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add libsql support

Browse Source
This commit is contained in:
xuu
2024-04-19 10:56:27 -06:00
parent 1f8b4ab24f
commit b1bff4cbf0
17 changed files with 609 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
mercury/sql/list-string.go
View File

@@ -43,9 +43,7 @@ func listScan(e *[]string, ends [2]rune) scanFn {
return nil
}
for _, s := range splitComma(string(str)) {
*e = append(*e, s)
}
*e = append(*e, splitComma(string(str))...)
return nil
}

2
mercury/sql/notify.go
View File

@@ -29,7 +29,7 @@ func (pgm *sqlHandler) GetNotify(ctx context.Context, event string) (lis mercury
Where(squirrel.Eq{"event": event}).
PlaceholderFormat(squirrel.Dollar).
RunWith(pgm.db).
QueryContext(context.TODO())
QueryContext(ctx)
if err != nil {
return nil, err

34
mercury/sql/otel.go
View File

@@ -2,6 +2,7 @@ package sql
import (
"database/sql"
"strings"
"go.nhat.io/otelsql"
semconv "go.opentelemetry.io/otel/semconv/v1.20.0"
@@ -9,25 +10,28 @@ import (
func openDB(driver, dsn string) (*sql.DB, error) {
system := semconv.DBSystemPostgreSQL
if driver == "sqlite" {
if driver == "sqlite" || strings.HasPrefix(driver, "libsql") {
system = semconv.DBSystemSqlite
}
// Register the otelsql wrapper for the provided postgres driver.
driverName, err := otelsql.Register(driver,
otelsql.AllowRoot(),
otelsql.TraceQueryWithoutArgs(),
otelsql.TraceRowsClose(),
otelsql.TraceRowsAffected(),
// otelsql.WithDatabaseName("my_database"), // Optional.
otelsql.WithSystem(system), // Optional.
)
if err != nil {
return nil, err
if driver == "postgres" {
var err error
// Register the otelsql wrapper for the provided postgres driver.
driver, err = otelsql.Register(driver,
otelsql.AllowRoot(),
otelsql.TraceQueryWithoutArgs(),
otelsql.TraceRowsClose(),
otelsql.TraceRowsAffected(),
// otelsql.WithDatabaseName("my_database"), // Optional.
otelsql.WithSystem(system), // Optional.
)
if err != nil {
return nil, err
}
}
// Connect to a Postgres database using the postgres driver wrapper.
db, err := sql.Open(driverName, dsn)
db, err := sql.Open(driver, dsn)
if err != nil {
return nil, err
}

280
mercury/sql/sql.go
View File

@@ -3,21 +3,27 @@ package sql
import (
"context"
"database/sql"
"errors"
"fmt"
"log"
"slices"
"strings"
sq "github.com/Masterminds/squirrel"
"go.sour.is/pkg/lg"
"go.sour.is/pkg/mercury"
"go.sour.is/pkg/rsql"
"golang.org/x/exp/maps"
)
var MAX_FILTER int = 40
type sqlHandler struct {
name string
db *sql.DB
paceholderFormat sq.PlaceholderFormat
listFormat [2]rune
readonly bool
getWhere func(search mercury.Search) (func(sq.SelectBuilder) sq.SelectBuilder, error)
}
var (
@@ -27,11 +33,13 @@ var (
_ mercury.WriteConfig = (*sqlHandler)(nil)
)
func Register() {
func Register() func(context.Context) error {
var hdlrs []*sqlHandler
mercury.Registry.Register("sql", func(s *mercury.Space) any {
var dsn string
var opts strings.Builder
var dbtype string
var readonly bool = slices.Contains(s.Tags, "readonly")
for _, c := range s.List {
if c.Name == "match" {
continue
@@ -49,7 +57,6 @@ func Register() {
if dsn == "" {
dsn = opts.String()
}
db, err := openDB(dbtype, dsn)
if err != nil {
return err
@@ -58,31 +65,47 @@ func Register() {
return err
}
switch dbtype {
case "sqlite":
return &sqlHandler{db, sq.Dollar, [2]rune{'[', ']'}}
case "sqlite", "libsql", "libsql+embed":
h := &sqlHandler{s.Space, db, sq.Question, [2]rune{'[', ']'}, readonly, GetWhereSQ}
hdlrs = append(hdlrs, h)
return h
case "postgres":
return &sqlHandler{db, sq.Dollar, [2]rune{'{', '}'}}
h := &sqlHandler{s.Space, db, sq.Dollar, [2]rune{'{', '}'}, readonly, GetWherePG}
hdlrs = append(hdlrs, h)
return h
default:
return fmt.Errorf("unsupported dbtype: %s", dbtype)
}
})
return func(ctx context.Context) error {
var errs error
for _, h := range hdlrs {
// if err = ctx.Err(); err != nil {
// return errors.Join(errs, err)
// }
errs = errors.Join(errs, h.db.Close())
}
return errs
}
}
type Space struct {
mercury.Space
ID uint64
id uint64
}
type Value struct {
mercury.Value
ID uint64
id uint64
}
func (p *sqlHandler) GetIndex(ctx context.Context, search mercury.NamespaceSearch, pgm *rsql.Program) (mercury.Config, error) {
func (p *sqlHandler) GetIndex(ctx context.Context, search mercury.Search) (mercury.Config, error) {
ctx, span := lg.Span(ctx)
defer span.End()
cols := rsql.GetDbColumns(mercury.Space{})
where, err := getWhere(search, cols)
where, err := p.getWhere(search)
if err != nil {
return nil, err
}
@@ -100,28 +123,40 @@ func (p *sqlHandler) GetIndex(ctx context.Context, search mercury.NamespaceSearc
return config, nil
}
func (p *sqlHandler) GetConfig(ctx context.Context, search mercury.NamespaceSearch, pgm *rsql.Program, fields []string) (mercury.Config, error) {
func (p *sqlHandler) GetConfig(ctx context.Context, search mercury.Search) (config mercury.Config, err error) {
ctx, span := lg.Span(ctx)
defer span.End()
idx, err := p.GetIndex(ctx, search, pgm)
where, err := p.getWhere(search)
if err != nil {
return nil, err
}
spaceMap := make(map[string]int, len(idx))
for u, s := range idx {
spaceMap[s.Space] = u
lis, err := p.listSpace(ctx, nil, where)
if err != nil {
log.Println(err)
return nil, err
}
where, err := getWhere(search, rsql.GetDbColumns(mercury.Value{}))
if err != nil {
return nil, err
if len(lis) == 0 {
return nil, nil
}
query := sq.Select(`"space"`, `"name"`, `"seq"`, `"notes"`, `"tags"`, `"values"`).
From("mercury_registry_vw").
Where(where).
OrderBy("space asc", "name asc").
spaceIDX := make([]uint64, len(lis))
spaceMap := make(map[uint64]int, len(lis))
config = make(mercury.Config, len(lis))
for i, s := range lis {
spaceIDX[i] = s.id
config[i] = &s.Space
spaceMap[s.id] = i
}
query := sq.Select(`"id"`, `"name"`, `"seq"`, `"notes"`, `"tags"`, `"values"`).
From("mercury_values").
Where(sq.Eq{"id": spaceIDX}).
OrderBy("id asc", "seq asc").
PlaceholderFormat(p.paceholderFormat)
span.AddEvent(p.name)
span.AddEvent(lg.LogQuery(query.ToSql()))
rows, err := query.RunWith(p.db).
QueryContext(ctx)
@@ -133,10 +168,10 @@ func (p *sqlHandler) GetConfig(ctx context.Context, search mercury.NamespaceSear
defer rows.Close()
for rows.Next() {
var s mercury.Value
var s Value
err = rows.Scan(
&s.Space,
&s.id,
&s.Name,
&s.Seq,
listScan(&s.Notes, p.listFormat),
@@ -146,19 +181,20 @@ func (p *sqlHandler) GetConfig(ctx context.Context, search mercury.NamespaceSear
if err != nil {
return nil, err
}
if u, ok := spaceMap[s.Space]; ok {
idx[u].List = append(idx[u].List, s)
if u, ok := spaceMap[s.id]; ok {
lis[u].List = append(lis[u].List, s.Value)
}
}
err = rows.Err()
span.RecordError(err)
span.AddEvent(fmt.Sprint("read index ", len(idx)))
return idx, err
span.AddEvent(fmt.Sprint("read index ", len(lis)))
// log.Println(config.String())
return config, err
}
func (p *sqlHandler) listSpace(ctx context.Context, tx sq.BaseRunner, where sq.Sqlizer) ([]*Space, error) {
func (p *sqlHandler) listSpace(ctx context.Context, tx sq.BaseRunner, where func(sq.SelectBuilder) sq.SelectBuilder) ([]*Space, error) {
ctx, span := lg.Span(ctx)
defer span.End()
@@ -168,9 +204,11 @@ func (p *sqlHandler) listSpace(ctx context.Context, tx sq.BaseRunner, where sq.S
query := sq.Select(`"id"`, `"space"`, `"notes"`, `"tags"`, `"trailer"`).
From("mercury_spaces").
Where(where).
OrderBy("space asc").
PlaceholderFormat(sq.Dollar)
PlaceholderFormat(p.paceholderFormat)
query = where(query)
span.AddEvent(p.name)
span.AddEvent(lg.LogQuery(query.ToSql()))
rows, err := query.RunWith(tx).
QueryContext(ctx)
@@ -185,7 +223,7 @@ func (p *sqlHandler) listSpace(ctx context.Context, tx sq.BaseRunner, where sq.S
for rows.Next() {
var s Space
err = rows.Scan(
&s.ID,
&s.id,
&s.Space.Space,
listScan(&s.Space.Notes, p.listFormat),
listScan(&s.Space.Tags, p.listFormat),
@@ -209,6 +247,10 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
ctx, span := lg.Span(ctx)
defer span.End()
if p.readonly {
return fmt.Errorf("readonly database")
}
// Delete spaces that are present in input but are empty.
deleteSpaces := make(map[string]struct{})
@@ -233,7 +275,8 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
}()
// get current spaces
lis, err := p.listSpace(ctx, tx, sq.Eq{"space": maps.Keys(names)})
where := func(qry sq.SelectBuilder) sq.SelectBuilder { return qry.Where(sq.Eq{"space": maps.Keys(names)}) }
lis, err := p.listSpace(ctx, tx, where)
if err != nil {
return
}
@@ -250,12 +293,12 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
currentNames[spaceName] = struct{}{}
if _, ok := deleteSpaces[spaceName]; ok {
deleteIDs = append(deleteIDs, s.ID)
deleteIDs = append(deleteIDs, s.id)
continue
}
updateSpaces = append(updateSpaces, config[names[spaceName]])
updateIDs = append(updateIDs, s.ID)
updateIDs = append(updateIDs, s.id)
}
for _, s := range config {
spaceName := s.Space
@@ -266,7 +309,7 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
// delete spaces
if ids := deleteIDs; len(ids) > 0 {
_, err = sq.Delete("mercury_spaces").Where(sq.Eq{"id": ids}).RunWith(tx).PlaceholderFormat(sq.Dollar).ExecContext(ctx)
_, err = sq.Delete("mercury_spaces").Where(sq.Eq{"id": ids}).RunWith(tx).PlaceholderFormat(p.paceholderFormat).ExecContext(ctx)
if err != nil {
return err
}
@@ -274,7 +317,7 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
// delete values
if ids := append(updateIDs, deleteIDs...); len(ids) > 0 {
_, err = sq.Delete("mercury_values").Where(sq.Eq{"id": ids}).RunWith(tx).PlaceholderFormat(sq.Dollar).ExecContext(ctx)
_, err = sq.Delete("mercury_values").Where(sq.Eq{"id": ids}).RunWith(tx).PlaceholderFormat(p.paceholderFormat).ExecContext(ctx)
if err != nil {
return err
}
@@ -289,7 +332,8 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
Set("tags", listValue(u.Tags, p.listFormat)).
Set("notes", listValue(u.Notes, p.listFormat)).
Set("trailer", listValue(u.Trailer, p.listFormat)).
PlaceholderFormat(sq.Dollar)
PlaceholderFormat(p.paceholderFormat)
span.AddEvent(p.name)
span.AddEvent(lg.LogQuery(query.ToSql()))
_, err := query.RunWith(tx).ExecContext(ctx)
@@ -298,7 +342,7 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
}
// log.Debugf("UPDATED %d SPACES", len(updateSpaces))
for _, v := range u.List {
newValues = append(newValues, &Value{Value: v, ID: updateIDs[i]})
newValues = append(newValues, &Value{Value: v, id: updateIDs[i]})
}
}
@@ -306,15 +350,16 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
for _, s := range insertSpaces {
var id uint64
query := sq.Insert("mercury_spaces").
PlaceholderFormat(sq.Dollar).
PlaceholderFormat(p.paceholderFormat).
Columns("space", "tags", "notes", "trailer").
Values(
s.Space,
listValue(s.Tags, p.listFormat),
s.Space,
listValue(s.Tags, p.listFormat),
listValue(s.Notes, p.listFormat),
listValue(s.Trailer, p.listFormat),
).
).
Suffix("RETURNING \"id\"")
span.AddEvent(p.name)
span.AddEvent(lg.LogQuery(query.ToSql()))
err := query.
@@ -322,12 +367,13 @@ func (p *sqlHandler) WriteConfig(ctx context.Context, config mercury.Config) (er
QueryRowContext(ctx).
Scan(&id)
if err != nil {
span.AddEvent(p.name)
s, v, _ := query.ToSql()
log.Println(s, v, err)
return err
}
for _, v := range s.List {
newValues = append(newValues, &Value{Value: v, ID: id})
newValues = append(newValues, &Value{Value: v, id: id})
}
}
@@ -353,7 +399,7 @@ func (p *sqlHandler) writeValues(ctx context.Context, tx sq.BaseRunner, lis []*V
newInsert := func() sq.InsertBuilder {
return sq.Insert("mercury_values").
RunWith(tx).
PlaceholderFormat(sq.Dollar).
PlaceholderFormat(p.paceholderFormat).
Columns(
`"id"`,
`"seq"`,
@@ -367,7 +413,7 @@ func (p *sqlHandler) writeValues(ctx context.Context, tx sq.BaseRunner, lis []*V
insert := newInsert()
for i, s := range lis {
insert = insert.Values(
s.ID,
s.id,
s.Seq,
s.Name,
listValue(s.Values, p.listFormat),
@@ -378,7 +424,7 @@ func (p *sqlHandler) writeValues(ctx context.Context, tx sq.BaseRunner, lis []*V
if i > 0 && i%chunk == 0 {
// log.Debugf("inserting %v rows into %v", i%chunk, d.Table)
// log.Debug(insert.ToSql())
span.AddEvent(p.name)
span.AddEvent(lg.LogQuery(insert.ToSql()))
_, err = insert.ExecContext(ctx)
@@ -392,7 +438,7 @@ func (p *sqlHandler) writeValues(ctx context.Context, tx sq.BaseRunner, lis []*V
}
if len(lis)%chunk > 0 {
// log.Debugf("inserting %v rows into %v", len(lis)%chunk, d.Table)
// log.Debug(insert.ToSql())
span.AddEvent(p.name)
span.AddEvent(lg.LogQuery(insert.ToSql()))
_, err = insert.ExecContext(ctx)
@@ -405,13 +451,11 @@ func (p *sqlHandler) writeValues(ctx context.Context, tx sq.BaseRunner, lis []*V
return
}
func getWhere(search mercury.NamespaceSearch, d *rsql.DbColumns) (sq.Sqlizer, error) {
func GetWherePG(search mercury.Search) (func(sq.SelectBuilder) sq.SelectBuilder, error) {
var where sq.Or
space, err := d.Col("space")
if err != nil {
return nil, err
}
for _, m := range search {
space := "space"
for _, m := range search.NamespaceSearch {
switch m.(type) {
case mercury.NamespaceNode:
where = append(where, sq.Eq{space: m.Value()})
@@ -422,5 +466,129 @@ func getWhere(search mercury.NamespaceSearch, d *rsql.DbColumns) (sq.Sqlizer, er
where = append(where, e)
}
}
return where, nil
var joins []sq.SelectBuilder
for i, o := range search.Find {
log.Println(o)
if i > MAX_FILTER {
err := fmt.Errorf("too many filters [%d]", MAX_FILTER)
return nil, err
}
q := sq.Select("DISTINCT id").From("mercury_values")
switch o.Op {
case "key":
q = q.Where(sq.Eq{"name": o.Left})
case "nkey":
q = q.Where(sq.NotEq{"name": o.Left})
case "eq":
q = q.Where("name = ? AND ? = any (values)", o.Left, o.Right)
case "neq":
q = q.Where("name = ? AND ? != any (values)", o.Left, o.Right)
case "gt":
q = q.Where("name = ? AND ? > any (values)", o.Left, o.Right)
case "lt":
q = q.Where("name = ? AND ? < any (values)", o.Left, o.Right)
case "ge":
q = q.Where("name = ? AND ? >= any (values)", o.Left, o.Right)
case "le":
q = q.Where("name = ? AND ? <= any (values)", o.Left, o.Right)
// case "like":
// q = q.Where("name = ? AND value LIKE ?", o.Left, o.Right)
// case "in":
// q = q.Where(sq.Eq{"name": o.Left, "value": strings.Split(o.Right, " ")})
}
joins = append(joins, q)
}
return func(s sq.SelectBuilder) sq.SelectBuilder {
for i, q := range joins {
s = s.JoinClause(q.Prefix("JOIN (").Suffix(fmt.Sprintf(`) r%03d USING (id)`, i)))
}
if search.Count > 0 {
s = s.Limit(search.Count)
}
return s.Where(where)
}, nil
}
func GetWhereSQ(search mercury.Search) (func(sq.SelectBuilder) sq.SelectBuilder, error) {
var where sq.Or
var errs error
id := "id"
space := "space"
name := "name"
values_each := `json_valid("values")`
values_valid := `json_valid("values")`
if errs != nil {
return nil, errs
}
for _, m := range search.NamespaceSearch {
switch m.(type) {
case mercury.NamespaceNode:
where = append(where, sq.Eq{space: m.Value()})
case mercury.NamespaceStar:
where = append(where, sq.Like{space: m.Value()})
case mercury.NamespaceTrace:
e := sq.Expr(`? LIKE `+space+` || '%'`, m.Value())
where = append(where, e)
}
}
var joins []sq.SelectBuilder
for i, o := range search.Find {
log.Println(o)
if i > MAX_FILTER {
err := fmt.Errorf("too many filters [%d]", MAX_FILTER)
return nil, err
}
q := sq.Select("DISTINCT " + id).From(`mercury_values mv, ` + values_each + ` vs`)
switch o.Op {
case "key":
q = q.Where(sq.Eq{name: o.Left})
case "nkey":
q = q.Where(sq.NotEq{name: o.Left})
case "eq":
q = q.Where(sq.And{sq.Expr(values_valid), sq.Eq{name: o.Left, `vs.value`: o.Right}})
case "neq":
q = q.Where(sq.And{sq.Expr(values_valid), sq.Eq{name: o.Left}, sq.NotEq{`vs.value`: o.Right}})
case "gt":
q = q.Where(sq.And{sq.Expr(values_valid), sq.Eq{name: o.Left}, sq.Gt{`vs.value`: o.Right}})
case "lt":
q = q.Where(sq.And{sq.Expr(values_valid), sq.Eq{name: o.Left}, sq.Lt{`vs.value`: o.Right}})
case "ge":
q = q.Where(sq.And{sq.Expr(values_valid), sq.Eq{name: o.Left}, sq.GtOrEq{`vs.value`: o.Right}})
case "le":
q = q.Where(sq.And{sq.Expr(values_valid), sq.Eq{name: o.Left}, sq.LtOrEq{`vs.value`: o.Right}})
case "like":
q = q.Where(sq.And{sq.Expr(values_valid), sq.Eq{name: o.Left}, sq.Like{`vs.value`: o.Right}})
case "in":
q = q.Where(sq.Eq{name: o.Left, "vs.value": strings.Split(o.Right, " ")})
}
joins = append(joins, q)
}
return func(s sq.SelectBuilder) sq.SelectBuilder {
for i, q := range joins {
s = s.JoinClause(q.Prefix("JOIN (").Suffix(fmt.Sprintf(`) r%03d USING (id)`, i)))
}
if search.Count > 0 {
s = s.Limit(search.Count)
}
if search.Offset > 0 {
s = s.Offset(search.Offset)
}
return s.Where(where)
}, nil
}