parser.go 40 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539
  1. // Copyright 2022 EMQ Technologies Co., Ltd.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package xsql
  15. import (
  16. "fmt"
  17. "github.com/golang-collections/collections/stack"
  18. "github.com/lf-edge/ekuiper/internal/binder/function"
  19. "github.com/lf-edge/ekuiper/pkg/ast"
  20. "github.com/lf-edge/ekuiper/pkg/message"
  21. "io"
  22. "math"
  23. "reflect"
  24. "strconv"
  25. "strings"
  26. )
  27. type Parser struct {
  28. s *Scanner
  29. i int // buffer index
  30. n int // buffer char count
  31. buf [3]struct {
  32. tok ast.Token
  33. lit string
  34. }
  35. inFunc string // currently parsing function name
  36. f int // anonymous field index number
  37. fn int // function index number
  38. clause string
  39. }
  40. func (p *Parser) ParseCondition() (ast.Expr, error) {
  41. if tok, _ := p.scanIgnoreWhitespace(); tok != ast.WHERE {
  42. p.unscan()
  43. return nil, nil
  44. }
  45. expr, err := p.ParseExpr()
  46. if err != nil {
  47. return nil, err
  48. }
  49. return expr, nil
  50. }
  51. func (p *Parser) scan() (tok ast.Token, lit string) {
  52. if p.n > 0 {
  53. p.n--
  54. return p.curr()
  55. }
  56. tok, lit = p.s.Scan()
  57. if tok != ast.WS && tok != ast.COMMENT {
  58. p.i = (p.i + 1) % len(p.buf)
  59. buf := &p.buf[p.i]
  60. buf.tok, buf.lit = tok, lit
  61. }
  62. return
  63. }
  64. func (p *Parser) curr() (ast.Token, string) {
  65. i := (p.i - p.n + len(p.buf)) % len(p.buf)
  66. buf := &p.buf[i]
  67. return buf.tok, buf.lit
  68. }
  69. func (p *Parser) scanIgnoreWhitespace() (tok ast.Token, lit string) {
  70. tok, lit = p.scan()
  71. for {
  72. if tok == ast.WS || tok == ast.COMMENT {
  73. tok, lit = p.scan()
  74. } else {
  75. break
  76. }
  77. }
  78. return tok, lit
  79. }
  80. func (p *Parser) unscan() { p.n++ }
  81. func NewParser(r io.Reader) *Parser {
  82. return &Parser{s: NewScanner(r)}
  83. }
  84. func (p *Parser) ParseQueries() ([]ast.SelectStatement, error) {
  85. var stmts []ast.SelectStatement
  86. if stmt, err := p.Parse(); err != nil {
  87. return nil, err
  88. } else {
  89. stmts = append(stmts, *stmt)
  90. }
  91. for {
  92. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.SEMICOLON {
  93. if stmt, err := p.Parse(); err != nil {
  94. return nil, err
  95. } else {
  96. if stmt != nil {
  97. stmts = append(stmts, *stmt)
  98. }
  99. }
  100. } else if tok == ast.EOF {
  101. break
  102. }
  103. }
  104. return stmts, nil
  105. }
  106. func (p *Parser) Parse() (*ast.SelectStatement, error) {
  107. selects := &ast.SelectStatement{}
  108. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.EOF {
  109. return nil, nil
  110. } else if tok != ast.SELECT {
  111. return nil, fmt.Errorf("Found %q, Expected SELECT.\n", lit)
  112. }
  113. p.clause = "select"
  114. if fields, err := p.parseFields(); err != nil {
  115. return nil, err
  116. } else {
  117. selects.Fields = fields
  118. }
  119. p.clause = "from"
  120. if src, err := p.parseSource(); err != nil {
  121. return nil, err
  122. } else {
  123. selects.Sources = src
  124. }
  125. p.clause = "join"
  126. if joins, err := p.parseJoins(); err != nil {
  127. return nil, err
  128. } else {
  129. selects.Joins = joins
  130. }
  131. p.clause = "where"
  132. if exp, err := p.ParseCondition(); err != nil {
  133. return nil, err
  134. } else {
  135. if exp != nil {
  136. selects.Condition = exp
  137. }
  138. }
  139. p.clause = "groupby"
  140. if dims, err := p.parseDimensions(); err != nil {
  141. return nil, err
  142. } else {
  143. selects.Dimensions = dims
  144. }
  145. p.clause = "having"
  146. if having, err := p.parseHaving(); err != nil {
  147. return nil, err
  148. } else {
  149. selects.Having = having
  150. }
  151. p.clause = "orderby"
  152. if sorts, err := p.parseSorts(); err != nil {
  153. return nil, err
  154. } else {
  155. selects.SortFields = sorts
  156. }
  157. p.clause = ""
  158. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.SEMICOLON {
  159. p.unscan()
  160. return selects, nil
  161. } else if tok != ast.EOF {
  162. return nil, fmt.Errorf("found %q, expected EOF.", lit)
  163. }
  164. if err := Validate(selects); err != nil {
  165. return nil, err
  166. }
  167. return selects, nil
  168. }
  169. func (p *Parser) parseSource() (ast.Sources, error) {
  170. var sources ast.Sources
  171. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.FROM {
  172. return nil, fmt.Errorf("found %q, expected FROM.", lit)
  173. }
  174. if src, alias, err := p.parseSourceLiteral(); err != nil {
  175. return nil, err
  176. } else {
  177. sources = append(sources, &ast.Table{Name: src, Alias: alias})
  178. }
  179. return sources, nil
  180. }
  181. //TODO Current func has problems when the source includes white space.
  182. func (p *Parser) parseSourceLiteral() (string, string, error) {
  183. var sourceSeg []string
  184. var alias string
  185. for {
  186. //HASH, DIV & ADD token is specially support for MQTT topic name patterns.
  187. if tok, lit := p.scanIgnoreWhitespace(); tok.AllowedSourceToken() {
  188. sourceSeg = append(sourceSeg, lit)
  189. if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.AS {
  190. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  191. alias = lit2
  192. } else {
  193. return "", "", fmt.Errorf("found %q, expected JOIN key word.", lit)
  194. }
  195. } else if tok1.AllowedSourceToken() {
  196. sourceSeg = append(sourceSeg, lit1)
  197. } else {
  198. p.unscan()
  199. break
  200. }
  201. } else {
  202. p.unscan()
  203. break
  204. }
  205. }
  206. return strings.Join(sourceSeg, ""), alias, nil
  207. }
  208. func (p *Parser) parseFieldNameSections() ([]string, error) {
  209. var fieldNameSects []string
  210. for {
  211. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.IDENT || tok == ast.ASTERISK {
  212. fieldNameSects = append(fieldNameSects, lit)
  213. if tok1, _ := p.scanIgnoreWhitespace(); !tok1.AllowedSFNToken() {
  214. p.unscan()
  215. break
  216. }
  217. } else {
  218. p.unscan()
  219. break
  220. }
  221. }
  222. if len(fieldNameSects) == 0 {
  223. return nil, fmt.Errorf("Cannot find any field name.\n")
  224. } else if len(fieldNameSects) > 2 {
  225. return nil, fmt.Errorf("Too many field names. Please use -> to reference keys in struct.\n")
  226. }
  227. return fieldNameSects, nil
  228. }
  229. func (p *Parser) parseJoins() (ast.Joins, error) {
  230. var joins ast.Joins
  231. for {
  232. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.INNER || tok == ast.LEFT || tok == ast.RIGHT || tok == ast.FULL || tok == ast.CROSS {
  233. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.JOIN {
  234. var jt = ast.INNER_JOIN
  235. switch tok {
  236. case ast.INNER:
  237. jt = ast.INNER_JOIN
  238. case ast.LEFT:
  239. jt = ast.LEFT_JOIN
  240. case ast.RIGHT:
  241. jt = ast.RIGHT_JOIN
  242. case ast.FULL:
  243. jt = ast.FULL_JOIN
  244. case ast.CROSS:
  245. jt = ast.CROSS_JOIN
  246. }
  247. if j, err := p.ParseJoin(jt); err != nil {
  248. return nil, err
  249. } else {
  250. joins = append(joins, *j)
  251. }
  252. } else {
  253. return nil, fmt.Errorf("found %q, expected JOIN key word.", lit)
  254. }
  255. } else {
  256. p.unscan()
  257. if len(joins) > 0 {
  258. return joins, nil
  259. }
  260. return nil, nil
  261. }
  262. }
  263. }
  264. func (p *Parser) ParseJoin(joinType ast.JoinType) (*ast.Join, error) {
  265. var j = &ast.Join{JoinType: joinType}
  266. if src, alias, err := p.parseSourceLiteral(); err != nil {
  267. return nil, err
  268. } else {
  269. j.Name = src
  270. j.Alias = alias
  271. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.ON {
  272. if ast.CROSS_JOIN == joinType {
  273. return nil, fmt.Errorf("On expression is not required for cross join type.\n")
  274. }
  275. if exp, err := p.ParseExpr(); err != nil {
  276. return nil, err
  277. } else {
  278. j.Expr = exp
  279. }
  280. } else {
  281. p.unscan()
  282. }
  283. }
  284. return j, nil
  285. }
  286. func (p *Parser) parseDimensions() (ast.Dimensions, error) {
  287. var ds ast.Dimensions
  288. if t, _ := p.scanIgnoreWhitespace(); t == ast.GROUP {
  289. if t1, l1 := p.scanIgnoreWhitespace(); t1 == ast.BY {
  290. for {
  291. if exp, err := p.ParseExpr(); err != nil {
  292. return nil, err
  293. } else {
  294. d := ast.Dimension{Expr: exp}
  295. ds = append(ds, d)
  296. }
  297. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.COMMA {
  298. continue
  299. } else {
  300. p.unscan()
  301. break
  302. }
  303. }
  304. } else {
  305. return nil, fmt.Errorf("found %q, expected BY statement.", l1)
  306. }
  307. } else {
  308. p.unscan()
  309. }
  310. return ds, nil
  311. }
  312. func (p *Parser) parseHaving() (ast.Expr, error) {
  313. if tok, _ := p.scanIgnoreWhitespace(); tok != ast.HAVING {
  314. p.unscan()
  315. return nil, nil
  316. }
  317. expr, err := p.ParseExpr()
  318. if err != nil {
  319. return nil, err
  320. }
  321. return expr, nil
  322. }
  323. func (p *Parser) parseSorts() (ast.SortFields, error) {
  324. var ss ast.SortFields
  325. if t, _ := p.scanIgnoreWhitespace(); t == ast.ORDER {
  326. if t1, l1 := p.scanIgnoreWhitespace(); t1 == ast.BY {
  327. for {
  328. if t1, l1 = p.scanIgnoreWhitespace(); t1 == ast.IDENT {
  329. s := ast.SortField{Ascending: true}
  330. p.unscan()
  331. if name, err := p.parseFieldNameSections(); err == nil {
  332. if len(name) == 2 {
  333. s.StreamName = ast.StreamName(name[0])
  334. s.Name = name[1]
  335. } else {
  336. s.Name = name[0]
  337. }
  338. s.Uname = strings.Join(name, ast.COLUMN_SEPARATOR)
  339. } else {
  340. return nil, err
  341. }
  342. p.unscan()
  343. if exp, err := p.ParseExpr(); err != nil {
  344. return nil, err
  345. } else {
  346. s.FieldExpr = exp
  347. }
  348. if t2, _ := p.scanIgnoreWhitespace(); t2 == ast.DESC {
  349. s.Ascending = false
  350. ss = append(ss, s)
  351. } else if t2 == ast.ASC {
  352. ss = append(ss, s)
  353. } else {
  354. ss = append(ss, s)
  355. p.unscan()
  356. continue
  357. }
  358. } else if t1 == ast.COMMA {
  359. continue
  360. } else {
  361. p.unscan()
  362. break
  363. }
  364. }
  365. } else {
  366. return nil, fmt.Errorf("found %q, expected BY keyword.", l1)
  367. }
  368. } else {
  369. p.unscan()
  370. }
  371. return ss, nil
  372. }
  373. func (p *Parser) parseFields() (ast.Fields, error) {
  374. var fields ast.Fields
  375. for {
  376. field, err := p.parseField()
  377. if err != nil {
  378. return nil, err
  379. } else {
  380. fields = append(fields, *field)
  381. }
  382. tok, _ := p.scanIgnoreWhitespace()
  383. if tok != ast.COMMA {
  384. p.unscan()
  385. break
  386. }
  387. }
  388. return fields, nil
  389. }
  390. func (p *Parser) parseField() (*ast.Field, error) {
  391. field := &ast.Field{}
  392. if exp, err := p.ParseExpr(); err != nil {
  393. return nil, err
  394. } else {
  395. field.Name = nameExpr(exp)
  396. field.Expr = exp
  397. }
  398. if alias, err := p.parseAlias(); err != nil {
  399. return nil, err
  400. } else {
  401. if alias != "" {
  402. if field.Name == "*" {
  403. return nil, fmt.Errorf("alias is not supported for *")
  404. }
  405. field.AName = alias
  406. }
  407. }
  408. if field.Name == "" && field.AName == "" {
  409. field.Name = DEFAULT_FIELD_NAME_PREFIX + strconv.Itoa(p.f)
  410. p.f += 1
  411. }
  412. return field, nil
  413. }
  414. func nameExpr(exp ast.Expr) string {
  415. switch e := exp.(type) {
  416. case *ast.FieldRef:
  417. return e.Name
  418. case *ast.Call:
  419. return e.Name
  420. case *ast.Wildcard:
  421. return ast.Tokens[ast.ASTERISK]
  422. default:
  423. return ""
  424. }
  425. }
  426. func (p *Parser) parseAlias() (string, error) {
  427. tok, lit := p.scanIgnoreWhitespace()
  428. if tok == ast.AS {
  429. if tok, lit = p.scanIgnoreWhitespace(); tok != ast.IDENT {
  430. return "", fmt.Errorf("found %q, expected as alias.", lit)
  431. } else {
  432. return lit, nil
  433. }
  434. }
  435. p.unscan()
  436. return "", nil
  437. }
  438. func (p *Parser) ParseExpr() (ast.Expr, error) {
  439. var err error
  440. root := &ast.BinaryExpr{}
  441. root.RHS, err = p.parseUnaryExpr(false)
  442. if err != nil {
  443. return nil, err
  444. }
  445. for {
  446. op, _ := p.scanIgnoreWhitespace()
  447. if !op.IsOperator() {
  448. p.unscan()
  449. return root.RHS, nil
  450. } else if op == ast.ASTERISK { //Change the asterisk to Mul token.
  451. op = ast.MUL
  452. } else if op == ast.LBRACKET { //LBRACKET is a special token, need to unscan
  453. op = ast.SUBSET
  454. p.unscan()
  455. } else if op == ast.IN { //IN is a special token, need to unscan
  456. p.unscan()
  457. } else if op == ast.NOT {
  458. afterNot, tk1 := p.scanIgnoreWhitespace()
  459. switch afterNot {
  460. case ast.IN: //IN is a special token, need to unscan
  461. op = ast.NOTIN
  462. p.unscan()
  463. break
  464. case ast.BETWEEN:
  465. op = ast.NOTBETWEEN
  466. node := root
  467. var lhs ast.Expr
  468. for {
  469. r, ok := node.RHS.(*ast.BinaryExpr)
  470. if !ok || r.OP.Precedence() >= op.Precedence() {
  471. lhs = node.RHS
  472. break
  473. }
  474. node = r
  475. }
  476. expr, err := p.parseBetween(lhs, ast.NOTBETWEEN)
  477. if err != nil {
  478. return nil, err
  479. }
  480. node.RHS = expr
  481. continue
  482. case ast.LIKE:
  483. op = ast.NOTLIKE
  484. default:
  485. return nil, fmt.Errorf("found %q, expected expression", tk1)
  486. }
  487. } else if op == ast.BETWEEN {
  488. node := root
  489. var lhs ast.Expr
  490. for {
  491. r, ok := node.RHS.(*ast.BinaryExpr)
  492. if !ok || r.OP.Precedence() >= op.Precedence() {
  493. lhs = node.RHS
  494. break
  495. }
  496. node = r
  497. }
  498. expr, err := p.parseBetween(lhs, op)
  499. if err != nil {
  500. return nil, err
  501. }
  502. node.RHS = expr
  503. continue
  504. }
  505. var rhs ast.Expr
  506. if rhs, err = p.parseUnaryExpr(op == ast.ARROW); err != nil {
  507. return nil, err
  508. }
  509. if op == ast.LIKE || op == ast.NOTLIKE {
  510. lp := &ast.LikePattern{
  511. Expr: rhs,
  512. }
  513. if l, ok := lp.Expr.(*ast.StringLiteral); ok {
  514. lp.Pattern, err = lp.Compile(l.Val)
  515. if err != nil {
  516. return nil, fmt.Errorf("invalid LIKE pattern: %s", err)
  517. }
  518. }
  519. rhs = lp
  520. }
  521. for node := root; ; {
  522. r, ok := node.RHS.(*ast.BinaryExpr)
  523. if !ok || r.OP.Precedence() >= op.Precedence() {
  524. node.RHS = &ast.BinaryExpr{LHS: node.RHS, RHS: rhs, OP: op}
  525. break
  526. }
  527. node = r
  528. }
  529. }
  530. }
  531. func (p *Parser) parseBetween(lhs ast.Expr, op ast.Token) (ast.Expr, error) {
  532. alhs, err := p.parseUnaryExpr(false)
  533. if err != nil {
  534. return nil, err
  535. }
  536. opp, _ := p.scanIgnoreWhitespace()
  537. if opp != ast.AND {
  538. return nil, fmt.Errorf("expect AND expression after between but found %s", opp)
  539. }
  540. arhs, err := p.parseUnaryExpr(false)
  541. if err != nil {
  542. return nil, err
  543. }
  544. return &ast.BinaryExpr{
  545. LHS: lhs,
  546. OP: op,
  547. RHS: &ast.BetweenExpr{
  548. Lower: alhs,
  549. Higher: arhs,
  550. },
  551. }, nil
  552. }
  553. func (p *Parser) parseUnaryExpr(isSubField bool) (ast.Expr, error) {
  554. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
  555. expr, err := p.ParseExpr()
  556. if err != nil {
  557. return nil, err
  558. }
  559. // Expect an RPAREN at the end.
  560. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 != ast.RPAREN {
  561. return nil, fmt.Errorf("found %q, expected right paren.", lit2)
  562. }
  563. return &ast.ParenExpr{Expr: expr}, nil
  564. } else if tok1 == ast.LBRACKET {
  565. return p.parseBracketExpr()
  566. } else if tok1 == ast.IN {
  567. return p.parseValueSetExpr()
  568. }
  569. p.unscan()
  570. tok, lit := p.scanIgnoreWhiteSpaceWithNegativeNum()
  571. if tok == ast.CASE {
  572. return p.parseCaseExpr()
  573. } else if tok == ast.IDENT {
  574. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
  575. return p.parseCall(lit)
  576. }
  577. p.unscan() //Back the Lparen token
  578. p.unscan() //Back the ident token
  579. if n, err := p.parseFieldNameSections(); err != nil {
  580. return nil, err
  581. } else {
  582. if p.inmeta() {
  583. if len(n) == 2 {
  584. return &ast.MetaRef{StreamName: ast.StreamName(n[0]), Name: n[1]}, nil
  585. }
  586. if isSubField {
  587. return &ast.JsonFieldRef{Name: n[0]}, nil
  588. }
  589. return &ast.MetaRef{StreamName: ast.DefaultStream, Name: n[0]}, nil
  590. } else {
  591. if len(n) == 2 {
  592. return &ast.FieldRef{StreamName: ast.StreamName(n[0]), Name: n[1]}, nil
  593. }
  594. if isSubField {
  595. return &ast.JsonFieldRef{Name: n[0]}, nil
  596. }
  597. return &ast.FieldRef{StreamName: ast.DefaultStream, Name: n[0]}, nil
  598. }
  599. }
  600. } else if tok == ast.STRING {
  601. return &ast.StringLiteral{Val: lit}, nil
  602. } else if tok == ast.INTEGER {
  603. val, _ := strconv.Atoi(lit)
  604. return &ast.IntegerLiteral{Val: val}, nil
  605. } else if tok == ast.NUMBER {
  606. if v, err := strconv.ParseFloat(lit, 64); err != nil {
  607. return nil, fmt.Errorf("found %q, invalid number value.", lit)
  608. } else {
  609. return &ast.NumberLiteral{Val: v}, nil
  610. }
  611. } else if tok == ast.TRUE || tok == ast.FALSE {
  612. if v, err := strconv.ParseBool(lit); err != nil {
  613. return nil, fmt.Errorf("found %q, invalid boolean value.", lit)
  614. } else {
  615. return &ast.BooleanLiteral{Val: v}, nil
  616. }
  617. } else if tok.IsTimeLiteral() {
  618. return &ast.TimeLiteral{Val: tok}, nil
  619. } else if tok == ast.ASTERISK {
  620. return p.parseAsterisk()
  621. }
  622. return nil, fmt.Errorf("found %q, expected expression.", lit)
  623. }
  624. func (p *Parser) parseValueSetExpr() (ast.Expr, error) {
  625. valsetExpr := &ast.ValueSetExpr{
  626. LiteralExprs: nil,
  627. ArrayExpr: nil,
  628. }
  629. // IN ("A", "B") or IN expression
  630. tk, _ := p.scanIgnoreWhitespace()
  631. if tk == ast.LPAREN {
  632. for {
  633. element, err := p.ParseExpr()
  634. if err != nil {
  635. return nil, fmt.Errorf("expect elements for IN expression, but %v", err)
  636. }
  637. valsetExpr.LiteralExprs = append(valsetExpr.LiteralExprs, element)
  638. if tok2, _ := p.scanIgnoreWhitespace(); tok2 != ast.COMMA {
  639. p.unscan()
  640. break
  641. }
  642. }
  643. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.RPAREN {
  644. return nil, fmt.Errorf("expect ) for IN expression, but got %q", lit)
  645. }
  646. return valsetExpr, nil
  647. } else {
  648. //back to IN
  649. p.unscan()
  650. }
  651. if exp, err := p.parseUnaryExpr(false); err != nil {
  652. return nil, fmt.Errorf("expect expression after IN, but got error %v", err)
  653. } else {
  654. return exp, nil
  655. }
  656. }
  657. func (p *Parser) parseBracketExpr() (ast.Expr, error) {
  658. tok2, lit2 := p.scanIgnoreWhiteSpaceWithNegativeNum()
  659. if tok2 == ast.RBRACKET {
  660. //field[]
  661. return &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 0}, End: &ast.IntegerLiteral{Val: math.MinInt32}}, nil
  662. } else if tok2 == ast.INTEGER {
  663. start, err := strconv.Atoi(lit2)
  664. if err != nil {
  665. return nil, fmt.Errorf("The start index %s is not an int value in bracket expression.", lit2)
  666. }
  667. if tok3, _ := p.scanIgnoreWhitespace(); tok3 == ast.RBRACKET {
  668. //Such as field[2]
  669. return &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: start}}, nil
  670. } else if tok3 == ast.COLON {
  671. //Such as field[2:] or field[2:4]
  672. return p.parseColonExpr(&ast.IntegerLiteral{Val: start})
  673. }
  674. } else if tok2 == ast.COLON {
  675. //Such as field[:3] or [:]
  676. return p.parseColonExpr(&ast.IntegerLiteral{Val: 0})
  677. } else {
  678. p.unscan()
  679. start, err := p.ParseExpr()
  680. if err != nil {
  681. return nil, fmt.Errorf("The start index %s is invalid in bracket expression.", lit2)
  682. }
  683. if tok3, _ := p.scanIgnoreWhitespace(); tok3 == ast.RBRACKET {
  684. //Such as field[2]
  685. return &ast.IndexExpr{Index: start}, nil
  686. } else if tok3 == ast.COLON {
  687. //Such as field[2:] or field[2:4]
  688. return p.parseColonExpr(start)
  689. }
  690. }
  691. return nil, fmt.Errorf("Unexpected token %q. when parsing bracket expressions.", lit2)
  692. }
  693. func (p *Parser) parseColonExpr(start ast.Expr) (ast.Expr, error) {
  694. tok, lit := p.scanIgnoreWhiteSpaceWithNegativeNum()
  695. if tok == ast.INTEGER {
  696. end, err := strconv.Atoi(lit)
  697. if err != nil {
  698. return nil, fmt.Errorf("The end index %s is not an int value in bracket expression.", lit)
  699. }
  700. if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.RBRACKET {
  701. return &ast.ColonExpr{Start: start, End: &ast.IntegerLiteral{Val: end}}, nil
  702. } else {
  703. return nil, fmt.Errorf("Found %q, expected right bracket.", lit1)
  704. }
  705. } else if tok == ast.RBRACKET {
  706. return &ast.ColonExpr{Start: start, End: &ast.IntegerLiteral{Val: math.MinInt32}}, nil
  707. }
  708. return nil, fmt.Errorf("Found %q, expected right bracket.", lit)
  709. }
  710. func (p *Parser) scanIgnoreWhiteSpaceWithNegativeNum() (ast.Token, string) {
  711. tok, lit := p.scanIgnoreWhitespace()
  712. if tok == ast.SUB {
  713. _, _ = p.s.ScanWhiteSpace()
  714. r := p.s.read()
  715. if isDigit(r) {
  716. p.s.unread()
  717. tok, lit = p.s.ScanNumber(false, true)
  718. }
  719. }
  720. return tok, lit
  721. }
  722. func (p *Parser) parseAs(f *ast.Field) (*ast.Field, error) {
  723. tok, lit := p.scanIgnoreWhitespace()
  724. if tok != ast.IDENT {
  725. return nil, fmt.Errorf("found %q, expected as alias.", lit)
  726. }
  727. f.AName = lit
  728. return f, nil
  729. }
  730. var WindowFuncs = map[string]struct{}{
  731. "tumblingwindow": {},
  732. "hoppingwindow": {},
  733. "sessionwindow": {},
  734. "slidingwindow": {},
  735. "countwindow": {},
  736. }
  737. func convFuncName(n string) (string, bool) {
  738. lname := strings.ToLower(n)
  739. if _, ok := WindowFuncs[lname]; ok {
  740. return lname, ok
  741. } else {
  742. return function.ConvName(n)
  743. }
  744. }
  745. func (p *Parser) parseCall(n string) (ast.Expr, error) {
  746. // Check if n function exists and convert it to lowercase for built-in func
  747. name, ok := convFuncName(n)
  748. if !ok {
  749. return nil, fmt.Errorf("function %s not found", n)
  750. }
  751. p.inFunc = name
  752. defer func() { p.inFunc = "" }()
  753. ft := function.GetFuncType(name)
  754. if ft == ast.FuncTypeCols && p.clause != "select" {
  755. return nil, fmt.Errorf("function %s can only be used inside the select clause", n)
  756. }
  757. var args []ast.Expr
  758. for {
  759. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.RPAREN {
  760. break
  761. } else {
  762. p.unscan()
  763. }
  764. if exp, err := p.ParseExpr(); err != nil {
  765. return nil, err
  766. } else {
  767. if ft == ast.FuncTypeCols {
  768. field := &ast.ColFuncField{Expr: exp, Name: nameExpr(exp)}
  769. args = append(args, field)
  770. } else {
  771. args = append(args, exp)
  772. }
  773. }
  774. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.COMMA {
  775. if tok != ast.RPAREN {
  776. return nil, fmt.Errorf("found function call %q, expected ), but with %q.", name, lit)
  777. }
  778. break
  779. }
  780. }
  781. if wt, err := validateWindows(name, args); wt == ast.NOT_WINDOW {
  782. if valErr := validateFuncs(name, args); valErr != nil {
  783. return nil, valErr
  784. }
  785. // Add context for some aggregate func
  786. if name == "deduplicate" {
  787. args = append([]ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}}, args...)
  788. }
  789. c := &ast.Call{Name: name, Args: args, FuncId: p.fn, FuncType: ft}
  790. p.fn += 1
  791. e := p.parseOverPartition(c)
  792. return c, e
  793. } else {
  794. if err != nil {
  795. return nil, err
  796. }
  797. win, err := p.ConvertToWindows(wt, args)
  798. if err != nil {
  799. return nil, err
  800. }
  801. // parse filter clause
  802. f, err := p.parseFilter()
  803. if err != nil {
  804. return nil, err
  805. } else if f != nil {
  806. win.Filter = f
  807. }
  808. return win, nil
  809. }
  810. }
  811. func (p *Parser) parseCaseExpr() (*ast.CaseExpr, error) {
  812. c := &ast.CaseExpr{}
  813. tok, _ := p.scanIgnoreWhitespace()
  814. p.unscan()
  815. if tok != ast.WHEN { // no condition value for case, additional validation needed
  816. if exp, err := p.ParseExpr(); err != nil {
  817. return nil, err
  818. } else {
  819. c.Value = exp
  820. }
  821. }
  822. loop:
  823. for {
  824. tok, _ := p.scanIgnoreWhitespace()
  825. switch tok {
  826. case ast.WHEN:
  827. if exp, err := p.ParseExpr(); err != nil {
  828. return nil, err
  829. } else {
  830. if c.WhenClauses == nil {
  831. c.WhenClauses = make([]*ast.WhenClause, 0)
  832. }
  833. if c.Value == nil && !ast.IsBooleanArg(exp) {
  834. return nil, fmt.Errorf("invalid CASE expression, WHEN expression must be a bool condition")
  835. }
  836. w := &ast.WhenClause{
  837. Expr: exp,
  838. }
  839. tokThen, _ := p.scanIgnoreWhitespace()
  840. if tokThen != ast.THEN {
  841. return nil, fmt.Errorf("invalid CASE expression, THEN expected after WHEN")
  842. } else {
  843. if expThen, err := p.ParseExpr(); err != nil {
  844. return nil, err
  845. } else {
  846. w.Result = expThen
  847. c.WhenClauses = append(c.WhenClauses, w)
  848. }
  849. }
  850. }
  851. case ast.ELSE:
  852. if c.WhenClauses != nil {
  853. if exp, err := p.ParseExpr(); err != nil {
  854. return nil, err
  855. } else {
  856. c.ElseClause = exp
  857. }
  858. } else {
  859. return nil, fmt.Errorf("invalid CASE expression, WHEN expected before ELSE")
  860. }
  861. case ast.END:
  862. if c.WhenClauses != nil {
  863. break loop
  864. } else {
  865. return nil, fmt.Errorf("invalid CASE expression, WHEN expected before END")
  866. }
  867. default:
  868. return nil, fmt.Errorf("invalid CASE expression, END expected")
  869. }
  870. }
  871. return c, nil
  872. }
  873. func validateWindows(fname string, args []ast.Expr) (ast.WindowType, error) {
  874. switch fname {
  875. case "tumblingwindow":
  876. if err := validateWindow(fname, 2, args); err != nil {
  877. return ast.TUMBLING_WINDOW, err
  878. }
  879. return ast.TUMBLING_WINDOW, nil
  880. case "hoppingwindow":
  881. if err := validateWindow(fname, 3, args); err != nil {
  882. return ast.HOPPING_WINDOW, err
  883. }
  884. return ast.HOPPING_WINDOW, nil
  885. case "sessionwindow":
  886. if err := validateWindow(fname, 3, args); err != nil {
  887. return ast.SESSION_WINDOW, err
  888. }
  889. return ast.SESSION_WINDOW, nil
  890. case "slidingwindow":
  891. if err := validateWindow(fname, 2, args); err != nil {
  892. return ast.SLIDING_WINDOW, err
  893. }
  894. return ast.SLIDING_WINDOW, nil
  895. case "countwindow":
  896. if len(args) == 1 {
  897. if para1, ok := args[0].(*ast.IntegerLiteral); ok && para1.Val > 0 {
  898. return ast.COUNT_WINDOW, nil
  899. } else {
  900. return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter value %s.", args[0])
  901. }
  902. } else if len(args) == 2 {
  903. if para1, ok1 := args[0].(*ast.IntegerLiteral); ok1 {
  904. if para2, ok2 := args[1].(*ast.IntegerLiteral); ok2 {
  905. if para1.Val < para2.Val {
  906. return ast.COUNT_WINDOW, fmt.Errorf("The second parameter value %d should be less than the first parameter %d.", para2.Val, para1.Val)
  907. } else {
  908. return ast.COUNT_WINDOW, nil
  909. }
  910. }
  911. }
  912. return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter value %s, %s.", args[0], args[1])
  913. } else {
  914. return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter count.")
  915. }
  916. }
  917. return ast.NOT_WINDOW, nil
  918. }
  919. func validateWindow(funcName string, expectLen int, args []ast.Expr) error {
  920. if len(args) != expectLen {
  921. return fmt.Errorf("The arguments for %s should be %d.\n", funcName, expectLen)
  922. }
  923. if _, ok := args[0].(*ast.TimeLiteral); !ok {
  924. return fmt.Errorf("The 1st argument for %s is expecting timer literal expression. One value of [dd|hh|mi|ss|ms].\n", funcName)
  925. }
  926. for i := 1; i < len(args); i++ {
  927. if _, ok := args[i].(*ast.IntegerLiteral); !ok {
  928. return fmt.Errorf("The %d argument for %s is expecting interger literal expression. \n", i, funcName)
  929. }
  930. }
  931. return nil
  932. }
  933. func (p *Parser) ConvertToWindows(wtype ast.WindowType, args []ast.Expr) (*ast.Window, error) {
  934. win := &ast.Window{WindowType: wtype}
  935. if wtype == ast.COUNT_WINDOW {
  936. win.Length = &ast.IntegerLiteral{Val: args[0].(*ast.IntegerLiteral).Val}
  937. if len(args) == 2 {
  938. win.Interval = &ast.IntegerLiteral{Val: args[1].(*ast.IntegerLiteral).Val}
  939. }
  940. return win, nil
  941. }
  942. var unit = 1
  943. v := args[0].(*ast.TimeLiteral).Val
  944. switch v {
  945. case ast.DD:
  946. unit = 24 * 3600 * 1000
  947. case ast.HH:
  948. unit = 3600 * 1000
  949. case ast.MI:
  950. unit = 60 * 1000
  951. case ast.SS:
  952. unit = 1000
  953. case ast.MS:
  954. unit = 1
  955. default:
  956. return nil, fmt.Errorf("Invalid timeliteral %s", v)
  957. }
  958. win.Length = &ast.IntegerLiteral{Val: args[1].(*ast.IntegerLiteral).Val * unit}
  959. if len(args) > 2 {
  960. win.Interval = &ast.IntegerLiteral{Val: args[2].(*ast.IntegerLiteral).Val * unit}
  961. } else {
  962. win.Interval = &ast.IntegerLiteral{Val: 0}
  963. }
  964. return win, nil
  965. }
  966. func (p *Parser) ParseCreateStmt() (ast.Statement, error) {
  967. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.CREATE {
  968. tok1, lit1 := p.scanIgnoreWhitespace()
  969. stmt := &ast.StreamStmt{}
  970. switch tok1 {
  971. case ast.STREAM:
  972. stmt.StreamType = ast.TypeStream
  973. case ast.TABLE:
  974. stmt.StreamType = ast.TypeTable
  975. default:
  976. return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
  977. }
  978. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  979. stmt.Name = ast.StreamName(lit2)
  980. if fields, err := p.parseStreamFields(); err != nil {
  981. return nil, err
  982. } else {
  983. stmt.StreamFields = fields
  984. }
  985. if opts, err := p.parseStreamOptions(); err != nil {
  986. return nil, err
  987. } else {
  988. stmt.Options = opts
  989. }
  990. if tok3, lit3 := p.scanIgnoreWhitespace(); tok3 == ast.SEMICOLON {
  991. p.unscan()
  992. } else if tok3 == ast.EOF {
  993. //Finish parsing create stream statement. Jump to validate
  994. } else {
  995. return nil, fmt.Errorf("found %q, expected semicolon or EOF.", lit3)
  996. }
  997. } else {
  998. return nil, fmt.Errorf("found %q, expected stream name.", lit2)
  999. }
  1000. if valErr := validateStream(stmt); valErr != nil {
  1001. return nil, valErr
  1002. }
  1003. return stmt, nil
  1004. } else {
  1005. p.unscan()
  1006. return nil, nil
  1007. }
  1008. }
  1009. // TODO more accurate validation for table
  1010. func validateStream(stmt *ast.StreamStmt) error {
  1011. f := stmt.Options.FORMAT
  1012. if f == "" {
  1013. f = message.FormatJson
  1014. }
  1015. lf := strings.ToLower(f)
  1016. switch lf {
  1017. case message.FormatBinary:
  1018. if stmt.StreamType == ast.TypeTable {
  1019. return fmt.Errorf("'binary' format is not supported for table")
  1020. }
  1021. switch len(stmt.StreamFields) {
  1022. case 0:
  1023. // do nothing for schemaless
  1024. case 1:
  1025. f := stmt.StreamFields[0]
  1026. if bt, ok := f.FieldType.(*ast.BasicType); ok {
  1027. if bt.Type == ast.BYTEA {
  1028. break
  1029. }
  1030. }
  1031. return fmt.Errorf("'binary' format stream can have only 'bytea' type field")
  1032. default:
  1033. return fmt.Errorf("'binary' format stream can have only one field")
  1034. }
  1035. default:
  1036. if !message.IsFormatSupported(lf) {
  1037. return fmt.Errorf("option 'format=%s' is invalid", f)
  1038. }
  1039. }
  1040. return nil
  1041. }
  1042. func (p *Parser) parseShowStmt() (ast.Statement, error) {
  1043. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.SHOW {
  1044. tok1, lit1 := p.scanIgnoreWhitespace()
  1045. switch tok1 {
  1046. case ast.STREAMS:
  1047. ss := &ast.ShowStreamsStatement{}
  1048. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EOF || tok2 == ast.SEMICOLON {
  1049. return ss, nil
  1050. } else {
  1051. return nil, fmt.Errorf("found %q, expected semecolon or EOF.", lit2)
  1052. }
  1053. case ast.TABLES:
  1054. ss := &ast.ShowTablesStatement{}
  1055. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EOF || tok2 == ast.SEMICOLON {
  1056. return ss, nil
  1057. } else {
  1058. return nil, fmt.Errorf("found %q, expected semecolon or EOF.", lit2)
  1059. }
  1060. default:
  1061. return nil, fmt.Errorf("found %q, expected keyword streams or tables.", lit1)
  1062. }
  1063. } else {
  1064. p.unscan()
  1065. return nil, nil
  1066. }
  1067. }
  1068. func (p *Parser) parseDescribeStmt() (ast.Statement, error) {
  1069. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.DESCRIBE {
  1070. tok1, lit1 := p.scanIgnoreWhitespace()
  1071. switch tok1 {
  1072. case ast.STREAM:
  1073. dss := &ast.DescribeStreamStatement{}
  1074. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1075. dss.Name = lit2
  1076. return dss, nil
  1077. } else {
  1078. return nil, fmt.Errorf("found %q, expected stream name.", lit2)
  1079. }
  1080. case ast.TABLE:
  1081. dss := &ast.DescribeTableStatement{}
  1082. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1083. dss.Name = lit2
  1084. return dss, nil
  1085. } else {
  1086. return nil, fmt.Errorf("found %q, expected table name.", lit2)
  1087. }
  1088. default:
  1089. return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
  1090. }
  1091. } else {
  1092. p.unscan()
  1093. return nil, nil
  1094. }
  1095. }
  1096. func (p *Parser) parseExplainStmt() (ast.Statement, error) {
  1097. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.EXPLAIN {
  1098. tok1, lit1 := p.scanIgnoreWhitespace()
  1099. switch tok1 {
  1100. case ast.STREAM:
  1101. ess := &ast.ExplainStreamStatement{}
  1102. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1103. ess.Name = lit2
  1104. return ess, nil
  1105. } else {
  1106. return nil, fmt.Errorf("found %q, expected stream name.", lit2)
  1107. }
  1108. case ast.TABLE:
  1109. ess := &ast.ExplainTableStatement{}
  1110. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1111. ess.Name = lit2
  1112. return ess, nil
  1113. } else {
  1114. return nil, fmt.Errorf("found %q, expected table name.", lit2)
  1115. }
  1116. default:
  1117. return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
  1118. }
  1119. } else {
  1120. p.unscan()
  1121. return nil, nil
  1122. }
  1123. }
  1124. func (p *Parser) parseDropStmt() (ast.Statement, error) {
  1125. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.DROP {
  1126. tok1, lit1 := p.scanIgnoreWhitespace()
  1127. switch tok1 {
  1128. case ast.STREAM:
  1129. ess := &ast.DropStreamStatement{}
  1130. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1131. ess.Name = lit2
  1132. return ess, nil
  1133. } else {
  1134. return nil, fmt.Errorf("found %q, expected stream name.", lit2)
  1135. }
  1136. case ast.TABLE:
  1137. ess := &ast.DropTableStatement{}
  1138. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1139. ess.Name = lit2
  1140. return ess, nil
  1141. } else {
  1142. return nil, fmt.Errorf("found %q, expected table name.", lit2)
  1143. }
  1144. default:
  1145. return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
  1146. }
  1147. } else {
  1148. p.unscan()
  1149. return nil, nil
  1150. }
  1151. }
  1152. func (p *Parser) parseStreamFields() (ast.StreamFields, error) {
  1153. lStack := &stack.Stack{}
  1154. var fields ast.StreamFields
  1155. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
  1156. lStack.Push(lit)
  1157. for {
  1158. //For the schemaless streams
  1159. //create stream demo () WITH (FORMAT="JSON", DATASOURCE="demo" TYPE="edgex")
  1160. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.RPAREN {
  1161. lStack.Pop()
  1162. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 != ast.WITH {
  1163. return nil, fmt.Errorf("found %q, expected is with.", lit2)
  1164. }
  1165. return fields, nil
  1166. } else {
  1167. p.unscan()
  1168. }
  1169. if f, err := p.parseStreamField(); err != nil {
  1170. return nil, err
  1171. } else {
  1172. fields = append(fields, *f)
  1173. }
  1174. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.RPAREN {
  1175. lStack.Pop()
  1176. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.WITH {
  1177. //Check the stack for LPAREN; If the stack for LPAREN is not zero, then it's not correct.
  1178. if lStack.Len() > 0 {
  1179. return nil, fmt.Errorf("Parenthesis is not matched.")
  1180. }
  1181. break
  1182. } else if tok2 == ast.COMMA {
  1183. if lStack.Len() > 0 {
  1184. return nil, fmt.Errorf("Parenthesis is in create record type not matched.")
  1185. }
  1186. p.unscan()
  1187. break
  1188. } else if tok2 == ast.RPAREN { //The nested type definition of ARRAY and Struct, such as "field ARRAY(STRUCT(f BIGINT))"
  1189. if lStack.Len() > 0 {
  1190. return nil, fmt.Errorf("Parenthesis is not matched.")
  1191. }
  1192. p.unscan()
  1193. break
  1194. } else {
  1195. if lStack.Len() == 0 {
  1196. return nil, fmt.Errorf("found %q, expected is with.", lit2)
  1197. }
  1198. p.unscan()
  1199. }
  1200. } else {
  1201. p.unscan()
  1202. }
  1203. }
  1204. } else {
  1205. return nil, fmt.Errorf("found %q, expected lparen after stream name.", lit)
  1206. }
  1207. return fields, nil
  1208. }
  1209. func (p *Parser) parseStreamField() (*ast.StreamField, error) {
  1210. field := &ast.StreamField{}
  1211. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.IDENT {
  1212. field.Name = lit
  1213. tok1, lit1 := p.scanIgnoreWhitespace()
  1214. if t := ast.GetDataType(tok1); t != ast.UNKNOWN && t.IsSimpleType() {
  1215. field.FieldType = &ast.BasicType{Type: t}
  1216. } else if t == ast.ARRAY {
  1217. if f, e := p.parseStreamArrayType(); e != nil {
  1218. return nil, e
  1219. } else {
  1220. field.FieldType = f
  1221. }
  1222. } else if t == ast.STRUCT {
  1223. if f, e := p.parseStreamStructType(); e != nil {
  1224. return nil, e
  1225. } else {
  1226. field.FieldType = f
  1227. }
  1228. } else if t == ast.UNKNOWN {
  1229. return nil, fmt.Errorf("found %q, expect valid stream field types(BIGINT | FLOAT | STRINGS | DATETIME | BOOLEAN | BYTEA | ARRAY | STRUCT).", lit1)
  1230. }
  1231. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.COMMA {
  1232. //Just consume the comma.
  1233. } else if tok2 == ast.RPAREN {
  1234. p.unscan()
  1235. } else {
  1236. return nil, fmt.Errorf("found %q, expect comma or rparen.", lit2)
  1237. }
  1238. } else {
  1239. return nil, fmt.Errorf("found %q, expect stream field name.", lit)
  1240. }
  1241. return field, nil
  1242. }
  1243. func (p *Parser) parseStreamArrayType() (ast.FieldType, error) {
  1244. lStack := &stack.Stack{}
  1245. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
  1246. lStack.Push(ast.LPAREN)
  1247. tok1, lit1 := p.scanIgnoreWhitespace()
  1248. if t := ast.GetDataType(tok1); t != ast.UNKNOWN && t.IsSimpleType() {
  1249. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.RPAREN {
  1250. lStack.Pop()
  1251. if lStack.Len() > 0 {
  1252. return nil, fmt.Errorf("Parenthesis is in array type not matched.")
  1253. }
  1254. return &ast.ArrayType{Type: t}, nil
  1255. } else {
  1256. return nil, fmt.Errorf("found %q, expect rparen in array type definition.", lit2)
  1257. }
  1258. } else if tok1 == ast.XSTRUCT {
  1259. if f, err := p.parseStreamStructType(); err != nil {
  1260. return nil, err
  1261. } else {
  1262. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.RPAREN {
  1263. lStack.Pop()
  1264. if lStack.Len() > 0 {
  1265. return nil, fmt.Errorf("Parenthesis is in struct of array type %q not matched.", tok1)
  1266. }
  1267. return &ast.ArrayType{Type: ast.STRUCT, FieldType: f}, nil
  1268. } else {
  1269. return nil, fmt.Errorf("found %q, expect rparen in struct of array type definition.", lit2)
  1270. }
  1271. }
  1272. } else if tok1 == ast.COMMA {
  1273. p.unscan()
  1274. } else {
  1275. return nil, fmt.Errorf("found %q, expect stream data types.", lit1)
  1276. }
  1277. } else {
  1278. }
  1279. return nil, nil
  1280. }
  1281. func (p *Parser) parseStreamStructType() (ast.FieldType, error) {
  1282. rf := &ast.RecType{}
  1283. if sfs, err := p.parseStreamFields(); err != nil {
  1284. return nil, err
  1285. } else {
  1286. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.COMMA {
  1287. rf.StreamFields = sfs
  1288. p.unscan()
  1289. } else if tok2 == ast.RPAREN {
  1290. rf.StreamFields = sfs
  1291. p.unscan()
  1292. } else {
  1293. return nil, fmt.Errorf("found %q, expect comma in create stream record statement.", lit2)
  1294. }
  1295. }
  1296. return rf, nil
  1297. }
  1298. func (p *Parser) parseStreamOptions() (*ast.Options, error) {
  1299. opts := &ast.Options{STRICT_VALIDATION: true}
  1300. v := reflect.ValueOf(opts)
  1301. lStack := &stack.Stack{}
  1302. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
  1303. lStack.Push(ast.LPAREN)
  1304. for {
  1305. if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.DATASOURCE || tok1 == ast.FORMAT || tok1 == ast.KEY || tok1 == ast.CONF_KEY || tok1 == ast.STRICT_VALIDATION || tok1 == ast.TYPE || tok1 == ast.TIMESTAMP || tok1 == ast.TIMESTAMP_FORMAT || tok1 == ast.RETAIN_SIZE || tok1 == ast.SHARED || tok1 == ast.SCHEMAID || tok1 == ast.KIND {
  1306. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EQ {
  1307. if tok3, lit3 := p.scanIgnoreWhitespace(); tok3 == ast.STRING {
  1308. switch tok1 {
  1309. case ast.STRICT_VALIDATION:
  1310. if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
  1311. return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, tok1)
  1312. } else {
  1313. opts.STRICT_VALIDATION = val == "TRUE"
  1314. }
  1315. case ast.RETAIN_SIZE:
  1316. if val, err := strconv.Atoi(lit3); err != nil {
  1317. return nil, fmt.Errorf("found %q, expect number value in %s option.", lit3, tok1)
  1318. } else {
  1319. opts.RETAIN_SIZE = val
  1320. }
  1321. case ast.SHARED:
  1322. if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
  1323. return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, tok1)
  1324. } else {
  1325. opts.SHARED = val == "TRUE"
  1326. }
  1327. case ast.KIND:
  1328. val := strings.ToLower(lit3)
  1329. opts.KIND = val
  1330. default:
  1331. f := v.Elem().FieldByName(lit1)
  1332. if f.IsValid() {
  1333. f.SetString(lit3)
  1334. } else { // should not happen
  1335. return nil, fmt.Errorf("invalid field %s.", lit1)
  1336. }
  1337. }
  1338. } else {
  1339. return nil, fmt.Errorf("found %q, expect string value in option.", lit3)
  1340. }
  1341. } else {
  1342. return nil, fmt.Errorf("found %q, expect equals(=) in options.", lit2)
  1343. }
  1344. } else if tok1 == ast.COMMA {
  1345. continue
  1346. } else if tok1 == ast.RPAREN {
  1347. if lStack.Pop(); lStack.Len() == 0 {
  1348. break
  1349. } else {
  1350. return nil, fmt.Errorf("Parenthesis is not matched in options definition.")
  1351. }
  1352. } else {
  1353. 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)
  1354. }
  1355. }
  1356. } else {
  1357. return nil, fmt.Errorf("found %q, expect stream options.", lit)
  1358. }
  1359. if opts.KIND == ast.StreamKindLookup && opts.TYPE == "memory" && opts.KEY == "" {
  1360. return nil, fmt.Errorf("Option \"key\" is required for memory lookup table.")
  1361. }
  1362. return opts, nil
  1363. }
  1364. // Only support filter on window now
  1365. func (p *Parser) parseFilter() (ast.Expr, error) {
  1366. if tok, _ := p.scanIgnoreWhitespace(); tok != ast.FILTER {
  1367. p.unscan()
  1368. return nil, nil
  1369. }
  1370. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.LPAREN {
  1371. return nil, fmt.Errorf("Found %q after FILTER, expect parentheses.", lit)
  1372. }
  1373. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.WHERE {
  1374. return nil, fmt.Errorf("Found %q after FILTER(, expect WHERE.", lit)
  1375. }
  1376. expr, err := p.ParseExpr()
  1377. if err != nil {
  1378. return nil, err
  1379. }
  1380. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.RPAREN {
  1381. return nil, fmt.Errorf("Found %q after FILTER, expect right parentheses.", lit)
  1382. }
  1383. return expr, nil
  1384. }
  1385. func (p *Parser) parseAsterisk() (ast.Expr, error) {
  1386. switch p.inFunc {
  1387. case "mqtt", "meta":
  1388. return &ast.MetaRef{StreamName: ast.DefaultStream, Name: "*"}, nil
  1389. default:
  1390. return &ast.Wildcard{Token: ast.ASTERISK}, nil
  1391. }
  1392. }
  1393. func (p *Parser) inmeta() bool {
  1394. return p.inFunc == "meta" || p.inFunc == "mqtt"
  1395. }
  1396. func (p *Parser) parseOverPartition(c *ast.Call) error {
  1397. if tok, _ := p.scanIgnoreWhitespace(); tok != ast.OVER {
  1398. p.unscan()
  1399. return nil
  1400. } else if function.IsAnalyticFunc(c.Name) {
  1401. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
  1402. if t, _ := p.scanIgnoreWhitespace(); t == ast.PARTITION {
  1403. if t1, l1 := p.scanIgnoreWhitespace(); t1 == ast.BY {
  1404. pe := &ast.PartitionExpr{}
  1405. for {
  1406. if exp, err := p.ParseExpr(); err != nil {
  1407. return err
  1408. } else {
  1409. pe.Exprs = append(pe.Exprs, exp)
  1410. }
  1411. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.COMMA {
  1412. continue
  1413. } else {
  1414. p.unscan()
  1415. break
  1416. }
  1417. }
  1418. if ttt, _ := p.scanIgnoreWhitespace(); ttt != ast.RPAREN {
  1419. return fmt.Errorf("found %q, expect right parentheses after PARTITION BY", ttt)
  1420. }
  1421. if len(pe.Exprs) == 0 {
  1422. return fmt.Errorf("PARTITION BY must have at least one expression.")
  1423. }
  1424. c.Partition = pe
  1425. return nil
  1426. } else {
  1427. return fmt.Errorf("found %q, expected by after partition.", l1)
  1428. }
  1429. } else {
  1430. return fmt.Errorf("Found %q after OVER (, expect partition by.", tok1)
  1431. }
  1432. } else {
  1433. return fmt.Errorf("Found %q after OVER, expect parentheses.", tok1)
  1434. }
  1435. } else {
  1436. return fmt.Errorf("Found OVER after non analytic function %s", c.Name)
  1437. }
  1438. }