parser.go 41 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579
  1. // Copyright 2022-2023 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. "io"
  18. "math"
  19. "reflect"
  20. "strconv"
  21. "strings"
  22. "github.com/golang-collections/collections/stack"
  23. "github.com/lf-edge/ekuiper/internal/binder/function"
  24. "github.com/lf-edge/ekuiper/pkg/ast"
  25. "github.com/lf-edge/ekuiper/pkg/message"
  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. }
  300. p.unscan()
  301. break
  302. }
  303. } else {
  304. return nil, fmt.Errorf("found %q, expected BY statement.", l1)
  305. }
  306. } else {
  307. p.unscan()
  308. }
  309. return ds, nil
  310. }
  311. func (p *Parser) parseHaving() (ast.Expr, error) {
  312. if tok, _ := p.scanIgnoreWhitespace(); tok != ast.HAVING {
  313. p.unscan()
  314. return nil, nil
  315. }
  316. expr, err := p.ParseExpr()
  317. if err != nil {
  318. return nil, err
  319. }
  320. return expr, nil
  321. }
  322. func (p *Parser) parseSorts() (ast.SortFields, error) {
  323. var ss ast.SortFields
  324. if t, _ := p.scanIgnoreWhitespace(); t == ast.ORDER {
  325. if t1, l1 := p.scanIgnoreWhitespace(); t1 == ast.BY {
  326. for {
  327. if t1, l1 = p.scanIgnoreWhitespace(); t1 == ast.IDENT {
  328. s := ast.SortField{Ascending: true}
  329. p.unscan()
  330. if name, err := p.parseFieldNameSections(); err == nil {
  331. if len(name) == 2 {
  332. s.StreamName = ast.StreamName(name[0])
  333. s.Name = name[1]
  334. } else {
  335. s.Name = name[0]
  336. }
  337. s.Uname = strings.Join(name, ast.COLUMN_SEPARATOR)
  338. } else {
  339. return nil, err
  340. }
  341. p.unscan()
  342. if exp, err := p.ParseExpr(); err != nil {
  343. return nil, err
  344. } else {
  345. s.FieldExpr = exp
  346. }
  347. if t2, _ := p.scanIgnoreWhitespace(); t2 == ast.DESC {
  348. s.Ascending = false
  349. ss = append(ss, s)
  350. } else if t2 == ast.ASC {
  351. ss = append(ss, s)
  352. } else {
  353. ss = append(ss, s)
  354. p.unscan()
  355. continue
  356. }
  357. } else if t1 == ast.COMMA {
  358. continue
  359. } else {
  360. p.unscan()
  361. break
  362. }
  363. }
  364. } else {
  365. return nil, fmt.Errorf("found %q, expected BY keyword.", l1)
  366. }
  367. } else {
  368. p.unscan()
  369. }
  370. return ss, nil
  371. }
  372. func (p *Parser) parseFields() (ast.Fields, error) {
  373. var fields ast.Fields
  374. for {
  375. field, err := p.parseField()
  376. if err != nil {
  377. return nil, err
  378. } else {
  379. fields = append(fields, *field)
  380. }
  381. tok, _ := p.scanIgnoreWhitespace()
  382. if tok != ast.COMMA {
  383. p.unscan()
  384. break
  385. }
  386. }
  387. return fields, nil
  388. }
  389. func (p *Parser) parseField() (*ast.Field, error) {
  390. field := &ast.Field{}
  391. if exp, err := p.ParseExpr(); err != nil {
  392. return nil, err
  393. } else {
  394. field.Name = nameExpr(exp)
  395. field.Expr = exp
  396. }
  397. if alias, err := p.parseAlias(); err != nil {
  398. return nil, err
  399. } else {
  400. if alias != "" {
  401. if field.Name == "*" {
  402. return nil, fmt.Errorf("alias is not supported for *")
  403. }
  404. field.AName = alias
  405. }
  406. }
  407. if field.Name == "" && field.AName == "" {
  408. field.Name = DEFAULT_FIELD_NAME_PREFIX + strconv.Itoa(p.f)
  409. p.f += 1
  410. }
  411. return field, nil
  412. }
  413. func nameExpr(exp ast.Expr) string {
  414. switch e := exp.(type) {
  415. case *ast.FieldRef:
  416. return e.Name
  417. case *ast.Call:
  418. return e.Name
  419. case *ast.Wildcard:
  420. return ast.Tokens[ast.ASTERISK]
  421. default:
  422. return ""
  423. }
  424. }
  425. func (p *Parser) parseAlias() (string, error) {
  426. tok, lit := p.scanIgnoreWhitespace()
  427. if tok == ast.AS {
  428. if tok, lit = p.scanIgnoreWhitespace(); tok != ast.IDENT {
  429. return "", fmt.Errorf("found %q, expected as alias.", lit)
  430. } else {
  431. return lit, nil
  432. }
  433. }
  434. p.unscan()
  435. return "", nil
  436. }
  437. func (p *Parser) ParseExpr() (ast.Expr, error) {
  438. var err error
  439. root := &ast.BinaryExpr{}
  440. root.RHS, err = p.parseUnaryExpr(false)
  441. if err != nil {
  442. return nil, err
  443. }
  444. for {
  445. op, _ := p.scanIgnoreWhitespace()
  446. if !op.IsOperator() {
  447. p.unscan()
  448. return root.RHS, nil
  449. } else if op == ast.ASTERISK { //Change the asterisk to Mul token.
  450. op = ast.MUL
  451. } else if op == ast.LBRACKET { //LBRACKET is a special token, need to unscan
  452. op = ast.SUBSET
  453. p.unscan()
  454. } else if op == ast.IN { //IN is a special token, need to unscan
  455. p.unscan()
  456. } else if op == ast.NOT {
  457. afterNot, tk1 := p.scanIgnoreWhitespace()
  458. switch afterNot {
  459. case ast.IN: //IN is a special token, need to unscan
  460. op = ast.NOTIN
  461. p.unscan()
  462. break
  463. case ast.BETWEEN:
  464. op = ast.NOTBETWEEN
  465. node := root
  466. var lhs ast.Expr
  467. for {
  468. r, ok := node.RHS.(*ast.BinaryExpr)
  469. if !ok || r.OP.Precedence() >= op.Precedence() {
  470. lhs = node.RHS
  471. break
  472. }
  473. node = r
  474. }
  475. expr, err := p.parseBetween(lhs, ast.NOTBETWEEN)
  476. if err != nil {
  477. return nil, err
  478. }
  479. node.RHS = expr
  480. continue
  481. case ast.LIKE:
  482. op = ast.NOTLIKE
  483. default:
  484. return nil, fmt.Errorf("found %q, expected expression", tk1)
  485. }
  486. } else if op == ast.BETWEEN {
  487. node := root
  488. var lhs ast.Expr
  489. for {
  490. r, ok := node.RHS.(*ast.BinaryExpr)
  491. if !ok || r.OP.Precedence() >= op.Precedence() {
  492. lhs = node.RHS
  493. break
  494. }
  495. node = r
  496. }
  497. expr, err := p.parseBetween(lhs, op)
  498. if err != nil {
  499. return nil, err
  500. }
  501. node.RHS = expr
  502. continue
  503. }
  504. var rhs ast.Expr
  505. if rhs, err = p.parseUnaryExpr(op == ast.ARROW); err != nil {
  506. return nil, err
  507. }
  508. if op == ast.LIKE || op == ast.NOTLIKE {
  509. lp := &ast.LikePattern{
  510. Expr: rhs,
  511. }
  512. if l, ok := lp.Expr.(*ast.StringLiteral); ok {
  513. lp.Pattern, err = lp.Compile(l.Val)
  514. if err != nil {
  515. return nil, fmt.Errorf("invalid LIKE pattern: %s", err)
  516. }
  517. }
  518. rhs = lp
  519. }
  520. for node := root; ; {
  521. r, ok := node.RHS.(*ast.BinaryExpr)
  522. if !ok || r.OP.Precedence() >= op.Precedence() {
  523. node.RHS = &ast.BinaryExpr{LHS: node.RHS, RHS: rhs, OP: op}
  524. break
  525. }
  526. node = r
  527. }
  528. }
  529. }
  530. func (p *Parser) parseBetween(lhs ast.Expr, op ast.Token) (ast.Expr, error) {
  531. alhs, err := p.parseUnaryExpr(false)
  532. if err != nil {
  533. return nil, err
  534. }
  535. opp, _ := p.scanIgnoreWhitespace()
  536. if opp != ast.AND {
  537. return nil, fmt.Errorf("expect AND expression after between but found %s", opp)
  538. }
  539. arhs, err := p.parseUnaryExpr(false)
  540. if err != nil {
  541. return nil, err
  542. }
  543. return &ast.BinaryExpr{
  544. LHS: lhs,
  545. OP: op,
  546. RHS: &ast.BetweenExpr{
  547. Lower: alhs,
  548. Higher: arhs,
  549. },
  550. }, nil
  551. }
  552. func (p *Parser) parseUnaryExpr(isSubField bool) (ast.Expr, error) {
  553. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
  554. expr, err := p.ParseExpr()
  555. if err != nil {
  556. return nil, err
  557. }
  558. // Expect an RPAREN at the end.
  559. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 != ast.RPAREN {
  560. return nil, fmt.Errorf("found %q, expected right paren.", lit2)
  561. }
  562. return &ast.ParenExpr{Expr: expr}, nil
  563. } else if tok1 == ast.LBRACKET {
  564. return p.parseBracketExpr()
  565. } else if tok1 == ast.IN {
  566. return p.parseValueSetExpr()
  567. }
  568. p.unscan()
  569. tok, lit := p.scanIgnoreWhiteSpaceWithNegativeNum()
  570. if tok == ast.CASE {
  571. return p.parseCaseExpr()
  572. } else if tok == ast.IDENT {
  573. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
  574. return p.parseCall(lit)
  575. }
  576. p.unscan() //Back the Lparen token
  577. p.unscan() //Back the ident token
  578. if n, err := p.parseFieldNameSections(); err != nil {
  579. return nil, err
  580. } else {
  581. if p.inmeta() {
  582. if len(n) == 2 {
  583. return &ast.MetaRef{StreamName: ast.StreamName(n[0]), Name: n[1]}, nil
  584. }
  585. if isSubField {
  586. return &ast.JsonFieldRef{Name: n[0]}, nil
  587. }
  588. return &ast.MetaRef{StreamName: ast.DefaultStream, Name: n[0]}, nil
  589. } else {
  590. if len(n) == 2 {
  591. return &ast.FieldRef{StreamName: ast.StreamName(n[0]), Name: n[1]}, nil
  592. }
  593. if isSubField {
  594. return &ast.JsonFieldRef{Name: n[0]}, nil
  595. }
  596. return &ast.FieldRef{StreamName: ast.DefaultStream, Name: n[0]}, nil
  597. }
  598. }
  599. } else if tok == ast.STRING {
  600. return &ast.StringLiteral{Val: lit}, nil
  601. } else if tok == ast.INTEGER {
  602. val, _ := strconv.Atoi(lit)
  603. return &ast.IntegerLiteral{Val: val}, nil
  604. } else if tok == ast.NUMBER {
  605. if v, err := strconv.ParseFloat(lit, 64); err != nil {
  606. return nil, fmt.Errorf("found %q, invalid number value.", lit)
  607. } else {
  608. return &ast.NumberLiteral{Val: v}, nil
  609. }
  610. } else if tok == ast.TRUE || tok == ast.FALSE {
  611. if v, err := strconv.ParseBool(lit); err != nil {
  612. return nil, fmt.Errorf("found %q, invalid boolean value.", lit)
  613. } else {
  614. return &ast.BooleanLiteral{Val: v}, nil
  615. }
  616. } else if tok.IsTimeLiteral() {
  617. return &ast.TimeLiteral{Val: tok}, nil
  618. } else if tok == ast.ASTERISK {
  619. return p.parseAsterisk()
  620. }
  621. return nil, fmt.Errorf("found %q, expected expression.", lit)
  622. }
  623. func (p *Parser) parseValueSetExpr() (ast.Expr, error) {
  624. valsetExpr := &ast.ValueSetExpr{
  625. LiteralExprs: nil,
  626. ArrayExpr: nil,
  627. }
  628. // IN ("A", "B") or IN expression
  629. tk, _ := p.scanIgnoreWhitespace()
  630. if tk == ast.LPAREN {
  631. for {
  632. element, err := p.ParseExpr()
  633. if err != nil {
  634. return nil, fmt.Errorf("expect elements for IN expression, but %v", err)
  635. }
  636. valsetExpr.LiteralExprs = append(valsetExpr.LiteralExprs, element)
  637. if tok2, _ := p.scanIgnoreWhitespace(); tok2 != ast.COMMA {
  638. p.unscan()
  639. break
  640. }
  641. }
  642. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.RPAREN {
  643. return nil, fmt.Errorf("expect ) for IN expression, but got %q", lit)
  644. }
  645. return valsetExpr, nil
  646. } else {
  647. //back to IN
  648. p.unscan()
  649. }
  650. if exp, err := p.parseUnaryExpr(false); err != nil {
  651. return nil, fmt.Errorf("expect expression after IN, but got error %v", err)
  652. } else {
  653. return exp, nil
  654. }
  655. }
  656. func (p *Parser) parseBracketExpr() (ast.Expr, error) {
  657. tok2, lit2 := p.scanIgnoreWhiteSpaceWithNegativeNum()
  658. if tok2 == ast.RBRACKET {
  659. //field[]
  660. return &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 0}, End: &ast.IntegerLiteral{Val: math.MinInt32}}, nil
  661. } else if tok2 == ast.INTEGER {
  662. start, err := strconv.Atoi(lit2)
  663. if err != nil {
  664. return nil, fmt.Errorf("The start index %s is not an int value in bracket expression.", lit2)
  665. }
  666. if tok3, _ := p.scanIgnoreWhitespace(); tok3 == ast.RBRACKET {
  667. //Such as field[2]
  668. return &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: start}}, nil
  669. } else if tok3 == ast.COLON {
  670. //Such as field[2:] or field[2:4]
  671. return p.parseColonExpr(&ast.IntegerLiteral{Val: start})
  672. }
  673. } else if tok2 == ast.COLON {
  674. //Such as field[:3] or [:]
  675. return p.parseColonExpr(&ast.IntegerLiteral{Val: 0})
  676. } else {
  677. p.unscan()
  678. start, err := p.ParseExpr()
  679. if err != nil {
  680. return nil, fmt.Errorf("The start index %s is invalid in bracket expression.", lit2)
  681. }
  682. if tok3, _ := p.scanIgnoreWhitespace(); tok3 == ast.RBRACKET {
  683. //Such as field[2]
  684. return &ast.IndexExpr{Index: start}, nil
  685. } else if tok3 == ast.COLON {
  686. //Such as field[2:] or field[2:4]
  687. return p.parseColonExpr(start)
  688. }
  689. }
  690. return nil, fmt.Errorf("Unexpected token %q. when parsing bracket expressions.", lit2)
  691. }
  692. func (p *Parser) parseColonExpr(start ast.Expr) (ast.Expr, error) {
  693. tok, lit := p.scanIgnoreWhiteSpaceWithNegativeNum()
  694. if tok == ast.INTEGER {
  695. end, err := strconv.Atoi(lit)
  696. if err != nil {
  697. return nil, fmt.Errorf("The end index %s is not an int value in bracket expression.", lit)
  698. }
  699. if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.RBRACKET {
  700. return &ast.ColonExpr{Start: start, End: &ast.IntegerLiteral{Val: end}}, nil
  701. } else {
  702. return nil, fmt.Errorf("Found %q, expected right bracket.", lit1)
  703. }
  704. } else if tok == ast.RBRACKET {
  705. return &ast.ColonExpr{Start: start, End: &ast.IntegerLiteral{Val: math.MinInt32}}, nil
  706. }
  707. p.unscan()
  708. end, err := p.ParseExpr()
  709. if err != nil {
  710. return nil, fmt.Errorf("The end index %s is invalid in bracket expression.", lit)
  711. }
  712. if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.RBRACKET {
  713. return &ast.ColonExpr{Start: start, End: end}, nil
  714. } else {
  715. return nil, fmt.Errorf("Found %q, expected right bracket.", lit1)
  716. }
  717. }
  718. func (p *Parser) scanIgnoreWhiteSpaceWithNegativeNum() (ast.Token, string) {
  719. tok, lit := p.scanIgnoreWhitespace()
  720. if tok == ast.SUB {
  721. _, _ = p.s.ScanWhiteSpace()
  722. r := p.s.read()
  723. if isDigit(r) {
  724. p.s.unread()
  725. tok, lit = p.s.ScanNumber(false, true)
  726. }
  727. }
  728. return tok, lit
  729. }
  730. func (p *Parser) parseAs(f *ast.Field) (*ast.Field, error) {
  731. tok, lit := p.scanIgnoreWhitespace()
  732. if tok != ast.IDENT {
  733. return nil, fmt.Errorf("found %q, expected as alias.", lit)
  734. }
  735. f.AName = lit
  736. return f, nil
  737. }
  738. var WindowFuncs = map[string]struct{}{
  739. "tumblingwindow": {},
  740. "hoppingwindow": {},
  741. "sessionwindow": {},
  742. "slidingwindow": {},
  743. "countwindow": {},
  744. }
  745. func convFuncName(n string) (string, bool) {
  746. lname := strings.ToLower(n)
  747. if _, ok := WindowFuncs[lname]; ok {
  748. return lname, ok
  749. } else {
  750. return function.ConvName(n)
  751. }
  752. }
  753. func (p *Parser) parseCall(n string) (ast.Expr, error) {
  754. // Check if n function exists and convert it to lowercase for built-in func
  755. name, ok := convFuncName(n)
  756. if !ok {
  757. return nil, fmt.Errorf("function %s not found", n)
  758. }
  759. p.inFunc = name
  760. defer func() { p.inFunc = "" }()
  761. ft := function.GetFuncType(name)
  762. if ft == ast.FuncTypeCols && p.clause != "select" {
  763. return nil, fmt.Errorf("function %s can only be used inside the select clause", n)
  764. }
  765. var args []ast.Expr
  766. for {
  767. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.RPAREN {
  768. break
  769. }
  770. p.unscan()
  771. if exp, err := p.ParseExpr(); err != nil {
  772. return nil, err
  773. } else {
  774. if ft == ast.FuncTypeCols {
  775. field := &ast.ColFuncField{Expr: exp, Name: nameExpr(exp)}
  776. args = append(args, field)
  777. } else {
  778. args = append(args, exp)
  779. }
  780. }
  781. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.COMMA {
  782. if tok != ast.RPAREN {
  783. return nil, fmt.Errorf("found function call %q, expected ), but with %q.", name, lit)
  784. }
  785. break
  786. }
  787. }
  788. if wt, err := validateWindows(name, args); wt == ast.NOT_WINDOW {
  789. if valErr := validateFuncs(name, args); valErr != nil {
  790. return nil, valErr
  791. }
  792. // Add context for some aggregate func
  793. if name == "deduplicate" {
  794. args = append([]ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}}, args...)
  795. }
  796. c := &ast.Call{Name: name, Args: args, FuncId: p.fn, FuncType: ft}
  797. p.fn += 1
  798. e := p.parseOver(c)
  799. return c, e
  800. } else {
  801. if err != nil {
  802. return nil, err
  803. }
  804. win, err := p.ConvertToWindows(wt, args)
  805. if err != nil {
  806. return nil, err
  807. }
  808. // parse filter clause
  809. f, err := p.parseFilter()
  810. if err != nil {
  811. return nil, err
  812. } else if f != nil {
  813. win.Filter = f
  814. }
  815. return win, nil
  816. }
  817. }
  818. func (p *Parser) parseCaseExpr() (*ast.CaseExpr, error) {
  819. c := &ast.CaseExpr{}
  820. tok, _ := p.scanIgnoreWhitespace()
  821. p.unscan()
  822. if tok != ast.WHEN { // no condition value for case, additional validation needed
  823. if exp, err := p.ParseExpr(); err != nil {
  824. return nil, err
  825. } else {
  826. c.Value = exp
  827. }
  828. }
  829. loop:
  830. for {
  831. tok, _ := p.scanIgnoreWhitespace()
  832. switch tok {
  833. case ast.WHEN:
  834. if exp, err := p.ParseExpr(); err != nil {
  835. return nil, err
  836. } else {
  837. if c.WhenClauses == nil {
  838. c.WhenClauses = make([]*ast.WhenClause, 0)
  839. }
  840. if c.Value == nil && !ast.IsBooleanArg(exp) {
  841. return nil, fmt.Errorf("invalid CASE expression, WHEN expression must be a bool condition")
  842. }
  843. w := &ast.WhenClause{
  844. Expr: exp,
  845. }
  846. tokThen, _ := p.scanIgnoreWhitespace()
  847. if tokThen != ast.THEN {
  848. return nil, fmt.Errorf("invalid CASE expression, THEN expected after WHEN")
  849. } else {
  850. if expThen, err := p.ParseExpr(); err != nil {
  851. return nil, err
  852. } else {
  853. w.Result = expThen
  854. c.WhenClauses = append(c.WhenClauses, w)
  855. }
  856. }
  857. }
  858. case ast.ELSE:
  859. if c.WhenClauses != nil {
  860. if exp, err := p.ParseExpr(); err != nil {
  861. return nil, err
  862. } else {
  863. c.ElseClause = exp
  864. }
  865. } else {
  866. return nil, fmt.Errorf("invalid CASE expression, WHEN expected before ELSE")
  867. }
  868. case ast.END:
  869. if c.WhenClauses != nil {
  870. break loop
  871. }
  872. return nil, fmt.Errorf("invalid CASE expression, WHEN expected before END")
  873. default:
  874. return nil, fmt.Errorf("invalid CASE expression, END expected")
  875. }
  876. }
  877. return c, nil
  878. }
  879. func validateWindows(fname string, args []ast.Expr) (ast.WindowType, error) {
  880. switch fname {
  881. case "tumblingwindow":
  882. if err := validateWindow(fname, 2, args); err != nil {
  883. return ast.TUMBLING_WINDOW, err
  884. }
  885. return ast.TUMBLING_WINDOW, nil
  886. case "hoppingwindow":
  887. if err := validateWindow(fname, 3, args); err != nil {
  888. return ast.HOPPING_WINDOW, err
  889. }
  890. return ast.HOPPING_WINDOW, nil
  891. case "sessionwindow":
  892. if err := validateWindow(fname, 3, args); err != nil {
  893. return ast.SESSION_WINDOW, err
  894. }
  895. return ast.SESSION_WINDOW, nil
  896. case "slidingwindow":
  897. if err := validateWindow(fname, 2, args); err != nil {
  898. return ast.SLIDING_WINDOW, err
  899. }
  900. return ast.SLIDING_WINDOW, nil
  901. case "countwindow":
  902. if len(args) == 1 {
  903. if para1, ok := args[0].(*ast.IntegerLiteral); ok && para1.Val > 0 {
  904. return ast.COUNT_WINDOW, nil
  905. } else {
  906. return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter value %s.", args[0])
  907. }
  908. } else if len(args) == 2 {
  909. if para1, ok1 := args[0].(*ast.IntegerLiteral); ok1 {
  910. if para2, ok2 := args[1].(*ast.IntegerLiteral); ok2 {
  911. if para1.Val < para2.Val {
  912. return ast.COUNT_WINDOW, fmt.Errorf("The second parameter value %d should be less than the first parameter %d.", para2.Val, para1.Val)
  913. } else {
  914. return ast.COUNT_WINDOW, nil
  915. }
  916. }
  917. }
  918. return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter value %s, %s.", args[0], args[1])
  919. } else {
  920. return ast.COUNT_WINDOW, fmt.Errorf("Invalid parameter count.")
  921. }
  922. }
  923. return ast.NOT_WINDOW, nil
  924. }
  925. func validateWindow(funcName string, expectLen int, args []ast.Expr) error {
  926. if len(args) != expectLen {
  927. return fmt.Errorf("The arguments for %s should be %d.\n", funcName, expectLen)
  928. }
  929. if _, ok := args[0].(*ast.TimeLiteral); !ok {
  930. return fmt.Errorf("The 1st argument for %s is expecting timer literal expression. One value of [dd|hh|mi|ss|ms].\n", funcName)
  931. }
  932. for i := 1; i < len(args); i++ {
  933. if _, ok := args[i].(*ast.IntegerLiteral); !ok {
  934. return fmt.Errorf("The %d argument for %s is expecting interger literal expression. \n", i, funcName)
  935. }
  936. }
  937. return nil
  938. }
  939. func (p *Parser) ConvertToWindows(wtype ast.WindowType, args []ast.Expr) (*ast.Window, error) {
  940. win := &ast.Window{WindowType: wtype}
  941. if wtype == ast.COUNT_WINDOW {
  942. win.Length = &ast.IntegerLiteral{Val: args[0].(*ast.IntegerLiteral).Val}
  943. if len(args) == 2 {
  944. win.Interval = &ast.IntegerLiteral{Val: args[1].(*ast.IntegerLiteral).Val}
  945. }
  946. return win, nil
  947. }
  948. var unit = 1
  949. v := args[0].(*ast.TimeLiteral).Val
  950. switch v {
  951. case ast.DD:
  952. unit = 24 * 3600 * 1000
  953. case ast.HH:
  954. unit = 3600 * 1000
  955. case ast.MI:
  956. unit = 60 * 1000
  957. case ast.SS:
  958. unit = 1000
  959. case ast.MS:
  960. unit = 1
  961. default:
  962. return nil, fmt.Errorf("Invalid timeliteral %s", v)
  963. }
  964. win.Length = &ast.IntegerLiteral{Val: args[1].(*ast.IntegerLiteral).Val * unit}
  965. if len(args) > 2 {
  966. win.Interval = &ast.IntegerLiteral{Val: args[2].(*ast.IntegerLiteral).Val * unit}
  967. } else {
  968. win.Interval = &ast.IntegerLiteral{Val: 0}
  969. }
  970. return win, nil
  971. }
  972. func (p *Parser) ParseCreateStmt() (ast.Statement, error) {
  973. _, lit := p.scanIgnoreWhitespace()
  974. lit = strings.ToUpper(lit)
  975. if lit == ast.CREATE {
  976. _, lit1 := p.scanIgnoreWhitespace()
  977. stmt := &ast.StreamStmt{}
  978. lit1 = strings.ToUpper(lit1)
  979. switch lit1 {
  980. case ast.STREAM:
  981. stmt.StreamType = ast.TypeStream
  982. case ast.TABLE:
  983. stmt.StreamType = ast.TypeTable
  984. default:
  985. return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
  986. }
  987. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  988. stmt.Name = ast.StreamName(lit2)
  989. if fields, err := p.parseStreamFields(); err != nil {
  990. return nil, err
  991. } else {
  992. stmt.StreamFields = fields
  993. }
  994. if opts, err := p.parseStreamOptions(); err != nil {
  995. return nil, err
  996. } else {
  997. stmt.Options = opts
  998. }
  999. if tok3, lit3 := p.scanIgnoreWhitespace(); tok3 == ast.SEMICOLON {
  1000. p.unscan()
  1001. } else if tok3 == ast.EOF {
  1002. //Finish parsing create stream statement. Jump to validate
  1003. } else {
  1004. return nil, fmt.Errorf("found %q, expected semicolon or EOF.", lit3)
  1005. }
  1006. } else {
  1007. return nil, fmt.Errorf("found %q, expected stream name.", lit2)
  1008. }
  1009. if valErr := validateStream(stmt); valErr != nil {
  1010. return nil, valErr
  1011. }
  1012. return stmt, nil
  1013. } else {
  1014. p.unscan()
  1015. return nil, nil
  1016. }
  1017. }
  1018. // TODO more accurate validation for table
  1019. func validateStream(stmt *ast.StreamStmt) error {
  1020. f := stmt.Options.FORMAT
  1021. if f == "" {
  1022. f = message.FormatJson
  1023. }
  1024. lf := strings.ToLower(f)
  1025. switch lf {
  1026. case message.FormatBinary:
  1027. if stmt.StreamType == ast.TypeTable {
  1028. return fmt.Errorf("'binary' format is not supported for table")
  1029. }
  1030. switch len(stmt.StreamFields) {
  1031. case 0:
  1032. // do nothing for schemaless
  1033. case 1:
  1034. f := stmt.StreamFields[0]
  1035. if bt, ok := f.FieldType.(*ast.BasicType); ok {
  1036. if bt.Type == ast.BYTEA {
  1037. break
  1038. }
  1039. }
  1040. return fmt.Errorf("'binary' format stream can have only 'bytea' type field")
  1041. default:
  1042. return fmt.Errorf("'binary' format stream can have only one field")
  1043. }
  1044. default:
  1045. if !message.IsFormatSupported(lf) {
  1046. return fmt.Errorf("option 'format=%s' is invalid", f)
  1047. }
  1048. }
  1049. return nil
  1050. }
  1051. func (p *Parser) parseShowStmt() (ast.Statement, error) {
  1052. _, lit := p.scanIgnoreWhitespace()
  1053. lit = strings.ToUpper(lit)
  1054. if lit == ast.SHOW {
  1055. _, lit1 := p.scanIgnoreWhitespace()
  1056. lit1 = strings.ToUpper(lit1)
  1057. switch lit1 {
  1058. case ast.STREAMS:
  1059. ss := &ast.ShowStreamsStatement{}
  1060. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EOF || tok2 == ast.SEMICOLON {
  1061. return ss, nil
  1062. } else {
  1063. return nil, fmt.Errorf("found %q, expected semecolon or EOF.", lit2)
  1064. }
  1065. case ast.TABLES:
  1066. ss := &ast.ShowTablesStatement{}
  1067. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EOF || tok2 == ast.SEMICOLON {
  1068. return ss, nil
  1069. } else {
  1070. return nil, fmt.Errorf("found %q, expected semecolon or EOF.", lit2)
  1071. }
  1072. default:
  1073. return nil, fmt.Errorf("found %q, expected keyword streams or tables.", lit1)
  1074. }
  1075. } else {
  1076. p.unscan()
  1077. return nil, nil
  1078. }
  1079. }
  1080. func (p *Parser) parseDescribeStmt() (ast.Statement, error) {
  1081. _, lit := p.scanIgnoreWhitespace()
  1082. lit = strings.ToUpper(lit)
  1083. if lit == ast.DESCRIBE {
  1084. _, lit1 := p.scanIgnoreWhitespace()
  1085. lit1 = strings.ToUpper(lit1)
  1086. switch lit1 {
  1087. case ast.STREAM:
  1088. dss := &ast.DescribeStreamStatement{}
  1089. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1090. dss.Name = lit2
  1091. return dss, nil
  1092. } else {
  1093. return nil, fmt.Errorf("found %q, expected stream name.", lit2)
  1094. }
  1095. case ast.TABLE:
  1096. dss := &ast.DescribeTableStatement{}
  1097. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1098. dss.Name = lit2
  1099. return dss, nil
  1100. } else {
  1101. return nil, fmt.Errorf("found %q, expected table name.", lit2)
  1102. }
  1103. default:
  1104. return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
  1105. }
  1106. } else {
  1107. p.unscan()
  1108. return nil, nil
  1109. }
  1110. }
  1111. func (p *Parser) parseExplainStmt() (ast.Statement, error) {
  1112. _, lit := p.scanIgnoreWhitespace()
  1113. lit = strings.ToUpper(lit)
  1114. if lit == ast.EXPLAIN {
  1115. _, lit1 := p.scanIgnoreWhitespace()
  1116. lit1 = strings.ToUpper(lit1)
  1117. switch lit1 {
  1118. case ast.STREAM:
  1119. ess := &ast.ExplainStreamStatement{}
  1120. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1121. ess.Name = lit2
  1122. return ess, nil
  1123. } else {
  1124. return nil, fmt.Errorf("found %q, expected stream name.", lit2)
  1125. }
  1126. case ast.TABLE:
  1127. ess := &ast.ExplainTableStatement{}
  1128. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1129. ess.Name = lit2
  1130. return ess, nil
  1131. } else {
  1132. return nil, fmt.Errorf("found %q, expected table name.", lit2)
  1133. }
  1134. default:
  1135. return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
  1136. }
  1137. } else {
  1138. p.unscan()
  1139. return nil, nil
  1140. }
  1141. }
  1142. func (p *Parser) parseDropStmt() (ast.Statement, error) {
  1143. _, lit := p.scanIgnoreWhitespace()
  1144. lit = strings.ToUpper(lit)
  1145. if lit == ast.DROP {
  1146. _, lit1 := p.scanIgnoreWhitespace()
  1147. lit1 = strings.ToUpper(lit1)
  1148. switch lit1 {
  1149. case ast.STREAM:
  1150. ess := &ast.DropStreamStatement{}
  1151. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1152. ess.Name = lit2
  1153. return ess, nil
  1154. } else {
  1155. return nil, fmt.Errorf("found %q, expected stream name.", lit2)
  1156. }
  1157. case ast.TABLE:
  1158. ess := &ast.DropTableStatement{}
  1159. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.IDENT {
  1160. ess.Name = lit2
  1161. return ess, nil
  1162. } else {
  1163. return nil, fmt.Errorf("found %q, expected table name.", lit2)
  1164. }
  1165. default:
  1166. return nil, fmt.Errorf("found %q, expected keyword stream or table.", lit1)
  1167. }
  1168. } else {
  1169. p.unscan()
  1170. return nil, nil
  1171. }
  1172. }
  1173. func (p *Parser) parseStreamFields() (ast.StreamFields, error) {
  1174. lStack := &stack.Stack{}
  1175. var fields ast.StreamFields
  1176. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
  1177. lStack.Push(lit)
  1178. for {
  1179. //For the schemaless streams
  1180. //create stream demo () WITH (FORMAT="JSON", DATASOURCE="demo" TYPE="edgex")
  1181. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.RPAREN {
  1182. lStack.Pop()
  1183. if _, lit2 := p.scanIgnoreWhitespace(); strings.ToUpper(lit2) != ast.WITH {
  1184. return nil, fmt.Errorf("found %q, expected is with.", lit2)
  1185. }
  1186. return fields, nil
  1187. } else {
  1188. p.unscan()
  1189. }
  1190. if f, err := p.parseStreamField(); err != nil {
  1191. return nil, err
  1192. } else {
  1193. fields = append(fields, *f)
  1194. }
  1195. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.RPAREN {
  1196. lStack.Pop()
  1197. tok2, lit2 := p.scanIgnoreWhitespace()
  1198. lit2 = strings.ToUpper(lit2)
  1199. if lit2 == ast.WITH {
  1200. //Check the stack for LPAREN; If the stack for LPAREN is not zero, then it's not correct.
  1201. if lStack.Len() > 0 {
  1202. return nil, fmt.Errorf("Parenthesis is not matched.")
  1203. }
  1204. break
  1205. } else if tok2 == ast.COMMA {
  1206. if lStack.Len() > 0 {
  1207. return nil, fmt.Errorf("Parenthesis is in create record type not matched.")
  1208. }
  1209. p.unscan()
  1210. break
  1211. } else if tok2 == ast.RPAREN { //The nested type definition of ARRAY and Struct, such as "field ARRAY(STRUCT(f BIGINT))"
  1212. if lStack.Len() > 0 {
  1213. return nil, fmt.Errorf("Parenthesis is not matched.")
  1214. }
  1215. p.unscan()
  1216. break
  1217. } else {
  1218. if lStack.Len() == 0 {
  1219. return nil, fmt.Errorf("found %q, expected is with.", lit2)
  1220. }
  1221. p.unscan()
  1222. }
  1223. } else {
  1224. p.unscan()
  1225. }
  1226. }
  1227. } else {
  1228. return nil, fmt.Errorf("found %q, expected lparen after stream name.", lit)
  1229. }
  1230. return fields, nil
  1231. }
  1232. func (p *Parser) parseStreamField() (*ast.StreamField, error) {
  1233. field := &ast.StreamField{}
  1234. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.IDENT {
  1235. field.Name = lit
  1236. _, lit1 := p.scanIgnoreWhitespace()
  1237. if t := ast.GetDataType(lit1); t != ast.UNKNOWN && t.IsSimpleType() {
  1238. field.FieldType = &ast.BasicType{Type: t}
  1239. } else if t == ast.ARRAY {
  1240. if f, e := p.parseStreamArrayType(); e != nil {
  1241. return nil, e
  1242. } else {
  1243. field.FieldType = f
  1244. }
  1245. } else if t == ast.STRUCT {
  1246. if f, e := p.parseStreamStructType(); e != nil {
  1247. return nil, e
  1248. } else {
  1249. field.FieldType = f
  1250. }
  1251. } else if t == ast.UNKNOWN {
  1252. return nil, fmt.Errorf("found %q, expect valid stream field types(BIGINT | FLOAT | STRINGS | DATETIME | BOOLEAN | BYTEA | ARRAY | STRUCT).", lit1)
  1253. }
  1254. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.COMMA {
  1255. //Just consume the comma.
  1256. } else if tok2 == ast.RPAREN {
  1257. p.unscan()
  1258. } else {
  1259. return nil, fmt.Errorf("found %q, expect comma or rparen.", lit2)
  1260. }
  1261. } else {
  1262. return nil, fmt.Errorf("found %q, expect stream field name.", lit)
  1263. }
  1264. return field, nil
  1265. }
  1266. func (p *Parser) parseStreamArrayType() (ast.FieldType, error) {
  1267. lStack := &stack.Stack{}
  1268. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
  1269. lStack.Push(ast.LPAREN)
  1270. tok1, lit1 := p.scanIgnoreWhitespace()
  1271. t := ast.GetDataType(lit1)
  1272. if t != ast.UNKNOWN && t.IsSimpleType() {
  1273. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.RPAREN {
  1274. lStack.Pop()
  1275. if lStack.Len() > 0 {
  1276. return nil, fmt.Errorf("Parenthesis is in array type not matched.")
  1277. }
  1278. return &ast.ArrayType{Type: t}, nil
  1279. } else {
  1280. return nil, fmt.Errorf("found %q, expect rparen in array type definition.", lit2)
  1281. }
  1282. } else if t == ast.STRUCT {
  1283. if f, err := p.parseStreamStructType(); err != nil {
  1284. return nil, err
  1285. } else {
  1286. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.RPAREN {
  1287. lStack.Pop()
  1288. if lStack.Len() > 0 {
  1289. return nil, fmt.Errorf("Parenthesis is in struct of array type %q not matched.", tok1)
  1290. }
  1291. return &ast.ArrayType{Type: ast.STRUCT, FieldType: f}, nil
  1292. } else {
  1293. return nil, fmt.Errorf("found %q, expect rparen in struct of array type definition.", lit2)
  1294. }
  1295. }
  1296. } else if tok1 == ast.COMMA {
  1297. p.unscan()
  1298. } else {
  1299. return nil, fmt.Errorf("found %q, expect stream data types.", lit1)
  1300. }
  1301. }
  1302. return nil, nil
  1303. }
  1304. func (p *Parser) parseStreamStructType() (ast.FieldType, error) {
  1305. rf := &ast.RecType{}
  1306. if sfs, err := p.parseStreamFields(); err != nil {
  1307. return nil, err
  1308. } else {
  1309. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.COMMA {
  1310. rf.StreamFields = sfs
  1311. p.unscan()
  1312. } else if tok2 == ast.RPAREN {
  1313. rf.StreamFields = sfs
  1314. p.unscan()
  1315. } else {
  1316. return nil, fmt.Errorf("found %q, expect comma in create stream record statement.", lit2)
  1317. }
  1318. }
  1319. return rf, nil
  1320. }
  1321. func (p *Parser) parseStreamOptions() (*ast.Options, error) {
  1322. opts := &ast.Options{STRICT_VALIDATION: false}
  1323. v := reflect.ValueOf(opts)
  1324. lStack := &stack.Stack{}
  1325. if tok, lit := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
  1326. lStack.Push(ast.LPAREN)
  1327. for {
  1328. tok1, lit1 := p.scanIgnoreWhitespace()
  1329. lit1 = strings.ToUpper(lit1)
  1330. if ast.IsStreamOptionKeyword(tok1, lit1) {
  1331. if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EQ {
  1332. if tok3, lit3 := p.scanIgnoreWhitespace(); tok3 == ast.STRING {
  1333. switch lit1 {
  1334. case ast.STRICT_VALIDATION:
  1335. if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
  1336. return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, lit1)
  1337. } else {
  1338. opts.STRICT_VALIDATION = val == "TRUE"
  1339. }
  1340. case ast.RETAIN_SIZE:
  1341. if val, err := strconv.Atoi(lit3); err != nil {
  1342. return nil, fmt.Errorf("found %q, expect number value in %s option.", lit3, lit1)
  1343. } else {
  1344. opts.RETAIN_SIZE = val
  1345. }
  1346. case ast.SHARED:
  1347. if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
  1348. return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, lit1)
  1349. } else {
  1350. opts.SHARED = val == "TRUE"
  1351. }
  1352. case ast.KIND:
  1353. val := strings.ToLower(lit3)
  1354. opts.KIND = val
  1355. default:
  1356. f := v.Elem().FieldByName(lit1)
  1357. if f.IsValid() {
  1358. f.SetString(lit3)
  1359. } else { // should not happen
  1360. return nil, fmt.Errorf("invalid field %s.", lit1)
  1361. }
  1362. }
  1363. } else {
  1364. return nil, fmt.Errorf("found %q, expect string value in option.", lit3)
  1365. }
  1366. } else {
  1367. return nil, fmt.Errorf("found %q, expect equals(=) in options.", lit2)
  1368. }
  1369. } else if tok1 == ast.COMMA {
  1370. continue
  1371. } else if tok1 == ast.RPAREN {
  1372. if lStack.Pop(); lStack.Len() == 0 {
  1373. break
  1374. }
  1375. return nil, fmt.Errorf("Parenthesis is not matched in options definition.")
  1376. } else {
  1377. 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)
  1378. }
  1379. }
  1380. } else {
  1381. return nil, fmt.Errorf("found %q, expect stream options.", lit)
  1382. }
  1383. if opts.KIND == ast.StreamKindLookup && opts.TYPE == "memory" && opts.KEY == "" {
  1384. return nil, fmt.Errorf("Option \"key\" is required for memory lookup table.")
  1385. }
  1386. return opts, nil
  1387. }
  1388. // Only support filter on window now
  1389. func (p *Parser) parseFilter() (ast.Expr, error) {
  1390. if tok, _ := p.scanIgnoreWhitespace(); tok != ast.FILTER {
  1391. p.unscan()
  1392. return nil, nil
  1393. }
  1394. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.LPAREN {
  1395. return nil, fmt.Errorf("Found %q after FILTER, expect parentheses.", lit)
  1396. }
  1397. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.WHERE {
  1398. return nil, fmt.Errorf("Found %q after FILTER(, expect WHERE.", lit)
  1399. }
  1400. expr, err := p.ParseExpr()
  1401. if err != nil {
  1402. return nil, err
  1403. }
  1404. if tok, lit := p.scanIgnoreWhitespace(); tok != ast.RPAREN {
  1405. return nil, fmt.Errorf("Found %q after FILTER, expect right parentheses.", lit)
  1406. }
  1407. return expr, nil
  1408. }
  1409. func (p *Parser) parseAsterisk() (ast.Expr, error) {
  1410. switch p.inFunc {
  1411. case "mqtt", "meta":
  1412. return &ast.MetaRef{StreamName: ast.DefaultStream, Name: "*"}, nil
  1413. default:
  1414. return &ast.Wildcard{Token: ast.ASTERISK}, nil
  1415. }
  1416. }
  1417. func (p *Parser) inmeta() bool {
  1418. return p.inFunc == "meta" || p.inFunc == "mqtt"
  1419. }
  1420. func (p *Parser) parseOver(c *ast.Call) error {
  1421. if tok, _ := p.scanIgnoreWhitespace(); tok != ast.OVER {
  1422. p.unscan()
  1423. return nil
  1424. } else if function.IsAnalyticFunc(c.Name) {
  1425. if tok1, _ := p.scanIgnoreWhitespace(); tok1 == ast.LPAREN {
  1426. if t, _ := p.scanIgnoreWhitespace(); t == ast.PARTITION {
  1427. if t1, l1 := p.scanIgnoreWhitespace(); t1 == ast.BY {
  1428. pe := &ast.PartitionExpr{}
  1429. for {
  1430. if exp, err := p.ParseExpr(); err != nil {
  1431. return err
  1432. } else {
  1433. pe.Exprs = append(pe.Exprs, exp)
  1434. }
  1435. if tok, _ := p.scanIgnoreWhitespace(); tok == ast.COMMA {
  1436. continue
  1437. }
  1438. p.unscan()
  1439. break
  1440. }
  1441. if len(pe.Exprs) == 0 {
  1442. return fmt.Errorf("PARTITION BY must have at least one expression.")
  1443. }
  1444. c.Partition = pe
  1445. } else {
  1446. return fmt.Errorf("found %q, expected by after partition.", l1)
  1447. }
  1448. } else {
  1449. p.unscan()
  1450. }
  1451. if t, _ := p.scanIgnoreWhitespace(); t == ast.WHEN {
  1452. if exp, err := p.ParseExpr(); err != nil {
  1453. return err
  1454. } else {
  1455. c.WhenExpr = exp
  1456. }
  1457. } else {
  1458. p.unscan()
  1459. }
  1460. if c.Partition != nil || c.WhenExpr != nil {
  1461. if ttt, _ := p.scanIgnoreWhitespace(); ttt != ast.RPAREN {
  1462. return fmt.Errorf("Found %q, expect right parentheses after OVER ", ttt)
  1463. }
  1464. }
  1465. if c.Partition == nil && c.WhenExpr == nil {
  1466. ttt, _ := p.scanIgnoreWhitespace()
  1467. return fmt.Errorf("Found %q after OVER (, expect partition by or when.", ttt)
  1468. }
  1469. return nil
  1470. } else {
  1471. return fmt.Errorf("Found %q after OVER, expect parentheses.", tok1)
  1472. }
  1473. } else {
  1474. return fmt.Errorf("Found OVER after non analytic function %s", c.Name)
  1475. }
  1476. }