~nilium/sql

cc8c8b8cb4992e5fe47a107d6f31e1d9c15d8ae3 — Noel Cower 8 months ago 0eb4a8b v0.2.0
Additional refactoring for use in chisel

Some additional refactoring to make it easier to use vdb in chisel
(working name, anyway). No significant changes at the moment, but
moving some marshaling code into vdb so it can be used with config
files more reliably.
3 files changed, 143 insertions(+), 53 deletions(-)

M main.go
M vdb/query.go
M vdb/value.go
M main.go => main.go +17 -26
@@ 120,9 120,7 @@ func main() {
	isolationLevel := sql.LevelDefault

	opts := vdb.QueryOptions{
		TryJSON:    false,
		TimeFormat: vdb.TimeString,
		Compact:    false,
	}

	fs := flag.NewFlagSet("sql", flag.ContinueOnError)


@@ 293,30 291,23 @@ Options:

	// -t, --time-format FORMAT
	timeFlag := cli.File(cli.NewFlagFunc("rfc3339", false, func(s string) error {
		switch s {
		case "ts", "rfc3339", "str":
			opts.TimeFormat = vdb.TimeString
		case "unix", "s", "sec":
			opts.TimeFormat = vdb.TimeUnix
		case "unixns", "ns", "nsec":
			opts.TimeFormat = vdb.TimeUnixNano
		case "unixus", "us", "usec":
			opts.TimeFormat = vdb.TimeUnixMicro
		case "unixms", "ms", "msec":
			opts.TimeFormat = vdb.TimeUnixMilli
		case "unixf", "fs", "fsec", "float":
			opts.TimeFormat = vdb.TimeFloat
		default:
			if strings.HasPrefix(s, "+") {
				opts.TimeFormat, opts.TimeLayout = vdb.TimeCustom, s[1:]
				break
			} else if strings.HasPrefix(s, "format:") {
				opts.TimeFormat, opts.TimeLayout = vdb.TimeCustom, s[7:]
				break
			}
		err := opts.TimeFormat.UnmarshalText([]byte(s))
		if err == nil {
			return nil
		}
		if opts.TimeFormat == vdb.TimeCustom {
			// Ignore TimeCustom because it's for programmatic use
			// (i.e., below).
			return fmt.Errorf("invalid time format: %q", s)
		}
		return nil
		if strings.HasPrefix(s, "+") {
			opts.TimeFormat, opts.TimeLayout = vdb.TimeCustom, s[1:]
			return nil
		} else if strings.HasPrefix(s, "format:") {
			opts.TimeFormat, opts.TimeLayout = vdb.TimeCustom, s[7:]
			return nil
		}
		return fmt.Errorf("invalid time format: %w", err)
	}))
	fs.Var(timeFlag, "t", "set the `format` of parsed times")
	fs.Var(timeFlag, "time-format", "set the `format` of parsed times")


@@ 463,7 454,7 @@ splitArgs:
		argSets = append(argSets, argv)
	}

	var queries []*vdb.Query
	var queries []*vdb.Execution
	for i, args := range argSets {
		sets := [][]string{}
		begin := 1


@@ 500,7 491,7 @@ splitArgs:
			panic(fmt.Errorf("error parsing query %d arguments: %w", i, err))
		}

		query := &vdb.Query{
		query := &vdb.Execution{
			Query:   queryStmt,
			Args:    qargs,
			Options: &opts,

M vdb/query.go => vdb/query.go +84 -27
@@ 19,6 19,7 @@ package vdb
import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	"github.com/jmoiron/sqlx"


@@ 26,16 27,61 @@ import (

type TimeFormat int

var ErrNilOptions = errors.New("options must not be nil")

const (
	TimeString TimeFormat = iota
	TimeFloat
	TimeUnixNano
	TimeUnixMicro
	TimeUnixMilli
	TimeUnix
	TimeCustom
	TimeString    TimeFormat = 0
	TimeFloat     TimeFormat = 1
	TimeUnixNano  TimeFormat = 2
	TimeUnixMicro TimeFormat = 3
	TimeUnixMilli TimeFormat = 4
	TimeUnix      TimeFormat = 5
	TimeCustom    TimeFormat = 6
)

func (t TimeFormat) MarshalText() ([]byte, error) {
	switch t {
	case TimeString:
		return []byte("rfc3339"), nil
	case TimeFloat:
		return []byte("fsec"), nil
	case TimeUnixNano:
		return []byte("unixns"), nil
	case TimeUnixMicro:
		return []byte("unixus"), nil
	case TimeUnixMilli:
		return []byte("unixms"), nil
	case TimeUnix:
		return []byte("unix"), nil
	case TimeCustom:
		return []byte("layout"), nil
	default:
		return nil, fmt.Errorf("unrecognized TimeFormat %d", t)
	}
}

func (t *TimeFormat) UnmarshalText(p []byte) error {
	switch s := string(p); s {
	case "ts", "rfc3339", "str":
		*t = TimeString
	case "unix", "s", "sec":
		*t = TimeUnix
	case "unixns", "ns", "nsec":
		*t = TimeUnixNano
	case "unixus", "us", "usec":
		*t = TimeUnixMicro
	case "unixms", "ms", "msec":
		*t = TimeUnixMilli
	case "unixf", "fs", "fsec", "float":
		*t = TimeFloat
	case "layout":
		*t = TimeCustom
	default:
		return fmt.Errorf("unrecognized time format %q", s)
	}
	return nil
}

type Results []Records

func (rs Results) Opaque() []interface{} {


@@ 61,33 107,34 @@ type Record map[string]*Value
func (r Record) Opaque() map[string]interface{} {
	m := make(map[string]interface{}, len(r))
	for k, v := range r {
		m[k] = v.Dest
		m[k] = v.Opaque()
	}
	return m
}

type QueryOptions struct {
	TryJSON    bool
	SkipJSON   bool
	TimeFormat TimeFormat
	TimeLayout string // Used if TimeFormat is TimeCustom.
	Compact    bool

	// Query formatting.
	BindType int
	TryJSON    bool       `json:"try_json"`
	SkipJSON   bool       `json:"skip_json"`
	TimeFormat TimeFormat `json:"time_format"`
	TimeLayout string     `json:"time_layout,omitempty"` // Used if TimeFormat is TimeCustom.
	Compact    bool       `json:"compact"`

	// Query formatting. This should be set depending on the driver, not
	// user-customizable.
	BindType int `json:"-"`
}

type Query struct {
type Execution struct {
	Query string
	Args  [][]interface{}

	Options *QueryOptions
}

func (q *Query) Exec(ctx context.Context, db DB) (results Results, err error) {
	results = make(Results, 0, len(q.Args)*2)
	for i, args := range q.Args {
		results, err = q.execArgSet(ctx, results, db, args)
func (e *Execution) Exec(ctx context.Context, db DB) (results Results, err error) {
	results = make(Results, 0, len(e.Args)*2)
	for i, args := range e.Args {
		results, err = execArgSet(ctx, results, e.Options, db, e.Query, args)
		if err != nil {
			return nil, fmt.Errorf("error executing query with %d arg set: %w", i+1, err)
		}


@@ 95,15 142,22 @@ func (q *Query) Exec(ctx context.Context, db DB) (results Results, err error) {
	return results, nil
}

func (q *Query) execArgSet(ctx context.Context, results Results, db DB, args []interface{}) (Results, error) {
	query, qargs, err := sqlx.In(q.Query, args...)
func Query(ctx context.Context, options *QueryOptions, db DB, query string, args ...interface{}) (Results, error) {
	return execArgSet(ctx, make(Results, 0, 1), options, db, query, args)
}

func execArgSet(ctx context.Context, results Results, options *QueryOptions, db DB, query string, args []interface{}) (Results, error) {
	if options == nil {
		return nil, ErrNilOptions
	}
	query, qargs, err := sqlx.In(query, args...)
	if err != nil {
		// This used to be a log message, but so far I've never seen it
		// come up, so now just return an error. If it shows up as an
		// error, something is probably wrong with an input query.
		return nil, fmt.Errorf("failed to expand query via sqlx.In: %w", err)
	} else {
		query = sqlx.Rebind(q.Options.BindType, query)
		query = sqlx.Rebind(options.BindType, query)
	}

	rows, err := db.QueryContext(ctx, query, qargs...)


@@ 113,7 167,7 @@ func (q *Query) execArgSet(ctx context.Context, results Results, db DB, args []i
	defer rows.Close()

	for i := 1; ; i++ {
		records, err := q.scanRows(ctx, rows)
		records, err := ScanRows(ctx, rows, options)
		if err != nil {
			return nil, fmt.Errorf("error scanning rows in statement %d: %w", i, err)
		}


@@ 132,7 186,10 @@ func (q *Query) execArgSet(ctx context.Context, results Results, db DB, args []i
	return results, nil
}

func (q *Query) scanRows(ctx context.Context, rows *sql.Rows) (Records, error) {
func ScanRows(ctx context.Context, rows *sql.Rows, options *QueryOptions) (Records, error) {
	if options == nil {
		return nil, ErrNilOptions
	}
	coltypes, err := rows.ColumnTypes()
	if err != nil {
		return nil, fmt.Errorf("error getting column types: %w", err)


@@ 145,7 202,7 @@ func (q *Query) scanRows(ctx context.Context, rows *sql.Rows) (Records, error) {
		record := make(map[string]*Value, len(coltypes))
		for ci, coltype := range coltypes {
			val := &Value{
				Options: q.Options,
				Options: options,
				Type:    coltype,
			}
			name := coltype.Name()

M vdb/value.go => vdb/value.go +42 -0
@@ 79,6 79,48 @@ func (v *Value) Value() (driver.Value, error) {
	return v.Dest, nil
}

func (v *Value) Opaque() interface{} {
	switch d := v.Dest.(type) {
	case *big.Float:
		// Convert to float64 for concrete value use.
		f, _ := d.Float64()
		return f
	case *big.Int:
		// Convert to int64 for concrete value use.
		return d.Int64()
	case time.Time:
		switch v.Options.TimeFormat {
		case TimeString:
			return d.Format(time.RFC3339Nano)
		case TimeFloat:
			t := d.Truncate(time.Second)
			sub := d.Sub(t)
			f := new(big.Float).SetPrec(0).SetFloat64(float64(sub))
			f = f.Quo(f, big.NewFloat(float64(time.Second)))
			f = f.Add(f, big.NewFloat(float64(t.Unix())))
			return f.Text('f', -1)
		case TimeUnix:
			return d.Unix()
		case TimeUnixNano, TimeUnixMicro, TimeUnixMilli:
			t := d.Truncate(time.Second)
			sub := d.Sub(t)
			b := big.NewInt(t.Unix())
			b = b.Mul(b, big.NewInt(int64(time.Second)))
			b = b.Add(b, big.NewInt(int64(sub)))
			switch v.Options.TimeFormat {
			case TimeUnixMicro:
				b = b.Quo(b, big.NewInt(int64(time.Microsecond)))
			case TimeUnixMilli:
				b = b.Quo(b, big.NewInt(int64(time.Millisecond)))
			}
			return b.String()
		case TimeCustom:
			return d.Format(v.Options.TimeLayout)
		}
	}
	return v.Dest
}

func (v *Value) parseFloat(src string) bool {
	b, ok := new(big.Int).SetString(src, 0)
	if ok {