feat(query): added MapR Queryer (#21)

Author: cnlangzi

.gitignore

@@ -10,6 +10,7 @@
 
 # Test binary, built with `go test -c`
 *.test
+*.db
 
 # Output of the go coverage tool, specifically when used with LiteIDE
 *.out

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.22
 	github.com/rs/zerolog v1.32.0
 	github.com/stretchr/testify v1.9.0
+	github.com/yaitoo/async v1.0.0
 )
 
 require (

go.sum

@@ -20,6 +20,8 @@ github.com/rs/zerolog v1.32.0 h1:keLypqrlIjaFsbmJOBdB/qvyF8KEtCWHwobLp5l/mQ0=
 github.com/rs/zerolog v1.32.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/yaitoo/async v1.0.0 h1:1of0INlIdF24Tz0Scg5Vd0Hnef5M+hSfQLdv2wFOHvc=
+github.com/yaitoo/async v1.0.0/go.mod h1:IpSO7Ei7AxiqLxFqDjN4rJaVlt8wm4ZxMXyyQaWmM1g=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

migrate/migrator.go

@@ -405,18 +405,16 @@ func (m *Migrator) migrate(ctx context.Context, db *sqle.DB) error {
 				switch s.Rotate {
 				case shardid.MonthlyRotate:
 					for t := s.RotateBegin; !t.After(s.RotateEnd); t = t.AddDate(0, 1, 0) {
-						rotations = append(rotations, "_"+t.Format("200601"))
+						rotations = append(rotations, shardid.FormatMonth(t))
 					}
 				case shardid.WeeklyRotate:
-					var week int
 					for t := s.RotateBegin; !t.After(s.RotateEnd); t = t.AddDate(0, 0, 7) {
-						_, week = t.ISOWeek() //1-53 week
-						rotations = append(rotations, "_"+t.Format("2006")+fmt.Sprintf("%03d", week))
+						rotations = append(rotations, shardid.FormatWeek(t))
 					}
 
 				case shardid.DailyRotate:
 					for t := s.RotateBegin; !t.After(s.RotateEnd); t = t.AddDate(0, 0, 1) {
-						rotations = append(rotations, "_"+t.Format("20060102"))
+						rotations = append(rotations, shardid.FormatDay(t))
 					}
 				}
 

query.go

@@ -0,0 +1,61 @@
+package sqle
+
+import (
+	"context"
+	"fmt"
+)
+
+type Errors struct {
+	items []error
+}
+
+func (e *Errors) Error() string {
+	return fmt.Sprint(e.items)
+}
+
+type Query[T any] struct {
+	db      *DB
+	queryer Queryer[T]
+	tables  []string
+}
+
+// NewQuery create a Query
+func NewQuery[T any](db *DB, options ...QueryOption[T]) Query[T] {
+	q := Query[T]{
+		db: db,
+	}
+
+	for _, opt := range options {
+		if opt != nil {
+			opt(&q)
+		}
+	}
+
+	if q.tables == nil {
+		q.tables = []string{""}
+	}
+
+	if q.queryer == nil {
+		q.queryer = &MapR[T]{
+			dbs: q.db.dbs,
+		}
+	}
+
+	return q
+}
+
+func (q *Query[T]) First(ctx context.Context, b *Builder) (T, error) {
+	return q.queryer.First(ctx, q.tables, b)
+}
+
+func (q *Query[T]) Count(ctx context.Context, b *Builder) (int, error) {
+	return q.queryer.Count(ctx, q.tables, b)
+}
+
+func (q *Query[T]) Query(ctx context.Context, b *Builder, less func(i, j T) bool) ([]T, error) {
+	return q.queryer.Query(ctx, q.tables, b, less)
+}
+
+func (q *Query[T]) QueryLimit(ctx context.Context, b *Builder, less func(i, j T) bool, limit int) ([]T, error) {
+	return q.queryer.QueryLimit(ctx, q.tables, b, less, limit)
+}

query_option.go

@@ -0,0 +1,39 @@
+package sqle
+
+import (
+	"time"
+
+	"github.com/yaitoo/sqle/shardid"
+)
+
+type QueryOption[T any] func(q *Query[T])
+
+func WithMonths[T any](start, end time.Time) QueryOption[T] {
+	return func(q *Query[T]) {
+		for t := start; !t.After(end); t = t.AddDate(0, 1, 0) {
+			q.tables = append(q.tables, shardid.FormatMonth(t))
+		}
+	}
+}
+
+func WithWeeks[T any](start, end time.Time) QueryOption[T] {
+	return func(q *Query[T]) {
+		for t := start; !t.After(end); t = t.AddDate(0, 0, 7) {
+			q.tables = append(q.tables, shardid.FormatWeek(t))
+		}
+	}
+}
+
+func WithDays[T any](start, end time.Time) QueryOption[T] {
+	return func(q *Query[T]) {
+		for t := start; !t.After(end); t = t.AddDate(0, 0, 1) {
+			q.tables = append(q.tables, shardid.FormatDay(t))
+		}
+	}
+}
+
+func WithQueryer[T any](qr Queryer[T]) QueryOption[T] {
+	return func(q *Query[T]) {
+		q.queryer = qr
+	}
+}

queryer.go

@@ -0,0 +1,10 @@
+package sqle
+
+import "context"
+
+type Queryer[T any] interface {
+	First(ctx context.Context, tables []string, b *Builder) (T, error)
+	Count(ctx context.Context, tables []string, b *Builder) (int, error)
+	Query(ctx context.Context, tables []string, b *Builder, less func(i, j T) bool) ([]T, error)
+	QueryLimit(ctx context.Context, tables []string, b *Builder, less func(i, j T) bool, limit int) ([]T, error)
+}

queryer_mapr.go

@@ -0,0 +1,156 @@
+package sqle
+
+import (
+	"context"
+	"sort"
+	"strconv"
+	"strings"
+
+	"github.com/yaitoo/async"
+)
+
+// MapR Map/Reduce Query
+type MapR[T any] struct {
+	dbs []*Context
+}
+
+func (q *MapR[T]) First(ctx context.Context, tables []string, b *Builder) (T, error) {
+	var it T
+	b.Input("rotate", "<rotate>") //lazy replace on async.Wait
+	query, args, err := b.Build()
+	if err != nil {
+		return it, err
+	}
+
+	w := async.New[T]()
+
+	for _, r := range tables {
+		qr := strings.ReplaceAll(query, "<rotate>", r)
+		for _, db := range q.dbs {
+			w.Add(func(db *Context, qr string) func(context.Context) (T, error) {
+				return func(ctx context.Context) (T, error) {
+					var t T
+					err := db.QueryRowContext(ctx, qr, args...).Bind(&t)
+					if err != nil {
+						return t, err
+					}
+
+					return t, nil
+				}
+			}(db, qr))
+		}
+	}
+
+	return w.WaitAny(ctx)
+}
+func (q *MapR[T]) Count(ctx context.Context, tables []string, b *Builder) (int, error) {
+	b.Input("rotate", "<rotate>") //lazy replace on async.Wait
+	query, args, err := b.Build()
+	if err != nil {
+		return 0, err
+	}
+
+	w := async.New[int]()
+
+	for _, r := range tables {
+		qr := strings.ReplaceAll(query, "<rotate>", r)
+		for _, db := range q.dbs {
+			w.Add(func(db *Context, qr string) func(context.Context) (int, error) {
+				return func(ctx context.Context) (int, error) {
+					var i int
+					err := db.QueryRowContext(ctx, qr, args...).Scan(&i)
+					if err != nil {
+						return i, err
+					}
+
+					return i, nil
+				}
+			}(db, qr))
+		}
+	}
+
+	items, err := w.Wait(ctx)
+
+	if err != nil {
+		return 0, err
+	}
+
+	var total int
+
+	for _, it := range items {
+		total += it
+	}
+
+	return total, nil
+}
+func (q *MapR[T]) Query(ctx context.Context, tables []string, b *Builder, less func(i, j T) bool) ([]T, error) {
+
+	b.Input("rotate", "<rotate>") //lazy replace on async.Wait
+	query, args, err := b.Build()
+	if err != nil {
+		return nil, err
+	}
+
+	w := async.New[[]T]()
+
+	for _, r := range tables {
+		qr := strings.ReplaceAll(query, "<rotate>", r)
+		for _, db := range q.dbs {
+			w.Add(func(db *Context, qr string) func(context.Context) ([]T, error) {
+				return func(context.Context) ([]T, error) {
+					var t []T
+					rows, err := db.QueryContext(ctx, qr, args...)
+					if err != nil {
+						return t, err
+					}
+
+					err = rows.Bind(&t)
+					if err != nil {
+						return t, err
+					}
+
+					return t, nil
+				}
+			}(db, qr))
+		}
+	}
+
+	items, err := w.Wait(ctx)
+
+	if err != nil {
+		return nil, err
+	}
+
+	var list []T
+
+	for _, it := range items {
+		if it != nil {
+			list = append(list, it...)
+		}
+	}
+
+	if less != nil {
+		sort.Slice(list, func(i, j int) bool {
+			return less(list[i], list[j])
+		})
+	}
+
+	return list, nil
+}
+func (q *MapR[T]) QueryLimit(ctx context.Context, tables []string, b *Builder, less func(i, j T) bool, limit int) ([]T, error) {
+
+	if limit > 0 {
+		b.SQL(" LIMIT " + strconv.Itoa(limit*len(q.dbs)))
+	}
+
+	list, err := q.Query(ctx, tables, b, less)
+	if err != nil {
+		return nil, err
+	}
+
+	if limit < len(list) {
+		return list[0:limit], nil
+	}
+
+	return list, nil
+}