parser.go 44 KB

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