scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

sql3 changes (#2211)

Browse source 
* first cut of working (slowly) bulk insert; table valued functions and a tuple data type to support time quantums

* oversight

* filter pushdown implementation; bulk insert

* addressed some linter issues
This commit is contained in:
pokeeffe-molecula 2022-09-06 11:16:11 -05:00 • committed by Christopher Lowenthal
parent 27365a5903
commit fcce9d05a9
47 changed files with 2883 additions and 373 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api.go
View file

@ -1793,6 +1793,8 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
subreq.Values = req.Values[start:i]
} else if req.FloatValues != nil {
subreq.FloatValues = req.FloatValues[start:i]
} else if req.TimestampValues != nil {
subreq.TimestampValues = req.TimestampValues[start:i]
}
guard <- struct{}{} // would block if guard channel is already filled
eg.Go(func() error {

28
ctl/cli.go
View file

@ -184,12 +184,26 @@ type response struct {
ExecutionTime int64 `json:"exec_time"`
}
func (r *response) WriteWarnings(w io.Writer) error {
if len(r.Warnings) > 0 {
if _, err := w.Write([]byte("\n")); err != nil {
return errors.Wrapf(err, "writing warning: %s", r.Error)
}
for _, warning := range r.Warnings {
if _, err := w.Write([]byte("Warning: " + warning + "\n")); err != nil {
return errors.Wrapf(err, "writing warning: %s", r.Error)
}
}
}
return nil
}
func (r *response) WriteOut(w io.Writer) error {
if r.Error != "" {
if _, err := w.Write([]byte("Error: " + r.Error + "\n")); err != nil {
return errors.Wrapf(err, "writing error: %s", r.Error)
}
return nil
return r.WriteWarnings(w)
}
t := table.NewWriter()
@ -211,15 +225,9 @@ func (r *response) WriteOut(w io.Writer) error {
}
t.Render()
if len(r.Warnings) > 0 {
if _, err := w.Write([]byte("\n")); err != nil {
return errors.Wrapf(err, "writing warning: %s", r.Error)
}
for _, warning := range r.Warnings {
if _, err := w.Write([]byte("Warning: " + warning + "\n")); err != nil {
return errors.Wrapf(err, "writing warning: %s", r.Error)
}
}
err := r.WriteWarnings(w)
if err != nil {
return err
}
lifeAffirmingMessage := ""
if r.ExecutionTime < 1000000 {

8
sql3/errors.go
View file

@ -42,6 +42,7 @@ const (
ErrIntegerLiteral errors.Code = "ErrIntegerLiteral"
ErrStringLiteral errors.Code = "ErrStringLiteral"
ErrLiteralEmptySetNotAllowed errors.Code = "ErrLiteralEmptySetNotAllowed"
ErrLiteralEmptyTupleNotAllowed errors.Code = "ErrLiteralEmptyTupleNotAllowed"
ErrSetLiteralMustContainIntOrString errors.Code = "ErrSetLiteralMustContainIntOrString"
ErrTypeAssignmentIncompatible errors.Code = "ErrTypeAssignmentIncompatible"
@ -179,6 +180,13 @@ func NewErrSetLiteralMustContainIntOrString(line, col int) error {
)
}
func NewErrLiteralEmptyTupleNotAllowed(line, col int) error {
return errors.New(
ErrLiteralEmptyTupleNotAllowed,
fmt.Sprintf("[%d:%d] tuple literal must contain at least one member", line, col),
)
}
func NewErrTypeIncompatibleWithBitwiseOperator(line, col int, operator, type1 string) error {
return errors.New(
ErrTypeIncompatibleWithBitwiseOperator,

194
sql3/parser/ast.go
View file

@ -77,8 +77,10 @@ func (*SavepointStatement) node() {}
func (*SelectStatement) node() {}
func (*ShardWidthOption) node() {}
func (*StringLit) node() {}
func (*TableValuedFunction) node() {}
func (*TimeUnitConstraint) node() {}
func (*TimeQuantumConstraint) node() {}
func (*TupleLiteralExpr) node() {}
func (*Type) node() {}
func (*UnaryExpr) node() {}
func (*UniqueConstraint) node() {}
@ -204,26 +206,27 @@ type Expr interface {
Pos() Pos
}
func (*BinaryExpr) expr() {}
func (*BoolLit) expr() {}
func (*Call) expr() {}
func (*CaseExpr) expr() {}
func (*CaseBlock) expr() {}
func (*CastExpr) expr() {}
func (*DateLit) expr() {}
func (*Exists) expr() {}
func (*ExprList) expr() {}
func (*Ident) expr() {}
func (*NullLit) expr() {}
func (*IntegerLit) expr() {}
func (*FloatLit) expr() {}
func (*ParenExpr) expr() {}
func (*SetLiteralExpr) expr() {}
func (*QualifiedRef) expr() {}
func (*Range) expr() {}
func (*StringLit) expr() {}
func (*UnaryExpr) expr() {}
func (*SelectStatement) expr() {}
func (*BinaryExpr) expr() {}
func (*BoolLit) expr() {}
func (*Call) expr() {}
func (*CaseExpr) expr() {}
func (*CaseBlock) expr() {}
func (*CastExpr) expr() {}
func (*DateLit) expr() {}
func (*Exists) expr() {}
func (*ExprList) expr() {}
func (*Ident) expr() {}
func (*NullLit) expr() {}
func (*IntegerLit) expr() {}
func (*FloatLit) expr() {}
func (*ParenExpr) expr() {}
func (*SetLiteralExpr) expr() {}
func (*TupleLiteralExpr) expr() {}
func (*QualifiedRef) expr() {}
func (*Range) expr() {}
func (*StringLit) expr() {}
func (*UnaryExpr) expr() {}
func (*SelectStatement) expr() {}
// CloneExpr returns a deep copy expr.
func CloneExpr(expr Expr) Expr {
@ -343,10 +346,11 @@ type Source interface {
OutputColumnQualifierNamed(qualifier string, name string) (*SourceOutputColumn, error)
}
func (*JoinClause) source() {}
func (*ParenSource) source() {}
func (*QualifiedTableName) source() {}
func (*SelectStatement) source() {}
func (*JoinClause) source() {}
func (*ParenSource) source() {}
func (*QualifiedTableName) source() {}
func (*TableValuedFunction) source() {}
func (*SelectStatement) source() {}
// CloneSource returns a deep copy src.
func CloneSource(src Source) Source {
@ -2715,15 +2719,30 @@ func (s *DropTriggerStatement) String() string {
return buf.String()
}
type BulkInsertIDMap struct {
Auto Pos // position of AUTO
ColumnExprs *ExprList
}
type BulkInsertMappedColumn struct {
SourceColumnOffset Expr
TargetColumn *Ident
}
type BulkInsertStatement struct {
Bulk Pos // position of BULK keyword
Insert Pos // position of INSERT keyword
Table *Ident // table name
From Pos // position of FROM keyword
DataFile Expr // data file name
With Pos // position of WITH keyword
From Pos // position of FROM keyword
DataFile Expr // data file name
With Pos // position of WITH keyword
BatchSize Expr
RowsLimit Expr
Format Expr
MapId *BulkInsertIDMap
ColumnMap []*BulkInsertMappedColumn
}
func (s *BulkInsertStatement) String() string {
@ -3520,6 +3539,79 @@ func (c *QualifiedTableName) OutputColumnQualifierNamed(qualifier string, name s
return nil, nil
}
type TableValuedFunction struct {
Name *Ident // table name
As Pos // position of AS keyword
Alias *Ident // optional table alias
Call *Call // call
OutputColumns []*SourceOutputColumn // output columns - populated during analysis
}
// TableName returns the name used to identify n.
// Returns the alias, if one is specified. Otherwise returns the name.
func (n *TableValuedFunction) TableName() string {
if s := IdentName(n.Alias); s != "" {
return s
}
return IdentName(n.Name)
}
func (n *TableValuedFunction) MatchesTablenameOrAlias(match string) bool {
return strings.EqualFold(IdentName(n.Alias), match) || strings.EqualFold(IdentName(n.Name), match)
}
// Clone returns a deep copy of n.
func (n *TableValuedFunction) Clone() *TableValuedFunction {
if n == nil {
return nil
}
other := *n
other.Name = n.Name.Clone()
other.Alias = n.Alias.Clone()
return &other
}
// String returns the string representation of the table name.
func (n *TableValuedFunction) String() string {
var buf bytes.Buffer
buf.WriteString(n.Name.String())
if n.Alias != nil {
fmt.Fprintf(&buf, " AS %s", n.Alias.String())
}
return buf.String()
}
func (c *TableValuedFunction) SourceFromAlias(alias string) Source {
if strings.EqualFold(IdentName(c.Alias), alias) {
return c
}
if strings.EqualFold(IdentName(c.Name), alias) {
return c
}
return nil
}
func (c *TableValuedFunction) PossibleOutputColumns() []*SourceOutputColumn {
return c.OutputColumns
}
func (c *TableValuedFunction) OutputColumnNamed(name string) (*SourceOutputColumn, error) {
for _, oc := range c.OutputColumns {
if strings.EqualFold(oc.ColumnName, name) {
return oc, nil
}
}
return nil, nil
}
func (c *TableValuedFunction) OutputColumnQualifierNamed(qualifier string, name string) (*SourceOutputColumn, error) {
if strings.EqualFold(IdentName(c.Alias), qualifier) || strings.EqualFold(IdentName(c.Name), qualifier) {
return c.OutputColumnNamed(name)
}
return nil, nil
}
type ParenSource struct {
Lparen Pos // position of left paren
X Source // nested source
@ -3911,6 +4003,54 @@ func (expr *SetLiteralExpr) String() string {
return buf.String()
}
type TupleLiteralExpr struct {
Lbrace Pos // position of left brace
Members []Expr // bracketed expression
Rbrace Pos // position of right brace
ResultDataType ExprDataType
}
func (expr *TupleLiteralExpr) IsLiteral() bool {
return true
}
func (expr *TupleLiteralExpr) DataType() ExprDataType {
return expr.ResultDataType
}
func (expr *TupleLiteralExpr) Pos() Pos {
return expr.Lbrace
}
// Clone returns a deep copy of expr.
func (expr *TupleLiteralExpr) Clone() *TupleLiteralExpr {
if expr == nil {
return nil
}
other := *expr
other.Members = cloneExprs(expr.Members)
return &other
}
// String returns the string representation of the expression.
func (expr *TupleLiteralExpr) String() string {
var buf bytes.Buffer
if len(expr.Members) != 0 {
buf.WriteString("{")
for i, col := range expr.Members {
if i != 0 {
buf.WriteString(", ")
}
buf.WriteString(col.String())
}
buf.WriteString("}")
}
return buf.String()
}
type Window struct {
Name *Ident // name of window
As Pos // position of AS keyword

111
sql3/parser/astdatatype.go
View file

@ -9,14 +9,16 @@ import (
)
const (
FieldTypeBool = "BOOL"
FieldTypeDecimal = "DECIMAL"
FieldTypeID = "ID"
FieldTypeIDSet = "IDSET"
FieldTypeInt = "INT"
FieldTypeString = "STRING"
FieldTypeStringSet = "STRINGSET"
FieldTypeTimestamp = "TIMESTAMP"
FieldTypeBool = "BOOL"
FieldTypeDecimal = "DECIMAL"
FieldTypeID = "ID"
FieldTypeIDSet = "IDSET"
FieldTypeIDSetQuantum = "IDSETQ"
FieldTypeInt = "INT"
FieldTypeString = "STRING"
FieldTypeStringSet = "STRINGSET"
FieldTypeStringSetQuantum = "STRINGSETQ"
FieldTypeTimestamp = "TIMESTAMP"
)
func IsValidTypeName(typeName string) bool {
@ -40,16 +42,20 @@ type ExprDataType interface {
TypeName() string
}
func (*DataTypeVoid) exprDataType() {}
func (*DataTypeRange) exprDataType() {}
func (*DataTypeBool) exprDataType() {}
func (*DataTypeDecimal) exprDataType() {}
func (*DataTypeID) exprDataType() {}
func (*DataTypeIDSet) exprDataType() {}
func (*DataTypeInt) exprDataType() {}
func (*DataTypeString) exprDataType() {}
func (*DataTypeStringSet) exprDataType() {}
func (*DataTypeTimestamp) exprDataType() {}
func (*DataTypeVoid) exprDataType() {}
func (*DataTypeRange) exprDataType() {}
func (*DataTypeTuple) exprDataType() {}
func (*DataTypeSubtable) exprDataType() {}
func (*DataTypeBool) exprDataType() {}
func (*DataTypeDecimal) exprDataType() {}
func (*DataTypeID) exprDataType() {}
func (*DataTypeIDSet) exprDataType() {}
func (*DataTypeIDSetQuantum) exprDataType() {}
func (*DataTypeInt) exprDataType() {}
func (*DataTypeString) exprDataType() {}
func (*DataTypeStringSet) exprDataType() {}
func (*DataTypeStringSetQuantum) exprDataType() {}
func (*DataTypeTimestamp) exprDataType() {}
type DataTypeVoid struct {
}
@ -76,6 +82,53 @@ func (dt *DataTypeRange) TypeName() string {
return fmt.Sprintf("RANGE(%s)", dt.SubscriptType.TypeName())
}
type DataTypeTuple struct {
Members []ExprDataType
}
func NewDataTypeTuple(members []ExprDataType) *DataTypeTuple {
return &DataTypeTuple{
Members: members,
}
}
func (dt *DataTypeTuple) TypeName() string {
ms := ""
for idx, m := range dt.Members {
ms = ms + m.TypeName()
if idx+1 < len(dt.Members) {
ms = ms + ", "
}
}
return fmt.Sprintf("TUPLE(%s)", ms)
}
type SubtableColumn struct {
Name string
DataType ExprDataType
}
type DataTypeSubtable struct {
Columns []*SubtableColumn
}
func NewDataTypeSubtable(columns []*SubtableColumn) *DataTypeSubtable {
return &DataTypeSubtable{
Columns: columns,
}
}
func (dt *DataTypeSubtable) TypeName() string {
ms := ""
for idx, m := range dt.Columns {
ms = ms + m.DataType.TypeName()
if idx+1 < len(dt.Columns) {
ms = ms + ", "
}
}
return fmt.Sprintf("SUBTABLE(%s)", ms)
}
type DataTypeBool struct {
}
@ -123,6 +176,17 @@ func (*DataTypeIDSet) TypeName() string {
return FieldTypeIDSet
}
type DataTypeIDSetQuantum struct {
}
func NewDataTypeIDSetQuantum() *DataTypeIDSetQuantum {
return &DataTypeIDSetQuantum{}
}
func (*DataTypeIDSetQuantum) TypeName() string {
return FieldTypeIDSetQuantum
}
type DataTypeInt struct {
}
@ -156,6 +220,17 @@ func (*DataTypeStringSet) TypeName() string {
return FieldTypeStringSet
}
type DataTypeStringSetQuantum struct {
}
func NewDataTypeStringSetQuantum() *DataTypeStringSetQuantum {
return &DataTypeStringSetQuantum{}
}
func (*DataTypeStringSetQuantum) TypeName() string {
return FieldTypeStringSetQuantum
}
type DataTypeTimestamp struct {
}

201
sql3/parser/parser.go
View file

@ -1404,9 +1404,114 @@ func (p *Parser) parseBulkInsertStatement() (_ *BulkInsertStatement, err error)
return nil, p.errorExpected(p.pos, p.tok, "literal")
}
if p.peek() == WITH {
stmt.With, _, _ = p.scan()
if !isBulkInsertOptionStartToken(p.peek(), p) {
return nil, p.errorExpected(p.pos, p.tok, "BATCHSIZE, ROWSLIMIT, FORMAT or MAP")
}
for {
err := p.parseBulkInsertOption(&stmt)
if err != nil {
return nil, err
}
if !isBulkInsertOptionStartToken(p.peek(), p) {
break
}
}
}
return &stmt, nil
}
func (p *Parser) parseBulkInsertOption(stmt *BulkInsertStatement) error {
switch p.peek() {
case IDENT:
ident, err := p.parseIdent("bulk insert option")
if err != nil {
return err
}
switch strings.ToUpper(ident.Name) {
case "BATCHSIZE":
if isLiteralToken(p.peek()) {
stmt.BatchSize = p.mustParseLiteral()
return nil
} else {
return p.errorExpected(p.pos, p.tok, "literal")
}
case "ROWSLIMIT":
if isLiteralToken(p.peek()) {
stmt.RowsLimit = p.mustParseLiteral()
return nil
} else {
return p.errorExpected(p.pos, p.tok, "literal")
}
case "FORMAT":
if isLiteralToken(p.peek()) {
stmt.Format = p.mustParseLiteral()
return nil
} else {
return p.errorExpected(p.pos, p.tok, "literal")
}
case "MAP":
if p.peek() == IDENT {
ident, err := p.parseIdent("bulk insert option")
if err != nil {
return err
}
switch strings.ToUpper(ident.Name) {
case "_ID":
stmt.MapId = &BulkInsertIDMap{}
if p.peek() != TO {
return p.errorExpected(p.pos, p.tok, "TO")
}
_, _, _ = p.scan()
if p.peek() == AUTOINCREMENT {
stmt.MapId.Auto, _, _ = p.scan()
return nil
}
stmt.MapId.ColumnExprs, err = p.parseExprList()
if err != nil {
return err
}
return nil
case "OFFSET":
columnMapItem := &BulkInsertMappedColumn{}
if isLiteralToken(p.peek()) {
columnMapItem.SourceColumnOffset = p.mustParseLiteral()
} else {
return p.errorExpected(p.pos, p.tok, "literal")
}
if p.peek() != TO {
return p.errorExpected(p.pos, p.tok, "TO")
}
_, _, _ = p.scan()
if p.peek() != IDENT {
return p.errorExpected(p.pos, p.tok, "IDENTIFIER")
}
columnMapItem.TargetColumn, err = p.parseIdent("bulk insert map offset option")
if err != nil {
return err
}
if stmt.ColumnMap == nil {
stmt.ColumnMap = []*BulkInsertMappedColumn{}
}
stmt.ColumnMap = append(stmt.ColumnMap, columnMapItem)
return nil
}
}
return p.errorExpected(p.pos, p.tok, "_ID or OFFSET")
}
}
return p.errorExpected(p.pos, p.tok, "BATCHSIZE, ROWSLIMIT, FORMAT or MAP")
}
func (p *Parser) parseInsertStatement(withClause *WithClause) (_ *InsertStatement, err error) {
if pk := p.peek(); pk != INSERT && pk != REPLACE {
return nil, p.errorExpected(p.pos, p.tok, "INSERT or REPLACE")
@ -1656,7 +1761,12 @@ func (p *Parser) parseUpdateStatement(withClause *WithClause) (_ *UpdateStatemen
}
}
if stmt.Table, err = p.parseQualifiedTableName(); err != nil {
ident, err := p.parseIdent("table name")
if err != nil {
return &stmt, err
}
stmt.Table, err = p.parseQualifiedTableName(ident)
if err != nil {
return &stmt, err
}
@ -1702,7 +1812,13 @@ func (p *Parser) parseDeleteStatement(withClause *WithClause) (_ *DeleteStatemen
return &stmt, p.errorExpected(p.pos, p.tok, "FROM")
}
stmt.From, _, _ = p.scan()
if stmt.Table, err = p.parseQualifiedTableName(); err != nil {
ident, err := p.parseIdent("table name")
if err != nil {
return &stmt, err
}
stmt.Table, err = p.parseQualifiedTableName(ident)
if err != nil {
return &stmt, err
}
@ -2089,7 +2205,14 @@ func (p *Parser) parseUnarySource() (source Source, err error) {
case LP:
return p.parseParenSource()
case IDENT, QIDENT:
return p.parseQualifiedTableName()
ident, err := p.parseIdent("table or function")
if err != nil {
return nil, err
}
if p.peek() == LP {
return p.parseTableValuedFunction(ident)
}
return p.parseQualifiedTableName(ident)
default:
return nil, p.errorExpected(p.pos, p.tok, "table name or left paren")
}
@ -2216,13 +2339,10 @@ func (p *Parser) parseParenSource() (_ *ParenSource, err error) {
return &source, nil
}
func (p *Parser) parseQualifiedTableName() (_ *QualifiedTableName, err error) {
func (p *Parser) parseQualifiedTableName(ident *Ident) (_ *QualifiedTableName, err error) {
var tbl QualifiedTableName
if !isIdentToken(p.peek()) {
return &tbl, p.errorExpected(p.pos, p.tok, "table name")
}
tbl.Name, _ = p.parseIdent("table name")
tbl.Name = ident
// Parse optional table alias ("AS alias" or just "alias").
if tok := p.peek(); tok == AS || isIdentToken(tok) {
@ -2257,6 +2377,28 @@ func (p *Parser) parseQualifiedTableName() (_ *QualifiedTableName, err error) {
return &tbl, nil
}
func (p *Parser) parseTableValuedFunction(ident *Ident) (_ *TableValuedFunction, err error) {
var tbl TableValuedFunction
tbl.Name = ident
tbl.Call, err = p.parseCall(ident)
if err != nil {
return &tbl, err
}
// Parse optional table alias ("AS alias" or just "alias").
if tok := p.peek(); tok == AS || isIdentToken(tok) {
if p.peek() == AS {
tbl.As, _, _ = p.scan()
}
if tbl.Alias, err = p.parseIdent("table alias"); err != nil {
return &tbl, err
}
}
return &tbl, nil
}
/*func (p *Parser) parseWithClause() (*WithClause, error) {
assert(p.peek() == WITH)
@ -2397,6 +2539,9 @@ func (p *Parser) parseOperand() (expr Expr, err error) {
case LB:
p.unscan()
return p.parseSetLiteralExpr()
case LBR:
p.unscan()
return p.parseTupleLiteralExpr()
case CAST:
p.unscan()
return p.parseCastExpr()
@ -2860,6 +3005,29 @@ func (p *Parser) parseSetLiteralExpr() (_ *SetLiteralExpr, err error) {
return &expr, nil
}
func (p *Parser) parseTupleLiteralExpr() (_ *TupleLiteralExpr, err error) {
var expr TupleLiteralExpr
expr.Lbrace, _, _ = p.scan()
for p.peek() != RBR {
x, err := p.ParseExpr()
if err != nil {
return &expr, err
}
expr.Members = append(expr.Members, x)
if p.peek() == RBR {
break
} else if p.peek() != COMMA {
return &expr, p.errorExpected(p.pos, p.tok, "comma or right brace")
}
p.scan()
}
expr.Rbrace, _, _ = p.scan()
return &expr, nil
}
func (p *Parser) parseCastExpr() (_ *CastExpr, err error) {
assert(p.peek() == CAST)
@ -3175,6 +3343,23 @@ func isTableOptionStartToken(tok Token) bool {
}
}
// isBulkInsertOptionStartToken returns true if tok is the initial token of a bulk insert option.
func isBulkInsertOptionStartToken(tok Token, p *Parser) bool {
switch tok {
case IDENT:
ident, err := p.parseIdent("bulk insert option")
defer p.unscan()
if err != nil {
return false
}
switch strings.ToUpper(ident.Name) {
case "BATCHSIZE", "ROWSLIMIT", "FORMAT", "MAP":
return true
}
}
return false
}
// isConstraintStartToken returns true if tok is the initial token of a constraint.
func isConstraintStartToken(tok Token, isTable bool) bool {
switch tok {

4
sql3/parser/scanner.go
View file

@ -54,6 +54,10 @@ func (s *Scanner) Scan() (pos Pos, token Token, lit string) {
return pos, LB, "["
case ']':
return pos, RB, "]"
case '{':
return pos, LBR, "{"
case '}':
return pos, RBR, "}"
case ',':
return pos, COMMA, ","
case '!':

4
sql3/parser/token.go
View file

@ -49,6 +49,8 @@ const (
RP // )
LB // [
RB // ]
LBR // {
RBR // }
COMMA // ,
NE // !=
EQ // =
@ -267,6 +269,8 @@ var tokens = [...]string{
RP: ")",
LB: "[",
RB: "]",
LBR: "{",
RBR: "}",
COMMA: ",",
NE: "!=",
EQ: "=",

188
sql3/planner/compilebulkinsert.go
View file

@ -4,6 +4,9 @@ package planner
import (
"context"
"os"
"strconv"
"strings"
pilosa "github.com/featurebasedb/featurebase/v3"
"github.com/featurebasedb/featurebase/v3/sql3"
@ -17,8 +20,7 @@ import (
func (p *ExecutionPlanner) compileBulkInsertStatement(stmt *parser.BulkInsertStatement) (_ types.PlanOperator, err error) {
tableName := parser.IdentName(stmt.Table)
/*table*/
_, err = p.schemaAPI.IndexInfo(context.Background(), tableName)
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return nil, sql3.NewErrTableNotFound(stmt.Table.NamePos.Line, stmt.Table.NamePos.Column, tableName)
@ -26,7 +28,105 @@ func (p *ExecutionPlanner) compileBulkInsertStatement(stmt *parser.BulkInsertSta
return nil, err
}
return NewPlanOpBulkInsert(p, tableName), nil
err = p.checkAccess(context.Background(), tableName, accessTypeWriteData)
if err != nil {
return nil, err
}
options := &bulkInsertOptions{
format: "CSV", //only format supported right now
}
sliteral, sok := stmt.DataFile.(*parser.StringLit)
if !sok {
return nil, sql3.NewErrInternalf("unexpected file name type '%T'", stmt.DataFile)
}
options.fileName = sliteral.Value
//file should exist
if _, err := os.Stat(options.fileName); errors.Is(err, os.ErrNotExist) {
// TODO (pok) need proper error
return nil, sql3.NewErrInternalf("file '%s' does not exist", stmt.DataFile)
}
literal, ok := stmt.BatchSize.(*parser.IntegerLit)
if !ok {
return nil, sql3.NewErrInternalf("unexpected batch size type '%T'", stmt.BatchSize)
}
i, err := strconv.ParseInt(literal.Value, 10, 64)
if err != nil {
return nil, err
}
options.batchSize = int(i)
literal, ok = stmt.RowsLimit.(*parser.IntegerLit)
if !ok {
return nil, sql3.NewErrInternalf("unexpected rowslimit type '%T'", stmt.RowsLimit)
}
i, err = strconv.ParseInt(literal.Value, 10, 64)
if err != nil {
return nil, err
}
options.rowsLimit = int(i)
options.idColumnMap = make([]interface{}, 0)
if stmt.MapId.ColumnExprs != nil {
for _, m := range stmt.MapId.ColumnExprs.Exprs {
literal, ok = m.(*parser.IntegerLit)
if !ok {
return nil, sql3.NewErrInternalf("unexpected id map expr type '%T'", m)
}
i, err = strconv.ParseInt(literal.Value, 10, 64)
if err != nil {
return nil, err
}
options.idColumnMap = append(options.idColumnMap, i)
}
}
if stmt.ColumnMap != nil {
options.columnMap = make([]*bulkInsertMappedColumn, 0)
for _, m := range stmt.ColumnMap {
literal, ok = m.SourceColumnOffset.(*parser.IntegerLit)
if !ok {
return nil, sql3.NewErrInternalf("unexpected column map expr type '%T'", m)
}
i, err = strconv.ParseInt(literal.Value, 10, 64)
if err != nil {
return nil, err
}
for _, fld := range table.Fields {
if strings.EqualFold(fld.Name, m.TargetColumn.Name) {
cm := &bulkInsertMappedColumn{
columnSource: i,
columnName: m.TargetColumn.Name,
columnDataType: fieldSQLDataType(fld),
}
options.columnMap = append(options.columnMap, cm)
break
}
}
}
} else {
options.columnMap = make([]*bulkInsertMappedColumn, 0)
//handle the case of a default mapping based on the table
i := 0
for _, fld := range table.Fields {
if strings.EqualFold(fld.Name, "_id") {
continue
}
cm := &bulkInsertMappedColumn{
columnSource: i,
columnName: fld.Name,
columnDataType: fieldSQLDataType(fld),
}
options.columnMap = append(options.columnMap, cm)
i += 1
}
}
return NewPlanOpBulkInsert(p, tableName, table.Options.Keys, options), nil
}
// analyzeBulkInsertStatement analyzes a BULK INSERT statement and returns an
@ -34,7 +134,7 @@ func (p *ExecutionPlanner) compileBulkInsertStatement(stmt *parser.BulkInsertSta
func (p *ExecutionPlanner) analyzeBulkInsertStatement(stmt *parser.BulkInsertStatement) error {
//check referred to table exists
tableName := parser.IdentName(stmt.Table)
/*table*/ _, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return sql3.NewErrTableNotFound(stmt.Table.NamePos.Line, stmt.Table.NamePos.Column, tableName)
@ -42,5 +142,85 @@ func (p *ExecutionPlanner) analyzeBulkInsertStatement(stmt *parser.BulkInsertSta
return err
}
// check filename
// file should be literal and a string
if !(stmt.DataFile.IsLiteral() && typeIsString(stmt.DataFile.DataType())) {
return sql3.NewErrStringLiteral(stmt.DataFile.Pos().Line, stmt.DataFile.Pos().Column)
}
// check options
// batch size should default to 1000
if stmt.BatchSize == nil {
stmt.BatchSize = &parser.IntegerLit{
Value: "1000",
}
}
// batch size should be literal and an int
if !(stmt.BatchSize.IsLiteral() && typeIsInteger(stmt.BatchSize.DataType())) {
return sql3.NewErrIntegerLiteral(stmt.BatchSize.Pos().Line, stmt.BatchSize.Pos().Column)
}
// rowslimit should default to 0
if stmt.RowsLimit == nil {
stmt.RowsLimit = &parser.IntegerLit{
Value: "0",
}
}
// rowslimit should be literal and an int
if !(stmt.RowsLimit.IsLiteral() && typeIsInteger(stmt.RowsLimit.DataType())) {
return sql3.NewErrIntegerLiteral(stmt.RowsLimit.Pos().Line, stmt.RowsLimit.Pos().Column)
}
// format should default to CSV
if stmt.Format == nil {
stmt.Format = &parser.StringLit{
Value: "CSV",
}
}
// format should be literal and a string
if !(stmt.Format.IsLiteral() && typeIsString(stmt.Format.DataType())) {
return sql3.NewErrStringLiteral(stmt.Format.Pos().Line, stmt.Format.Pos().Column)
}
//CSV is the only format supported right now
format, ok := stmt.Format.(*parser.StringLit)
if !ok {
return sql3.NewErrInternalf("unexpected format type '%T'", stmt.Format)
}
if !strings.EqualFold(format.Value, "CSV") {
//TODO (pok) - proper error needed here
return sql3.NewErrInternalf("unexpected format '%s'", format.Value)
}
// if we have an id map, check expressions are literals and ints
if stmt.MapId.ColumnExprs != nil {
for _, im := range stmt.MapId.ColumnExprs.Exprs {
if !(im.IsLiteral() && typeIsInteger(im.DataType())) {
return sql3.NewErrIntegerLiteral(im.Pos().Line, im.Pos().Column)
}
}
}
//if we have a column map, check offset expressions and target column names
if stmt.ColumnMap != nil {
for _, cm := range stmt.ColumnMap {
if !(cm.SourceColumnOffset.IsLiteral() && typeIsInteger(cm.SourceColumnOffset.DataType())) {
return sql3.NewErrIntegerLiteral(cm.SourceColumnOffset.Pos().Line, cm.SourceColumnOffset.Pos().Column)
}
found := false
for _, fld := range table.Fields {
if strings.EqualFold(cm.TargetColumn.Name, fld.Name) {
found = true
break
}
}
if !found {
return sql3.NewErrColumnNotFound(cm.TargetColumn.NamePos.Line, cm.TargetColumn.NamePos.Line, cm.TargetColumn.Name)
}
}
}
return nil
}

4
sql3/planner/compileinsert.go
View file

@ -165,10 +165,6 @@ func (p *ExecutionPlanner) analyzeInsertStatement(stmt *parser.InsertStatement)
return err
}
if !e.IsLiteral() {
return sql3.NewErrLiteralExpected(expr.Pos().Line, expr.Pos().Column)
}
// Type check against same ordinal position in column type list.
if !typesAreAssignmentCompatible(typeNames[i], e.DataType()) {
return sql3.NewErrTypeAssignmentIncompatible(expr.Pos().Line, expr.Pos().Column, e.DataType().TypeName(), typeNames[i].TypeName())

113
sql3/planner/compileselect.go
View file

@ -4,6 +4,7 @@ package planner
import (
"context"
"strings"
pilosa "github.com/featurebasedb/featurebase/v3"
"github.com/featurebasedb/featurebase/v3/sql3"
@ -15,10 +16,9 @@ import (
// compileSelectStatment compiles a parser.SelectStatment AST into a PlanOperator
func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement, isSubquery bool) (types.PlanOperator, error) {
query := NewPlanOpQuery(NewPlanOpNullTable(), p.sql)
//p.pushPlannerScope(query)
p.scopeStack.push(query)
// handle select list
// handle projections
projections := make([]types.PlanExpression, 0)
for _, c := range stmt.Columns {
planExpr, err := p.compileExpr(c.Expr)
@ -52,13 +52,24 @@ func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement,
query.AddWarning("DISTINCT not yet implemented")
}
// source expression last
source, err := p.compileSelectSource(query, stmt.WhereExpr, stmt.Source)
// handle the where clause
where, err := p.compileExpr(stmt.WhereExpr)
if err != nil {
return nil, err
}
//do we have straight projection or a group by?
// source expression
source, err := p.compileSelectSource(query, stmt.Source)
if err != nil {
return nil, err
}
// if we did have a where, insert the filter op
if where != nil {
source = NewPlanOpFilter(p, where, source)
}
// do we have straight projection or a group by?
var compiledOp types.PlanOperator
if len(query.aggregates) > 0 {
compiledOp = NewPlanOpProjection(projections, NewPlanOpGroupBy(query.aggregates, groupByExprs, source))
@ -86,7 +97,7 @@ func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement,
compiledOp = NewPlanOpOrderBy(orderByFields, compiledOp)
}
//insert the top operator if it exists
// insert the top operator if it exists
if stmt.Top.IsValid() {
topExpr, err := p.compileExpr(stmt.TopExpr)
if err != nil {
@ -95,11 +106,10 @@ func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement,
compiledOp = NewPlanOpTop(topExpr, compiledOp)
}
//pop the scope
//p.popPlannerScope()
// pop the scope
_ = p.scopeStack.pop()
//if it is a subquery, don't wrap in a PlanOpQuery
// if it is a subquery, don't wrap in a PlanOpQuery
if isSubquery {
return compiledOp, nil
}
@ -109,18 +119,18 @@ func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement,
return query.WithChildren(children...)
}
func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, whereExpr parser.Expr, source parser.Source) (types.PlanOperator, error) {
func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, source parser.Source) (types.PlanOperator, error) {
if source == nil {
return NewPlanOpNullTable(), nil
}
switch sourceExpr := source.(type) {
case *parser.JoinClause:
topOp, err := p.compileSelectSource(scope, whereExpr, sourceExpr.X)
topOp, err := p.compileSelectSource(scope, sourceExpr.X)
if err != nil {
return nil, err
}
bottomOp, err := p.compileSelectSource(scope, whereExpr, sourceExpr.Y)
bottomOp, err := p.compileSelectSource(scope, sourceExpr.Y)
if err != nil {
return nil, err
}
@ -131,7 +141,7 @@ func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, whereExpr par
return NewPlanOpNestedLoops(topOp, bottomOp), nil
case *parser.QualifiedTableName:
//get all the qualified refs that refer to this table
// get all the qualified refs that refer to this table
extractColumns := []types.PlanExpression{}
for _, r := range scope.referenceList {
@ -140,19 +150,39 @@ func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, whereExpr par
}
}
// handle the where clause
where, err := p.compileExpr(whereExpr)
tableName := parser.IdentName(sourceExpr.Name)
if sourceExpr.Alias != nil {
aliasName := parser.IdentName(sourceExpr.Alias)
return NewPlanOpRelAlias(aliasName, NewPlanOpPQLTableScan(p, tableName, extractColumns)), nil
}
return NewPlanOpPQLTableScan(p, tableName, extractColumns), nil
case *parser.TableValuedFunction:
callExpr, err := p.compileCallExpr(sourceExpr.Call)
if err != nil {
return nil, err
}
//get for the table name
tableName := parser.IdentName(sourceExpr.Name)
if sourceExpr.Alias != nil {
aliasName := parser.IdentName(sourceExpr.Alias)
return NewPlanOpRelAlias(aliasName, NewPlanOpTableValuedFunction(p, callExpr)), nil
}
return NewPlanOpPQLTableScan(p, tableName, extractColumns, where), nil
return NewPlanOpTableValuedFunction(p, callExpr), nil
case *parser.ParenSource:
return p.compileSelectSource(scope, whereExpr, sourceExpr.X)
if sourceExpr.Alias != nil {
aliasName := parser.IdentName(sourceExpr.Alias)
op, err := p.compileSelectSource(scope, sourceExpr.X)
if err != nil {
return nil, err
}
return NewPlanOpRelAlias(aliasName, op), nil
}
return p.compileSelectSource(scope, sourceExpr.X)
case *parser.SelectStatement:
subQuery, err := p.compileSelectStatement(sourceExpr, true)
@ -166,31 +196,31 @@ func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, whereExpr par
}
}
func (p *ExecutionPlanner) analyzeSource(source parser.Source) error {
func (p *ExecutionPlanner) analyzeSource(source parser.Source, scope parser.Statement) error {
if source == nil {
return nil
}
switch source := source.(type) {
case *parser.JoinClause:
err := p.analyzeSource(source.X)
err := p.analyzeSource(source.X, scope)
if err != nil {
return err
}
err = p.analyzeSource(source.Y)
err = p.analyzeSource(source.Y, scope)
if err != nil {
return err
}
return nil
case *parser.ParenSource:
err := p.analyzeSource(source.X)
err := p.analyzeSource(source.X, scope)
if err != nil {
return err
}
return nil
case *parser.QualifiedTableName:
//check table exists
// check table exists
tableName := parser.IdentName(source.Name)
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
if err != nil {
@ -213,6 +243,37 @@ func (p *ExecutionPlanner) analyzeSource(source parser.Source) error {
return nil
case *parser.TableValuedFunction:
//check it actually is a table valued function - we only support one right now; subtable()
switch strings.ToUpper(source.Name.Name) {
case "SUBTABLE":
_, err := p.analyzeCallExpression(source.Call, scope)
if err != nil {
return err
}
tvfResultType, ok := source.Call.ResultDataType.(*parser.DataTypeSubtable)
if !ok {
return sql3.NewErrInternalf("unexepected tvf return type")
}
// populate the output columns from the source
for idx, member := range tvfResultType.Columns {
soc := &parser.SourceOutputColumn{
TableName: "", // TODO (pok) use the tq column actually referenced as the table name
ColumnName: member.Name,
ColumnIndex: idx,
Datatype: member.DataType,
}
source.OutputColumns = append(source.OutputColumns, soc)
}
default:
return sql3.NewErrInternalf("table valued function expected")
}
return nil
case *parser.SelectStatement:
err := p.analyzeSelectStatement(source)
if err != nil {
@ -226,8 +287,8 @@ func (p *ExecutionPlanner) analyzeSource(source parser.Source) error {
}
func (p *ExecutionPlanner) analyzeSelectStatement(stmt *parser.SelectStatement) error {
//analyze source first - needed for name resolution
err := p.analyzeSource(stmt.Source)
// analyze source first - needed for name resolution
err := p.analyzeSource(stmt.Source, stmt)
if err != nil {
return err
}

1
sql3/planner/executionplanner_test.go
View file

@ -589,6 +589,7 @@ func TestPlanner_AlterTable(t *testing.T) {
})
}
func TestPlanner_DropTable(t *testing.T) {
c := test.MustRunCluster(t, 1)
defer c.Close()

119
sql3/planner/expression.go
View file

@ -91,6 +91,31 @@ func coerceValue(sourceType parser.ExprDataType, targetType parser.ExprDataType,
switch targetType.(type) {
case *parser.DataTypeIDSet:
return value, nil
case *parser.DataTypeIDSetQuantum:
return []interface{}{
nil, //no timestamp
value,
}, nil
}
case *parser.DataTypeStringSet:
switch targetType.(type) {
case *parser.DataTypeStringSet:
return value, nil
case *parser.DataTypeStringSetQuantum:
return []interface{}{
nil, //no timestamp
value,
}, nil
}
case *parser.DataTypeTuple:
switch targetType.(type) {
case *parser.DataTypeIDSetQuantum:
return value, nil
case *parser.DataTypeStringSetQuantum:
return value, nil
}
default:
@ -1381,7 +1406,7 @@ func (n *callPlanExpression) WithChildren(children ...types.PlanExpression) (typ
// aliasPlanExpression is a alias ref
type aliasPlanExpression struct {
types.SchemaIdentifiable
types.IdentifiableByName
aliasName string
expr types.PlanExpression
}
@ -1397,12 +1422,12 @@ func (n *aliasPlanExpression) Name() string {
return n.aliasName
}
//evaluates expression based on current row and column
// evaluates expression based on current row and column
func (n *aliasPlanExpression) Evaluate(currentRow []interface{}) (interface{}, error) {
return n.expr.Evaluate(currentRow)
}
//returns the type of the expression
// returns the type of the expression
func (n *aliasPlanExpression) Type() parser.ExprDataType {
return n.expr.Type()
}
@ -1431,7 +1456,7 @@ func (n *aliasPlanExpression) WithChildren(children ...types.PlanExpression) (ty
// qualifiedRefPlanExpression is a qualified ref
type qualifiedRefPlanExpression struct {
types.SchemaIdentifiable
types.IdentifiableByName
tableName string
columnName string
columnIndex int
@ -2049,6 +2074,81 @@ func (n *exprSetLiteralPlanExpression) WithChildren(children ...types.PlanExpres
return newExprSetLiteralPlanExpression(children, n.dataType), nil
}
// exprTupleLiteralPlanExpression is a tuple literal
type exprTupleLiteralPlanExpression struct {
members []types.PlanExpression
dataType parser.ExprDataType
}
func newExprTupleLiteralPlanExpression(members []types.PlanExpression, dataType parser.ExprDataType) *exprTupleLiteralPlanExpression {
return &exprTupleLiteralPlanExpression{
members: members,
dataType: dataType,
}
}
func (n *exprTupleLiteralPlanExpression) Evaluate(currentRow []interface{}) (interface{}, error) {
timestampEval, err := n.members[0].Evaluate(currentRow)
if err != nil {
return nil, err
}
//if it is a string, do a coercion
val, ok := timestampEval.(string)
if ok {
if tm, err := time.ParseInLocation(time.RFC3339Nano, val, time.UTC); err == nil {
timestampEval = tm
} else if tm, err := time.ParseInLocation(time.RFC3339, val, time.UTC); err == nil {
timestampEval = tm
} else if tm, err := time.ParseInLocation("2006-01-02", val, time.UTC); err == nil {
timestampEval = tm
} else {
return nil, sql3.NewErrInvalidTypeCoercion(0, 0, val, n.members[0].Type().TypeName())
}
}
setEval, err := n.members[1].Evaluate(currentRow)
if err != nil {
return nil, err
}
// nil if anything is nil
if timestampEval == nil || setEval == nil {
return nil, nil
}
return []interface{}{
timestampEval,
setEval,
}, nil
}
func (n *exprTupleLiteralPlanExpression) Type() parser.ExprDataType {
return n.dataType
}
func (n *exprTupleLiteralPlanExpression) Plan() map[string]interface{} {
result := make(map[string]interface{})
result["_expr"] = fmt.Sprintf("%T", n)
ps := make([]interface{}, 0)
for _, e := range n.members {
ps = append(ps, e.Plan())
}
result["members"] = ps
return result
}
func (n *exprTupleLiteralPlanExpression) Children() []types.PlanExpression {
return n.members
}
func (n *exprTupleLiteralPlanExpression) WithChildren(children ...types.PlanExpression) (types.PlanExpression, error) {
if len(children) != len(n.members) {
return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children))
}
return newExprTupleLiteralPlanExpression(children, n.dataType), nil
}
// compileExpr returns a types.PlanExpression tree for a given parser.Expr
func (p *ExecutionPlanner) compileExpr(expr parser.Expr) (_ types.PlanExpression, err error) {
if expr == nil {
@ -2101,6 +2201,17 @@ func (p *ExecutionPlanner) compileExpr(expr parser.Expr) (_ types.PlanExpression
}
return newExprSetLiteralPlanExpression(exprList, expr.DataType()), nil
case *parser.TupleLiteralExpr:
exprList := []types.PlanExpression{}
for _, e := range expr.Members {
listExpr, err := p.compileExpr(e)
if err != nil {
return nil, err
}
exprList = append(exprList, listExpr)
}
return newExprTupleLiteralPlanExpression(exprList, expr.DataType()), nil
case *parser.Ident:
return nil, sql3.NewErrInternal("identifiers are not supported")

31
sql3/planner/expressionanalyzer.go
View file

@ -148,6 +148,24 @@ func (p *ExecutionPlanner) analyzeExpression(expr parser.Expr, scope parser.Stat
return e, nil
case *parser.TupleLiteralExpr:
memberTypes := make([]parser.ExprDataType, 0)
for i, ex := range e.Members {
memberExpr, err := p.analyzeExpression(ex, scope)
if err != nil {
return nil, err
}
e.Members[i] = memberExpr
memberTypes = append(memberTypes, memberExpr.DataType())
}
if len(e.Members) == 0 {
return nil, sql3.NewErrLiteralEmptyTupleNotAllowed(e.Lbrace.Line, e.Lbrace.Column)
}
e.ResultDataType = parser.NewDataTypeTuple(memberTypes)
return e, nil
case *parser.QualifiedRef:
switch sc := scope.(type) {
case *parser.SelectStatement:
@ -359,6 +377,19 @@ func (p *ExecutionPlanner) analyzeBinaryExpression(expr *parser.BinaryExpr, scop
if !typeIsCompatibleWithComparisonOperator(x.DataType()) {
return nil, sql3.NewErrTypeIncompatibleWithComparisonOperator(x.Pos().Line, x.Pos().Column, op.String(), x.DataType().TypeName())
}
if typeIsTimestamp(x.DataType()) && y.IsLiteral() && typeIsString(y.DataType()) {
// we have a string literal on the rhs being compared to a date so
// try to convert to a date literal
rhs, ok := y.(*parser.StringLit)
if !ok {
return nil, sql3.NewErrInternalf("unexpected expression type '%T'", y)
}
newRhs := rhs.ConvertToTimestamp()
if newRhs != nil {
expr.Y = newRhs
y = newRhs
}
}
if !typeIsCompatibleWithComparisonOperator(y.DataType()) {
return nil, sql3.NewErrTypeIncompatibleWithComparisonOperator(y.Pos().Line, y.Pos().Column, op.String(), y.DataType().TypeName())
}

3
sql3/planner/expressionanalyzercall.go
View file

@ -239,6 +239,9 @@ func (p *ExecutionPlanner) analyzeCallExpression(call *parser.Call, scope parser
case "DATEPART":
return p.analyzeFunctionDatePart(call, scope)
case "SUBTABLE":
return p.analyzeFunctionSubtable(call, scope)
default:
return nil, sql3.NewErrCallUnknownFunction(call.Name.NamePos.Line, call.Name.NamePos.Column, call.Name.Name)
}

10
sql3/planner/expressionpql.go
View file

@ -169,6 +169,14 @@ func (p *ExecutionPlanner) generatePQLCallFromBinaryExpr(ctx context.Context, ex
},
}, nil
case *parser.DataTypeTimestamp:
return &pql.Call{
Name: "Row",
Args: map[string]interface{}{
lhs.columnName: pqlValue,
},
}, nil
default:
return nil, sql3.NewErrInternalf("unsupported type for binary expression: %v (%T)", typ, typ)
}
@ -220,6 +228,8 @@ func planExprToValue(expr types.PlanExpression) (interface{}, error) {
return strconv.ParseInt(expr.value, 10, 64)
case *stringLiteralPlanExpression:
return expr.value, nil
case *dateLiteralPlanExpression:
return expr.value, nil
default:
return nil, sql3.NewErrInternalf("cannot convert SQL expression %T to a literal value", expr)
}

75
sql3/planner/expressiontypes.go
View file

@ -57,7 +57,14 @@ func fieldSQLDataType(f *pilosa.FieldInfo) parser.ExprDataType {
case pilosa.FieldTypeDecimal:
return parser.NewDataTypeDecimal(f.Options.Scale)
case pilosa.FieldTypeTime, pilosa.FieldTypeTimestamp:
case pilosa.FieldTypeTime:
if f.Options.Keys {
return parser.NewDataTypeStringSetQuantum()
} else {
return parser.NewDataTypeIDSetQuantum()
}
case pilosa.FieldTypeTimestamp:
return parser.NewDataTypeTimestamp()
default:
@ -253,6 +260,33 @@ func typesAreAssignmentCompatible(targetType parser.ExprDataType, sourceType par
default:
return false
}
case *parser.DataTypeStringSetQuantum:
switch source := sourceType.(type) {
case *parser.DataTypeStringSetQuantum:
return true
case *parser.DataTypeStringSet:
return true
case *parser.DataTypeTuple:
// if we are assigning to a time quantum, a tuple is allowed, but members
// have to be a timestamp (or coercable to a timestamp) and and stringset
if len(source.Members) != 2 {
return false
}
_, ok := source.Members[0].(*parser.DataTypeTimestamp)
if !ok {
_, ok = source.Members[0].(*parser.DataTypeString)
if !ok {
return false
}
}
_, ok = source.Members[1].(*parser.DataTypeStringSet)
if !ok {
return false
}
return true
default:
return false
}
case *parser.DataTypeIDSet:
switch sourceType.(type) {
case *parser.DataTypeIDSet:
@ -260,6 +294,33 @@ func typesAreAssignmentCompatible(targetType parser.ExprDataType, sourceType par
default:
return false
}
case *parser.DataTypeIDSetQuantum:
switch source := sourceType.(type) {
case *parser.DataTypeIDSetQuantum:
return true
case *parser.DataTypeIDSet:
return true
case *parser.DataTypeTuple:
// if we are assigning to a time quantum, a tuple is allowed, but members
// have to be a timestamp (or coercable to a timestamp) and and idset
if len(source.Members) != 2 {
return false
}
_, ok := source.Members[0].(*parser.DataTypeTimestamp)
if !ok {
_, ok = source.Members[0].(*parser.DataTypeString)
if !ok {
return false
}
}
_, ok = source.Members[1].(*parser.DataTypeIDSet)
if !ok {
return false
}
return true
default:
return false
}
case *parser.DataTypeDecimal:
switch rhs := sourceType.(type) {
case *parser.DataTypeDecimal:
@ -368,6 +429,18 @@ func typeIsSet(testType parser.ExprDataType) (bool, parser.ExprDataType) {
}
}
// returns true if the type is a timequantum type
func typeIsTimeQuantum(testType parser.ExprDataType) (bool, parser.ExprDataType) {
switch testType.(type) {
case *parser.DataTypeIDSetQuantum:
return true, parser.NewDataTypeIDSet()
case *parser.DataTypeStringSet:
return true, parser.NewDataTypeStringSet()
default:
return false, nil
}
}
// returns true if the type is bool
func typeIsBool(testType parser.ExprDataType) bool {
switch testType.(type) {

66
sql3/planner/inbuiltfunctionstable.go Normal file
View file

@ -0,0 +1,66 @@
package planner
import (
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
)
// TODO (pok) this needs to go somewhere
/*func (p *ExecutionPlanner) analyzeFunctionRecord(call *parser.Call, scope parser.Statement) (parser.Expr, error) {
if len(call.Args) != 2 {
return nil, sql3.NewErrCallParameterCountMismatch(call.Rparen.Line, call.Rparen.Column, call.Name.Name, 2, len(call.Args))
}
// timestamp
timestampType := parser.NewDataTypeTimestamp()
if !typesAreAssignmentCompatible(timestampType, call.Args[0].DataType()) {
return nil, sql3.NewErrParameterTypeMistmatch(call.Args[0].Pos().Line, call.Args[0].Pos().Column, call.Args[0].DataType().TypeName(), timestampType.TypeName())
}
// set
ok, _ := typeIsSet(call.Args[1].DataType())
if !ok {
return nil, sql3.NewErrSetExpressionExpected(call.Args[1].Pos().Line, call.Args[1].Pos().Column)
}
//return record
call.ResultDataType = parser.NewDataTypeSubtable([]*parser.SubtableColumn{
{
Name: "",
DataType: timestampType,
},
{
Name: "",
DataType: call.Args[1].DataType(),
},
})
return call, nil
}
*/
func (p *ExecutionPlanner) analyzeFunctionSubtable(call *parser.Call, scope parser.Statement) (parser.Expr, error) {
if len(call.Args) != 1 {
return nil, sql3.NewErrCallParameterCountMismatch(call.Rparen.Line, call.Rparen.Column, call.Name.Name, 2, len(call.Args))
}
// set
ok, _ := typeIsTimeQuantum(call.Args[0].DataType())
if !ok {
// TODO (pok) send back the right error
return nil, sql3.NewErrSetExpressionExpected(call.Args[1].Pos().Line, call.Args[1].Pos().Column)
}
call.ResultDataType = parser.NewDataTypeSubtable([]*parser.SubtableColumn{
{
Name: "_id",
DataType: parser.NewDataTypeID(),
},
{
Name: "timestamp",
DataType: parser.NewDataTypeTimestamp(),
},
{
Name: "value",
DataType: call.Args[0].DataType(),
},
})
return call, nil
}

1
sql3/planner/opaltertable.go
View file

@ -29,6 +29,7 @@ func NewPlanOpAlterTable(p *ExecutionPlanner, tableName string, operation alterO
oldColumnName: oldColumnName,
newColumnName: newColumnName,
columnDef: columnDef,
warnings: make([]string, 0),
}
}

588
sql3/planner/opbulkinsert.go
View file

@ -4,22 +4,66 @@ package planner
import (
"context"
"encoding/csv"
"fmt"
"io"
"log"
"os"
"strconv"
"time"
"github.com/featurebasedb/featurebase/v3/sql3/planner/types"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
)
// bulkInsertMappedColumn specifies a mapping from the source
// data to a target column name
type bulkInsertMappedColumn struct {
// source expression
// using an interface for so we have flexibility in data types as format changes
columnSource interface{}
// name of the target column
columnName string
// data type of the target column
columnDataType parser.ExprDataType
}
// bulkInsertOptions contains options for bulk insert
type bulkInsertOptions struct {
// name of the file we're going to read
fileName string
// number of rows in a batch
batchSize int
// stop after this many rows
rowsLimit int
// format specifier (CSV is the only one right now)
format string
// the column map in the source data to use as the _id value
// if empty or nill auto increment
// using an interface so we have flexibility in data types as format changes
idColumnMap []interface{}
// column mappings
columnMap []*bulkInsertMappedColumn
}
// PlanOpBulkInsert plan operator to handle INSERT.
type PlanOpBulkInsert struct {
planner *ExecutionPlanner
tableName string
isKeyed bool
options *bulkInsertOptions
warnings []string
}
func NewPlanOpBulkInsert(p *ExecutionPlanner, tableName string) *PlanOpBulkInsert {
func NewPlanOpBulkInsert(p *ExecutionPlanner, tableName string, isKeyed bool, options *bulkInsertOptions) *PlanOpBulkInsert {
return &PlanOpBulkInsert{
planner: p,
tableName: tableName,
isKeyed: isKeyed,
options: options,
warnings: make([]string, 0),
}
}
@ -32,6 +76,26 @@ func (p *PlanOpBulkInsert) Plan() map[string]interface{} {
}
result["_schema"] = sc
result["tableName"] = p.tableName
options := make(map[string]interface{})
options["batchsize"] = p.options.batchSize
options["rowslimit"] = p.options.rowsLimit
options["format"] = p.options.format
if len(p.options.idColumnMap) > 0 {
options["idColumnMap"] = p.options.idColumnMap
} else {
options["idColumnMap"] = "autoincrement"
}
colMap := make([]interface{}, 0)
for _, m := range p.options.columnMap {
cm := make(map[string]interface{})
cm["columnSource"] = m.columnSource
cm["columnName"] = m.columnName
colMap = append(colMap, cm)
}
options["columnMap"] = colMap
result["options"] = options
return result
}
@ -56,24 +120,534 @@ func (p *PlanOpBulkInsert) Children() []types.PlanOperator {
}
func (p *PlanOpBulkInsert) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
return &bulkInsertRowIter{
return &bulkInsertCSVRowIter{
planner: p.planner,
tableName: p.tableName,
isKeyed: p.isKeyed,
options: p.options,
}, nil
}
func (p *PlanOpBulkInsert) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
return NewPlanOpBulkInsert(p.planner, p.tableName), nil
return NewPlanOpBulkInsert(p.planner, p.tableName, p.isKeyed, p.options), nil
}
type bulkInsertRowIter struct {
type bulkInsertCSVRowIter struct {
planner *ExecutionPlanner
tableName string
isKeyed bool
options *bulkInsertOptions
latch *struct{}
currentBatch []interface{}
lastKeyValue uint64
}
var _ types.RowIterator = (*bulkInsertRowIter)(nil)
var _ types.RowIterator = (*bulkInsertCSVRowIter)(nil)
func (i *bulkInsertRowIter) Next(ctx context.Context) (types.Row, error) {
func (i *bulkInsertCSVRowIter) Next(ctx context.Context) (types.Row, error) {
if i.latch == nil {
i.latch = &struct{}{}
i.lastKeyValue = 0
f, err := os.Open(i.options.fileName)
if err != nil {
return nil, err
}
defer f.Close()
linesRead := 0
csvReader := csv.NewReader(f)
for {
rec, err := csvReader.Read()
if err == io.EOF {
break
}
if err != nil {
return nil, err
}
// do something with read line
err = i.processCSVLine(ctx, rec)
if err != nil {
return nil, err
}
linesRead += 1
// bail if we have a rows limit and we've hit it
if i.options.rowsLimit > 0 && linesRead >= i.options.rowsLimit {
break
}
}
}
return nil, types.ErrNoMoreRows
}
func (i *bulkInsertCSVRowIter) processCSVLine(ctx context.Context, line []string) error {
if i.currentBatch == nil {
i.currentBatch = make([]interface{}, 0)
}
i.currentBatch = append(i.currentBatch, line)
if len(i.currentBatch) >= i.options.batchSize {
log.Printf("BULK INSERT: processing batch (%d)", len(i.currentBatch))
err := i.processBatch(ctx)
log.Printf("BULK INSERT: batch processed")
if err != nil {
return err
}
}
return nil
}
func (i *bulkInsertCSVRowIter) processBatch(ctx context.Context) error {
batchLen := len(i.currentBatch)
colIDs := make([]uint64, batchLen)
colKeys := make([]string, batchLen)
insertData := make([]interface{}, len(i.options.columnMap))
addColID := func(index int, v interface{}) error {
switch id := v.(type) {
case int64:
colIDs[index] = uint64(id)
case uint64:
colIDs[index] = id
case string:
colKeys[index] = id
default:
return sql3.NewErrInternalf("unhandled _id data type '%T'", id)
}
return nil
}
// make objects for each column depending on data type
for cidx, mc := range i.options.columnMap {
switch targetType := mc.columnDataType.(type) {
case *parser.DataTypeID:
vals := make([]uint64, batchLen)
insertData[cidx] = vals
case *parser.DataTypeInt:
vals := make([]int64, batchLen)
insertData[cidx] = vals
case *parser.DataTypeString:
vals := make([]string, batchLen)
insertData[cidx] = vals
case *parser.DataTypeTimestamp:
vals := make([]time.Time, batchLen)
insertData[cidx] = vals
default:
return sql3.NewErrInternalf("unhandled target type '%T'", targetType)
}
}
// for each row in the batch add value to each mapped column
log.Printf("BULK INSERT: building batch...")
for rowIdx, row := range i.currentBatch {
csvRow, ok := row.([]string)
if !ok {
return sql3.NewErrInternalf("unexpected row type '%T'", row)
}
//handle each column
for colIdx, mc := range i.options.columnMap {
columnPosition, ok := mc.columnSource.(int64)
if !ok {
return sql3.NewErrInternalf("unexpected columnPosition type '%T'", mc.columnSource)
}
switch targetType := mc.columnDataType.(type) {
case *parser.DataTypeID:
valueStr := csvRow[columnPosition]
insertValue, err := strconv.ParseUint(valueStr, 10, 64)
if err != nil {
return err
}
columnData, ok := insertData[colIdx].([]uint64)
if !ok {
return sql3.NewErrInternalf("unexpected columnData type '%T'", insertData[colIdx])
}
columnData[rowIdx] = insertValue
case *parser.DataTypeInt:
valueStr := csvRow[columnPosition]
insertValue, err := strconv.ParseInt(valueStr, 10, 64)
if err != nil {
return err
}
columnData, ok := insertData[colIdx].([]int64)
if !ok {
return sql3.NewErrInternalf("unexpected columnData type '%T'", insertData[colIdx])
}
columnData[rowIdx] = insertValue
case *parser.DataTypeString:
insertValue := csvRow[columnPosition]
columnData, ok := insertData[colIdx].([]string)
if !ok {
return sql3.NewErrInternalf("unexpected columnData type '%T'", insertData[colIdx])
}
columnData[rowIdx] = insertValue
case *parser.DataTypeTimestamp:
valueStr := csvRow[columnPosition]
var insertValue time.Time
if tm, err := time.ParseInLocation(time.RFC3339Nano, valueStr, time.UTC); err == nil {
insertValue = tm
} else if tm, err := time.ParseInLocation(time.RFC3339, valueStr, time.UTC); err == nil {
insertValue = tm
} else if tm, err := time.ParseInLocation("2006-01-02 15:04:05", valueStr, time.UTC); err == nil {
insertValue = tm
} else if tm, err := time.ParseInLocation("2006-01-02", valueStr, time.UTC); err == nil {
insertValue = tm
} else {
return err
}
columnData, ok := insertData[colIdx].([]time.Time)
if !ok {
return sql3.NewErrInternalf("unexpected columnData type '%T'", insertData[colIdx])
}
columnData[rowIdx] = insertValue
default:
return sql3.NewErrInternalf("unhandled target type '%T'", targetType)
}
}
// add _id
if len(i.options.idColumnMap) > 0 {
return sql3.NewErrInternalf("not yet implemented")
} else {
// if the table is keyed, use the string representation of an integer key value
if i.isKeyed {
//auto increment
err := addColID(rowIdx, fmt.Sprintf("%d", i.lastKeyValue))
if err != nil {
return err
}
} else {
//auto increment
err := addColID(rowIdx, i.lastKeyValue)
if err != nil {
return err
}
}
i.lastKeyValue += 1
}
}
log.Printf("BULK INSERT: building batch complete")
// now loop again and actually do the insert
log.Printf("BULK INSERT: inserting columns...")
qcx := i.planner.computeAPI.Txf().NewQcx()
//nil out colids if the table is keyed
for colIdx, mc := range i.options.columnMap {
log.Printf("BULK INSERT: inserting column '%s'...", mc.columnName)
if i.isKeyed {
colIDs = nil
}
switch targetType := mc.columnDataType.(type) {
case *parser.DataTypeID:
vals, ok := insertData[colIdx].([]uint64)
if !ok {
return sql3.NewErrInternalf("unexpected insert data type '%T'", insertData[colIdx])
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: mc.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowIDs: vals,
}
err := i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return err
}
case *parser.DataTypeInt:
vals, ok := insertData[colIdx].([]int64)
if !ok {
return sql3.NewErrInternalf("unexpected insert data type '%T'", insertData[colIdx])
}
req := &pilosa.ImportValueRequest{
Index: i.tableName,
Field: mc.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
Values: vals,
}
err := i.planner.computeAPI.ImportValue(ctx, qcx, req)
if err != nil {
return err
}
case *parser.DataTypeString:
vals, ok := insertData[colIdx].([]string)
if !ok {
return sql3.NewErrInternalf("unexpected insert data type '%T'", insertData[colIdx])
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: mc.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowKeys: vals,
}
err := i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return err
}
case *parser.DataTypeTimestamp:
vals, ok := insertData[colIdx].([]time.Time)
if !ok {
return sql3.NewErrInternalf("unexpected insert data type '%T'", insertData[colIdx])
}
// TODO (pok) - getting and error for timestamp columns
// 'Error: local import after remote imports: number of columns (1) and number of values (0) do not match'
req := &pilosa.ImportValueRequest{
Index: i.tableName,
Field: mc.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
TimestampValues: vals,
}
err := i.planner.computeAPI.ImportValue(ctx, qcx, req)
if err != nil {
return err
}
default:
return sql3.NewErrInternalf("unhandled target type '%T'", targetType)
}
log.Printf("BULK INSERT: inserting column '%s' complete.", mc.columnName)
}
log.Printf("BULK INSERT: inserting columns complete.")
/*
//eval all the expressions and do the insert
for idx, iv := range i.insertValues {
sourceType := iv.Type()
switch targetType := i.targetColumns[idx].dataType.(type) {
case *parser.DataTypeBool:
err = addColID(columnID)
if err != nil {
return nil, err
}
val := eval.(bool)
vals := make([]uint64, 1)
if val {
vals[0] = 1
} else {
vals[0] = 0
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: targetColumn.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowIDs: vals,
}
err = i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return nil, err
}
case *parser.DataTypeDecimal:
err = addColID(columnID)
if err != nil {
return nil, err
}
vals := make([]float64, 1)
vals[0] = eval.(pql.Decimal).Float64()
req := &pilosa.ImportValueRequest{
Index: i.tableName,
Field: targetColumn.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
FloatValues: vals,
}
err = i.planner.computeAPI.ImportValue(ctx, qcx, req)
if err != nil {
return nil, err
}
case *parser.DataTypeIDSet:
rowIDs := make([]uint64, 0)
rowSet := eval.([]int64)
for k := range rowSet {
err = addColID(columnID)
if err != nil {
return nil, err
}
rowIDs = append(rowIDs, uint64(rowSet[k]))
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: targetColumn.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowIDs: rowIDs,
}
err = i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return nil, err
}
case *parser.DataTypeIDSetQuantum:
rowIDs := make([]uint64, 0)
timestamps := make([]int64, 0)
coercedVal, err := coerceValue(sourceType, targetType, eval, parser.Pos{Line: 0, Column: 0})
if err != nil {
return nil, err
}
record := coercedVal.([]interface{})
rowSet := record[1].([]int64)
for k := range rowSet {
err = addColID(columnID)
if err != nil {
return nil, err
}
rowIDs = append(rowIDs, uint64(rowSet[k]))
}
if record[0] == nil {
timestamps = nil
} else {
timestamp, ok := record[0].(time.Time)
if !ok {
return nil, sql3.NewErrInternalf("unexpected type '%T'", record[0])
}
for _ = range rowSet {
timestamps = append(timestamps, timestamp.Unix())
}
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: targetColumn.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowIDs: rowIDs,
Timestamps: timestamps,
}
err = i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return nil, err
}
case *parser.DataTypeStringSet:
rowKeys := make([]string, 0)
rowSet := eval.([]string)
for k := range rowSet {
err = addColID(columnID)
if err != nil {
return nil, err
}
rowKeys = append(rowKeys, rowSet[k])
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: targetColumn.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowKeys: rowKeys,
}
err = i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return nil, err
}
case *parser.DataTypeStringSetQuantum:
rowKeys := make([]string, 0)
timestamps := make([]int64, 0)
coercedVal, err := coerceValue(sourceType, targetType, eval, parser.Pos{Line: 0, Column: 0})
if err != nil {
return nil, err
}
record := coercedVal.([]interface{})
rowSet := record[1].([]string)
for k := range rowSet {
err = addColID(columnID)
if err != nil {
return nil, err
}
rowKeys = append(rowKeys, rowSet[k])
}
if record[0] == nil {
timestamps = nil
} else {
timestamp, ok := record[0].(time.Time)
if !ok {
return nil, sql3.NewErrInternalf("unexpected type '%T'", record[0])
}
for _ = range rowSet {
timestamps = append(timestamps, timestamp.Unix())
}
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: targetColumn.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowKeys: rowKeys,
Timestamps: timestamps,
}
err = i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return nil, err
}
default:
return nil, sql3.NewErrInternalf("unhandled data type '%T'", targetType)
}
}*/
// done with current batch
i.currentBatch = nil
return nil
}

1
sql3/planner/opcreatetable.go
View file

@ -30,6 +30,7 @@ func NewPlanOpCreateTable(p *ExecutionPlanner, tableName string, failIfExists bo
isKeyed: isKeyed,
keyPartitions: keyPartitions,
columns: columns,
warnings: make([]string, 0),
}
}

5
sql3/planner/opdistinct.go
View file

@ -17,8 +17,9 @@ type PlanOpDistinct struct {
func NewPlanOpDistinct(p *ExecutionPlanner, source types.PlanOperator) *PlanOpDistinct {
return &PlanOpDistinct{
planner: p,
source: source,
planner: p,
source: source,
warnings: make([]string, 0),
}
}

5
sql3/planner/opdroptable.go
View file

@ -19,8 +19,9 @@ type PlanOpDropTable struct {
func NewPlanOpDropTable(p *ExecutionPlanner, index *pilosa.IndexInfo) *PlanOpDropTable {
return &PlanOpDropTable{
planner: p,
index: index,
planner: p,
index: index,
warnings: make([]string, 0),
}
}

3
sql3/planner/opfeaturebasecolumns.go
View file

@ -20,7 +20,8 @@ type PlanOpFeatureBaseColumns struct {
func NewPlanOpFeatureBaseColumns(index *pilosa.IndexInfo) *PlanOpFeatureBaseColumns {
node := &PlanOpFeatureBaseColumns{
index: index,
index: index,
warnings: make([]string, 0),
}
return node
}

1
sql3/planner/opfeaturebasetables.go
View file

@ -22,6 +22,7 @@ type PlanOpFeatureBaseTables struct {
func NewPlanOpFeatureBaseTables(indexInfo []*pilosa.IndexInfo) *PlanOpFeatureBaseTables {
return &PlanOpFeatureBaseTables{
indexInfo: indexInfo,
warnings: make([]string, 0),
}
}

97
sql3/planner/opfilter.go Normal file
View file

@ -0,0 +1,97 @@
// Copyright 2022 Molecula Corp. All rights reserved.
package planner
import (
"context"
"fmt"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/planner/types"
)
// PlanOpFilter is a filter operator
type PlanOpFilter struct {
planner *ExecutionPlanner
ChildOp types.PlanOperator
Predicate types.PlanExpression
warnings []string
}
func NewPlanOpFilter(planner *ExecutionPlanner, predicate types.PlanExpression, child types.PlanOperator) *PlanOpFilter {
return &PlanOpFilter{
planner: planner,
Predicate: predicate,
ChildOp: child,
warnings: make([]string, 0),
}
}
func (p *PlanOpFilter) Schema() types.Schema {
return p.ChildOp.Schema()
}
func (p *PlanOpFilter) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
i, err := p.ChildOp.Iterator(ctx, row)
if err != nil {
return nil, err
}
return newFilterIterator(ctx, p.Predicate, i), nil
}
func (p *PlanOpFilter) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
if len(children) != 1 {
return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children))
}
return NewPlanOpFilter(p.planner, p.Predicate, children[0]), nil
}
func (p *PlanOpFilter) Children() []types.PlanOperator {
return []types.PlanOperator{
p.ChildOp,
}
}
func (p *PlanOpFilter) Plan() map[string]interface{} {
result := make(map[string]interface{})
result["_op"] = fmt.Sprintf("%T", p)
ps := make([]string, 0)
for _, e := range p.Schema() {
ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName()))
}
result["_schema"] = ps
result["child"] = p.ChildOp.Plan()
return result
}
func (p *PlanOpFilter) String() string {
return ""
}
func (p *PlanOpFilter) AddWarning(warning string) {
p.warnings = append(p.warnings, warning)
}
func (p *PlanOpFilter) Warnings() []string {
return p.warnings
}
type filterIterator struct {
predicate types.PlanExpression
child types.RowIterator
ctx context.Context
}
func newFilterIterator(ctx context.Context, predicate types.PlanExpression, child types.RowIterator) *filterIterator {
return &filterIterator{
ctx: ctx,
predicate: predicate,
child: child,
}
}
func (i *filterIterator) Next(ctx context.Context) (types.Row, error) {
//TODO (pok) - actually implement the filter
return i.child.Next(ctx)
}

1
sql3/planner/opgroupby.go
View file

@ -25,6 +25,7 @@ func NewPlanOpGroupBy(aggregates []types.PlanExpression, groupByExprs []types.Pl
ChildOp: child,
Aggregates: aggregates,
GroupByExprs: groupByExprs,
warnings: make([]string, 0),
}
}

93
sql3/planner/opinsert.go
View file

@ -30,6 +30,7 @@ func NewPlanOpInsert(p *ExecutionPlanner, tableName string, targetColumns []*qua
tableName: tableName,
targetColumns: targetColumns,
insertValues: insertValues,
warnings: make([]string, 0),
}
}
@ -281,6 +282,51 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) {
return nil, err
}
case *parser.DataTypeIDSetQuantum:
rowIDs := make([]uint64, 0)
timestamps := make([]int64, 0)
coercedVal, err := coerceValue(sourceType, targetType, eval, parser.Pos{Line: 0, Column: 0})
if err != nil {
return nil, err
}
record := coercedVal.([]interface{})
rowSet := record[1].([]int64)
for k := range rowSet {
err = addColID(columnID)
if err != nil {
return nil, err
}
rowIDs = append(rowIDs, uint64(rowSet[k]))
}
if record[0] == nil {
timestamps = nil
} else {
timestamp, ok := record[0].(time.Time)
if !ok {
return nil, sql3.NewErrInternalf("unexpected type '%T'", record[0])
}
for range rowSet {
timestamps = append(timestamps, timestamp.Unix())
}
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: targetColumn.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowIDs: rowIDs,
Timestamps: timestamps,
}
err = i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return nil, err
}
case *parser.DataTypeString:
err = addColID(columnID)
if err != nil {
@ -327,6 +373,51 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) {
return nil, err
}
case *parser.DataTypeStringSetQuantum:
rowKeys := make([]string, 0)
timestamps := make([]int64, 0)
coercedVal, err := coerceValue(sourceType, targetType, eval, parser.Pos{Line: 0, Column: 0})
if err != nil {
return nil, err
}
record := coercedVal.([]interface{})
rowSet := record[1].([]string)
for k := range rowSet {
err = addColID(columnID)
if err != nil {
return nil, err
}
rowKeys = append(rowKeys, rowSet[k])
}
if record[0] == nil {
timestamps = nil
} else {
timestamp, ok := record[0].(time.Time)
if !ok {
return nil, sql3.NewErrInternalf("unexpected type '%T'", record[0])
}
for range rowSet {
timestamps = append(timestamps, timestamp.Unix())
}
}
req := &pilosa.ImportRequest{
Index: i.tableName,
Field: targetColumn.columnName,
Shard: 0, //TODO: handle non-0 shards
ColumnIDs: colIDs,
ColumnKeys: colKeys,
RowKeys: rowKeys,
Timestamps: timestamps,
}
err = i.planner.computeAPI.Import(ctx, qcx, req)
if err != nil {
return nil, err
}
case *parser.DataTypeTimestamp:
err = addColID(columnID)
if err != nil {
@ -356,7 +447,7 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) {
}
default:
return nil, sql3.NewErrInternalf("unhandled data type '%T'", iv.Type())
return nil, sql3.NewErrInternalf("unhandled data type '%T'", targetType)
}
}

165
sql3/planner/opnestedloops.go
View file

@ -15,13 +15,15 @@ import (
type PlanOpNestedLoops struct {
top types.PlanOperator
bottom types.PlanOperator
cond types.PlanExpression
warnings []string
}
func NewPlanOpNestedLoops(top, bottom types.PlanOperator) *PlanOpNestedLoops {
return &PlanOpNestedLoops{
top: top,
bottom: bottom,
top: top,
bottom: bottom,
warnings: make([]string, 0),
}
}
@ -81,6 +83,14 @@ func (p *PlanOpNestedLoops) WithChildren(children ...types.PlanOperator) (types.
return NewPlanOpNestedLoops(children[0], children[1]), nil
}
func (p *PlanOpNestedLoops) NewWithExpressions(exprs ...types.PlanExpression) (types.PlanOperator, error) {
if len(exprs) != 1 {
return nil, sql3.NewErrInternalf("unexpected number of exprs '%d'", len(exprs))
}
p.cond = exprs[0]
return p, nil
}
type joinType byte
const (
@ -89,25 +99,6 @@ const (
joinTypeRight
)
// joinMode defines the mode in which a join will be performed.
type joinMode byte
const (
// unknownMode is the default mode. It will start iterating without really
// knowing in which mode it will end up computing the join. If it
// iterates the right side fully one time and so far it fits in memory,
// then it will switch to memory mode. Otherwise, if at some point during
// this first iteration it finds that it does not fit in memory, will
// switch to multipass mode.
unknownMode joinMode = iota
// memoryMode computes all the join directly in memory iterating each
// side of the join exactly once.
//memoryMode
// multipassMode computes the join by iterating the left side once,
// and the right side one time for each row in the left side.
multipassMode
)
type nestedLoopsIter struct {
typ joinType
@ -116,35 +107,33 @@ type nestedLoopsIter struct {
ctx context.Context
cond types.PlanExpression
secondaryProvider types.RowIterable
bottomProvider types.RowIterable
primaryRow types.Row
topRow types.Row
foundMatch bool
rowSize int
originalRow types.Row
scopeLen int
mode joinMode
secondaryRows RowCache
bottomRows RowCache
}
func newNestedLoopsIter(ctx context.Context, jt joinType, top types.RowIterator, bottom types.RowIterable, scopeRow types.Row, joinCondition types.PlanExpression, rowWidth int, originalRow types.Row) *nestedLoopsIter {
return &nestedLoopsIter{
typ: jt,
top: top,
secondaryProvider: bottom,
cond: joinCondition,
rowSize: rowWidth,
originalRow: originalRow,
secondaryRows: newInMemoryRowCache(),
ctx: ctx,
typ: jt,
top: top,
bottomProvider: bottom,
cond: joinCondition,
rowSize: rowWidth,
originalRow: originalRow,
bottomRows: newInMemoryRowCache(),
ctx: ctx,
}
}
func (i *nestedLoopsIter) loadPrimary(ctx context.Context) error {
// If primary has already been loaded, it's safe to no-op.
if i.primaryRow != nil {
func (i *nestedLoopsIter) loadTop(ctx context.Context) error {
if i.topRow != nil {
return nil
}
@ -152,119 +141,35 @@ func (i *nestedLoopsIter) loadPrimary(ctx context.Context) error {
if err != nil {
return err
}
i.primaryRow = i.originalRow.Append(r)
i.topRow = i.originalRow.Append(r)
i.foundMatch = false
return nil
}
/*func (i *nestedLoopsIter) loadSecondaryInMemory(ctx context.Context) error {
iter, err := i.secondaryProvider.Iterator(ctx, i.primaryRow)
if err != nil {
return err
}
for {
row, err := iter.Next(ctx)
if err == types.ErrNoMoreRows {
break
}
if err != nil {
//iter.Close(ctx)
return err
}
if err := i.secondaryRows.Add(row); err != nil {
//iter.Close(ctx)
return err
}
}
//err = iter.Close(ctx)
//if err != nil {
// return err
//}
if len(i.secondaryRows.Get()) == 0 {
return types.ErrNoMoreRows
}
return nil
}*/
func (i *nestedLoopsIter) loadSecondary(ctx context.Context) (row types.Row, err error) {
/*if i.mode == memoryMode {
if len(i.secondaryRows.Get()) == 0 {
if err = i.loadSecondaryInMemory(ctx); err != nil {
if err == types.ErrNoMoreRows {
i.primaryRow = nil
i.pos = 0
}
return nil, err
}
}
if i.pos >= len(i.secondaryRows.Get()) {
i.primaryRow = nil
i.pos = 0
return nil, types.ErrNoMoreRows
}
row := i.secondaryRows.Get()[i.pos]
i.pos++
return row, nil
}*/
func (i *nestedLoopsIter) loadBottom(ctx context.Context) (row types.Row, err error) {
if i.bottom == nil {
var iter types.RowIterator
iter, err = i.secondaryProvider.Iterator(ctx, i.primaryRow)
iter, err = i.bottomProvider.Iterator(ctx, i.topRow)
if err != nil {
return nil, err
}
i.bottom = iter
}
rightRow, err := i.bottom.Next(ctx)
if err != nil {
if err == types.ErrNoMoreRows {
//err = i.bottom.Close(ctx)
i.bottom = nil
//if err != nil {
// return nil, err
//}
i.primaryRow = nil
// If we got to this point and the mode is still unknown it means
// the right side fits in memory, so the mode changes to memory
// join.
//if i.mode == unknownMode {
// i.mode = memoryMode
//}
i.topRow = nil
return nil, types.ErrNoMoreRows
}
return nil, err
}
if i.mode == unknownMode {
var switchToMultipass bool
//if !ctx.Memory.HasAvailable() {
//switchToMultipass = true
//} else {
err := i.secondaryRows.Add(rightRow)
if err != nil { //&& !sql.ErrNoMemoryAvailable.Is(err) {
return nil, err
}
//}
if switchToMultipass {
//i.Dispose()
i.secondaryRows = nil
i.mode = multipassMode
}
err = i.bottomRows.Add(rightRow)
if err != nil {
return nil, err
}
return rightRow, nil
}
@ -307,12 +212,12 @@ func conditionIsTrue(ctx context.Context, row types.Row, cond types.PlanExpressi
func (i *nestedLoopsIter) Next(ctx context.Context) (types.Row, error) {
for {
if err := i.loadPrimary(ctx); err != nil {
if err := i.loadTop(ctx); err != nil {
return nil, err
}
primary := i.primaryRow
secondary, err := i.loadSecondary(ctx)
primary := i.topRow
secondary, err := i.loadBottom(ctx)
if err != nil {
if err == types.ErrNoMoreRows {
if !i.foundMatch && (i.typ == joinTypeLeft || i.typ == joinTypeRight) {

4
sql3/planner/opnulltable.go
View file

@ -16,7 +16,9 @@ type PlanOpNullTable struct {
}
func NewPlanOpNullTable() *PlanOpNullTable {
return &PlanOpNullTable{}
return &PlanOpNullTable{
warnings: make([]string, 0),
}
}
func (p *PlanOpNullTable) Schema() types.Schema {

1
sql3/planner/oporderby.go
View file

@ -48,6 +48,7 @@ func NewPlanOpOrderBy(orderByFields []*OrderByExpression, child types.PlanOperat
return &PlanOpOrderBy{
ChildOp: child,
orderByFields: orderByFields,
warnings: make([]string, 0),
}
}

1
sql3/planner/oppqlaggregate.go
View file

@ -29,6 +29,7 @@ func NewPlanOpPQLAggregate(p *ExecutionPlanner, tableName string, aggregate type
tableName: tableName,
filter: filter,
aggregate: aggregate,
warnings: make([]string, 0),
}
}

3
sql3/planner/oppqlgroupby.go
View file

@ -31,6 +31,7 @@ func NewPlanOpPQLGroupBy(p *ExecutionPlanner, tableName string, groupByExprs []t
groupByExprs: groupByExprs,
filter: filter,
aggregate: aggregate,
warnings: make([]string, 0),
}
}
@ -146,7 +147,7 @@ func (i *pqlGroupByRowIter) Next(ctx context.Context) (types.Row, error) {
Args: map[string]interface{}{},
}
for _, c := range i.groupByColumns {
ref, ok := c.(types.SchemaIdentifiable)
ref, ok := c.(types.IdentifiableByName)
if !ok {
return nil, sql3.NewErrInternalf("unexpected expression type in group by list '%T'", c)
}

1
sql3/planner/oppqlmultiaggregate.go
View file

@ -20,6 +20,7 @@ func NewPlanOpPQLMultiAggregate(p *ExecutionPlanner, operators []*PlanOpPQLAggre
return &PlanOpPQLMultiAggregate{
planner: p,
operators: operators,
warnings: make([]string, 0),
}
}

1
sql3/planner/oppqlmultigroupby.go
View file

@ -25,6 +25,7 @@ func NewPlanOpPQLMultiGroupBy(p *ExecutionPlanner, operators []*PlanOpPQLGroupBy
planner: p,
operators: operators,
groupByExprs: groupByExprs,
warnings: make([]string, 0),
}
}

20
sql3/planner/optablescan.go → sql3/planner/oppqltablescan.go
View file

@ -24,12 +24,12 @@ type PlanOpPQLTableScan struct {
warnings []string
}
func NewPlanOpPQLTableScan(p *ExecutionPlanner, tableName string, columns []types.PlanExpression, filter types.PlanExpression) *PlanOpPQLTableScan {
func NewPlanOpPQLTableScan(p *ExecutionPlanner, tableName string, columns []types.PlanExpression) *PlanOpPQLTableScan {
return &PlanOpPQLTableScan{
planner: p,
tableName: tableName,
columns: columns,
filter: filter,
warnings: make([]string, 0),
}
}
@ -71,10 +71,19 @@ func (p *PlanOpPQLTableScan) Warnings() []string {
return p.warnings
}
func (p *PlanOpPQLTableScan) Name() string {
return p.tableName
}
func (p *PlanOpPQLTableScan) UpdateFilters(filterCondition types.PlanExpression) (types.PlanOperator, error) {
p.filter = filterCondition
return p, nil
}
func (p *PlanOpPQLTableScan) Schema() types.Schema {
result := make(types.Schema, 0)
for _, col := range p.columns {
si, ok := col.(types.SchemaIdentifiable)
si, ok := col.(types.IdentifiableByName)
if ok {
result = append(result, &types.PlannerColumn{
Name: si.Name(),
@ -105,7 +114,6 @@ func (p *PlanOpPQLTableScan) WithChildren(children ...types.PlanOperator) (types
}
// TODO(pok) remove the name mapping here and do it by ordinal position
type tableScanRowIter struct {
planner *ExecutionPlanner
tableName string
@ -172,7 +180,7 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) {
call := &pql.Call{Name: "Extract", Children: []*pql.Call{cond}}
for _, c := range i.columns {
col, ok := c.(types.SchemaIdentifiable)
col, ok := c.(types.IdentifiableByName)
if !ok {
return nil, sql3.NewErrInternalf("unexpected column type '%T'", c)
}
@ -210,7 +218,7 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) {
for _, c := range i.columns {
result := i.result[0]
col, ok := c.(types.SchemaIdentifiable)
col, ok := c.(types.IdentifiableByName)
if !ok {
return nil, sql3.NewErrInternalf("unexpected column type '%T'", c)
}

9
sql3/planner/opprojection.go
View file

@ -21,6 +21,7 @@ func NewPlanOpProjection(expressions []types.PlanExpression, child types.PlanOpe
return &PlanOpProjection{
ChildOp: child,
Projections: expressions,
warnings: make([]string, 0),
}
}
@ -96,16 +97,16 @@ func (p *PlanOpProjection) Warnings() []string {
func ExpressionToColumn(e types.PlanExpression) *types.PlannerColumn {
var name string
if n, ok := e.(types.SchemaIdentifiable); ok {
if n, ok := e.(types.IdentifiableByName); ok {
name = n.Name()
} else {
//TODO(pok) - work out what this should be
//TODO(pok) - implement this
name = "" //e.String()
}
var table string
if t, ok := e.(types.SchemaObject); ok {
table = t.ObjectName()
if t, ok := e.(types.IdentifiableByName); ok {
table = t.Name()
}
return &types.PlannerColumn{

8
sql3/planner/opquery.go
View file

@ -26,7 +26,8 @@ var _ types.PlanOperator = (*PlanOpQuery)(nil)
func NewPlanOpQuery(child types.PlanOperator, sql string) *PlanOpQuery {
return &PlanOpQuery{
ChildOp: child,
ChildOp: child,
warnings: make([]string, 0),
}
}
@ -52,7 +53,10 @@ func (p *PlanOpQuery) WithChildren(children ...types.PlanOperator) (types.PlanOp
if len(children) != 1 {
return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children))
}
return NewPlanOpQuery(children[0], p.sql), nil
op := NewPlanOpQuery(children[0], p.sql)
op.warnings = append(op.warnings, p.warnings...)
return op, nil
}
func (p *PlanOpQuery) Plan() map[string]interface{} {

80
sql3/planner/oprelalias.go Normal file
View file

@ -0,0 +1,80 @@
// Copyright 2022 Molecula Corp. All rights reserved.
package planner
import (
"context"
"fmt"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/planner/types"
)
// PlanOpRelAlias implements an alias for a relation
type PlanOpRelAlias struct {
ChildOp types.PlanOperator
alias string
warnings []string
}
func NewPlanOpRelAlias(alias string, child types.PlanOperator) *PlanOpRelAlias {
return &PlanOpRelAlias{
ChildOp: child,
alias: alias,
warnings: make([]string, 0),
}
}
func (p *PlanOpRelAlias) Schema() types.Schema {
return p.ChildOp.Schema()
}
func (p *PlanOpRelAlias) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
return p.ChildOp.Iterator(ctx, row)
}
func (p *PlanOpRelAlias) Children() []types.PlanOperator {
return []types.PlanOperator{
p.ChildOp,
}
}
func (p *PlanOpRelAlias) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
if len(children) != 1 {
return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children))
}
return NewPlanOpRelAlias(p.alias, children[0]), nil
}
func (p *PlanOpRelAlias) Plan() map[string]interface{} {
result := make(map[string]interface{})
result["_op"] = fmt.Sprintf("%T", p)
sc := make([]string, 0)
for _, e := range p.Schema() {
sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName()))
}
result["_schema"] = sc
result["alias"] = p.alias
result["child"] = p.ChildOp.Plan()
return result
}
func (p *PlanOpRelAlias) String() string {
return ""
}
func (p *PlanOpRelAlias) AddWarning(warning string) {
p.warnings = append(p.warnings, warning)
}
func (p *PlanOpRelAlias) Warnings() []string {
var w []string
w = append(w, p.warnings...)
w = append(w, p.ChildOp.Warnings()...)
return w
}
func (p *PlanOpRelAlias) Name() string {
return p.alias
}

8
sql3/planner/opsubquery.go
View file

@ -17,7 +17,8 @@ type PlanOpSubquery struct {
func NewPlanOpSubquery(child types.PlanOperator) *PlanOpSubquery {
return &PlanOpSubquery{
ChildOp: child,
ChildOp: child,
warnings: make([]string, 0),
}
}
@ -67,5 +68,8 @@ func (p *PlanOpSubquery) Warnings() []string {
w = append(w, p.ChildOp.Warnings()...)
}
return w
}
func (p *PlanOpSubquery) Name() string {
return ""
}

82
sql3/planner/optablevaluedfunction.go Normal file
View file

@ -0,0 +1,82 @@
// Copyright 2022 Molecula Corp. All rights reserved.
package planner
import (
"context"
"fmt"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
)
// PlanOpTableValuedFunction is an operator for a subquery
type PlanOpTableValuedFunction struct {
planner *ExecutionPlanner
callExpr types.PlanExpression
warnings []string
}
func NewPlanOpTableValuedFunction(p *ExecutionPlanner, callExpr types.PlanExpression) *PlanOpTableValuedFunction {
return &PlanOpTableValuedFunction{
planner: p,
callExpr: callExpr,
warnings: make([]string, 0),
}
}
func (p *PlanOpTableValuedFunction) Schema() types.Schema {
result := make(types.Schema, 0)
tvfResultType, ok := p.callExpr.Type().(*parser.DataTypeSubtable)
if !ok {
return result
}
for _, member := range tvfResultType.Columns {
result = append(result, &types.PlannerColumn{
Name: member.Name,
Table: "",
Type: member.DataType,
})
}
return result
}
func (p *PlanOpTableValuedFunction) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
return nil, sql3.NewErrInternalf("table valued functions are not yet implemented")
}
func (p *PlanOpTableValuedFunction) Children() []types.PlanOperator {
return []types.PlanOperator{}
}
func (p *PlanOpTableValuedFunction) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
return nil, nil
}
func (p *PlanOpTableValuedFunction) Plan() map[string]interface{} {
result := make(map[string]interface{})
result["_op"] = fmt.Sprintf("%T", p)
sc := make([]string, 0)
for _, e := range p.Schema() {
sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.Name, e.Table, e.Type.TypeName()))
}
result["_schema"] = sc
return result
}
func (p *PlanOpTableValuedFunction) String() string {
return ""
}
func (p *PlanOpTableValuedFunction) AddWarning(warning string) {
p.warnings = append(p.warnings, warning)
}
func (p *PlanOpTableValuedFunction) Warnings() []string {
var w []string
w = append(w, p.warnings...)
return w
}

5
sql3/planner/optop.go
View file

@ -18,8 +18,9 @@ type PlanOpTop struct {
func NewPlanOpTop(expr types.PlanExpression, child types.PlanOperator) *PlanOpTop {
return &PlanOpTop{
ChildOp: child,
expr: expr,
ChildOp: child,
expr: expr,
warnings: make([]string, 0),
}
}

627
sql3/planner/planoptimizer.go
View file

@ -5,19 +5,28 @@ package planner
import (
"context"
"fmt"
"log"
"reflect"
"strings"
"github.com/featurebasedb/featurebase/v3/sql3"
"github.com/featurebasedb/featurebase/v3/sql3/planner/types"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
)
//TODO(pok) push order by down as far as possible
//TODO(pok) handle the case of the order by expressions not being in a projection list
//TODO(pok) you can't group by _id in PQL, so we need to not use a PQL group by operator here
//TODO(pok) move constant folding to the here
//TODO(pok) move constant folding to in here
// a function prototype for all optimizer rules
type OptimizerFunc func(context.Context, *ExecutionPlanner, types.PlanOperator, *OptimizerScope) (types.PlanOperator, bool, error)
// a list of optimzer rules; order can be important important
var optimizerFunctions = []OptimizerFunc{
// push down filter predicates as far as possible,
pushdownFilters,
// if we have a group by that has one TableScanOperator,
// no Top or TopN or Distincts, try to use a PQL(multi)
// groupby operator instead
@ -28,6 +37,10 @@ var optimizerFunctions = []OptimizerFunc{
// to use a PQL aggregate operators instead
tryToReplaceGroupByWithPQLAggregate,
// if we have a subtable call on a timequantum type
// take the join out and use the appropriate PQL operator instead
tryToRewriteSubtableJoins,
// update the columnIdx for all the references in the projections
// based on the child operator for a projection
fixGroupByProjections,
@ -41,11 +54,11 @@ var optimizerFunctions = []OptimizerFunc{
pushdownPQLTop,
}
// this will be used in future for symbol resolution when CTEs and subquery support matures
// and we need to introduce the concept of scope to symbol resolution
type OptimizerScope struct {
}
type OptimizerFunc func(context.Context, *ExecutionPlanner, types.PlanOperator, *OptimizerScope) (types.PlanOperator, bool, error)
// optimizePlan takes a plan from the compiler and executes a series of transforms on it to optimize it
func (p *ExecutionPlanner) optimizePlan(ctx context.Context, plan types.PlanOperator) (types.PlanOperator, error) {
var err error
@ -70,6 +83,417 @@ func (p *ExecutionPlanner) optimizeNode(ctx context.Context, node types.PlanOper
return node, nil
}
// a set of filters for a operator graph
type filterSet struct {
filterConditions []types.PlanExpression
filtersByRelation map[string][]types.PlanExpression
handledFilters []types.PlanExpression
relationAliases RelationAliasesMap
}
func newFilterSet(filter types.PlanExpression, filtersByTable map[string][]types.PlanExpression, tableAliases RelationAliasesMap) *filterSet {
return &filterSet{
filterConditions: splitOnAnd(filter),
filtersByRelation: filtersByTable,
relationAliases: tableAliases,
}
}
func (fs *filterSet) availableFiltersForTable(table string) []types.PlanExpression {
filters, ok := fs.filtersByRelation[table]
if !ok {
return nil
}
return remainingExpressions(filters, fs.handledFilters)
}
func (fs *filterSet) handledCount() int {
return len(fs.handledFilters)
}
func (fs *filterSet) markFiltersHandled(exprs ...types.PlanExpression) {
fs.handledFilters = append(fs.handledFilters, exprs...)
}
func (fs *filterSet) unhandledPredicates(ctx context.Context) []types.PlanExpression {
var available []types.PlanExpression
for _, e := range fs.filterConditions {
available = append(available, remainingExpressions([]types.PlanExpression{e}, fs.handledFilters)...)
}
return available
}
func remainingExpressions(allExprs, lessExprs []types.PlanExpression) []types.PlanExpression {
var remainder []types.PlanExpression
for _, e := range allExprs {
var found bool
for _, s := range lessExprs {
if reflect.DeepEqual(e, s) {
found = true
break
}
}
if !found {
remainder = append(remainder, e)
}
}
return remainder
}
// RelationAliasesMap is a map of aliases to Relations
type RelationAliasesMap map[string]types.IdentifiableByName
func (ta RelationAliasesMap) addAlias(alias types.IdentifiableByName, target types.IdentifiableByName) error {
lowerName := strings.ToLower(alias.Name())
if _, ok := ta[lowerName]; ok {
return sql3.NewErrInternalf("unexpected duplicate alias name")
}
ta[lowerName] = target
return nil
}
func getRelationAliases(n types.PlanOperator, scope *OptimizerScope) (RelationAliasesMap, error) {
var aliases RelationAliasesMap
var aliasFn func(node types.PlanOperator) bool
var inspectErr error
aliasFn = func(node types.PlanOperator) bool {
if node == nil {
return false
}
if at, ok := node.(*PlanOpRelAlias); ok {
switch t := at.ChildOp.(type) {
case *PlanOpPQLTableScan:
inspectErr = aliases.addAlias(at, t)
case *PlanOpSubquery:
inspectErr = aliases.addAlias(at, t)
default:
panic(fmt.Sprintf("unexpected child node '%T'", at.ChildOp))
}
return false
}
switch node := node.(type) {
case *PlanOpPQLTableScan:
inspectErr = aliases.addAlias(node, node)
return false
}
return true
}
aliases = make(RelationAliasesMap)
InspectPlan(n, aliasFn)
if inspectErr != nil {
return nil, inspectErr
}
return aliases, inspectErr
}
// governs how far down filter push down can go
func filterPushdownChildSelector(c ParentContext) bool {
switch c.Parent.(type) {
case *PlanOpRelAlias:
//definitely don't go any further than alias
return false
}
return true
}
// governs how far down filter push down above tables can go
func filterPushdownAboveTablesChildSelector(c ParentContext) bool {
if !filterPushdownChildSelector(c) {
return false
}
switch c.Parent.(type) {
case *PlanOpFilter:
switch c.Operator.(type) {
case *PlanOpRelAlias, *PlanOpPQLTableScan:
return false
}
}
return true
}
// returns an expression given a list of expressions, if the list is > 2 expressions, all the individual
// expressions are ANDed together
func joinExprsWithAnd(exprs ...types.PlanExpression) types.PlanExpression {
switch len(exprs) {
case 0:
return nil
case 1:
return exprs[0]
default:
result := newBinOpPlanExpression(exprs[0], parser.AND, exprs[1], parser.NewDataTypeBool())
for _, e := range exprs[2:] {
result = newBinOpPlanExpression(result, parser.AND, e, parser.NewDataTypeBool())
}
return result
}
}
func removePushedDownConditions(ctx context.Context, a *ExecutionPlanner, node *PlanOpFilter, filters *filterSet) (types.PlanOperator, bool, error) {
if filters.handledCount() == 0 {
return node, true, nil
}
unhandled := filters.unhandledPredicates(ctx)
if len(unhandled) == 0 {
return node.ChildOp, false, nil
}
joinedExpr := joinExprsWithAnd(unhandled...)
return NewPlanOpFilter(a, joinedExpr, node.ChildOp), false, nil
}
func getRelation(node types.PlanOperator) types.IdentifiableByName {
var relation types.IdentifiableByName
InspectPlan(node, func(node types.PlanOperator) bool {
switch n := node.(type) {
case *PlanOpPQLTableScan:
relation = n
return false
}
return true
})
return relation
}
func pushdownFiltersToFilterableRelations(ctx context.Context, a *ExecutionPlanner, tableNode types.PlanOperator, scope *OptimizerScope, filters *filterSet, tableAliases RelationAliasesMap) (types.PlanOperator, bool, error) {
// only do this if it is an alias or a pql table scan
switch tableNode.(type) {
case *PlanOpRelAlias, *PlanOpPQLTableScan:
// continue
default:
return nil, true, sql3.NewErrInternalf("unexpected op type '%T'", tableNode)
}
table := getRelation(tableNode)
if table == nil {
return tableNode, true, nil
}
ft, ok := table.(types.FilteredRelation)
if !ok {
return tableNode, true, nil
}
// do we have any filters for this table? if not, bail...
tableFilters := filters.availableFiltersForTable(table.Name())
if len(tableFilters) == 0 {
return tableNode, true, nil
}
filters.markFiltersHandled(tableFilters...)
tableFilters, _, err := fixFieldRefIndexesOnExpressions(ctx, scope, a, tableNode.Schema(), tableFilters...)
if err != nil {
return nil, true, err
}
newOp, err := ft.UpdateFilters(joinExprsWithAnd(tableFilters...))
if err != nil {
return nil, true, err
}
return newOp, false, nil
}
func pushdownFiltersToAboveRelation(ctx context.Context, a *ExecutionPlanner, tableNode types.PlanOperator, scope *OptimizerScope, filters *filterSet) (types.PlanOperator, bool, error) {
table := getRelation(tableNode)
if table == nil {
return tableNode, true, nil
}
// reposition any remaining filters for a table to directly above the table itself
var pushedDownFilterExpression types.PlanExpression
if tableFilters := filters.availableFiltersForTable(table.Name()); len(tableFilters) > 0 {
filters.markFiltersHandled(tableFilters...)
handled, _, err := fixFieldRefIndexesOnExpressions(ctx, scope, a, tableNode.Schema(), tableFilters...)
if err != nil {
return nil, true, err
}
pushedDownFilterExpression = joinExprsWithAnd(handled...)
}
switch tableNode.(type) {
case *PlanOpRelAlias, *PlanOpPQLTableScan:
node := tableNode
if pushedDownFilterExpression != nil {
return NewPlanOpFilter(a, pushedDownFilterExpression, node), false, nil
}
return node, false, nil
default:
return nil, true, sql3.NewErrInternalf("unexpected op type '%T'", tableNode)
}
}
func pushdownFilters(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) {
tableAliases, err := getRelationAliases(n, scope)
if err != nil {
return nil, true, err
}
pushdownFiltersForFilterableRelations := func(n *PlanOpFilter, filters *filterSet) (types.PlanOperator, bool, error) {
return TransformPlanOpWithParent(n, filterPushdownChildSelector, func(c ParentContext) (types.PlanOperator, bool, error) {
switch node := c.Operator.(type) {
case *PlanOpFilter:
n, samePred, err := removePushedDownConditions(ctx, a, node, filters)
if err != nil {
return nil, true, err
}
n, sameFix, err := fixFieldRefIndexesForOperator(ctx, a, n, scope)
if err != nil {
return nil, true, err
}
return n, samePred && sameFix, nil
case *PlanOpRelAlias, *PlanOpPQLTableScan:
n, samePred, err := pushdownFiltersToFilterableRelations(ctx, a, node, scope, filters, tableAliases)
if err != nil {
return nil, true, err
}
n, sameFix, err := fixFieldRefIndexesForOperator(ctx, a, n, scope)
if err != nil {
return nil, true, err
}
return n, samePred && sameFix, nil
default:
return fixFieldRefIndexesForOperator(ctx, a, node, scope)
}
})
}
pushdownFiltersCloseToRelations := func(n types.PlanOperator, filters *filterSet) (types.PlanOperator, bool, error) {
return TransformPlanOpWithParent(n, filterPushdownAboveTablesChildSelector, func(c ParentContext) (types.PlanOperator, bool, error) {
switch node := c.Operator.(type) {
case *PlanOpFilter:
n, same, err := removePushedDownConditions(ctx, a, node, filters)
if err != nil {
return nil, true, err
}
if same {
return n, true, nil
}
n, _, err = fixFieldRefIndexesForOperator(ctx, a, n, scope)
if err != nil {
return nil, true, err
}
return n, false, nil
case *PlanOpRelAlias, *PlanOpPQLTableScan:
table, same, err := pushdownFiltersToAboveRelation(ctx, a, node, scope, filters)
if err != nil {
return nil, true, err
}
if same {
return node, true, nil
}
node, _, err = fixFieldRefIndexesForOperator(ctx, a, table, scope)
if err != nil {
return nil, true, err
}
return node, false, nil
default:
return fixFieldRefIndexesForOperator(ctx, a, node, scope)
}
})
}
return TransformPlanOp(n, func(node types.PlanOperator) (types.PlanOperator, bool, error) {
switch n := node.(type) {
case *PlanOpFilter:
filtersByTable := getFiltersByRelation(n)
filters := newFilterSet(n.Predicate, filtersByTable, tableAliases)
// first push down filters to any op that implements FilteredRelation
node, sameA, err := pushdownFiltersForFilterableRelations(n, filters)
if err != nil {
return nil, true, err
}
// second push down filters as close as possible to the relations they apply to
node, sameB, err := pushdownFiltersCloseToRelations(node, filters)
if err != nil {
return nil, true, err
}
return node, sameA && sameB, nil
default:
return n, true, nil
}
})
}
// getFiltersByRelation returns a map of relations name to filter expressions for the op provided
func getFiltersByRelation(n types.PlanOperator) map[string][]types.PlanExpression {
filters := make(map[string][]types.PlanExpression)
InspectPlan(n, func(node types.PlanOperator) bool {
switch nd := node.(type) {
case *PlanOpFilter:
fs := exprToRelationFilters(nd.Predicate)
for k, exprs := range fs {
filters[k] = append(filters[k], exprs...)
}
}
return true
})
return filters
}
// exprToRelationFilters returns a map of relation name to filter expressions for the expression
// passed after the expression is split on AND.
func exprToRelationFilters(expr types.PlanExpression) map[string][]types.PlanExpression {
filters := make(map[string][]types.PlanExpression)
for _, expr := range splitOnAnd(expr) {
var seenTables = make(map[string]bool)
var lastTable string
hasSubquery := false
InspectExpression(expr, func(e types.PlanExpression) bool {
f, ok := e.(*qualifiedRefPlanExpression)
if ok {
if !seenTables[f.tableName] {
seenTables[f.tableName] = true
lastTable = f.tableName
}
} else if _, isSubquery := e.(*subqueryPlanExpression); isSubquery {
hasSubquery = true
return false
}
return true
})
if len(seenTables) == 1 && !hasSubquery {
filters[lastTable] = append(filters[lastTable], expr)
}
}
return filters
}
// splitOnAnd breaks binops that are AND expressions into a list recursively
func splitOnAnd(expr types.PlanExpression) []types.PlanExpression {
binOp, ok := expr.(*binOpPlanExpression)
if !ok || binOp.op != parser.AND {
return []types.PlanExpression{
expr,
}
}
return append(
splitOnAnd(binOp.lhs),
splitOnAnd(binOp.rhs)...,
)
}
func tryToReplaceGroupByWithPQLAggregate(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) {
//bail if there are any joins
scans, err := hasOnlyTableScans(ctx, a, n, scope)
@ -189,6 +613,71 @@ func tryToReplaceGroupByWithPQLGroupBy(ctx context.Context, a *ExecutionPlanner,
return n, true, nil
}
// the semantic for accessing a timequantum field is to use the subtable() table valued function in a join
// rewrite queries that use this pattern to use the appropriate PQL call
func tryToRewriteSubtableJoins(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) {
//bail if there are no joins
joins := getNestedLoopOperators(ctx, a, n, scope)
if len(joins) == 0 {
return n, true, nil
}
//get the projections, we're going to need them later
projections := getPlanOpProjectionOperators(ctx, a, n, scope)
return TransformPlanOp(n, func(node types.PlanOperator) (types.PlanOperator, bool, error) {
switch nl := node.(type) {
case *PlanOpNestedLoops:
var tvf *PlanOpTableValuedFunction
// bail if the join does not have a tvf as one of the operators
tvftop, topok := nl.top.(*PlanOpTableValuedFunction)
tvfbottom, bottomok := nl.bottom.(*PlanOpTableValuedFunction)
//bail if both sides of the join are a tvf
if topok && bottomok {
return nl, true, nil
}
if topok {
tvf = tvftop
}
if bottomok {
tvf = tvfbottom
}
//if tvf == nil, then neither side is a tvf
if tvf == nil {
return nl, true, nil
}
//check it is the subtable() tvf
tvfCall, ok := tvf.callExpr.(*callPlanExpression)
if !ok {
return nl, true, nil
}
if !strings.EqualFold(tvfCall.name, "subtable") {
return nl, true, nil
}
// if there is no join condition, it's an extract; replace the 'value' reference
// with a reference with the first argument and remove the join
if nl.cond == nil {
// get the first argument column from the tvf
// for each of the projection operators, for each of the projections
// transform each of the referenced values with a the first arg
log.Printf("%T", projections)
}
// there is a join condition, make sure it is one that is permissible (range queries only?)
log.Printf("%T", tvf)
return nl, true, nil
default:
return nl, true, nil
}
})
}
func pushdownPQLTop(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) {
//bail if there are any joins
hasOnlyScans, err := hasOnlyTableScans(ctx, a, n, scope)
@ -501,3 +990,131 @@ func getTableScanOperators(ctx context.Context, a *ExecutionPlanner, n types.Pla
})
return tables
}
// inspects a plan op tree and returns a list (or error) of all the PlanOpProjection operators
func getPlanOpProjectionOperators(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) []*PlanOpProjection {
var projs []*PlanOpProjection
InspectPlan(n, func(node types.PlanOperator) bool {
switch nd := node.(type) {
case *PlanOpProjection:
projs = append(projs, nd)
return false
}
return true
})
return projs
}
// inspects a plan op tree and returns a list (or error) of all the PlanOpNestedLoops operators
func getNestedLoopOperators(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) []*PlanOpNestedLoops {
var joins []*PlanOpNestedLoops
InspectPlan(n, func(node types.PlanOperator) bool {
switch nd := node.(type) {
case *PlanOpNestedLoops:
joins = append(joins, nd)
return false
}
return true
})
return joins
}
func fixFieldRefIndexes(ctx context.Context, scope *OptimizerScope, a *ExecutionPlanner, schema types.Schema, exp types.PlanExpression) (types.PlanExpression, bool, error) {
return TransformExpr(exp, func(e types.PlanExpression) (types.PlanExpression, bool, error) {
switch e := e.(type) {
case *qualifiedRefPlanExpression:
for i, col := range schema {
newIndex := i
if e.Name() == col.Name && e.tableName == col.Table {
if newIndex != e.columnIndex {
// update the column index
e.columnIndex = newIndex
}
return e, true, nil
}
}
return nil, true, sql3.NewErrColumnNotFound(0, 0, e.Name())
}
return e, true, nil
})
}
func fixFieldRefIndexesOnExpressions(ctx context.Context, scope *OptimizerScope, a *ExecutionPlanner, schema types.Schema, expressions ...types.PlanExpression) ([]types.PlanExpression, bool, error) {
var result []types.PlanExpression
var res types.PlanExpression
var same bool
var err error
for i := range expressions {
e := expressions[i]
res, same, err = fixFieldRefIndexes(ctx, scope, a, schema, e)
if err != nil {
return nil, true, err
}
if !same {
if result == nil {
result = make([]types.PlanExpression, len(expressions))
copy(result, expressions)
}
result[i] = res
}
}
if len(result) > 0 {
return result, false, nil
}
return expressions, true, nil
}
func fixFieldRefIndexesForOperator(ctx context.Context, a *ExecutionPlanner, node types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) {
if _, ok := node.(types.ContainsExpressions); !ok {
return node, true, nil
}
var schemas []types.Schema
for _, child := range node.Children() {
schemas = append(schemas, child.Schema())
}
if len(schemas) < 1 {
return node, true, nil
}
n, sameC, err := TransformPlanOpExprsWithPlanOp(node, func(_ types.PlanOperator, e types.PlanExpression) (types.PlanExpression, bool, error) {
for _, schema := range schemas {
fixed, same, err := fixFieldRefIndexes(ctx, scope, a, schema, e)
if err == nil {
return fixed, same, nil
}
if strings.Contains(err.Error(), "unexpected!") {
continue
}
return nil, true, err
}
return e, true, nil
})
if err != nil {
return nil, true, err
}
sameJ := true
var cond types.PlanExpression
switch j := n.(type) {
case *PlanOpNestedLoops:
cond, sameJ, err = fixFieldRefIndexes(ctx, scope, a, j.Schema(), j.cond)
if err != nil {
return nil, true, err
}
if !sameJ {
n, err = j.NewWithExpressions(cond)
if err != nil {
return nil, true, err
}
}
}
return n, sameC && sameJ, nil
}

194
sql3/planner/planwalker.go
View file

@ -31,9 +31,9 @@ func PlanWalk(v PlanVisitor, op types.PlanOperator) {
v.VisitOperator(nil)
}
type planInspector func(types.PlanOperator) bool
type planInspectionFunction func(types.PlanOperator) bool
func (f planInspector) VisitOperator(op types.PlanOperator) PlanVisitor {
func (f planInspectionFunction) VisitOperator(op types.PlanOperator) PlanVisitor {
if f(op) {
return f
}
@ -43,7 +43,7 @@ func (f planInspector) VisitOperator(op types.PlanOperator) PlanVisitor {
// InspectPlan traverses the plan op graph depth-first order
// if f(op) returns true, InspectPlan invokes f recursively for each of the children of op,
// followed by a call of f(nil).
func InspectPlan(op types.PlanOperator, f planInspector) {
func InspectPlan(op types.PlanOperator, f planInspectionFunction) {
PlanWalk(f, op)
}
@ -78,10 +78,10 @@ func (f exprInspector) VisitExpr(e types.PlanExpression) ExprVisitor {
return nil
}
// WalkExpressions traverses the plan and calls sql.Walk on any expression it finds.
func WalkExpressions(v ExprVisitor, node types.PlanOperator) {
InspectPlan(node, func(node types.PlanOperator) bool {
if n, ok := node.(types.ContainsExpressions); ok {
// WalkExpressions traverses the plan and calls ExprWalk on any expression it finds
func WalkExpressions(v ExprVisitor, op types.PlanOperator) {
InspectPlan(op, func(operator types.PlanOperator) bool {
if n, ok := operator.(types.ContainsExpressions); ok {
for _, e := range n.Expressions() {
ExprWalk(v, e)
}
@ -92,8 +92,13 @@ func WalkExpressions(v ExprVisitor, node types.PlanOperator) {
// InspectExpressions traverses the plan and calls WalkExpressions on any
// expression it finds.
func InspectExpressions(node types.PlanOperator, f exprInspector) {
WalkExpressions(f, node)
func InspectExpressions(op types.PlanOperator, f exprInspector) {
WalkExpressions(f, op)
}
// InspectExpression traverses expressoins in depth-first order
func InspectExpression(expr types.PlanExpression, f func(expr types.PlanExpression) bool) {
ExprWalk(exprInspector(f), expr)
}
//-----------------------------------------------------------------------------
@ -103,19 +108,19 @@ func InspectExpressions(node types.PlanOperator, f exprInspector) {
type PlanOpExprVisitor interface {
// VisitPlanOpExpr method is invoked for each expr encountered by Walk. If the result Visitor is not nil, Walk visits each of
// the children of the expr with that visitor, followed by a call of VisitPlanOpExpr(nil, nil) to the returned visitor.
VisitPlanOpExpr(node types.PlanOperator, expression types.PlanExpression) PlanOpExprVisitor
VisitPlanOpExpr(op types.PlanOperator, expression types.PlanExpression) PlanOpExprVisitor
}
// ExprWithPlanOpWalk traverses the expression tree in depth-first order. It starts by calling v.VisitPlanOpExpr(op, expr); expr must
// not be nil. If the visitor returned by v.VisitPlanOpExpr(op, expr) is not nil, Walk is invoked recursively with the returned
// visitor for each children of the expr, followed by a call of v.VisitPlanOpExpr(nil, nil) to the returned visitor.
func ExprWithPlanOpWalk(v PlanOpExprVisitor, n types.PlanOperator, expr types.PlanExpression) {
if v = v.VisitPlanOpExpr(n, expr); v == nil {
func ExprWithPlanOpWalk(v PlanOpExprVisitor, op types.PlanOperator, expr types.PlanExpression) {
if v = v.VisitPlanOpExpr(op, expr); v == nil {
return
}
for _, child := range expr.Children() {
ExprWithPlanOpWalk(v, n, child)
ExprWithPlanOpWalk(v, op, child)
}
v.VisitPlanOpExpr(nil, nil)
@ -123,19 +128,19 @@ func ExprWithPlanOpWalk(v PlanOpExprVisitor, n types.PlanOperator, expr types.Pl
type exprWithNodeInspector func(types.PlanOperator, types.PlanExpression) bool
func (f exprWithNodeInspector) VisitPlanOpExpr(n types.PlanOperator, e types.PlanExpression) PlanOpExprVisitor {
if f(n, e) {
func (f exprWithNodeInspector) VisitPlanOpExpr(op types.PlanOperator, expr types.PlanExpression) PlanOpExprVisitor {
if f(op, expr) {
return f
}
return nil
}
// WalkExpressionsWithPlanOp traverses the plan and calls ExprWithPlanOpWalk on any expression it finds.
func WalkExpressionsWithPlanOp(v PlanOpExprVisitor, n types.PlanOperator) {
InspectPlan(n, func(n types.PlanOperator) bool {
if expressioner, ok := n.(types.ContainsExpressions); ok {
func WalkExpressionsWithPlanOp(v PlanOpExprVisitor, op types.PlanOperator) {
InspectPlan(op, func(operator types.PlanOperator) bool {
if expressioner, ok := operator.(types.ContainsExpressions); ok {
for _, e := range expressioner.Expressions() {
ExprWithPlanOpWalk(v, n, e)
ExprWithPlanOpWalk(v, operator, e)
}
}
return true
@ -143,25 +148,23 @@ func WalkExpressionsWithPlanOp(v PlanOpExprVisitor, n types.PlanOperator) {
}
// InspectExpressionsWithPlanOp traverses the plan and calls f on any expression it finds.
func InspectExpressionsWithPlanOp(node types.PlanOperator, f exprWithNodeInspector) {
WalkExpressionsWithPlanOp(f, node)
func InspectExpressionsWithPlanOp(op types.PlanOperator, f exprWithNodeInspector) {
WalkExpressionsWithPlanOp(f, op)
}
// PlanOpFunc is a function that given a plan op will return either a transformed plan op or the original plan op.
// PlanOpTransformFunc is a function that given a plan op will return either a transformed plan op or the original plan op.
// If there was a transformation, the bool will be true, and an error if there was an error
type PlanOpFunc func(n types.PlanOperator) (types.PlanOperator, bool, error)
type PlanOpTransformFunc func(op types.PlanOperator) (types.PlanOperator, bool, error)
// TransformPlanOp applies a transformation function to the given plan op graph
func TransformPlanOp(op types.PlanOperator, f PlanOpFunc) (types.PlanOperator, bool, error) {
func TransformPlanOp(op types.PlanOperator, f PlanOpTransformFunc) (types.PlanOperator, bool, error) {
children := op.Children()
if len(children) == 0 {
return f(op)
}
var (
newChildren []types.PlanOperator
)
var newChildren []types.PlanOperator
for i := range children {
child := children[i]
@ -179,56 +182,117 @@ func TransformPlanOp(op types.PlanOperator, f PlanOpFunc) (types.PlanOperator, b
}
var err error
sameC := true
sameChildren := true
if len(newChildren) > 0 {
sameC = false
sameChildren = false
op, err = op.WithChildren(newChildren...)
if err != nil {
return nil, true, err
}
}
op, sameN, err := f(op)
op, sameOperator, err := f(op)
if err != nil {
return nil, true, err
}
return op, sameC && sameN, nil
return op, sameChildren && sameOperator, nil
}
// ParentContext is a struct that enables transformation functions to include a parent operator
type ParentContext struct {
Operator types.PlanOperator
Parent types.PlanOperator
ChildCount int
}
type ParentContextFunc func(c ParentContext) (types.PlanOperator, bool, error)
type ParentSelectorFunc func(c ParentContext) bool
func TransformPlanOpWithParent(op types.PlanOperator, s ParentSelectorFunc, f ParentContextFunc) (types.PlanOperator, bool, error) {
return planOpWithParentHelper(ParentContext{op, nil, -1}, s, f)
}
func planOpWithParentHelper(c ParentContext, s ParentSelectorFunc, f ParentContextFunc) (types.PlanOperator, bool, error) {
operator := c.Operator
children := operator.Children()
if len(children) == 0 {
return f(c)
}
var (
newChildren []types.PlanOperator
err error
)
for i := range children {
child := children[i]
cc := ParentContext{child, operator, i}
if s == nil || s(cc) {
child, same, err := planOpWithParentHelper(cc, s, f)
if err != nil {
return nil, true, err
}
if !same {
if newChildren == nil {
newChildren = make([]types.PlanOperator, len(children))
copy(newChildren, children)
}
newChildren[i] = child
}
}
}
sameChildren := true
if len(newChildren) > 0 {
sameChildren = false
operator, err = operator.WithChildren(newChildren...)
if err != nil {
return nil, true, err
}
}
operator, sameOperator, err := f(ParentContext{operator, c.Parent, c.ChildCount})
if err != nil {
return nil, true, err
}
return operator, sameChildren && sameOperator, nil
}
// ExprWithPlanOpFunc is a function that given an expression and the node
// that contains it, will return that expression as is or transformed
// along with an error, if any.
type ExprWithPlanOpFunc func(types.PlanOperator, types.PlanExpression) (types.PlanExpression, bool, error)
type ExprWithPlanOpFunc func(op types.PlanOperator, expr types.PlanExpression) (types.PlanExpression, bool, error)
// ExprFunc is a function that given an expression will return that
// expression as is or transformed, or bool to indicate
// whether the expression was modified, and an error or nil.
type ExprFunc func(e types.PlanExpression) (types.PlanExpression, bool, error)
type ExprFunc func(expr types.PlanExpression) (types.PlanExpression, bool, error)
// TransformPlanOpExprsWithPlanOp applies a transformation function to all expressions on the given plan operator from the bottom up in the context of the plan operator
func TransformPlanOpExprsWithPlanOp(op types.PlanOperator, f ExprWithPlanOpFunc) (types.PlanOperator, bool, error) {
return TransformPlanOp(op, func(n types.PlanOperator) (types.PlanOperator, bool, error) {
return SinglePlanOpExprsWithPlanOp(n, f)
return TransformSinglePlanOpExprsInPlanOpContext(n, f)
})
}
// TransformPlanOpExprs applies a transformation function to all expressions on the given plan operator from the bottom up
func TransformPlanOpExprs(op types.PlanOperator, f ExprFunc) (types.PlanOperator, bool, error) {
return TransformPlanOpExprsWithPlanOp(op, func(n types.PlanOperator, e types.PlanExpression) (types.PlanExpression, bool, error) {
return f(e)
return TransformPlanOpExprsWithPlanOp(op, func(operator types.PlanOperator, expr types.PlanExpression) (types.PlanExpression, bool, error) {
return f(expr)
})
}
// SinglePlanOpExprsWithPlanOp applies a transformation function to all expressions on a given plan operator in the context of that plan operator
func SinglePlanOpExprsWithPlanOp(n types.PlanOperator, f ExprWithPlanOpFunc) (types.PlanOperator, bool, error) {
ne, ok := n.(types.ContainsExpressions)
// TransformSinglePlanOpExprsInPlanOpContext applies a transformation function to all expressions on a given plan operator in the context of that plan operator
func TransformSinglePlanOpExprsInPlanOpContext(op types.PlanOperator, f ExprWithPlanOpFunc) (types.PlanOperator, bool, error) {
ne, ok := op.(types.ContainsExpressions)
if !ok {
return n, true, nil
return op, true, nil
}
exprs := ne.Expressions()
if len(exprs) == 0 {
return n, true, nil
return op, true, nil
}
var (
@ -238,7 +302,7 @@ func SinglePlanOpExprsWithPlanOp(n types.PlanOperator, f ExprWithPlanOpFunc) (ty
for i := range exprs {
e := exprs[i]
e, same, err := TransformExprWithPlanOp(n, e, f)
e, same, err := TransformExprWithPlanOp(op, e, f)
if err != nil {
return nil, true, err
}
@ -252,25 +316,25 @@ func SinglePlanOpExprsWithPlanOp(n types.PlanOperator, f ExprWithPlanOpFunc) (ty
}
if len(newExprs) > 0 {
n, err = ne.WithExpressions(newExprs...)
op, err = ne.NewWithExpressions(newExprs...)
if err != nil {
return nil, true, err
}
return n, false, nil
return op, false, nil
}
return n, true, nil
return op, true, nil
}
// TransformSinglePlanOpExpressions applies a transformation function to all expressions on the given plan operator
func TransformSinglePlanOpExpressions(o types.PlanOperator, f ExprFunc) (types.PlanOperator, bool, error) {
e, ok := o.(types.ContainsExpressions)
func TransformSinglePlanOpExpressions(op types.PlanOperator, f ExprFunc) (types.PlanOperator, bool, error) {
e, ok := op.(types.ContainsExpressions)
if !ok {
return o, true, nil
return op, true, nil
}
exprs := e.Expressions()
if len(exprs) == 0 {
return o, true, nil
return op, true, nil
}
var newExprs []types.PlanExpression
@ -289,20 +353,20 @@ func TransformSinglePlanOpExpressions(o types.PlanOperator, f ExprFunc) (types.P
}
}
if len(newExprs) > 0 {
n, err := e.WithExpressions(newExprs...)
n, err := e.NewWithExpressions(newExprs...)
if err != nil {
return nil, true, err
}
return n, false, nil
}
return o, true, nil
return op, true, nil
}
// TransformExpr applies a transformation function to an expression
func TransformExpr(e types.PlanExpression, f ExprFunc) (types.PlanExpression, bool, error) {
children := e.Children()
func TransformExpr(expr types.PlanExpression, f ExprFunc) (types.PlanExpression, bool, error) {
children := expr.Children()
if len(children) == 0 {
return f(e)
return f(expr)
}
var (
@ -325,20 +389,20 @@ func TransformExpr(e types.PlanExpression, f ExprFunc) (types.PlanExpression, bo
}
}
sameC := true
sameChildren := true
if len(newChildren) > 0 {
sameC = false
e, err = e.WithChildren(newChildren...)
sameChildren = false
expr, err = expr.WithChildren(newChildren...)
if err != nil {
return nil, true, err
}
}
e, sameN, err := f(e)
expr, sameExpr, err := f(expr)
if err != nil {
return nil, true, err
}
return e, sameC && sameN, nil
return expr, sameChildren && sameExpr, nil
}
// TransformExprWithPlanOp applies a transformation function to an expression in the context of a plan operator
@ -355,11 +419,11 @@ func TransformExprWithPlanOp(n types.PlanOperator, e types.PlanExpression, f Exp
for i := 0; i < len(children); i++ {
c := children[i]
c, sameC, err := TransformExprWithPlanOp(n, c, f)
c, same, err := TransformExprWithPlanOp(n, c, f)
if err != nil {
return nil, true, err
}
if !sameC {
if !same {
if newChildren == nil {
newChildren = make([]types.PlanExpression, len(children))
copy(newChildren, children)
@ -368,18 +432,18 @@ func TransformExprWithPlanOp(n types.PlanOperator, e types.PlanExpression, f Exp
}
}
sameC := true
sameChilren := true
if len(newChildren) > 0 {
sameC = false
sameChilren = false
e, err = e.WithChildren(newChildren...)
if err != nil {
return nil, true, err
}
}
e, sameN, err := f(n, e)
e, sameExpr, err := f(n, e)
if err != nil {
return nil, true, err
}
return e, sameC && sameN, nil
return e, sameChilren && sameExpr, nil
}

20
sql3/planner/types/operator.go
View file

@ -41,13 +41,8 @@ type ContainsExpressions interface {
// returns the list of expressions contained by the plan operator
Expressions() []PlanExpression
// WithExpressions returns a new operator with expressions replaced
WithExpressions(...PlanExpression) (PlanOperator, error)
}
// SchemaObject exposes an ObjectName() for operators that iterate on schema objects
type SchemaObject interface {
ObjectName() string
// NewWithExpressions returns a new operator with expressions replaced
NewWithExpressions(exprs ...PlanExpression) (PlanOperator, error)
}
// PlannerColumn is the definition of a column returned as a set from each operator
@ -57,6 +52,17 @@ type PlannerColumn struct {
Type parser.ExprDataType
}
// Relation is an interface to something that can be treated as a relation
type Relation interface {
Name() string
}
// FilteredRelation is an interface to something that can be treated as a relation that can be filtered
type FilteredRelation interface {
Relation
UpdateFilters(filterCondition PlanExpression) (PlanOperator, error)
}
// Schema is the definition a set of columns from each operator
type Schema []*PlannerColumn

6
sql3/planner/types/planexpression.go
View file

@ -6,7 +6,7 @@ import (
"github.com/featurebasedb/featurebase/v3/sql3/parser"
)
//TODO(pok) we can get rid of this - we have expression types for all of these now...
// TODO(pok) we can get rid of this - we have expression types for all of these now...
type AggregateFunctionType int
// The list of AggregateFunction.
@ -56,7 +56,7 @@ type Aggregable interface {
AggAdditionalExpr() []PlanExpression
}
// Interface to an expression that is a reference to a schema object
type SchemaIdentifiable interface {
// Interface to something that can be identified by a name
type IdentifiableByName interface {
Name() string
}

64
sql3/sql3.ebnf
View file

@ -16,6 +16,7 @@ statement = show_tables
| alter_table_stmt
| select_stmt
| insert_stmt
| bulk_insert_stmt
| delete_stmt ;
(*
@ -111,7 +112,28 @@ rename_column = "RENAME", [ "COLUMN" ], identifier, "TO", identifier ;
Inserts data into a FeatureBase table.
*)
insert_stmt = "INSERT", "INTO", identifier, "(", identifier, { identifier, "," }, ")", "VALUES", "(", expr, { expr, "," }, ")" ;
insert_stmt = ( "INSERT" | "REPLACE" ), "INTO", identifier, [ "(", identifier, { identifier, "," }, ")" ], "VALUES", "(", expr, { expr, "," }, ")" ;
(*
BULK INSERT
----------------
Bulk inserts data into a FeatureBase table.
*)
bulk_insert_stmt = "BULK", "INSERT", identifier, "FROM", string_literal, [ "WITH", bulk_insert_option, { bulk_insert_option } ];
bulk_insert_option =
"BATCHSIZE", integer_literal
| "ROWSLIMIT", integer_literal
| "FORMAT", "CSV"
| bulk_insert_map_option ;
bulk_insert_map_option = "MAP", map_id_option | map_column_option ;
map_id_option = "_ID", "TO", ( "AUTOINCREMENT" | expr, { ",", expr } );
map_column_option = "OFFSET", integer_literal, "TO", column_name ;
(*
@ -141,7 +163,8 @@ column_alias = identifier ;
from_clause = "FROM", table_or_subquery, { ",", table_or_subquery } ;
table_or_subquery = identifier, [ [ "AS" ], table_alias ], [ table_option ]
table_or_subquery =
( identifier | table_valued_function ), [ [ "AS" ], table_alias ], [ table_option ]
| "(", select_stmt, ")", [ [ "AS" ], table_alias ] ;
table_alias = identifier ;
@ -156,39 +179,13 @@ group_by_clause = "GROUP", "BY", expr, { ",", expr }, [ "HAVING", expr ] ;
Expressions
----------------
timequantums - modelling
* let stringsetcolq refer to a set with a timequantum column
* the expression stringsetcolq is an array/table? of tuple of (timestamp, set)
* when you refer to stringsetcolq by name, this is shorthand for
"the set value for the latest timestamp"
* if we were to use dotted notation to refer to the components of the tuple,
we could refer to the components of the tuple e.g.
- stringsetcolq.timestamp
- stringsetcolq.value
* but stringsetcolq is an array of tuples, so assuming some array type notation
- stringsetcolq[subscript].timestamp
- stringsetcolq[subscript].value
* this is dumb
* what if, when you refer to stringsetcolq by name, this is shorthand for
"the set value for the latest timestamp" - the tuple not the array
* the underlying 'table' could be modelled as (_id, timestamp, set)
* if you want access to that table, we could use a table valued function that would enable us to use
cross apply etc.
* you want to do range queries on the timestamp
* you want to be able to see what timestamps you have
* you want to be able to see what sets and set values you have
* do we ever want to be able to do this with any arbitary range queryable type, right
now we have one additional dimension, would we ever want more? would we want additional values?
*)
expr = integer_literal
| string_literal
| decimal_literal
| set_literal
| tuple_literal
| date_literal
| [ table_name, "." ], column_name
| unary_op, expr
@ -228,6 +225,8 @@ binary_op = "="
set_literal = "[", expr, { ",", expr }, "]" ;
tuple_literal = "{", expr, { ",", expr }, "}" ;
date_literal = rfc_3339
| "CURRENT_DATE"
| "CURRENT_TIMESTAMP" ;
@ -237,7 +236,8 @@ table_name = identifier ;
column_name = identifier ;
function_call = agg_function
| non_agg_function ;
| non_agg_function
| table_valued_function ;
agg_function = ( "AVG" | "COUNT" | "MAX" | "MIN" | "SUM" ), "(", ( ( [ "DISTINCT" ], expr ) | "*" ), ")"
| "PERCENTILE", "(", expr, ",", expr, ")" ;
@ -248,6 +248,10 @@ non_agg_function =
| "SETCONTAINSANY" , "(", expr, ",", expr, ")"
| "DATEPART" , "(", expr, ",", expr, ")" ;
table_valued_function =
"SUBTABLE" , "(", table_name, ".", column_name, ")";
identifier = letter , { letter | digit | "_" } ;
letter = "A" | "B" | "C" | "D" | "E" | "F" | "G"