From fcce9d05a9dd6f2f192bca72ec3764244601a9c3 Mon Sep 17 00:00:00 2001 From: pokeeffe-molecula <85502298+pokeeffe-molecula@users.noreply.github.com> Date: Tue, 6 Sep 2022 11:16:11 -0500 Subject: [PATCH] sql3 changes (#2211) * 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 --- api.go | 2 + ctl/cli.go | 28 +- sql3/errors.go | 8 + sql3/parser/ast.go | 194 +++++- sql3/parser/astdatatype.go | 111 +++- sql3/parser/parser.go | 201 +++++- sql3/parser/scanner.go | 4 + sql3/parser/token.go | 4 + sql3/planner/compilebulkinsert.go | 188 +++++- sql3/planner/compileinsert.go | 4 - sql3/planner/compileselect.go | 113 +++- sql3/planner/executionplanner_test.go | 1 + sql3/planner/expression.go | 119 +++- sql3/planner/expressionanalyzer.go | 31 + sql3/planner/expressionanalyzercall.go | 3 + sql3/planner/expressionpql.go | 10 + sql3/planner/expressiontypes.go | 75 ++- sql3/planner/inbuiltfunctionstable.go | 66 ++ sql3/planner/opaltertable.go | 1 + sql3/planner/opbulkinsert.go | 588 +++++++++++++++- sql3/planner/opcreatetable.go | 1 + sql3/planner/opdistinct.go | 5 +- sql3/planner/opdroptable.go | 5 +- sql3/planner/opfeaturebasecolumns.go | 3 +- sql3/planner/opfeaturebasetables.go | 1 + sql3/planner/opfilter.go | 97 +++ sql3/planner/opgroupby.go | 1 + sql3/planner/opinsert.go | 93 ++- sql3/planner/opnestedloops.go | 165 +---- sql3/planner/opnulltable.go | 4 +- sql3/planner/oporderby.go | 1 + sql3/planner/oppqlaggregate.go | 1 + sql3/planner/oppqlgroupby.go | 3 +- sql3/planner/oppqlmultiaggregate.go | 1 + sql3/planner/oppqlmultigroupby.go | 1 + .../{optablescan.go => oppqltablescan.go} | 20 +- sql3/planner/opprojection.go | 9 +- sql3/planner/opquery.go | 8 +- sql3/planner/oprelalias.go | 80 +++ sql3/planner/opsubquery.go | 8 +- sql3/planner/optablevaluedfunction.go | 82 +++ sql3/planner/optop.go | 5 +- sql3/planner/planoptimizer.go | 627 +++++++++++++++++- sql3/planner/planwalker.go | 194 ++++-- sql3/planner/types/operator.go | 20 +- sql3/planner/types/planexpression.go | 6 +- sql3/sql3.ebnf | 64 +- 47 files changed, 2883 insertions(+), 373 deletions(-) create mode 100644 sql3/planner/inbuiltfunctionstable.go create mode 100644 sql3/planner/opfilter.go rename sql3/planner/{optablescan.go => oppqltablescan.go} (93%) create mode 100644 sql3/planner/oprelalias.go create mode 100644 sql3/planner/optablevaluedfunction.go diff --git a/api.go b/api.go index e5ac71d55..27c8b23c3 100644 --- a/api.go +++ b/api.go @@ -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 { diff --git a/ctl/cli.go b/ctl/cli.go index 922645435..2cea97bdd 100644 --- a/ctl/cli.go +++ b/ctl/cli.go @@ -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 { diff --git a/sql3/errors.go b/sql3/errors.go index c75af0f98..4e03873ee 100644 --- a/sql3/errors.go +++ b/sql3/errors.go @@ -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, diff --git a/sql3/parser/ast.go b/sql3/parser/ast.go index 0ebaca648..39bb3141d 100644 --- a/sql3/parser/ast.go +++ b/sql3/parser/ast.go @@ -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 diff --git a/sql3/parser/astdatatype.go b/sql3/parser/astdatatype.go index c08ce2720..3297c3ad9 100644 --- a/sql3/parser/astdatatype.go +++ b/sql3/parser/astdatatype.go @@ -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 { } diff --git a/sql3/parser/parser.go b/sql3/parser/parser.go index f4c3ce643..d8227894d 100644 --- a/sql3/parser/parser.go +++ b/sql3/parser/parser.go @@ -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 { diff --git a/sql3/parser/scanner.go b/sql3/parser/scanner.go index 69fb912c3..3f07f9274 100644 --- a/sql3/parser/scanner.go +++ b/sql3/parser/scanner.go @@ -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 '!': diff --git a/sql3/parser/token.go b/sql3/parser/token.go index 17c18022c..7b6c68b6b 100644 --- a/sql3/parser/token.go +++ b/sql3/parser/token.go @@ -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: "=", diff --git a/sql3/planner/compilebulkinsert.go b/sql3/planner/compilebulkinsert.go index 6bfe0d9d7..e7960e3b6 100644 --- a/sql3/planner/compilebulkinsert.go +++ b/sql3/planner/compilebulkinsert.go @@ -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 } diff --git a/sql3/planner/compileinsert.go b/sql3/planner/compileinsert.go index e8fcaa5b6..ce4dec97d 100644 --- a/sql3/planner/compileinsert.go +++ b/sql3/planner/compileinsert.go @@ -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()) diff --git a/sql3/planner/compileselect.go b/sql3/planner/compileselect.go index b209e5ecc..8d80170ab 100644 --- a/sql3/planner/compileselect.go +++ b/sql3/planner/compileselect.go @@ -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 } diff --git a/sql3/planner/executionplanner_test.go b/sql3/planner/executionplanner_test.go index 59b12ad76..1cfb719e7 100644 --- a/sql3/planner/executionplanner_test.go +++ b/sql3/planner/executionplanner_test.go @@ -589,6 +589,7 @@ func TestPlanner_AlterTable(t *testing.T) { }) } + func TestPlanner_DropTable(t *testing.T) { c := test.MustRunCluster(t, 1) defer c.Close() diff --git a/sql3/planner/expression.go b/sql3/planner/expression.go index 5244f9168..5cad8d227 100644 --- a/sql3/planner/expression.go +++ b/sql3/planner/expression.go @@ -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") diff --git a/sql3/planner/expressionanalyzer.go b/sql3/planner/expressionanalyzer.go index 46b4a20ce..58d056ecd 100644 --- a/sql3/planner/expressionanalyzer.go +++ b/sql3/planner/expressionanalyzer.go @@ -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()) } diff --git a/sql3/planner/expressionanalyzercall.go b/sql3/planner/expressionanalyzercall.go index c95288e40..d667f5652 100644 --- a/sql3/planner/expressionanalyzercall.go +++ b/sql3/planner/expressionanalyzercall.go @@ -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) } diff --git a/sql3/planner/expressionpql.go b/sql3/planner/expressionpql.go index 7baf32312..74052b0b3 100644 --- a/sql3/planner/expressionpql.go +++ b/sql3/planner/expressionpql.go @@ -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) } diff --git a/sql3/planner/expressiontypes.go b/sql3/planner/expressiontypes.go index b8c4353c2..118b2425e 100644 --- a/sql3/planner/expressiontypes.go +++ b/sql3/planner/expressiontypes.go @@ -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) { diff --git a/sql3/planner/inbuiltfunctionstable.go b/sql3/planner/inbuiltfunctionstable.go new file mode 100644 index 000000000..41e7e55c6 --- /dev/null +++ b/sql3/planner/inbuiltfunctionstable.go @@ -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 +} diff --git a/sql3/planner/opaltertable.go b/sql3/planner/opaltertable.go index 237d29112..a8beddf98 100644 --- a/sql3/planner/opaltertable.go +++ b/sql3/planner/opaltertable.go @@ -29,6 +29,7 @@ func NewPlanOpAlterTable(p *ExecutionPlanner, tableName string, operation alterO oldColumnName: oldColumnName, newColumnName: newColumnName, columnDef: columnDef, + warnings: make([]string, 0), } } diff --git a/sql3/planner/opbulkinsert.go b/sql3/planner/opbulkinsert.go index 3bc67e545..2809f76ad 100644 --- a/sql3/planner/opbulkinsert.go +++ b/sql3/planner/opbulkinsert.go @@ -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 +} diff --git a/sql3/planner/opcreatetable.go b/sql3/planner/opcreatetable.go index 6fa4af132..29ed595b9 100644 --- a/sql3/planner/opcreatetable.go +++ b/sql3/planner/opcreatetable.go @@ -30,6 +30,7 @@ func NewPlanOpCreateTable(p *ExecutionPlanner, tableName string, failIfExists bo isKeyed: isKeyed, keyPartitions: keyPartitions, columns: columns, + warnings: make([]string, 0), } } diff --git a/sql3/planner/opdistinct.go b/sql3/planner/opdistinct.go index 03ede8067..288ff6346 100644 --- a/sql3/planner/opdistinct.go +++ b/sql3/planner/opdistinct.go @@ -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), } } diff --git a/sql3/planner/opdroptable.go b/sql3/planner/opdroptable.go index 14bfd96c6..8a80d4ed8 100644 --- a/sql3/planner/opdroptable.go +++ b/sql3/planner/opdroptable.go @@ -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), } } diff --git a/sql3/planner/opfeaturebasecolumns.go b/sql3/planner/opfeaturebasecolumns.go index ac4e6151c..a4ab8deab 100644 --- a/sql3/planner/opfeaturebasecolumns.go +++ b/sql3/planner/opfeaturebasecolumns.go @@ -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 } diff --git a/sql3/planner/opfeaturebasetables.go b/sql3/planner/opfeaturebasetables.go index 15c6511ac..eb2080c43 100644 --- a/sql3/planner/opfeaturebasetables.go +++ b/sql3/planner/opfeaturebasetables.go @@ -22,6 +22,7 @@ type PlanOpFeatureBaseTables struct { func NewPlanOpFeatureBaseTables(indexInfo []*pilosa.IndexInfo) *PlanOpFeatureBaseTables { return &PlanOpFeatureBaseTables{ indexInfo: indexInfo, + warnings: make([]string, 0), } } diff --git a/sql3/planner/opfilter.go b/sql3/planner/opfilter.go new file mode 100644 index 000000000..dff39e24a --- /dev/null +++ b/sql3/planner/opfilter.go @@ -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) +} diff --git a/sql3/planner/opgroupby.go b/sql3/planner/opgroupby.go index 978e76785..3e74e54d1 100644 --- a/sql3/planner/opgroupby.go +++ b/sql3/planner/opgroupby.go @@ -25,6 +25,7 @@ func NewPlanOpGroupBy(aggregates []types.PlanExpression, groupByExprs []types.Pl ChildOp: child, Aggregates: aggregates, GroupByExprs: groupByExprs, + warnings: make([]string, 0), } } diff --git a/sql3/planner/opinsert.go b/sql3/planner/opinsert.go index ea2ea17b2..b981897aa 100644 --- a/sql3/planner/opinsert.go +++ b/sql3/planner/opinsert.go @@ -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) } } diff --git a/sql3/planner/opnestedloops.go b/sql3/planner/opnestedloops.go index 364261a2d..2a415412e 100644 --- a/sql3/planner/opnestedloops.go +++ b/sql3/planner/opnestedloops.go @@ -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) { diff --git a/sql3/planner/opnulltable.go b/sql3/planner/opnulltable.go index 7825bfd62..3ab206afe 100644 --- a/sql3/planner/opnulltable.go +++ b/sql3/planner/opnulltable.go @@ -16,7 +16,9 @@ type PlanOpNullTable struct { } func NewPlanOpNullTable() *PlanOpNullTable { - return &PlanOpNullTable{} + return &PlanOpNullTable{ + warnings: make([]string, 0), + } } func (p *PlanOpNullTable) Schema() types.Schema { diff --git a/sql3/planner/oporderby.go b/sql3/planner/oporderby.go index 23e26f6f3..4f22e629c 100644 --- a/sql3/planner/oporderby.go +++ b/sql3/planner/oporderby.go @@ -48,6 +48,7 @@ func NewPlanOpOrderBy(orderByFields []*OrderByExpression, child types.PlanOperat return &PlanOpOrderBy{ ChildOp: child, orderByFields: orderByFields, + warnings: make([]string, 0), } } diff --git a/sql3/planner/oppqlaggregate.go b/sql3/planner/oppqlaggregate.go index 62183457d..bcfff76cc 100644 --- a/sql3/planner/oppqlaggregate.go +++ b/sql3/planner/oppqlaggregate.go @@ -29,6 +29,7 @@ func NewPlanOpPQLAggregate(p *ExecutionPlanner, tableName string, aggregate type tableName: tableName, filter: filter, aggregate: aggregate, + warnings: make([]string, 0), } } diff --git a/sql3/planner/oppqlgroupby.go b/sql3/planner/oppqlgroupby.go index 3cab51ae5..010f71c99 100644 --- a/sql3/planner/oppqlgroupby.go +++ b/sql3/planner/oppqlgroupby.go @@ -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) } diff --git a/sql3/planner/oppqlmultiaggregate.go b/sql3/planner/oppqlmultiaggregate.go index 72f718f0c..bd681f826 100644 --- a/sql3/planner/oppqlmultiaggregate.go +++ b/sql3/planner/oppqlmultiaggregate.go @@ -20,6 +20,7 @@ func NewPlanOpPQLMultiAggregate(p *ExecutionPlanner, operators []*PlanOpPQLAggre return &PlanOpPQLMultiAggregate{ planner: p, operators: operators, + warnings: make([]string, 0), } } diff --git a/sql3/planner/oppqlmultigroupby.go b/sql3/planner/oppqlmultigroupby.go index 5392e7ff6..13bfc73b1 100644 --- a/sql3/planner/oppqlmultigroupby.go +++ b/sql3/planner/oppqlmultigroupby.go @@ -25,6 +25,7 @@ func NewPlanOpPQLMultiGroupBy(p *ExecutionPlanner, operators []*PlanOpPQLGroupBy planner: p, operators: operators, groupByExprs: groupByExprs, + warnings: make([]string, 0), } } diff --git a/sql3/planner/optablescan.go b/sql3/planner/oppqltablescan.go similarity index 93% rename from sql3/planner/optablescan.go rename to sql3/planner/oppqltablescan.go index 73c5856e6..06602ae32 100644 --- a/sql3/planner/optablescan.go +++ b/sql3/planner/oppqltablescan.go @@ -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) } diff --git a/sql3/planner/opprojection.go b/sql3/planner/opprojection.go index 9ec703bcc..3d0c3d256 100644 --- a/sql3/planner/opprojection.go +++ b/sql3/planner/opprojection.go @@ -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{ diff --git a/sql3/planner/opquery.go b/sql3/planner/opquery.go index c755919c3..27a8e956a 100644 --- a/sql3/planner/opquery.go +++ b/sql3/planner/opquery.go @@ -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{} { diff --git a/sql3/planner/oprelalias.go b/sql3/planner/oprelalias.go new file mode 100644 index 000000000..bcf3b42ce --- /dev/null +++ b/sql3/planner/oprelalias.go @@ -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 +} diff --git a/sql3/planner/opsubquery.go b/sql3/planner/opsubquery.go index 743166c96..5cf616be4 100644 --- a/sql3/planner/opsubquery.go +++ b/sql3/planner/opsubquery.go @@ -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 "" } diff --git a/sql3/planner/optablevaluedfunction.go b/sql3/planner/optablevaluedfunction.go new file mode 100644 index 000000000..23cee50b7 --- /dev/null +++ b/sql3/planner/optablevaluedfunction.go @@ -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 + +} diff --git a/sql3/planner/optop.go b/sql3/planner/optop.go index 1562036fe..cd5836956 100644 --- a/sql3/planner/optop.go +++ b/sql3/planner/optop.go @@ -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), } } diff --git a/sql3/planner/planoptimizer.go b/sql3/planner/planoptimizer.go index 742b129ab..08cca55d1 100644 --- a/sql3/planner/planoptimizer.go +++ b/sql3/planner/planoptimizer.go @@ -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 +} diff --git a/sql3/planner/planwalker.go b/sql3/planner/planwalker.go index fe1c2d89b..84e0d6e68 100644 --- a/sql3/planner/planwalker.go +++ b/sql3/planner/planwalker.go @@ -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 } diff --git a/sql3/planner/types/operator.go b/sql3/planner/types/operator.go index 1b1a548fe..14bf8c073 100644 --- a/sql3/planner/types/operator.go +++ b/sql3/planner/types/operator.go @@ -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 diff --git a/sql3/planner/types/planexpression.go b/sql3/planner/types/planexpression.go index 0e6390a9b..2d537842f 100644 --- a/sql3/planner/types/planexpression.go +++ b/sql3/planner/types/planexpression.go @@ -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 } diff --git a/sql3/sql3.ebnf b/sql3/sql3.ebnf index 802e2f1ae..e62e9b41e 100644 --- a/sql3/sql3.ebnf +++ b/sql3/sql3.ebnf @@ -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"