project_test.go 39 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580
  1. package plans
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "fmt"
  6. "github.com/emqx/kuiper/common"
  7. "github.com/emqx/kuiper/xsql"
  8. "github.com/emqx/kuiper/xstream/contexts"
  9. "reflect"
  10. "strings"
  11. "testing"
  12. )
  13. func TestProjectPlan_Apply1(t *testing.T) {
  14. var tests = []struct {
  15. sql string
  16. data *xsql.Tuple
  17. result []map[string]interface{}
  18. }{
  19. {
  20. sql: "SELECT a FROM test",
  21. data: &xsql.Tuple{
  22. Emitter: "test",
  23. Message: xsql.Message{
  24. "a": "val_a",
  25. },
  26. },
  27. result: []map[string]interface{}{{
  28. "a": "val_a",
  29. }},
  30. },
  31. {
  32. sql: "SELECT b FROM test",
  33. data: &xsql.Tuple{
  34. Emitter: "test",
  35. Message: xsql.Message{
  36. "a": "val_a",
  37. },
  38. },
  39. result: []map[string]interface{}{{}},
  40. },
  41. {
  42. sql: "SELECT ts FROM test",
  43. data: &xsql.Tuple{
  44. Emitter: "test",
  45. Message: xsql.Message{
  46. "a": "val_a",
  47. "ts": common.TimeFromUnixMilli(1568854573431),
  48. },
  49. },
  50. result: []map[string]interface{}{{
  51. "ts": "2019-09-19T00:56:13.431Z",
  52. }},
  53. },
  54. //Schemaless may return a message without selecting column
  55. {
  56. sql: "SELECT ts FROM test",
  57. data: &xsql.Tuple{
  58. Emitter: "test",
  59. Message: xsql.Message{
  60. "a": "val_a",
  61. "ts2": common.TimeFromUnixMilli(1568854573431),
  62. },
  63. },
  64. result: []map[string]interface{}{{}},
  65. },
  66. {
  67. sql: "SELECT A FROM test",
  68. data: &xsql.Tuple{
  69. Emitter: "test",
  70. Message: xsql.Message{
  71. "a": "val_a",
  72. },
  73. },
  74. result: []map[string]interface{}{{
  75. "A": "val_a",
  76. }},
  77. },
  78. {
  79. sql: `SELECT "value" FROM test`,
  80. data: &xsql.Tuple{
  81. Emitter: "test",
  82. Message: xsql.Message{},
  83. },
  84. result: []map[string]interface{}{{
  85. DEFAULT_FIELD_NAME_PREFIX + "0": "value",
  86. }},
  87. },
  88. {
  89. sql: `SELECT 3.4 FROM test`,
  90. data: &xsql.Tuple{
  91. Emitter: "test",
  92. Message: xsql.Message{},
  93. },
  94. result: []map[string]interface{}{{
  95. DEFAULT_FIELD_NAME_PREFIX + "0": 3.4,
  96. }},
  97. },
  98. {
  99. sql: `SELECT 5 FROM test`,
  100. data: &xsql.Tuple{
  101. Emitter: "test",
  102. Message: xsql.Message{},
  103. },
  104. result: []map[string]interface{}{{
  105. DEFAULT_FIELD_NAME_PREFIX + "0": 5.0,
  106. }},
  107. },
  108. {
  109. sql: `SELECT a, "value" AS b FROM test`,
  110. data: &xsql.Tuple{
  111. Emitter: "test",
  112. Message: xsql.Message{
  113. "a": "val_a",
  114. },
  115. },
  116. result: []map[string]interface{}{{
  117. "a": "val_a",
  118. "b": "value",
  119. }},
  120. },
  121. {
  122. sql: `SELECT a, "value" AS b, 3.14 as Pi, 0 as Zero FROM test`,
  123. data: &xsql.Tuple{
  124. Emitter: "test",
  125. Message: xsql.Message{
  126. "a": "val_a",
  127. },
  128. },
  129. result: []map[string]interface{}{{
  130. "a": "val_a",
  131. "b": "value",
  132. "Pi": 3.14,
  133. "Zero": 0.0,
  134. }},
  135. },
  136. {
  137. sql: `SELECT a->b AS ab FROM test`,
  138. data: &xsql.Tuple{
  139. Emitter: "test",
  140. Message: xsql.Message{
  141. "a": map[string]interface{}{"b": "hello"},
  142. },
  143. },
  144. result: []map[string]interface{}{{
  145. "ab": "hello",
  146. }},
  147. },
  148. {
  149. sql: `SELECT a->b AS ab FROM test`,
  150. data: &xsql.Tuple{
  151. Emitter: "test",
  152. Message: xsql.Message{
  153. "a": map[string]interface{}(nil),
  154. },
  155. },
  156. result: []map[string]interface{}{{}},
  157. },
  158. {
  159. sql: `SELECT a->b AS ab FROM test`,
  160. data: &xsql.Tuple{
  161. Emitter: "test",
  162. Message: xsql.Message{
  163. "name": "name",
  164. },
  165. },
  166. result: []map[string]interface{}{{}},
  167. },
  168. {
  169. sql: `SELECT a->b AS ab FROM test`,
  170. data: &xsql.Tuple{
  171. Emitter: "test",
  172. Message: xsql.Message{
  173. "a": "commonstring",
  174. },
  175. },
  176. result: []map[string]interface{}{{}},
  177. },
  178. {
  179. sql: `SELECT a[0]->b AS ab FROM test`,
  180. data: &xsql.Tuple{
  181. Emitter: "test",
  182. Message: xsql.Message{
  183. "a": []interface{}{
  184. map[string]interface{}{"b": "hello1"},
  185. map[string]interface{}{"b": "hello2"},
  186. },
  187. },
  188. },
  189. result: []map[string]interface{}{{
  190. "ab": "hello1",
  191. }},
  192. },
  193. {
  194. sql: `SELECT a[0]->b AS ab FROM test`,
  195. data: &xsql.Tuple{
  196. Emitter: "test",
  197. Message: xsql.Message{
  198. "a": []map[string]interface{}{
  199. {"b": "hello1"},
  200. {"b": "hello2"},
  201. },
  202. },
  203. },
  204. result: []map[string]interface{}{{
  205. "ab": "hello1",
  206. }},
  207. },
  208. {
  209. sql: `SELECT a[2:4] AS ab FROM test`,
  210. data: &xsql.Tuple{
  211. Emitter: "test",
  212. Message: xsql.Message{
  213. "a": []map[string]interface{}{
  214. {"b": "hello1"},
  215. {"b": "hello2"},
  216. {"b": "hello3"},
  217. {"b": "hello4"},
  218. {"b": "hello5"},
  219. },
  220. },
  221. },
  222. result: []map[string]interface{}{{
  223. "ab": []interface{}{
  224. map[string]interface{}{"b": "hello3"},
  225. map[string]interface{}{"b": "hello4"},
  226. },
  227. }},
  228. },
  229. {
  230. sql: `SELECT a->c->d AS f1 FROM test`,
  231. data: &xsql.Tuple{
  232. Emitter: "test",
  233. Message: xsql.Message{
  234. "a": map[string]interface{}{
  235. "b": "hello",
  236. "c": map[string]interface{}{
  237. "d": 35.2,
  238. },
  239. },
  240. },
  241. },
  242. result: []map[string]interface{}{{
  243. "f1": 35.2,
  244. }},
  245. },
  246. {
  247. sql: `SELECT a->c->d AS f1 FROM test`,
  248. data: &xsql.Tuple{
  249. Emitter: "test",
  250. Message: xsql.Message{
  251. "a": map[string]interface{}{
  252. "b": "hello",
  253. "c": map[string]interface{}{
  254. "e": 35.2,
  255. },
  256. },
  257. },
  258. },
  259. result: []map[string]interface{}{{}},
  260. },
  261. {
  262. sql: `SELECT a->c->d AS f1 FROM test`,
  263. data: &xsql.Tuple{
  264. Emitter: "test",
  265. Message: xsql.Message{
  266. "a": map[string]interface{}{
  267. "b": "hello",
  268. },
  269. },
  270. },
  271. result: []map[string]interface{}{{}},
  272. },
  273. //The int type is not supported yet, the json parser returns float64 for int values
  274. {
  275. sql: `SELECT a->c->d AS f1 FROM test`,
  276. data: &xsql.Tuple{
  277. Emitter: "test",
  278. Message: xsql.Message{
  279. "a": map[string]interface{}{
  280. "b": "hello",
  281. "c": map[string]interface{}{
  282. "d": float64(35),
  283. },
  284. },
  285. },
  286. },
  287. result: []map[string]interface{}{{
  288. "f1": float64(35),
  289. }},
  290. },
  291. {
  292. sql: "SELECT a FROM test",
  293. data: &xsql.Tuple{
  294. Emitter: "test",
  295. Message: xsql.Message{},
  296. },
  297. result: []map[string]interface{}{
  298. {},
  299. },
  300. },
  301. {
  302. sql: "SELECT * FROM test",
  303. data: &xsql.Tuple{
  304. Emitter: "test",
  305. Message: xsql.Message{},
  306. },
  307. result: []map[string]interface{}{
  308. {},
  309. },
  310. },
  311. {
  312. sql: `SELECT * FROM test`,
  313. data: &xsql.Tuple{
  314. Emitter: "test",
  315. Message: xsql.Message{
  316. "a": map[string]interface{}{
  317. "b": "hello",
  318. "c": map[string]interface{}{
  319. "d": 35.2,
  320. },
  321. },
  322. },
  323. },
  324. result: []map[string]interface{}{{
  325. "a": map[string]interface{}{
  326. "b": "hello",
  327. "c": map[string]interface{}{
  328. "d": 35.2,
  329. },
  330. },
  331. }},
  332. },
  333. {
  334. sql: `SELECT * FROM test`,
  335. data: &xsql.Tuple{
  336. Emitter: "test",
  337. Message: xsql.Message{
  338. "a": "val1",
  339. "b": 3.14,
  340. },
  341. },
  342. result: []map[string]interface{}{{
  343. "a": "val1",
  344. "b": 3.14,
  345. }},
  346. },
  347. {
  348. sql: `SELECT 3*4 AS f1 FROM test`,
  349. data: &xsql.Tuple{
  350. Emitter: "test",
  351. Message: xsql.Message{},
  352. },
  353. result: []map[string]interface{}{{
  354. "f1": float64(12),
  355. }},
  356. },
  357. {
  358. sql: `SELECT 4.5*2 AS f1 FROM test`,
  359. data: &xsql.Tuple{
  360. Emitter: "test",
  361. Message: xsql.Message{},
  362. },
  363. result: []map[string]interface{}{{
  364. "f1": float64(9),
  365. }},
  366. },
  367. }
  368. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  369. contextLogger := common.Log.WithField("rule", "TestProjectPlan_Apply1")
  370. ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
  371. for i, tt := range tests {
  372. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  373. pp := &ProjectPlan{Fields: stmt.Fields}
  374. pp.isTest = true
  375. result := pp.Apply(ctx, tt.data)
  376. var mapRes []map[string]interface{}
  377. if v, ok := result.([]byte); ok {
  378. err := json.Unmarshal(v, &mapRes)
  379. if err != nil {
  380. t.Errorf("Failed to parse the input into map.\n")
  381. continue
  382. }
  383. //fmt.Printf("%t\n", mapRes["rengine_field_0"])
  384. if !reflect.DeepEqual(tt.result, mapRes) {
  385. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  386. }
  387. } else {
  388. t.Errorf("%d. The returned result is not type of []byte\n", i)
  389. }
  390. }
  391. }
  392. func TestProjectPlan_MultiInput(t *testing.T) {
  393. var tests = []struct {
  394. sql string
  395. data interface{}
  396. result []map[string]interface{}
  397. }{
  398. {
  399. sql: "SELECT * FROM tbl WHERE abc*2+3 > 12 AND abc < 20",
  400. data: &xsql.Tuple{
  401. Emitter: "tbl",
  402. Message: xsql.Message{
  403. "abc": int64(6),
  404. },
  405. },
  406. result: []map[string]interface{}{{
  407. "abc": float64(6), //json marshall problem
  408. }},
  409. },
  410. {
  411. sql: "SELECT abc FROM tbl WHERE abc*2+3 > 12 OR def = \"hello\"",
  412. data: &xsql.Tuple{
  413. Emitter: "tbl",
  414. Message: xsql.Message{
  415. "abc": int64(34),
  416. "def": "hello",
  417. },
  418. },
  419. result: []map[string]interface{}{{
  420. "abc": float64(34),
  421. }},
  422. },
  423. {
  424. sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  425. data: xsql.WindowTuplesSet{
  426. xsql.WindowTuples{
  427. Emitter: "src1",
  428. Tuples: []xsql.Tuple{
  429. {
  430. Emitter: "src1",
  431. Message: xsql.Message{"id1": 1, "f1": "v1"},
  432. }, {
  433. Emitter: "src1",
  434. Message: xsql.Message{"id1": 2, "f1": "v2"},
  435. }, {
  436. Emitter: "src1",
  437. Message: xsql.Message{"id1": 3, "f1": "v1"},
  438. },
  439. },
  440. },
  441. },
  442. result: []map[string]interface{}{{
  443. "id1": float64(1),
  444. }, {
  445. "id1": float64(2),
  446. }, {
  447. "id1": float64(3),
  448. }},
  449. },
  450. {
  451. sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  452. data: xsql.WindowTuplesSet{
  453. xsql.WindowTuples{
  454. Emitter: "src1",
  455. Tuples: []xsql.Tuple{
  456. {
  457. Emitter: "src1",
  458. Message: xsql.Message{"id1": 1, "f1": "v1"},
  459. }, {
  460. Emitter: "src1",
  461. Message: xsql.Message{"id2": 2, "f1": "v2"},
  462. }, {
  463. Emitter: "src1",
  464. Message: xsql.Message{"id1": 3, "f1": "v1"},
  465. },
  466. },
  467. },
  468. },
  469. result: []map[string]interface{}{{
  470. "id1": float64(1),
  471. }, {}, {
  472. "id1": float64(3),
  473. }},
  474. },
  475. {
  476. sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  477. data: xsql.WindowTuplesSet{
  478. xsql.WindowTuples{
  479. Emitter: "src1",
  480. Tuples: []xsql.Tuple{
  481. {
  482. Emitter: "src1",
  483. Message: xsql.Message{"id1": 1, "f1": "v1"},
  484. }, {
  485. Emitter: "src1",
  486. Message: xsql.Message{"id1": 2, "f1": "v2"},
  487. }, {
  488. Emitter: "src1",
  489. Message: xsql.Message{"id1": 3, "f1": "v1"},
  490. },
  491. },
  492. },
  493. },
  494. result: []map[string]interface{}{{
  495. "id1": float64(1),
  496. "f1": "v1",
  497. }, {
  498. "id1": float64(2),
  499. "f1": "v2",
  500. }, {
  501. "id1": float64(3),
  502. "f1": "v1",
  503. }},
  504. },
  505. {
  506. sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  507. data: xsql.WindowTuplesSet{
  508. xsql.WindowTuples{
  509. Emitter: "src1",
  510. Tuples: []xsql.Tuple{
  511. {
  512. Emitter: "src1",
  513. Message: xsql.Message{"id1": 1, "f1": "v1"},
  514. }, {
  515. Emitter: "src1",
  516. Message: xsql.Message{"id2": 2, "f2": "v2"},
  517. }, {
  518. Emitter: "src1",
  519. Message: xsql.Message{"id1": 3, "f1": "v1"},
  520. },
  521. },
  522. },
  523. },
  524. result: []map[string]interface{}{{
  525. "id1": float64(1),
  526. "f1": "v1",
  527. }, {
  528. "id2": float64(2),
  529. "f2": "v2",
  530. }, {
  531. "id1": float64(3),
  532. "f1": "v1",
  533. }},
  534. },
  535. {
  536. sql: "SELECT src1.* FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  537. data: xsql.WindowTuplesSet{
  538. xsql.WindowTuples{
  539. Emitter: "src1",
  540. Tuples: []xsql.Tuple{
  541. {
  542. Emitter: "src1",
  543. Message: xsql.Message{"id1": 1, "f1": "v1"},
  544. }, {
  545. Emitter: "src1",
  546. Message: xsql.Message{"id1": 2, "f1": "v2"},
  547. }, {
  548. Emitter: "src1",
  549. Message: xsql.Message{"id1": 3, "f1": "v1"},
  550. },
  551. },
  552. },
  553. },
  554. result: []map[string]interface{}{{
  555. "id1": float64(1),
  556. "f1": "v1",
  557. }, {
  558. "id1": float64(2),
  559. "f1": "v2",
  560. }, {
  561. "id1": float64(3),
  562. "f1": "v1",
  563. }},
  564. },
  565. {
  566. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  567. data: xsql.JoinTupleSets{
  568. xsql.JoinTuple{
  569. Tuples: []xsql.Tuple{
  570. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  571. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  572. },
  573. },
  574. xsql.JoinTuple{
  575. Tuples: []xsql.Tuple{
  576. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  577. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  578. },
  579. },
  580. xsql.JoinTuple{
  581. Tuples: []xsql.Tuple{
  582. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  583. },
  584. },
  585. },
  586. result: []map[string]interface{}{{
  587. "id1": float64(1),
  588. }, {
  589. "id1": float64(2),
  590. }, {
  591. "id1": float64(3),
  592. }},
  593. },
  594. {
  595. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  596. data: xsql.JoinTupleSets{
  597. xsql.JoinTuple{
  598. Tuples: []xsql.Tuple{
  599. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  600. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  601. },
  602. },
  603. xsql.JoinTuple{
  604. Tuples: []xsql.Tuple{
  605. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  606. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  607. },
  608. },
  609. xsql.JoinTuple{
  610. Tuples: []xsql.Tuple{
  611. {Emitter: "src1", Message: xsql.Message{"id2": 3, "f1": "v1"}},
  612. },
  613. },
  614. },
  615. result: []map[string]interface{}{{
  616. "id1": float64(1),
  617. }, {
  618. "id1": float64(2),
  619. }, {}},
  620. },
  621. {
  622. sql: "SELECT abc FROM tbl group by abc",
  623. data: xsql.GroupedTuplesSet{
  624. {
  625. &xsql.Tuple{
  626. Emitter: "tbl",
  627. Message: xsql.Message{
  628. "abc": int64(6),
  629. "def": "hello",
  630. },
  631. },
  632. },
  633. },
  634. result: []map[string]interface{}{{
  635. "abc": float64(6),
  636. }},
  637. },
  638. {
  639. sql: "SELECT abc FROM tbl group by abc",
  640. data: xsql.GroupedTuplesSet{
  641. {
  642. &xsql.Tuple{
  643. Emitter: "tbl",
  644. Message: xsql.Message{
  645. "def": "hello",
  646. },
  647. },
  648. },
  649. },
  650. result: []map[string]interface{}{{}},
  651. },
  652. {
  653. sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  654. data: xsql.GroupedTuplesSet{
  655. {
  656. &xsql.Tuple{
  657. Emitter: "src1",
  658. Message: xsql.Message{"id1": 1, "f1": "v1"},
  659. },
  660. &xsql.Tuple{
  661. Emitter: "src1",
  662. Message: xsql.Message{"id1": 3, "f1": "v1"},
  663. },
  664. },
  665. {
  666. &xsql.Tuple{
  667. Emitter: "src1",
  668. Message: xsql.Message{"id1": 2, "f1": "v2"},
  669. },
  670. },
  671. },
  672. result: []map[string]interface{}{{
  673. "id1": float64(1),
  674. }, {
  675. "id1": float64(2),
  676. }},
  677. },
  678. {
  679. sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  680. data: xsql.GroupedTuplesSet{
  681. {
  682. &xsql.Tuple{
  683. Emitter: "src1",
  684. Message: xsql.Message{"id1": 1, "f1": "v1"},
  685. },
  686. &xsql.Tuple{
  687. Emitter: "src1",
  688. Message: xsql.Message{"id1": 3, "f1": "v1"},
  689. },
  690. },
  691. {
  692. &xsql.Tuple{
  693. Emitter: "src1",
  694. Message: xsql.Message{"id2": 2, "f1": "v2"},
  695. },
  696. },
  697. },
  698. result: []map[string]interface{}{{
  699. "id1": float64(1),
  700. }, {}},
  701. },
  702. {
  703. sql: "SELECT src2.id2 FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  704. data: xsql.GroupedTuplesSet{
  705. {
  706. &xsql.JoinTuple{
  707. Tuples: []xsql.Tuple{
  708. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  709. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  710. },
  711. },
  712. },
  713. {
  714. &xsql.JoinTuple{
  715. Tuples: []xsql.Tuple{
  716. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  717. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  718. },
  719. },
  720. },
  721. {
  722. &xsql.JoinTuple{
  723. Tuples: []xsql.Tuple{
  724. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  725. },
  726. },
  727. },
  728. },
  729. result: []map[string]interface{}{{
  730. "id2": float64(2),
  731. }, {
  732. "id2": float64(4),
  733. }, {}},
  734. },
  735. {
  736. sql: "SELECT src1.*, f2 FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
  737. data: xsql.JoinTupleSets{
  738. xsql.JoinTuple{
  739. Tuples: []xsql.Tuple{
  740. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  741. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  742. },
  743. },
  744. xsql.JoinTuple{
  745. Tuples: []xsql.Tuple{
  746. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  747. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  748. },
  749. },
  750. xsql.JoinTuple{
  751. Tuples: []xsql.Tuple{
  752. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  753. },
  754. },
  755. },
  756. result: []map[string]interface{}{{
  757. "id1": float64(1),
  758. "f1": "v1",
  759. "f2": "w2",
  760. }, {
  761. "id1": float64(2),
  762. "f1": "v2",
  763. "f2": "w3",
  764. }, {
  765. "id1": float64(3),
  766. "f1": "v1",
  767. }},
  768. },
  769. {
  770. sql: "SELECT * FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
  771. data: xsql.JoinTupleSets{
  772. xsql.JoinTuple{
  773. Tuples: []xsql.Tuple{
  774. {Emitter: "src1", Message: xsql.Message{"id": 1, "f1": "v1"}},
  775. {Emitter: "src2", Message: xsql.Message{"id": 2, "f2": "w2"}},
  776. },
  777. },
  778. xsql.JoinTuple{
  779. Tuples: []xsql.Tuple{
  780. {Emitter: "src1", Message: xsql.Message{"id": 2, "f1": "v2"}},
  781. {Emitter: "src2", Message: xsql.Message{"id": 4, "f2": "w3"}},
  782. },
  783. },
  784. xsql.JoinTuple{
  785. Tuples: []xsql.Tuple{
  786. {Emitter: "src1", Message: xsql.Message{"id": 3, "f1": "v1"}},
  787. },
  788. },
  789. },
  790. result: []map[string]interface{}{{
  791. "id": float64(1),
  792. "f1": "v1",
  793. "f2": "w2",
  794. }, {
  795. "id": float64(2),
  796. "f1": "v2",
  797. "f2": "w3",
  798. }, {
  799. "id": float64(3),
  800. "f1": "v1",
  801. }},
  802. },
  803. {
  804. sql: "SELECT src1.* FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  805. data: xsql.GroupedTuplesSet{
  806. {
  807. &xsql.Tuple{
  808. Emitter: "src1",
  809. Message: xsql.Message{"id1": 1, "f1": "v1"},
  810. },
  811. &xsql.Tuple{
  812. Emitter: "src1",
  813. Message: xsql.Message{"id1": 3, "f1": "v1"},
  814. },
  815. },
  816. {
  817. &xsql.Tuple{
  818. Emitter: "src1",
  819. Message: xsql.Message{"id1": 2, "f1": "v2"},
  820. },
  821. },
  822. },
  823. result: []map[string]interface{}{{
  824. "id1": float64(1),
  825. "f1": "v1",
  826. }, {
  827. "id1": float64(2),
  828. "f1": "v2",
  829. }},
  830. },
  831. {
  832. sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  833. data: xsql.GroupedTuplesSet{
  834. {
  835. &xsql.JoinTuple{
  836. Tuples: []xsql.Tuple{
  837. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  838. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  839. },
  840. },
  841. },
  842. {
  843. &xsql.JoinTuple{
  844. Tuples: []xsql.Tuple{
  845. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  846. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  847. },
  848. },
  849. },
  850. {
  851. &xsql.JoinTuple{
  852. Tuples: []xsql.Tuple{
  853. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  854. },
  855. },
  856. },
  857. },
  858. result: []map[string]interface{}{{
  859. "id2": float64(2),
  860. "id1": float64(1),
  861. "f1": "v1",
  862. }, {
  863. "id2": float64(4),
  864. "id1": float64(2),
  865. "f1": "v2",
  866. }, {
  867. "id1": float64(3),
  868. "f1": "v1",
  869. }},
  870. },
  871. {
  872. sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  873. data: xsql.GroupedTuplesSet{
  874. {
  875. &xsql.JoinTuple{
  876. Tuples: []xsql.Tuple{
  877. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  878. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  879. },
  880. },
  881. },
  882. {
  883. &xsql.JoinTuple{
  884. Tuples: []xsql.Tuple{
  885. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  886. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  887. },
  888. },
  889. },
  890. {
  891. &xsql.JoinTuple{
  892. Tuples: []xsql.Tuple{
  893. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  894. },
  895. },
  896. },
  897. },
  898. result: []map[string]interface{}{{
  899. "id2": float64(2),
  900. "id1": float64(1),
  901. "f1": "v1",
  902. }, {
  903. "id2": float64(4),
  904. "id1": float64(2),
  905. "f1": "v2",
  906. }, {
  907. "id1": float64(3),
  908. "f1": "v1",
  909. }},
  910. },
  911. }
  912. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  913. contextLogger := common.Log.WithField("rule", "TestProjectPlan_MultiInput")
  914. ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
  915. for i, tt := range tests {
  916. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  917. pp := &ProjectPlan{Fields: stmt.Fields}
  918. pp.isTest = true
  919. result := pp.Apply(ctx, tt.data)
  920. var mapRes []map[string]interface{}
  921. if v, ok := result.([]byte); ok {
  922. err := json.Unmarshal(v, &mapRes)
  923. if err != nil {
  924. t.Errorf("Failed to parse the input into map.\n")
  925. continue
  926. }
  927. //fmt.Printf("%t\n", mapRes["rengine_field_0"])
  928. if !reflect.DeepEqual(tt.result, mapRes) {
  929. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  930. }
  931. } else {
  932. t.Errorf("The returned result is not type of []byte\n")
  933. }
  934. }
  935. }
  936. func TestProjectPlan_Funcs(t *testing.T) {
  937. var tests = []struct {
  938. sql string
  939. data interface{}
  940. result []map[string]interface{}
  941. }{
  942. {
  943. sql: "SELECT round(a) as r FROM test",
  944. data: &xsql.Tuple{
  945. Emitter: "test",
  946. Message: xsql.Message{
  947. "a": 47.5,
  948. },
  949. },
  950. result: []map[string]interface{}{{
  951. "r": float64(48),
  952. }},
  953. }, {
  954. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  955. data: xsql.WindowTuplesSet{
  956. xsql.WindowTuples{
  957. Emitter: "test",
  958. Tuples: []xsql.Tuple{
  959. {
  960. Emitter: "src1",
  961. Message: xsql.Message{"a": 53.1},
  962. }, {
  963. Emitter: "src1",
  964. Message: xsql.Message{"a": 27.4},
  965. }, {
  966. Emitter: "src1",
  967. Message: xsql.Message{"a": 123123.7},
  968. },
  969. },
  970. },
  971. },
  972. result: []map[string]interface{}{{
  973. "r": float64(53),
  974. }, {
  975. "r": float64(27),
  976. }, {
  977. "r": float64(123124),
  978. }},
  979. }, {
  980. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  981. data: xsql.WindowTuplesSet{
  982. xsql.WindowTuples{
  983. Emitter: "test",
  984. Tuples: []xsql.Tuple{
  985. {
  986. Emitter: "src1",
  987. Message: xsql.Message{"a": 53.1},
  988. }, {
  989. Emitter: "src1",
  990. Message: xsql.Message{"a": 27.4},
  991. }, {
  992. Emitter: "src1",
  993. Message: xsql.Message{"a": 123123.7},
  994. },
  995. },
  996. },
  997. },
  998. result: []map[string]interface{}{{
  999. "r": float64(53),
  1000. }, {
  1001. "r": float64(27),
  1002. }, {
  1003. "r": float64(123124),
  1004. }},
  1005. }, {
  1006. sql: "SELECT round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1007. data: xsql.JoinTupleSets{
  1008. xsql.JoinTuple{
  1009. Tuples: []xsql.Tuple{
  1010. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
  1011. {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1012. },
  1013. },
  1014. xsql.JoinTuple{
  1015. Tuples: []xsql.Tuple{
  1016. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
  1017. {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1018. },
  1019. },
  1020. xsql.JoinTuple{
  1021. Tuples: []xsql.Tuple{
  1022. {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
  1023. {Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1024. },
  1025. },
  1026. },
  1027. result: []map[string]interface{}{{
  1028. "r": float64(66),
  1029. }, {
  1030. "r": float64(73),
  1031. }, {
  1032. "r": float64(89),
  1033. }},
  1034. }, {
  1035. sql: "SELECT CONCAT(test.id, test.a, test1.b) as concat FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1036. data: xsql.JoinTupleSets{
  1037. xsql.JoinTuple{
  1038. Tuples: []xsql.Tuple{
  1039. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
  1040. {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1041. },
  1042. },
  1043. xsql.JoinTuple{
  1044. Tuples: []xsql.Tuple{
  1045. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
  1046. {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1047. },
  1048. },
  1049. xsql.JoinTuple{
  1050. Tuples: []xsql.Tuple{
  1051. {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
  1052. {Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1053. },
  1054. },
  1055. },
  1056. result: []map[string]interface{}{{
  1057. "concat": "165.5512",
  1058. }, {
  1059. "concat": "273.49934",
  1060. }, {
  1061. "concat": "388.886",
  1062. }},
  1063. }, {
  1064. sql: "SELECT count(a) as r FROM test",
  1065. data: &xsql.Tuple{
  1066. Emitter: "test",
  1067. Message: xsql.Message{
  1068. "a": 47.5,
  1069. },
  1070. },
  1071. result: []map[string]interface{}{{
  1072. "r": float64(1),
  1073. }},
  1074. }, {
  1075. sql: "SELECT meta(test.device) as d FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1076. data: xsql.JoinTupleSets{
  1077. xsql.JoinTuple{
  1078. Tuples: []xsql.Tuple{
  1079. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}, Metadata: xsql.Metadata{"device": "devicea"}},
  1080. {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1081. },
  1082. },
  1083. xsql.JoinTuple{
  1084. Tuples: []xsql.Tuple{
  1085. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1086. {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1087. },
  1088. },
  1089. xsql.JoinTuple{
  1090. Tuples: []xsql.Tuple{
  1091. {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}, Metadata: xsql.Metadata{"device": "devicec"}},
  1092. {Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1093. },
  1094. },
  1095. },
  1096. result: []map[string]interface{}{{
  1097. "d": "devicea",
  1098. }, {
  1099. "d": "deviceb",
  1100. }, {
  1101. "d": "devicec",
  1102. }},
  1103. },
  1104. }
  1105. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1106. contextLogger := common.Log.WithField("rule", "TestProjectPlan_Funcs")
  1107. ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
  1108. for i, tt := range tests {
  1109. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1110. if err != nil {
  1111. t.Error(err)
  1112. }
  1113. pp := &ProjectPlan{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
  1114. pp.isTest = true
  1115. result := pp.Apply(ctx, tt.data)
  1116. var mapRes []map[string]interface{}
  1117. if v, ok := result.([]byte); ok {
  1118. err := json.Unmarshal(v, &mapRes)
  1119. if err != nil {
  1120. t.Errorf("Failed to parse the input into map.\n")
  1121. continue
  1122. }
  1123. //fmt.Printf("%t\n", mapRes["rengine_field_0"])
  1124. if !reflect.DeepEqual(tt.result, mapRes) {
  1125. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  1126. }
  1127. } else {
  1128. t.Errorf("%d. The returned result is not type of []byte\n", i)
  1129. }
  1130. }
  1131. }
  1132. func TestProjectPlan_AggFuncs(t *testing.T) {
  1133. var tests = []struct {
  1134. sql string
  1135. data interface{}
  1136. result []map[string]interface{}
  1137. }{
  1138. {
  1139. sql: "SELECT count(*) as c, round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1140. data: xsql.GroupedTuplesSet{
  1141. {
  1142. &xsql.JoinTuple{
  1143. Tuples: []xsql.Tuple{
  1144. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1145. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1146. },
  1147. },
  1148. &xsql.JoinTuple{
  1149. Tuples: []xsql.Tuple{
  1150. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1151. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1152. },
  1153. },
  1154. },
  1155. {
  1156. &xsql.JoinTuple{
  1157. Tuples: []xsql.Tuple{
  1158. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1159. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1160. },
  1161. },
  1162. &xsql.JoinTuple{
  1163. Tuples: []xsql.Tuple{
  1164. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1165. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1166. },
  1167. },
  1168. },
  1169. },
  1170. result: []map[string]interface{}{{
  1171. "c": float64(2),
  1172. "r": float64(122),
  1173. }, {
  1174. "c": float64(2),
  1175. "r": float64(89),
  1176. }},
  1177. },
  1178. {
  1179. sql: "SELECT avg(a) as avg FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1180. data: xsql.GroupedTuplesSet{
  1181. {
  1182. &xsql.JoinTuple{
  1183. Tuples: []xsql.Tuple{
  1184. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1185. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1186. },
  1187. },
  1188. &xsql.JoinTuple{
  1189. Tuples: []xsql.Tuple{
  1190. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
  1191. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1192. },
  1193. },
  1194. &xsql.JoinTuple{
  1195. Tuples: []xsql.Tuple{
  1196. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 98.31}},
  1197. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
  1198. },
  1199. },
  1200. &xsql.JoinTuple{
  1201. Tuples: []xsql.Tuple{
  1202. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
  1203. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1204. },
  1205. },
  1206. },
  1207. {
  1208. &xsql.JoinTuple{
  1209. Tuples: []xsql.Tuple{
  1210. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1211. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1212. },
  1213. },
  1214. &xsql.JoinTuple{
  1215. Tuples: []xsql.Tuple{
  1216. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1217. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1218. },
  1219. },
  1220. },
  1221. },
  1222. result: []map[string]interface{}{{
  1223. "avg": 116.68,
  1224. }, {
  1225. "avg": 51.815,
  1226. }},
  1227. },
  1228. {
  1229. sql: "SELECT max(a) as max FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1230. data: xsql.GroupedTuplesSet{
  1231. {
  1232. &xsql.JoinTuple{
  1233. Tuples: []xsql.Tuple{
  1234. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1235. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1236. },
  1237. },
  1238. &xsql.JoinTuple{
  1239. Tuples: []xsql.Tuple{
  1240. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1241. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1242. },
  1243. },
  1244. &xsql.JoinTuple{
  1245. Tuples: []xsql.Tuple{
  1246. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1247. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1248. },
  1249. },
  1250. },
  1251. {
  1252. &xsql.JoinTuple{
  1253. Tuples: []xsql.Tuple{
  1254. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1255. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1256. },
  1257. },
  1258. &xsql.JoinTuple{
  1259. Tuples: []xsql.Tuple{
  1260. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1261. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1262. },
  1263. },
  1264. },
  1265. },
  1266. result: []map[string]interface{}{{
  1267. "max": 177.51,
  1268. }, {
  1269. "max": 89.03,
  1270. }},
  1271. },
  1272. {
  1273. sql: "SELECT min(a) as min FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1274. data: xsql.JoinTupleSets{
  1275. xsql.JoinTuple{
  1276. Tuples: []xsql.Tuple{
  1277. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1278. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1279. },
  1280. },
  1281. xsql.JoinTuple{
  1282. Tuples: []xsql.Tuple{
  1283. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1284. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1285. },
  1286. },
  1287. xsql.JoinTuple{
  1288. Tuples: []xsql.Tuple{
  1289. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1290. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1291. },
  1292. },
  1293. },
  1294. result: []map[string]interface{}{{
  1295. "min": 68.55,
  1296. }},
  1297. }, {
  1298. sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
  1299. data: xsql.WindowTuplesSet{
  1300. xsql.WindowTuples{
  1301. Emitter: "test",
  1302. Tuples: []xsql.Tuple{
  1303. {
  1304. Emitter: "src1",
  1305. Message: xsql.Message{"a": 53},
  1306. }, {
  1307. Emitter: "src1",
  1308. Message: xsql.Message{"a": 27},
  1309. }, {
  1310. Emitter: "src1",
  1311. Message: xsql.Message{"a": 123123},
  1312. },
  1313. },
  1314. },
  1315. },
  1316. result: []map[string]interface{}{{
  1317. "sum": float64(123203),
  1318. }},
  1319. }, {
  1320. sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
  1321. data: xsql.WindowTuplesSet{
  1322. xsql.WindowTuples{
  1323. Emitter: "test",
  1324. Tuples: []xsql.Tuple{
  1325. {
  1326. Emitter: "src1",
  1327. Message: xsql.Message{"a": 53},
  1328. }, {
  1329. Emitter: "src1",
  1330. Message: xsql.Message{"a": 27},
  1331. }, {
  1332. Emitter: "src1",
  1333. Message: xsql.Message{"a": 123123},
  1334. },
  1335. },
  1336. },
  1337. },
  1338. result: []map[string]interface{}{{
  1339. "sum": float64(123203),
  1340. }},
  1341. },
  1342. {
  1343. sql: "SELECT count(*) as c, meta(test1.device) as d FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1344. data: xsql.GroupedTuplesSet{
  1345. {
  1346. &xsql.JoinTuple{
  1347. Tuples: []xsql.Tuple{
  1348. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1349. {Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
  1350. },
  1351. },
  1352. &xsql.JoinTuple{
  1353. Tuples: []xsql.Tuple{
  1354. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1355. {Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1356. },
  1357. },
  1358. },
  1359. {
  1360. &xsql.JoinTuple{
  1361. Tuples: []xsql.Tuple{
  1362. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1363. {Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
  1364. },
  1365. },
  1366. &xsql.JoinTuple{
  1367. Tuples: []xsql.Tuple{
  1368. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1369. {Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
  1370. },
  1371. },
  1372. },
  1373. },
  1374. result: []map[string]interface{}{{
  1375. "c": float64(2),
  1376. "d": "devicea",
  1377. }, {
  1378. "c": float64(2),
  1379. "d": "devicec",
  1380. }},
  1381. },
  1382. }
  1383. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1384. contextLogger := common.Log.WithField("rule", "TestProjectPlan_AggFuncs")
  1385. ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
  1386. for i, tt := range tests {
  1387. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1388. if err != nil {
  1389. t.Error(err)
  1390. }
  1391. pp := &ProjectPlan{Fields: stmt.Fields, IsAggregate: true}
  1392. pp.isTest = true
  1393. result := pp.Apply(ctx, tt.data)
  1394. var mapRes []map[string]interface{}
  1395. if v, ok := result.([]byte); ok {
  1396. err := json.Unmarshal(v, &mapRes)
  1397. if err != nil {
  1398. t.Errorf("Failed to parse the input into map.\n")
  1399. continue
  1400. }
  1401. //fmt.Printf("%t\n", mapRes["rengine_field_0"])
  1402. if !reflect.DeepEqual(tt.result, mapRes) {
  1403. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  1404. }
  1405. } else {
  1406. t.Errorf("%d. %q\n\nThe returned result is not type of []byte: %#v\n", i, tt.sql, result)
  1407. }
  1408. }
  1409. }
  1410. func TestProjectPlanError(t *testing.T) {
  1411. var tests = []struct {
  1412. sql string
  1413. data interface{}
  1414. result interface{}
  1415. }{
  1416. {
  1417. sql: "SELECT a FROM test",
  1418. data: errors.New("an error from upstream"),
  1419. result: errors.New("an error from upstream"),
  1420. }, {
  1421. sql: "SELECT a * 5 FROM test",
  1422. data: &xsql.Tuple{
  1423. Emitter: "test",
  1424. Message: xsql.Message{
  1425. "a": "val_a",
  1426. },
  1427. },
  1428. result: errors.New("run Select error: invalid operation string(val_a) * int64(5)"),
  1429. }, {
  1430. sql: `SELECT a[0]->b AS ab FROM test`,
  1431. data: &xsql.Tuple{
  1432. Emitter: "test",
  1433. Message: xsql.Message{
  1434. "a": "common string",
  1435. },
  1436. },
  1437. result: errors.New("run Select error: invalid operation string(common string) [] *xsql.BracketEvalResult(&{0 0})"),
  1438. }, {
  1439. sql: `SELECT round(a) as r FROM test`,
  1440. data: &xsql.Tuple{
  1441. Emitter: "test",
  1442. Message: xsql.Message{
  1443. "a": "common string",
  1444. },
  1445. },
  1446. result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
  1447. }, {
  1448. sql: `SELECT round(a) as r FROM test`,
  1449. data: &xsql.Tuple{
  1450. Emitter: "test",
  1451. Message: xsql.Message{
  1452. "abc": "common string",
  1453. },
  1454. },
  1455. result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
  1456. }, {
  1457. sql: "SELECT avg(a) as avg FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1458. data: xsql.GroupedTuplesSet{
  1459. {
  1460. &xsql.JoinTuple{
  1461. Tuples: []xsql.Tuple{
  1462. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1463. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1464. },
  1465. },
  1466. &xsql.JoinTuple{
  1467. Tuples: []xsql.Tuple{
  1468. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
  1469. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1470. },
  1471. },
  1472. &xsql.JoinTuple{
  1473. Tuples: []xsql.Tuple{
  1474. {Emitter: "test", Message: xsql.Message{"id": 4, "a": "dde"}},
  1475. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
  1476. },
  1477. },
  1478. &xsql.JoinTuple{
  1479. Tuples: []xsql.Tuple{
  1480. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
  1481. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1482. },
  1483. },
  1484. },
  1485. {
  1486. &xsql.JoinTuple{
  1487. Tuples: []xsql.Tuple{
  1488. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1489. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1490. },
  1491. },
  1492. &xsql.JoinTuple{
  1493. Tuples: []xsql.Tuple{
  1494. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1495. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1496. },
  1497. },
  1498. },
  1499. },
  1500. result: errors.New("run Select error: call func avg error: requires float64 but found string(dde)"),
  1501. }, {
  1502. sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
  1503. data: xsql.WindowTuplesSet{
  1504. xsql.WindowTuples{
  1505. Emitter: "test",
  1506. Tuples: []xsql.Tuple{
  1507. {
  1508. Emitter: "src1",
  1509. Message: xsql.Message{"a": 53},
  1510. }, {
  1511. Emitter: "src1",
  1512. Message: xsql.Message{"a": "ddd"},
  1513. }, {
  1514. Emitter: "src1",
  1515. Message: xsql.Message{"a": 123123},
  1516. },
  1517. },
  1518. },
  1519. },
  1520. result: errors.New("run Select error: call func sum error: requires int but found string(ddd)"),
  1521. }, {
  1522. sql: `SELECT a[0]->b AS ab FROM test`,
  1523. data: &xsql.Tuple{
  1524. Emitter: "test",
  1525. Message: xsql.Message{
  1526. "a": []map[string]interface{}(nil),
  1527. },
  1528. },
  1529. result: errors.New("run Select error: out of index: 0 of 0"),
  1530. },
  1531. }
  1532. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1533. contextLogger := common.Log.WithField("rule", "TestProjectPlanError")
  1534. ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
  1535. for i, tt := range tests {
  1536. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1537. pp := &ProjectPlan{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
  1538. pp.isTest = true
  1539. result := pp.Apply(ctx, tt.data)
  1540. if !reflect.DeepEqual(tt.result, result) {
  1541. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1542. }
  1543. }
  1544. }