sql

README

SQLFlow

Build, Run, Test

Before running the unit tests, we need to build and run a Docker container that hosts a MySQL database following this guide.

To build the parser using goyacc and run all unit tests, use the following command:

go get -d ./... && goyacc -p sql -o parser.go sql.y && go test -v

Documentation

Index

• Constants
• Variables
• func InferFeatureColumns(slct *standardSelect, parsedFeatureColumns map[string][]columns.FeatureColumn, ...) (FeatureColumnMap, ColumnSpecMap, error)
• func Pipe() (*PipeReader, *PipeWriter)
• func SplitMultipleSQL(statements string) ([]string, error)
• func ValidateAttributes(attrs map[string]*attribute) error
• type ColumnSpecMap
• type DB
  • func NewDB(datasource string) (*DB, error)
• type Estimator
• type FeatureColumnMap
• type FeatureMeta
• type PipeReader
  • func Run(slct string, db *DB, modelDir string, session *pb.Session) *PipeReader
  • func (r *PipeReader) Close()
  • func (r *PipeReader) ReadAll() chan interface{}
• type PipeWriter
  • func (w *PipeWriter) Close()
  • func (w *PipeWriter) Write(item interface{}) error

Constants

const ANALYZE = 57352

const AND = 57363

const AS = 57359

const COLUMN = 57354

const FOR = 57358

const FROM = 57347

const GE = 57365

const IDENT = 57360

const INTO = 57357

const LABEL = 57355

const LE = 57366

const LIMIT = 57349

const NE = 57367

const NOT = 57368

const NUMBER = 57361

const OR = 57364

const POWER = 57369

const PREDICT = 57351

const SELECT = 57346

const STRING = 57362

const TRAIN = 57350

const UMINUS = 57370

const USING = 57356

const WHERE = 57348

const WITH = 57353

Variables

var ErrClosedPipe = errors.New("pipe: write on closed pipe")

ErrClosedPipe will occure when manipulating an already closed pipe

Functions

func InferFeatureColumns

func InferFeatureColumns(slct *standardSelect,
	parsedFeatureColumns map[string][]columns.FeatureColumn,
	parsedColumnSpecs map[string][]*columns.ColumnSpec,
	connConfig *connectionConfig) (FeatureColumnMap, ColumnSpecMap, error)

InferFeatureColumns fill up featureColumn and columnSpec structs for all fields.

func Pipe

func Pipe() (*PipeReader, *PipeWriter)

Pipe creates a synchronous in-memory pipe.

It is safe to call Read and Write in parallel with each other or with Close. Parallel calls to Read and parallel calls to Write are also safe: the individual calls will be gated sequentially.

func SplitMultipleSQL

func SplitMultipleSQL(statements string) ([]string, error)

SplitMultipleSQL returns a list of SQL statements if the input statements contains mutiple SQL statements separated by ;

func ValidateAttributes

func ValidateAttributes(attrs map[string]*attribute) error

ValidateAttributes validates the attributes are legal.

Types

type ColumnSpecMap

type ColumnSpecMap map[string]*columns.ColumnSpec

ColumnSpecMap is a mappign from column name to ColumnSpec struct

type DB

type DB struct {
	*sql.DB
	// contains filtered or unexported fields
}

DB extends sql.DB

func NewDB

func NewDB(datasource string) (*DB, error)

NewDB returns a DB object with verifying the datasource name.

type Estimator

type Estimator struct {
	IsTrain              bool
	TrainingDatasetSQL   string // IsTrain == true
	ValidationDatasetSQL string // IsTrain == true
	PredictionDatasetSQL string // IsTrain != true
	X                    []*FeatureMeta
	Y                    *FeatureMeta
	TableName            string
	// contains filtered or unexported fields
}

Estimator describes estimator meta data

type FeatureColumnMap

type FeatureColumnMap map[string]map[string]columns.FeatureColumn

FeatureColumnMap is like: target -> key -> FeatureColumn

type FeatureMeta

type FeatureMeta struct {
	FeatureName string
	Dtype       string
	Delimiter   string
	InputShape  string
	IsSparse    bool
}

FeatureMeta describes feature column meta data

type PipeReader

type PipeReader struct {
	// contains filtered or unexported fields
}

PipeReader reads real data

func Run

func Run(slct string, db *DB, modelDir string, session *pb.Session) *PipeReader

Run executes a SQL query and returns a stream of rows or messages

func (*PipeReader) Close

func (r *PipeReader) Close()

Close closes the reader; subsequent writes to the

func (*PipeReader) ReadAll

func (r *PipeReader) ReadAll() chan interface{}

ReadAll returns the data chan. The caller should use it as `for r := range pr.ReadAll()`

type PipeWriter

type PipeWriter struct {
	// contains filtered or unexported fields
}

PipeWriter writes real data

func (*PipeWriter) Close

func (w *PipeWriter) Close()

Close closes the writer; subsequent ReadAll from the read half of the pipe will return a closed channel.

func (*PipeWriter) Write

func (w *PipeWriter) Write(item interface{}) error

Write writes the item to the underlying data stream. It returns ErrClosedPipe when the data stream is closed.