parser.go 45 KB

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