1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579 |
- // Copyright 2022-2023 EMQ Technologies Co., Ltd.
- //
- // Licensed under the Apache License, Version 2.0 (the "License");
- // you may not use this file except in compliance with the License.
- // You may obtain a copy of the License at
- //
- // http://www.apache.org/licenses/LICENSE-2.0
- //
- // Unless required by applicable law or agreed to in writing, software
- // distributed under the License is distributed on an "AS IS" BASIS,
- // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- // See the License for the specific language governing permissions and
- // limitations under the License.
- package xsql
- import (
- "fmt"
- "github.com/golang-collections/collections/stack"
- "github.com/lf-edge/ekuiper/internal/binder/function"
- "github.com/lf-edge/ekuiper/pkg/ast"
- "github.com/lf-edge/ekuiper/pkg/message"
- "io"
- "math"
- "reflect"
- "strconv"
- "strings"
- )
- type Parser struct {
- s *Scanner
- i int // buffer index
- n int // buffer char count
- buf [3]struct {
- tok ast.Token
- lit string
- }
- inFunc string // currently parsing function name
- f int // anonymous field index number
- fn int // function index number
- clause string
- }
- func (p *Parser) ParseCondition() (ast.Expr, error) {
- if tok, _ := p.scanIgnoreWhitespace(); tok != ast.WHERE {
- p.unscan()
- return nil, nil
- }
- expr, err := p.ParseExpr()
- if err != nil {
- return nil, err
- }
- return expr, nil
- }
- func (p *Parser) scan() (tok ast.Token, lit string) {
- if p.n > 0 {
- p.n--
- return p.curr()
- }
- tok, lit = p.s.Scan()
- if tok != ast.WS && tok != ast.COMMENT {
- p.i = (p.i + 1) % len(p.buf)
- buf := &p.buf[p.i]
- buf.tok, buf.lit = tok, lit
- }
- return
- }
- func (p *Parser) curr() (ast.Token, string) {
- i := (p.i - p.n + len(p.buf)) % len(p.buf)
- buf := &p.buf[i]
- return buf.tok, buf.lit
- }
- func (p *Parser) scanIgnoreWhitespace() (tok ast.Token, lit string) {
- tok, lit = p.scan()
- for {
- if tok == ast.WS || tok == ast.COMMENT {
- tok, lit = p.scan()
- } else {
- break
- }
- }
- return tok, lit
- }
- func (p *Parser) unscan() { p.n++ }
- func NewParser(r io.Reader) *Parser {
- return &Parser{s: NewScanner(r)}
- }
- func (p *Parser) ParseQueries() ([]ast.SelectStatement, error) {
- var stmts []ast.SelectStatement
- if stmt, err := p.Parse(); err != nil {
- return nil, err
- } else {
- stmts = append(stmts, *stmt)
- }
- for {
- if tok, _ := p.scanIgnoreWhitespace(); tok == ast.SEMICOLON {
- if stmt, err := p.Parse(); err != nil {
- return nil, err
- } else {
- if stmt != nil {
- stmts = append(stmts, *stmt)
- }
- }
- } else if tok == ast.EOF {
- break
- }
- }
- return stmts, nil
- }
- func (p *Parser) Parse() (*ast.SelectStatement, error) {
- selects := &ast.SelectStatement{}
- if tok, lit := p.scanIgnoreWhitespace(); tok == ast.EOF {
- return nil, nil
- } else if tok != ast.SELECT {
- return nil, fmt.Errorf("Found %q, Expected SELECT.\n", lit)
- }
- p.clause = "select"
- if fields, err := p.parseFields(); err != nil {
- return nil, err
- } else {
- selects.Fields = fields
- }
- p.clause = "from"
- if src, err := p.parseSource(); err != nil {
- return nil, err
- } else {
- selects.Sources = src
- }
- p.clause = "join"
- if joins, err := p.parseJoins(); err != nil {
- return nil, err
- } else {
- selects.Joins = joins
- }
- p.clause = "where"
- if exp, err := p.ParseCondition(); err != nil {
- return nil, err
- } else {
- if exp != nil {
- selects.Condition = exp
- }
- }
- p.clause = "groupby"
- if dims, err := p.parseDimensions(); err != nil {
- return nil, err
- } else {
- selects.Dimensions = dims
- }
- p.clause = "having"
- if having, err := p.parseHaving(); err != nil {
- return nil, err
- } else {
- selects.Having = having
- }
- p.clause = "orderby"
- if sorts, err := p.parseSorts(); err != nil {
- return nil, err
- } else {
- selects.SortFields = sorts
- }
- p.clause = ""
- if tok, lit := p.scanIgnoreWhitespace(); tok == ast.SEMICOLON {
- p.unscan()
- return selects, nil
- } else if tok != ast.EOF {
- return nil, fmt.Errorf("found %q, expected EOF.", lit)
- }
- if err := Validate(selects); err != nil {
- return nil, err
- }
- return selects, nil
- }
- func (p *Parser) parseSource() (ast.Sources, error) {
- var sources ast.Sources
- if tok, lit := p.scanIgnoreWhitespace(); tok != ast.FROM {
- return nil, fmt.Errorf("found %q, expected FROM.", lit)
- }
- if src, alias, err := p.parseSourceLiteral(); err != nil {
- return nil, err
- } else {
- sources = append(sources, &ast.Table{Name: src, Alias: alias})
- }
- return sources, nil
- }
- // TODO Current func has problems when the source includes white space.
- func (p *Parser) parseSourceLiteral() (string, string, error) {
- var sourceSeg []string
- var alias string
- for {
- //HASH, DIV & ADD token is specially support for MQTT topic name patterns.
- if tok, lit := p.scanIgnoreWhitespace(); tok.AllowedSourceToken() {
- sourceSeg = append(sourceSeg, lit)
- if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.AS {
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
- alias = lit2
- } else {
- return "", "", fmt.Errorf("found %q, expected JOIN key word.", lit)
- }
- } else if tok1.AllowedSourceToken() {
- sourceSeg = append(sourceSeg, lit1)
- } else {
- p.unscan()
- break
- }
- } else {
- p.unscan()
- break
- }
- }
- return strings.Join(sourceSeg, ""), alias, nil
- }
- func (p *Parser) parseFieldNameSections() ([]string, error) {
- var fieldNameSects []string
- for {
- if tok, lit := p.scanIgnoreWhitespace(); tok == ast.IDENT || tok == ast.ASTERISK {
- fieldNameSects = append(fieldNameSects, lit)
- if tok1, _ := p.scanIgnoreWhitespace(); !tok1.AllowedSFNToken() {
- p.unscan()
- break
- }
- } else {
- p.unscan()
- break
- }
- }
- if len(fieldNameSects) == 0 {
- return nil, fmt.Errorf("Cannot find any field name.\n")
- } else if len(fieldNameSects) > 2 {
- return nil, fmt.Errorf("Too many field names. Please use -> to reference keys in struct.\n")
- }
- return fieldNameSects, nil
- }
- func (p *Parser) parseJoins() (ast.Joins, error) {
- var joins ast.Joins
- for {
- if tok, lit := p.scanIgnoreWhitespace(); tok == ast.INNER || tok == ast.LEFT || tok == ast.RIGHT || tok == ast.FULL || tok == ast.CROSS {
- if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.JOIN {
- var jt = ast.INNER_JOIN
- switch tok {
- case ast.INNER:
- jt = ast.INNER_JOIN
- case ast.LEFT:
- jt = ast.LEFT_JOIN
- case ast.RIGHT:
- jt = ast.RIGHT_JOIN
- case ast.FULL:
- jt = ast.FULL_JOIN
- case ast.CROSS:
- jt = ast.CROSS_JOIN
- }
- if j, err := p.ParseJoin(jt); err != nil {
- return nil, err
- } else {
- joins = append(joins, *j)
- }
- } else {
- return nil, fmt.Errorf("found %q, expected JOIN key word.", lit)
- }
- } else {
- p.unscan()
- if len(joins) > 0 {
- return joins, nil
- }
- return nil, nil
- }
- }
- }
- func (p *Parser) ParseJoin(joinType ast.JoinType) (*ast.Join, error) {
- var j = &ast.Join{JoinType: joinType}
- if src, alias, err := p.parseSourceLiteral(); err != nil {
- return nil, err
- } else {
- j.Name = src
- j.Alias = alias
- if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.ON {
- if ast.CROSS_JOIN == joinType {
- return nil, fmt.Errorf("On expression is not required for cross join type.\n")
- }
- if exp, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- j.Expr = exp
- }
- } else {
- p.unscan()
- }
- }
- return j, nil
- }
- func (p *Parser) parseDimensions() (ast.Dimensions, error) {
- var ds ast.Dimensions
- if t, _ := p.scanIgnoreWhitespace(); t == ast.GROUP {
- if t1, l1 := p.scanIgnoreWhitespace(); t1 == ast.BY {
- for {
- if exp, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- d := ast.Dimension{Expr: exp}
- ds = append(ds, d)
- }
- if tok, _ := p.scanIgnoreWhitespace(); tok == ast.COMMA {
- continue
- }
- p.unscan()
- break
- }
- } else {
- return nil, fmt.Errorf("found %q, expected BY statement.", l1)
- }
- } else {
- p.unscan()
- }
- return ds, nil
- }
- func (p *Parser) parseHaving() (ast.Expr, error) {
- if tok, _ := p.scanIgnoreWhitespace(); tok != ast.HAVING {
- p.unscan()
- return nil, nil
- }
- expr, err := p.ParseExpr()
- if err != nil {
- return nil, err
- }
- return expr, nil
- }
- func (p *Parser) parseSorts() (ast.SortFields, error) {
- var ss ast.SortFields
- if t, _ := p.scanIgnoreWhitespace(); t == ast.ORDER {
- if t1, l1 := p.scanIgnoreWhitespace(); t1 == ast.BY {
- for {
- if t1, l1 = p.scanIgnoreWhitespace(); t1 == ast.IDENT {
- s := ast.SortField{Ascending: true}
- p.unscan()
- if name, err := p.parseFieldNameSections(); err == nil {
- if len(name) == 2 {
- s.StreamName = ast.StreamName(name[0])
- s.Name = name[1]
- } else {
- s.Name = name[0]
- }
- s.Uname = strings.Join(name, ast.COLUMN_SEPARATOR)
- } else {
- return nil, err
- }
- p.unscan()
- if exp, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- s.FieldExpr = exp
- }
- if t2, _ := p.scanIgnoreWhitespace(); t2 == ast.DESC {
- s.Ascending = false
- ss = append(ss, s)
- } else if t2 == ast.ASC {
- ss = append(ss, s)
- } else {
- ss = append(ss, s)
- p.unscan()
- continue
- }
- } else if t1 == ast.COMMA {
- continue
- } else {
- p.unscan()
- break
- }
- }
- } else {
- return nil, fmt.Errorf("found %q, expected BY keyword.", l1)
- }
- } else {
- p.unscan()
- }
- return ss, nil
- }
- func (p *Parser) parseFields() (ast.Fields, error) {
- var fields ast.Fields
- for {
- field, err := p.parseField()
- if err != nil {
- return nil, err
- } else {
- fields = append(fields, *field)
- }
- tok, _ := p.scanIgnoreWhitespace()
- if tok != ast.COMMA {
- p.unscan()
- break
- }
- }
- return fields, nil
- }
- func (p *Parser) parseField() (*ast.Field, error) {
- field := &ast.Field{}
- if exp, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- field.Name = nameExpr(exp)
- field.Expr = exp
- }
- if alias, err := p.parseAlias(); err != nil {
- return nil, err
- } else {
- if alias != "" {
- if field.Name == "*" {
- return nil, fmt.Errorf("alias is not supported for *")
- }
- field.AName = alias
- }
- }
- if field.Name == "" && field.AName == "" {
- field.Name = DEFAULT_FIELD_NAME_PREFIX + strconv.Itoa(p.f)
- p.f += 1
- }
- return field, nil
- }
- func nameExpr(exp ast.Expr) string {
- switch e := exp.(type) {
- case *ast.FieldRef:
- return e.Name
- case *ast.Call:
- return e.Name
- case *ast.Wildcard:
- return ast.Tokens[ast.ASTERISK]
- default:
- return ""
- }
- }
- func (p *Parser) parseAlias() (string, error) {
- tok, lit := p.scanIgnoreWhitespace()
- if tok == ast.AS {
- if tok, lit = p.scanIgnoreWhitespace(); tok != ast.IDENT {
- return "", fmt.Errorf("found %q, expected as alias.", lit)
- } else {
- return lit, nil
- }
- }
- p.unscan()
- return "", nil
- }
- func (p *Parser) ParseExpr() (ast.Expr, error) {
- var err error
- root := &ast.BinaryExpr{}
- root.RHS, err = p.parseUnaryExpr(false)
- if err != nil {
- return nil, err
- }
- for {
- op, _ := p.scanIgnoreWhitespace()
- if !op.IsOperator() {
- p.unscan()
- return root.RHS, nil
- } else if op == ast.ASTERISK { //Change the asterisk to Mul token.
- op = ast.MUL
- } else if op == ast.LBRACKET { //LBRACKET is a special token, need to unscan
- op = ast.SUBSET
- p.unscan()
- } else if op == ast.IN { //IN is a special token, need to unscan
- p.unscan()
- } else if op == ast.NOT {
- afterNot, tk1 := p.scanIgnoreWhitespace()
- switch afterNot {
- case ast.IN: //IN is a special token, need to unscan
- op = ast.NOTIN
- p.unscan()
- break
- case ast.BETWEEN:
- op = ast.NOTBETWEEN
- node := root
- var lhs ast.Expr
- for {
- r, ok := node.RHS.(*ast.BinaryExpr)
- if !ok || r.OP.Precedence() >= op.Precedence() {
- lhs = node.RHS
- break
- }
- node = r
- }
- expr, err := p.parseBetween(lhs, ast.NOTBETWEEN)
- if err != nil {
- return nil, err
- }
- node.RHS = expr
- continue
- case ast.LIKE:
- op = ast.NOTLIKE
- default:
- return nil, fmt.Errorf("found %q, expected expression", tk1)
- }
- } else if op == ast.BETWEEN {
- node := root
- var lhs ast.Expr
- for {
- r, ok := node.RHS.(*ast.BinaryExpr)
- if !ok || r.OP.Precedence() >= op.Precedence() {
- lhs = node.RHS
- break
- }
- node = r
- }
- expr, err := p.parseBetween(lhs, op)
- if err != nil {
- return nil, err
- }
- node.RHS = expr
- continue
- }
- var rhs ast.Expr
- if rhs, err = p.parseUnaryExpr(op == ast.ARROW); err != nil {
- return nil, err
- }
- if op == ast.LIKE || op == ast.NOTLIKE {
- lp := &ast.LikePattern{
- Expr: rhs,
- }
- if l, ok := lp.Expr.(*ast.StringLiteral); ok {
- lp.Pattern, err = lp.Compile(l.Val)
- if err != nil {
- return nil, fmt.Errorf("invalid LIKE pattern: %s", err)
- }
- }
- rhs = lp
- }
- for node := root; ; {
- r, ok := node.RHS.(*ast.BinaryExpr)
- if !ok || r.OP.Precedence() >= op.Precedence() {
- node.RHS = &ast.BinaryExpr{LHS: node.RHS, RHS: rhs, OP: op}
- break
- }
- node = r
- }
- }
- }
- func (p *Parser) parseBetween(lhs ast.Expr, op ast.Token) (ast.Expr, error) {
- alhs, err := p.parseUnaryExpr(false)
- if err != nil {
- return nil, err
- }
- opp, _ := p.scanIgnoreWhitespace()
- if opp != ast.AND {
- return nil, fmt.Errorf("expect AND expression after between but found %s", opp)
- }
- arhs, err := p.parseUnaryExpr(false)
- if err != nil {
- return nil, err
- }
- return &ast.BinaryExpr{
- LHS: lhs,
- OP: op,
- RHS: &ast.BetweenExpr{
- Lower: alhs,
- Higher: arhs,
- },
- }, nil
- }
- func (p *Parser) parseUnaryExpr(isSubField bool) (ast.Expr, error) {
- if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
- expr, err := p.ParseExpr()
- if err != nil {
- return nil, err
- }
- // Expect an RPAREN at the end.
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 != ast.RPAREN {
- return nil, fmt.Errorf("found %q, expected right paren.", lit2)
- }
- return &ast.ParenExpr{Expr: expr}, nil
- } else if tok1 == ast.LBRACKET {
- return p.parseBracketExpr()
- } else if tok1 == ast.IN {
- return p.parseValueSetExpr()
- }
- p.unscan()
- tok, lit := p.scanIgnoreWhiteSpaceWithNegativeNum()
- if tok == ast.CASE {
- return p.parseCaseExpr()
- } else if tok == ast.IDENT {
- if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
- return p.parseCall(lit)
- }
- p.unscan() //Back the Lparen token
- p.unscan() //Back the ident token
- if n, err := p.parseFieldNameSections(); err != nil {
- return nil, err
- } else {
- if p.inmeta() {
- if len(n) == 2 {
- return &ast.MetaRef{StreamName: ast.StreamName(n[0]), Name: n[1]}, nil
- }
- if isSubField {
- return &ast.JsonFieldRef{Name: n[0]}, nil
- }
- return &ast.MetaRef{StreamName: ast.DefaultStream, Name: n[0]}, nil
- } else {
- if len(n) == 2 {
- return &ast.FieldRef{StreamName: ast.StreamName(n[0]), Name: n[1]}, nil
- }
- if isSubField {
- return &ast.JsonFieldRef{Name: n[0]}, nil
- }
- return &ast.FieldRef{StreamName: ast.DefaultStream, Name: n[0]}, nil
- }
- }
- } else if tok == ast.STRING {
- return &ast.StringLiteral{Val: lit}, nil
- } else if tok == ast.INTEGER {
- val, _ := strconv.Atoi(lit)
- return &ast.IntegerLiteral{Val: val}, nil
- } else if tok == ast.NUMBER {
- if v, err := strconv.ParseFloat(lit, 64); err != nil {
- return nil, fmt.Errorf("found %q, invalid number value.", lit)
- } else {
- return &ast.NumberLiteral{Val: v}, nil
- }
- } else if tok == ast.TRUE || tok == ast.FALSE {
- if v, err := strconv.ParseBool(lit); err != nil {
- return nil, fmt.Errorf("found %q, invalid boolean value.", lit)
- } else {
- return &ast.BooleanLiteral{Val: v}, nil
- }
- } else if tok.IsTimeLiteral() {
- return &ast.TimeLiteral{Val: tok}, nil
- } else if tok == ast.ASTERISK {
- return p.parseAsterisk()
- }
- return nil, fmt.Errorf("found %q, expected expression.", lit)
- }
- func (p *Parser) parseValueSetExpr() (ast.Expr, error) {
- valsetExpr := &ast.ValueSetExpr{
- LiteralExprs: nil,
- ArrayExpr: nil,
- }
- // IN ("A", "B") or IN expression
- tk, _ := p.scanIgnoreWhitespace()
- if tk == ast.LPAREN {
- for {
- element, err := p.ParseExpr()
- if err != nil {
- return nil, fmt.Errorf("expect elements for IN expression, but %v", err)
- }
- valsetExpr.LiteralExprs = append(valsetExpr.LiteralExprs, element)
- if tok2, _ := p.scanIgnoreWhitespace(); tok2 != ast.COMMA {
- p.unscan()
- break
- }
- }
- if tok, lit := p.scanIgnoreWhitespace(); tok != ast.RPAREN {
- return nil, fmt.Errorf("expect ) for IN expression, but got %q", lit)
- }
- return valsetExpr, nil
- } else {
- //back to IN
- p.unscan()
- }
- if exp, err := p.parseUnaryExpr(false); err != nil {
- return nil, fmt.Errorf("expect expression after IN, but got error %v", err)
- } else {
- return exp, nil
- }
- }
- func (p *Parser) parseBracketExpr() (ast.Expr, error) {
- tok2, lit2 := p.scanIgnoreWhiteSpaceWithNegativeNum()
- if tok2 == ast.RBRACKET {
- //field[]
- return &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 0}, End: &ast.IntegerLiteral{Val: math.MinInt32}}, nil
- } else if tok2 == ast.INTEGER {
- start, err := strconv.Atoi(lit2)
- if err != nil {
- return nil, fmt.Errorf("The start index %s is not an int value in bracket expression.", lit2)
- }
- if tok3, _ := p.scanIgnoreWhitespace(); tok3 == ast.RBRACKET {
- //Such as field[2]
- return &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: start}}, nil
- } else if tok3 == ast.COLON {
- //Such as field[2:] or field[2:4]
- return p.parseColonExpr(&ast.IntegerLiteral{Val: start})
- }
- } else if tok2 == ast.COLON {
- //Such as field[:3] or [:]
- return p.parseColonExpr(&ast.IntegerLiteral{Val: 0})
- } else {
- p.unscan()
- start, err := p.ParseExpr()
- if err != nil {
- return nil, fmt.Errorf("The start index %s is invalid in bracket expression.", lit2)
- }
- if tok3, _ := p.scanIgnoreWhitespace(); tok3 == ast.RBRACKET {
- //Such as field[2]
- return &ast.IndexExpr{Index: start}, nil
- } else if tok3 == ast.COLON {
- //Such as field[2:] or field[2:4]
- return p.parseColonExpr(start)
- }
- }
- return nil, fmt.Errorf("Unexpected token %q. when parsing bracket expressions.", lit2)
- }
- func (p *Parser) parseColonExpr(start ast.Expr) (ast.Expr, error) {
- tok, lit := p.scanIgnoreWhiteSpaceWithNegativeNum()
- if tok == ast.INTEGER {
- end, err := strconv.Atoi(lit)
- if err != nil {
- return nil, fmt.Errorf("The end index %s is not an int value in bracket expression.", lit)
- }
- if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.RBRACKET {
- return &ast.ColonExpr{Start: start, End: &ast.IntegerLiteral{Val: end}}, nil
- } else {
- return nil, fmt.Errorf("Found %q, expected right bracket.", lit1)
- }
- } else if tok == ast.RBRACKET {
- return &ast.ColonExpr{Start: start, End: &ast.IntegerLiteral{Val: math.MinInt32}}, nil
- }
- p.unscan()
- end, err := p.ParseExpr()
- if err != nil {
- return nil, fmt.Errorf("The end index %s is invalid in bracket expression.", lit)
- }
- if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.RBRACKET {
- return &ast.ColonExpr{Start: start, End: end}, nil
- } else {
- return nil, fmt.Errorf("Found %q, expected right bracket.", lit1)
- }
- }
- func (p *Parser) scanIgnoreWhiteSpaceWithNegativeNum() (ast.Token, string) {
- tok, lit := p.scanIgnoreWhitespace()
- if tok == ast.SUB {
- _, _ = p.s.ScanWhiteSpace()
- r := p.s.read()
- if isDigit(r) {
- p.s.unread()
- tok, lit = p.s.ScanNumber(false, true)
- }
- }
- return tok, lit
- }
- func (p *Parser) parseAs(f *ast.Field) (*ast.Field, error) {
- tok, lit := p.scanIgnoreWhitespace()
- if tok != ast.IDENT {
- return nil, fmt.Errorf("found %q, expected as alias.", lit)
- }
- f.AName = lit
- return f, nil
- }
- var WindowFuncs = map[string]struct{}{
- "tumblingwindow": {},
- "hoppingwindow": {},
- "sessionwindow": {},
- "slidingwindow": {},
- "countwindow": {},
- }
- func convFuncName(n string) (string, bool) {
- lname := strings.ToLower(n)
- if _, ok := WindowFuncs[lname]; ok {
- return lname, ok
- } else {
- return function.ConvName(n)
- }
- }
- func (p *Parser) parseCall(n string) (ast.Expr, error) {
- // Check if n function exists and convert it to lowercase for built-in func
- name, ok := convFuncName(n)
- if !ok {
- return nil, fmt.Errorf("function %s not found", n)
- }
- p.inFunc = name
- defer func() { p.inFunc = "" }()
- ft := function.GetFuncType(name)
- if ft == ast.FuncTypeCols && p.clause != "select" {
- return nil, fmt.Errorf("function %s can only be used inside the select clause", n)
- }
- var args []ast.Expr
- for {
- if tok, _ := p.scanIgnoreWhitespace(); tok == ast.RPAREN {
- break
- }
- p.unscan()
- if exp, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- if ft == ast.FuncTypeCols {
- field := &ast.ColFuncField{Expr: exp, Name: nameExpr(exp)}
- args = append(args, field)
- } else {
- args = append(args, exp)
- }
- }
- if tok, lit := p.scanIgnoreWhitespace(); tok != ast.COMMA {
- if tok != ast.RPAREN {
- return nil, fmt.Errorf("found function call %q, expected ), but with %q.", name, lit)
- }
- break
- }
- }
- if wt, err := validateWindows(name, args); wt == ast.NOT_WINDOW {
- if valErr := validateFuncs(name, args); valErr != nil {
- return nil, valErr
- }
- // Add context for some aggregate func
- if name == "deduplicate" {
- args = append([]ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}}, args...)
- }
- c := &ast.Call{Name: name, Args: args, FuncId: p.fn, FuncType: ft}
- p.fn += 1
- e := p.parseOver(c)
- return c, e
- } else {
- if err != nil {
- return nil, err
- }
- win, err := p.ConvertToWindows(wt, args)
- if err != nil {
- return nil, err
- }
- // parse filter clause
- f, err := p.parseFilter()
- if err != nil {
- return nil, err
- } else if f != nil {
- win.Filter = f
- }
- return win, nil
- }
- }
- func (p *Parser) parseCaseExpr() (*ast.CaseExpr, error) {
- c := &ast.CaseExpr{}
- tok, _ := p.scanIgnoreWhitespace()
- p.unscan()
- if tok != ast.WHEN { // no condition value for case, additional validation needed
- if exp, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- c.Value = exp
- }
- }
- loop:
- for {
- tok, _ := p.scanIgnoreWhitespace()
- switch tok {
- case ast.WHEN:
- if exp, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- if c.WhenClauses == nil {
- c.WhenClauses = make([]*ast.WhenClause, 0)
- }
- if c.Value == nil && !ast.IsBooleanArg(exp) {
- return nil, fmt.Errorf("invalid CASE expression, WHEN expression must be a bool condition")
- }
- w := &ast.WhenClause{
- Expr: exp,
- }
- tokThen, _ := p.scanIgnoreWhitespace()
- if tokThen != ast.THEN {
- return nil, fmt.Errorf("invalid CASE expression, THEN expected after WHEN")
- } else {
- if expThen, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- w.Result = expThen
- c.WhenClauses = append(c.WhenClauses, w)
- }
- }
- }
- case ast.ELSE:
- if c.WhenClauses != nil {
- if exp, err := p.ParseExpr(); err != nil {
- return nil, err
- } else {
- c.ElseClause = exp
- }
- } else {
- return nil, fmt.Errorf("invalid CASE expression, WHEN expected before ELSE")
- }
- case ast.END:
- if c.WhenClauses != nil {
- break loop
- }
- return nil, fmt.Errorf("invalid CASE expression, WHEN expected before END")
- default:
- return nil, fmt.Errorf("invalid CASE expression, END expected")
- }
- }
- return c, nil
- }
- func validateWindows(fname string, args []ast.Expr) (ast.WindowType, error) {
- switch fname {
- case "tumblingwindow":
- if err := validateWindow(fname, 2, args); err != nil {
- return ast.TUMBLING_WINDOW, err
- }
- return ast.TUMBLING_WINDOW, nil
- case "hoppingwindow":
- if err := validateWindow(fname, 3, args); err != nil {
- return ast.HOPPING_WINDOW, err
- }
- return ast.HOPPING_WINDOW, nil
- case "sessionwindow":
- if err := validateWindow(fname, 3, args); err != nil {
- return ast.SESSION_WINDOW, err
- }
- return ast.SESSION_WINDOW, nil
- case "slidingwindow":
- if err := validateWindow(fname, 2, args); err != nil {
- return ast.SLIDING_WINDOW, err
- }
- return ast.SLIDING_WINDOW, nil
- case "countwindow":
- if len(args) == 1 {
- if para1, ok := args[0].(*ast.IntegerLiteral); ok && para1.Val > 0 {
- return ast.COUNT_WINDOW, nil
- } else {
- return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter value %s.", args[0])
- }
- } else if len(args) == 2 {
- if para1, ok1 := args[0].(*ast.IntegerLiteral); ok1 {
- if para2, ok2 := args[1].(*ast.IntegerLiteral); ok2 {
- if para1.Val < para2.Val {
- return ast.COUNT_WINDOW, fmt.Errorf("The second parameter value %d should be less than the first parameter %d.", para2.Val, para1.Val)
- } else {
- return ast.COUNT_WINDOW, nil
- }
- }
- }
- return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter value %s, %s.", args[0], args[1])
- } else {
- return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter count.")
- }
- }
- return ast.NOT_WINDOW, nil
- }
- func validateWindow(funcName string, expectLen int, args []ast.Expr) error {
- if len(args) != expectLen {
- return fmt.Errorf("The arguments for %s should be %d.\n", funcName, expectLen)
- }
- if _, ok := args[0].(*ast.TimeLiteral); !ok {
- return fmt.Errorf("The 1st argument for %s is expecting timer literal expression. One value of [dd|hh|mi|ss|ms].\n", funcName)
- }
- for i := 1; i < len(args); i++ {
- if _, ok := args[i].(*ast.IntegerLiteral); !ok {
- return fmt.Errorf("The %d argument for %s is expecting interger literal expression. \n", i, funcName)
- }
- }
- return nil
- }
- func (p *Parser) ConvertToWindows(wtype ast.WindowType, args []ast.Expr) (*ast.Window, error) {
- win := &ast.Window{WindowType: wtype}
- if wtype == ast.COUNT_WINDOW {
- win.Length = &ast.IntegerLiteral{Val: args[0].(*ast.IntegerLiteral).Val}
- if len(args) == 2 {
- win.Interval = &ast.IntegerLiteral{Val: args[1].(*ast.IntegerLiteral).Val}
- }
- return win, nil
- }
- var unit = 1
- v := args[0].(*ast.TimeLiteral).Val
- switch v {
- case ast.DD:
- unit = 24 * 3600 * 1000
- case ast.HH:
- unit = 3600 * 1000
- case ast.MI:
- unit = 60 * 1000
- case ast.SS:
- unit = 1000
- case ast.MS:
- unit = 1
- default:
- return nil, fmt.Errorf("Invalid timeliteral %s", v)
- }
- win.Length = &ast.IntegerLiteral{Val: args[1].(*ast.IntegerLiteral).Val * unit}
- if len(args) > 2 {
- win.Interval = &ast.IntegerLiteral{Val: args[2].(*ast.IntegerLiteral).Val * unit}
- } else {
- win.Interval = &ast.IntegerLiteral{Val: 0}
- }
- return win, nil
- }
- func (p *Parser) ParseCreateStmt() (ast.Statement, error) {
- _, lit := p.scanIgnoreWhitespace()
- lit = strings.ToUpper(lit)
- if lit == ast.CREATE {
- _, lit1 := p.scanIgnoreWhitespace()
- stmt := &ast.StreamStmt{}
- lit1 = strings.ToUpper(lit1)
- switch lit1 {
- case ast.STREAM:
- stmt.StreamType = ast.TypeStream
- case ast.TABLE:
- stmt.StreamType = ast.TypeTable
- default:
- return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
- }
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
- stmt.Name = ast.StreamName(lit2)
- if fields, err := p.parseStreamFields(); err != nil {
- return nil, err
- } else {
- stmt.StreamFields = fields
- }
- if opts, err := p.parseStreamOptions(); err != nil {
- return nil, err
- } else {
- stmt.Options = opts
- }
- if tok3, lit3 := p.scanIgnoreWhitespace(); tok3 == ast.SEMICOLON {
- p.unscan()
- } else if tok3 == ast.EOF {
- //Finish parsing create stream statement. Jump to validate
- } else {
- return nil, fmt.Errorf("found %q, expected semicolon or EOF.", lit3)
- }
- } else {
- return nil, fmt.Errorf("found %q, expected stream name.", lit2)
- }
- if valErr := validateStream(stmt); valErr != nil {
- return nil, valErr
- }
- return stmt, nil
- } else {
- p.unscan()
- return nil, nil
- }
- }
- // TODO more accurate validation for table
- func validateStream(stmt *ast.StreamStmt) error {
- f := stmt.Options.FORMAT
- if f == "" {
- f = message.FormatJson
- }
- lf := strings.ToLower(f)
- switch lf {
- case message.FormatBinary:
- if stmt.StreamType == ast.TypeTable {
- return fmt.Errorf("'binary' format is not supported for table")
- }
- switch len(stmt.StreamFields) {
- case 0:
- // do nothing for schemaless
- case 1:
- f := stmt.StreamFields[0]
- if bt, ok := f.FieldType.(*ast.BasicType); ok {
- if bt.Type == ast.BYTEA {
- break
- }
- }
- return fmt.Errorf("'binary' format stream can have only 'bytea' type field")
- default:
- return fmt.Errorf("'binary' format stream can have only one field")
- }
- default:
- if !message.IsFormatSupported(lf) {
- return fmt.Errorf("option 'format=%s' is invalid", f)
- }
- }
- return nil
- }
- func (p *Parser) parseShowStmt() (ast.Statement, error) {
- _, lit := p.scanIgnoreWhitespace()
- lit = strings.ToUpper(lit)
- if lit == ast.SHOW {
- _, lit1 := p.scanIgnoreWhitespace()
- lit1 = strings.ToUpper(lit1)
- switch lit1 {
- case ast.STREAMS:
- ss := &ast.ShowStreamsStatement{}
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EOF || tok2 == ast.SEMICOLON {
- return ss, nil
- } else {
- return nil, fmt.Errorf("found %q, expected semecolon or EOF.", lit2)
- }
- case ast.TABLES:
- ss := &ast.ShowTablesStatement{}
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EOF || tok2 == ast.SEMICOLON {
- return ss, nil
- } else {
- return nil, fmt.Errorf("found %q, expected semecolon or EOF.", lit2)
- }
- default:
- return nil, fmt.Errorf("found %q, expected keyword streams or tables.", lit1)
- }
- } else {
- p.unscan()
- return nil, nil
- }
- }
- func (p *Parser) parseDescribeStmt() (ast.Statement, error) {
- _, lit := p.scanIgnoreWhitespace()
- lit = strings.ToUpper(lit)
- if lit == ast.DESCRIBE {
- _, lit1 := p.scanIgnoreWhitespace()
- lit1 = strings.ToUpper(lit1)
- switch lit1 {
- case ast.STREAM:
- dss := &ast.DescribeStreamStatement{}
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
- dss.Name = lit2
- return dss, nil
- } else {
- return nil, fmt.Errorf("found %q, expected stream name.", lit2)
- }
- case ast.TABLE:
- dss := &ast.DescribeTableStatement{}
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
- dss.Name = lit2
- return dss, nil
- } else {
- return nil, fmt.Errorf("found %q, expected table name.", lit2)
- }
- default:
- return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
- }
- } else {
- p.unscan()
- return nil, nil
- }
- }
- func (p *Parser) parseExplainStmt() (ast.Statement, error) {
- _, lit := p.scanIgnoreWhitespace()
- lit = strings.ToUpper(lit)
- if lit == ast.EXPLAIN {
- _, lit1 := p.scanIgnoreWhitespace()
- lit1 = strings.ToUpper(lit1)
- switch lit1 {
- case ast.STREAM:
- ess := &ast.ExplainStreamStatement{}
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
- ess.Name = lit2
- return ess, nil
- } else {
- return nil, fmt.Errorf("found %q, expected stream name.", lit2)
- }
- case ast.TABLE:
- ess := &ast.ExplainTableStatement{}
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
- ess.Name = lit2
- return ess, nil
- } else {
- return nil, fmt.Errorf("found %q, expected table name.", lit2)
- }
- default:
- return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
- }
- } else {
- p.unscan()
- return nil, nil
- }
- }
- func (p *Parser) parseDropStmt() (ast.Statement, error) {
- _, lit := p.scanIgnoreWhitespace()
- lit = strings.ToUpper(lit)
- if lit == ast.DROP {
- _, lit1 := p.scanIgnoreWhitespace()
- lit1 = strings.ToUpper(lit1)
- switch lit1 {
- case ast.STREAM:
- ess := &ast.DropStreamStatement{}
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
- ess.Name = lit2
- return ess, nil
- } else {
- return nil, fmt.Errorf("found %q, expected stream name.", lit2)
- }
- case ast.TABLE:
- ess := &ast.DropTableStatement{}
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
- ess.Name = lit2
- return ess, nil
- } else {
- return nil, fmt.Errorf("found %q, expected table name.", lit2)
- }
- default:
- return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
- }
- } else {
- p.unscan()
- return nil, nil
- }
- }
- func (p *Parser) parseStreamFields() (ast.StreamFields, error) {
- lStack := &stack.Stack{}
- var fields ast.StreamFields
- if tok, lit := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
- lStack.Push(lit)
- for {
- //For the schemaless streams
- //create stream demo () WITH (FORMAT="JSON", DATASOURCE="demo" TYPE="edgex")
- if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.RPAREN {
- lStack.Pop()
- if _, lit2 := p.scanIgnoreWhitespace(); strings.ToUpper(lit2) != ast.WITH {
- return nil, fmt.Errorf("found %q, expected is with.", lit2)
- }
- return fields, nil
- } else {
- p.unscan()
- }
- if f, err := p.parseStreamField(); err != nil {
- return nil, err
- } else {
- fields = append(fields, *f)
- }
- if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.RPAREN {
- lStack.Pop()
- tok2, lit2 := p.scanIgnoreWhitespace()
- lit2 = strings.ToUpper(lit2)
- if lit2 == ast.WITH {
- //Check the stack for LPAREN; If the stack for LPAREN is not zero, then it's not correct.
- if lStack.Len() > 0 {
- return nil, fmt.Errorf("Parenthesis is not matched.")
- }
- break
- } else if tok2 == ast.COMMA {
- if lStack.Len() > 0 {
- return nil, fmt.Errorf("Parenthesis is in create record type not matched.")
- }
- p.unscan()
- break
- } else if tok2 == ast.RPAREN { //The nested type definition of ARRAY and Struct, such as "field ARRAY(STRUCT(f BIGINT))"
- if lStack.Len() > 0 {
- return nil, fmt.Errorf("Parenthesis is not matched.")
- }
- p.unscan()
- break
- } else {
- if lStack.Len() == 0 {
- return nil, fmt.Errorf("found %q, expected is with.", lit2)
- }
- p.unscan()
- }
- } else {
- p.unscan()
- }
- }
- } else {
- return nil, fmt.Errorf("found %q, expected lparen after stream name.", lit)
- }
- return fields, nil
- }
- func (p *Parser) parseStreamField() (*ast.StreamField, error) {
- field := &ast.StreamField{}
- if tok, lit := p.scanIgnoreWhitespace(); tok == ast.IDENT {
- field.Name = lit
- _, lit1 := p.scanIgnoreWhitespace()
- if t := ast.GetDataType(lit1); t != ast.UNKNOWN && t.IsSimpleType() {
- field.FieldType = &ast.BasicType{Type: t}
- } else if t == ast.ARRAY {
- if f, e := p.parseStreamArrayType(); e != nil {
- return nil, e
- } else {
- field.FieldType = f
- }
- } else if t == ast.STRUCT {
- if f, e := p.parseStreamStructType(); e != nil {
- return nil, e
- } else {
- field.FieldType = f
- }
- } else if t == ast.UNKNOWN {
- return nil, fmt.Errorf("found %q, expect valid stream field types(BIGINT | FLOAT | STRINGS | DATETIME | BOOLEAN | BYTEA | ARRAY | STRUCT).", lit1)
- }
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.COMMA {
- //Just consume the comma.
- } else if tok2 == ast.RPAREN {
- p.unscan()
- } else {
- return nil, fmt.Errorf("found %q, expect comma or rparen.", lit2)
- }
- } else {
- return nil, fmt.Errorf("found %q, expect stream field name.", lit)
- }
- return field, nil
- }
- func (p *Parser) parseStreamArrayType() (ast.FieldType, error) {
- lStack := &stack.Stack{}
- if tok, _ := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
- lStack.Push(ast.LPAREN)
- tok1, lit1 := p.scanIgnoreWhitespace()
- t := ast.GetDataType(lit1)
- if t != ast.UNKNOWN && t.IsSimpleType() {
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.RPAREN {
- lStack.Pop()
- if lStack.Len() > 0 {
- return nil, fmt.Errorf("Parenthesis is in array type not matched.")
- }
- return &ast.ArrayType{Type: t}, nil
- } else {
- return nil, fmt.Errorf("found %q, expect rparen in array type definition.", lit2)
- }
- } else if t == ast.STRUCT {
- if f, err := p.parseStreamStructType(); err != nil {
- return nil, err
- } else {
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.RPAREN {
- lStack.Pop()
- if lStack.Len() > 0 {
- return nil, fmt.Errorf("Parenthesis is in struct of array type %q not matched.", tok1)
- }
- return &ast.ArrayType{Type: ast.STRUCT, FieldType: f}, nil
- } else {
- return nil, fmt.Errorf("found %q, expect rparen in struct of array type definition.", lit2)
- }
- }
- } else if tok1 == ast.COMMA {
- p.unscan()
- } else {
- return nil, fmt.Errorf("found %q, expect stream data types.", lit1)
- }
- } else {
- }
- return nil, nil
- }
- func (p *Parser) parseStreamStructType() (ast.FieldType, error) {
- rf := &ast.RecType{}
- if sfs, err := p.parseStreamFields(); err != nil {
- return nil, err
- } else {
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.COMMA {
- rf.StreamFields = sfs
- p.unscan()
- } else if tok2 == ast.RPAREN {
- rf.StreamFields = sfs
- p.unscan()
- } else {
- return nil, fmt.Errorf("found %q, expect comma in create stream record statement.", lit2)
- }
- }
- return rf, nil
- }
- func (p *Parser) parseStreamOptions() (*ast.Options, error) {
- opts := &ast.Options{STRICT_VALIDATION: false}
- v := reflect.ValueOf(opts)
- lStack := &stack.Stack{}
- if tok, lit := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
- lStack.Push(ast.LPAREN)
- for {
- tok1, lit1 := p.scanIgnoreWhitespace()
- lit1 = strings.ToUpper(lit1)
- if ast.IsStreamOptionKeyword(tok1, lit1) {
- if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EQ {
- if tok3, lit3 := p.scanIgnoreWhitespace(); tok3 == ast.STRING {
- switch lit1 {
- case ast.STRICT_VALIDATION:
- if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
- return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, lit1)
- } else {
- opts.STRICT_VALIDATION = val == "TRUE"
- }
- case ast.RETAIN_SIZE:
- if val, err := strconv.Atoi(lit3); err != nil {
- return nil, fmt.Errorf("found %q, expect number value in %s option.", lit3, lit1)
- } else {
- opts.RETAIN_SIZE = val
- }
- case ast.SHARED:
- if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
- return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, lit1)
- } else {
- opts.SHARED = val == "TRUE"
- }
- case ast.KIND:
- val := strings.ToLower(lit3)
- opts.KIND = val
- default:
- f := v.Elem().FieldByName(lit1)
- if f.IsValid() {
- f.SetString(lit3)
- } else { // should not happen
- return nil, fmt.Errorf("invalid field %s.", lit1)
- }
- }
- } else {
- return nil, fmt.Errorf("found %q, expect string value in option.", lit3)
- }
- } else {
- return nil, fmt.Errorf("found %q, expect equals(=) in options.", lit2)
- }
- } else if tok1 == ast.COMMA {
- continue
- } else if tok1 == ast.RPAREN {
- if lStack.Pop(); lStack.Len() == 0 {
- break
- }
- return nil, fmt.Errorf("Parenthesis is not matched in options definition.")
- } else {
- return nil, fmt.Errorf("found %q, unknown option keys(DATASOURCE|FORMAT|KEY|CONF_KEY|SHARED|STRICT_VALIDATION|TYPE|TIMESTAMP|TIMESTAMP_FORMAT|RETAIN_SIZE|SCHEMAID).", lit1)
- }
- }
- } else {
- return nil, fmt.Errorf("found %q, expect stream options.", lit)
- }
- if opts.KIND == ast.StreamKindLookup && opts.TYPE == "memory" && opts.KEY == "" {
- return nil, fmt.Errorf("Option \"key\" is required for memory lookup table.")
- }
- return opts, nil
- }
- // Only support filter on window now
- func (p *Parser) parseFilter() (ast.Expr, error) {
- if tok, _ := p.scanIgnoreWhitespace(); tok != ast.FILTER {
- p.unscan()
- return nil, nil
- }
- if tok, lit := p.scanIgnoreWhitespace(); tok != ast.LPAREN {
- return nil, fmt.Errorf("Found %q after FILTER, expect parentheses.", lit)
- }
- if tok, lit := p.scanIgnoreWhitespace(); tok != ast.WHERE {
- return nil, fmt.Errorf("Found %q after FILTER(, expect WHERE.", lit)
- }
- expr, err := p.ParseExpr()
- if err != nil {
- return nil, err
- }
- if tok, lit := p.scanIgnoreWhitespace(); tok != ast.RPAREN {
- return nil, fmt.Errorf("Found %q after FILTER, expect right parentheses.", lit)
- }
- return expr, nil
- }
- func (p *Parser) parseAsterisk() (ast.Expr, error) {
- switch p.inFunc {
- case "mqtt", "meta":
- return &ast.MetaRef{StreamName: ast.DefaultStream, Name: "*"}, nil
- default:
- return &ast.Wildcard{Token: ast.ASTERISK}, nil
- }
- }
- func (p *Parser) inmeta() bool {
- return p.inFunc == "meta" || p.inFunc == "mqtt"
- }
- func (p *Parser) parseOver(c *ast.Call) error {
- if tok, _ := p.scanIgnoreWhitespace(); tok != ast.OVER {
- p.unscan()
- return nil
- } else if function.IsAnalyticFunc(c.Name) {
- if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
- if t, _ := p.scanIgnoreWhitespace(); t == ast.PARTITION {
- if t1, l1 := p.scanIgnoreWhitespace(); t1 == ast.BY {
- pe := &ast.PartitionExpr{}
- for {
- if exp, err := p.ParseExpr(); err != nil {
- return err
- } else {
- pe.Exprs = append(pe.Exprs, exp)
- }
- if tok, _ := p.scanIgnoreWhitespace(); tok == ast.COMMA {
- continue
- }
- p.unscan()
- break
- }
- if len(pe.Exprs) == 0 {
- return fmt.Errorf("PARTITION BY must have at least one expression.")
- }
- c.Partition = pe
- } else {
- return fmt.Errorf("found %q, expected by after partition.", l1)
- }
- } else {
- p.unscan()
- }
- if t, _ := p.scanIgnoreWhitespace(); t == ast.WHEN {
- if exp, err := p.ParseExpr(); err != nil {
- return err
- } else {
- c.WhenExpr = exp
- }
- } else {
- p.unscan()
- }
- if c.Partition != nil || c.WhenExpr != nil {
- if ttt, _ := p.scanIgnoreWhitespace(); ttt != ast.RPAREN {
- return fmt.Errorf("Found %q, expect right parentheses after OVER ", ttt)
- }
- }
- if c.Partition == nil && c.WhenExpr == nil {
- ttt, _ := p.scanIgnoreWhitespace()
- return fmt.Errorf("Found %q after OVER (, expect partition by or when.", ttt)
- }
- return nil
- } else {
- return fmt.Errorf("Found %q after OVER, expect parentheses.", tok1)
- }
- } else {
- return fmt.Errorf("Found OVER after non analytic function %s", c.Name)
- }
- }
|