project_test.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083
  1. package plans
  2. import (
  3. "encoding/json"
  4. "engine/common"
  5. "engine/xsql"
  6. "fmt"
  7. "reflect"
  8. "strings"
  9. "testing"
  10. )
  11. func TestProjectPlan_Apply1(t *testing.T) {
  12. var tests = []struct {
  13. sql string
  14. data *xsql.Tuple
  15. result []map[string]interface{}
  16. }{
  17. {
  18. sql: "SELECT a FROM test",
  19. data: &xsql.Tuple{
  20. Emitter: "test",
  21. Message: xsql.Message{
  22. "a" : "val_a",
  23. },
  24. },
  25. result: []map[string]interface{}{{
  26. "a": "val_a",
  27. }},
  28. },
  29. {
  30. sql: "SELECT ts FROM test",
  31. data: &xsql.Tuple{
  32. Emitter: "test",
  33. Message: xsql.Message{
  34. "a" : "val_a",
  35. "ts" : common.TimeFromUnixMilli(1568854573431),
  36. },
  37. },
  38. result: []map[string]interface{}{{
  39. "ts": "2019-09-19T00:56:13.000000431Z",
  40. }},
  41. },
  42. {
  43. sql: "SELECT A FROM test",
  44. data: &xsql.Tuple{
  45. Emitter: "test",
  46. Message: xsql.Message{
  47. "a" : "val_a",
  48. },
  49. },
  50. result: []map[string]interface{}{{
  51. "A": "val_a",
  52. }},
  53. },
  54. {
  55. sql: `SELECT "value" FROM test`,
  56. data: &xsql.Tuple{
  57. Emitter: "test",
  58. Message: xsql.Message{
  59. },
  60. },
  61. result: []map[string]interface{}{{
  62. DEFAULT_FIELD_NAME_PREFIX + "0" : "value",
  63. }},
  64. },
  65. {
  66. sql: `SELECT 3.4 FROM test`,
  67. data: &xsql.Tuple{
  68. Emitter: "test",
  69. Message: xsql.Message{
  70. },
  71. },
  72. result: []map[string]interface{}{{
  73. DEFAULT_FIELD_NAME_PREFIX + "0" : 3.4,
  74. }},
  75. },
  76. {
  77. sql: `SELECT 5 FROM test`,
  78. data: &xsql.Tuple{
  79. Emitter: "test",
  80. Message: xsql.Message{
  81. },
  82. },
  83. result: []map[string]interface{}{{
  84. DEFAULT_FIELD_NAME_PREFIX + "0": 5.0,
  85. }},
  86. },
  87. {
  88. sql: `SELECT a, "value" AS b FROM test`,
  89. data: &xsql.Tuple{
  90. Emitter: "test",
  91. Message: xsql.Message{
  92. "a" : "val_a",
  93. },
  94. },
  95. result: []map[string]interface{}{{
  96. "a" : "val_a",
  97. "b" : "value",
  98. }},
  99. },
  100. {
  101. sql: `SELECT a, "value" AS b, 3.14 as Pi, 0 as Zero FROM test`,
  102. data: &xsql.Tuple{
  103. Emitter: "test",
  104. Message: xsql.Message{
  105. "a" : "val_a",
  106. },
  107. },
  108. result: []map[string]interface{}{{
  109. "a" : "val_a",
  110. "b" : "value",
  111. "Pi" : 3.14,
  112. "Zero" : 0.0,
  113. }},
  114. },
  115. {
  116. sql: `SELECT a->b AS ab FROM test`,
  117. data: &xsql.Tuple{
  118. Emitter: "test",
  119. Message: xsql.Message{
  120. "a" : map[string]interface{}{"b" : "hello"},
  121. },
  122. },
  123. result: []map[string]interface{}{{
  124. "ab" : "hello",
  125. }},
  126. },
  127. {
  128. sql: `SELECT a[0]->b AS ab FROM test`,
  129. data: &xsql.Tuple{
  130. Emitter: "test",
  131. Message: xsql.Message{
  132. "a" : []interface{}{
  133. map[string]interface{}{"b" : "hello1"},
  134. map[string]interface{}{"b" : "hello2"},
  135. },
  136. },
  137. },
  138. result: []map[string]interface{}{{
  139. "ab" : "hello1",
  140. }},
  141. },
  142. {
  143. sql: `SELECT a->c->d AS f1 FROM test`,
  144. data: &xsql.Tuple{
  145. Emitter: "test",
  146. Message: xsql.Message{
  147. "a" : map[string]interface{}{
  148. "b" : "hello",
  149. "c" : map[string]interface{}{
  150. "d": 35.2,
  151. },
  152. },
  153. },
  154. },
  155. result: []map[string]interface{}{{
  156. "f1" : 35.2,
  157. }},
  158. },
  159. //The int type is not supported yet, the json parser returns float64 for int values
  160. {
  161. sql: `SELECT a->c->d AS f1 FROM test`,
  162. data: &xsql.Tuple{
  163. Emitter: "test",
  164. Message: xsql.Message{
  165. "a" : map[string]interface{}{
  166. "b" : "hello",
  167. "c" : map[string]interface{}{
  168. "d": float64(35),
  169. },
  170. },
  171. },
  172. },
  173. result: []map[string]interface{}{{
  174. "f1" : float64(35),
  175. }},
  176. },
  177. {
  178. sql: "SELECT a FROM test",
  179. data: &xsql.Tuple{
  180. Emitter: "test",
  181. Message: xsql.Message{
  182. },
  183. },
  184. result: []map[string]interface{}{
  185. {},
  186. },
  187. },
  188. {
  189. sql: "SELECT * FROM test",
  190. data: &xsql.Tuple{
  191. Emitter: "test",
  192. Message: xsql.Message{
  193. },
  194. },
  195. result: []map[string]interface{}{
  196. {},
  197. },
  198. },
  199. {
  200. sql: `SELECT * FROM test`,
  201. data: &xsql.Tuple{
  202. Emitter: "test",
  203. Message: xsql.Message{
  204. "a" : map[string]interface{}{
  205. "b" : "hello",
  206. "c" : map[string]interface{}{
  207. "d": 35.2,
  208. },
  209. },
  210. },
  211. },
  212. result: []map[string]interface{}{{
  213. "a" : map[string]interface{} {
  214. "b" : "hello",
  215. "c" : map[string]interface{} {
  216. "d" : 35.2,
  217. },
  218. },
  219. }},
  220. },
  221. {
  222. sql: `SELECT * FROM test`,
  223. data: &xsql.Tuple{
  224. Emitter: "test",
  225. Message: xsql.Message{
  226. "a" : "val1",
  227. "b" : 3.14,
  228. },
  229. },
  230. result: []map[string]interface{}{{
  231. "a" : "val1",
  232. "b" : 3.14,
  233. }},
  234. },
  235. {
  236. sql: `SELECT 3*4 AS f1 FROM test`,
  237. data: &xsql.Tuple{
  238. Emitter: "test",
  239. Message: xsql.Message{
  240. },
  241. },
  242. result: []map[string]interface{}{{
  243. "f1" : float64(12),
  244. }},
  245. },
  246. {
  247. sql: `SELECT 4.5*2 AS f1 FROM test`,
  248. data: &xsql.Tuple{
  249. Emitter: "test",
  250. Message: xsql.Message{
  251. },
  252. },
  253. result: []map[string]interface{}{{
  254. "f1" : float64(9),
  255. }},
  256. },
  257. }
  258. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  259. for i, tt := range tests {
  260. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  261. pp := &ProjectPlan{Fields:stmt.Fields}
  262. result := pp.Apply(nil, tt.data)
  263. var mapRes []map[string]interface{}
  264. if v, ok := result.([]byte); ok {
  265. err := json.Unmarshal(v, &mapRes)
  266. if err != nil {
  267. t.Errorf("Failed to parse the input into map.\n")
  268. continue
  269. }
  270. //fmt.Printf("%t\n", mapRes["rengine_field_0"])
  271. if !reflect.DeepEqual(tt.result, mapRes) {
  272. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  273. }
  274. } else {
  275. t.Errorf("The returned result is not type of []byte\n")
  276. }
  277. }
  278. }
  279. func TestProjectPlan_MultiInput(t *testing.T) {
  280. var tests = []struct {
  281. sql string
  282. data interface{}
  283. result []map[string]interface{}
  284. }{
  285. {
  286. sql: "SELECT * FROM tbl WHERE abc*2+3 > 12 AND abc < 20",
  287. data: &xsql.Tuple{
  288. Emitter: "tbl",
  289. Message: xsql.Message{
  290. "abc" : int64(6),
  291. },
  292. },
  293. result: []map[string]interface{}{{
  294. "abc" : float64(6), //json marshall problem
  295. }},
  296. },
  297. {
  298. sql: "SELECT abc FROM tbl WHERE abc*2+3 > 12 OR def = \"hello\"",
  299. data: &xsql.Tuple{
  300. Emitter: "tbl",
  301. Message: xsql.Message{
  302. "abc" : int64(34),
  303. "def" : "hello",
  304. },
  305. },
  306. result: []map[string]interface{}{{
  307. "abc" : float64(34),
  308. }},
  309. },
  310. {
  311. sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  312. data: xsql.WindowTuplesSet{
  313. xsql.WindowTuples{
  314. Emitter:"src1",
  315. Tuples:[]xsql.Tuple{
  316. {
  317. Emitter: "src1",
  318. Message: xsql.Message{ "id1" : 1, "f1" : "v1", },
  319. },{
  320. Emitter: "src1",
  321. Message: xsql.Message{ "id1" : 2, "f1" : "v2", },
  322. },{
  323. Emitter: "src1",
  324. Message: xsql.Message{ "id1" : 3, "f1" : "v1", },
  325. },
  326. },
  327. },
  328. },
  329. result: []map[string]interface{}{{
  330. "id1" : float64(1),
  331. },{
  332. "id1" : float64(2),
  333. },{
  334. "id1" : float64(3),
  335. }},
  336. },
  337. {
  338. sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  339. data: xsql.WindowTuplesSet{
  340. xsql.WindowTuples{
  341. Emitter:"src1",
  342. Tuples:[]xsql.Tuple{
  343. {
  344. Emitter: "src1",
  345. Message: xsql.Message{ "id1" : 1, "f1" : "v1", },
  346. },{
  347. Emitter: "src1",
  348. Message: xsql.Message{ "id1" : 2, "f1" : "v2", },
  349. },{
  350. Emitter: "src1",
  351. Message: xsql.Message{ "id1" : 3, "f1" : "v1", },
  352. },
  353. },
  354. },
  355. },
  356. result: []map[string]interface{}{{
  357. "id1" : float64(1),
  358. "f1" : "v1",
  359. },{
  360. "id1" : float64(2),
  361. "f1" : "v2",
  362. },{
  363. "id1" : float64(3),
  364. "f1" : "v1",
  365. }},
  366. },
  367. {
  368. sql: "SELECT src1.* FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  369. data: xsql.WindowTuplesSet{
  370. xsql.WindowTuples{
  371. Emitter:"src1",
  372. Tuples:[]xsql.Tuple{
  373. {
  374. Emitter: "src1",
  375. Message: xsql.Message{ "id1" : 1, "f1" : "v1", },
  376. },{
  377. Emitter: "src1",
  378. Message: xsql.Message{ "id1" : 2, "f1" : "v2", },
  379. },{
  380. Emitter: "src1",
  381. Message: xsql.Message{ "id1" : 3, "f1" : "v1", },
  382. },
  383. },
  384. },
  385. },
  386. result: []map[string]interface{}{{
  387. "id1" : float64(1),
  388. "f1" : "v1",
  389. },{
  390. "id1" : float64(2),
  391. "f1" : "v2",
  392. },{
  393. "id1" : float64(3),
  394. "f1" : "v1",
  395. }},
  396. },
  397. {
  398. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  399. data: xsql.JoinTupleSets{
  400. xsql.JoinTuple{
  401. Tuples: []xsql.Tuple{
  402. {Emitter: "src1", Message: xsql.Message{ "id1" : 1, "f1" : "v1",},},
  403. {Emitter: "src2", Message: xsql.Message{ "id2" : 2, "f2" : "w2",},},
  404. },
  405. },
  406. xsql.JoinTuple{
  407. Tuples: []xsql.Tuple{
  408. {Emitter: "src1", Message: xsql.Message{ "id1" : 2, "f1" : "v2",},},
  409. {Emitter: "src2", Message: xsql.Message{ "id2" : 4, "f2" : "w3",},},
  410. },
  411. },
  412. xsql.JoinTuple{
  413. Tuples: []xsql.Tuple{
  414. {Emitter: "src1", Message: xsql.Message{ "id1" : 3, "f1" : "v1",},},
  415. },
  416. },
  417. },
  418. result: []map[string]interface{}{{
  419. "id1" : float64(1),
  420. },{
  421. "id1" : float64(2),
  422. },{
  423. "id1" : float64(3),
  424. }},
  425. },
  426. {
  427. sql: "SELECT abc FROM tbl group by abc",
  428. data: xsql.GroupedTuplesSet{
  429. {
  430. &xsql.Tuple{
  431. Emitter: "tbl",
  432. Message: xsql.Message{
  433. "abc" : int64(6),
  434. "def" : "hello",
  435. },
  436. },
  437. },
  438. },
  439. result: []map[string]interface{}{{
  440. "abc" : float64(6),
  441. },},
  442. },
  443. {
  444. sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  445. data: xsql.GroupedTuplesSet{
  446. {
  447. &xsql.Tuple{
  448. Emitter: "src1",
  449. Message: xsql.Message{ "id1" : 1, "f1" : "v1", },
  450. },
  451. &xsql.Tuple{
  452. Emitter: "src1",
  453. Message: xsql.Message{ "id1" : 3, "f1" : "v1", },
  454. },
  455. },
  456. {
  457. &xsql.Tuple{
  458. Emitter: "src1",
  459. Message: xsql.Message{ "id1" : 2, "f1" : "v2", },
  460. },
  461. },
  462. },
  463. result: []map[string]interface{}{{
  464. "id1": float64(1),
  465. },{
  466. "id1": float64(2),
  467. },},
  468. },
  469. {
  470. sql: "SELECT src2.id2 FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  471. data: xsql.GroupedTuplesSet{
  472. {
  473. &xsql.JoinTuple{
  474. Tuples: []xsql.Tuple{
  475. {Emitter: "src1", Message: xsql.Message{ "id1" : 1, "f1" : "v1",},},
  476. {Emitter: "src2", Message: xsql.Message{ "id2" : 2, "f2" : "w2",},},
  477. },
  478. },
  479. },
  480. {
  481. &xsql.JoinTuple{
  482. Tuples: []xsql.Tuple{
  483. {Emitter: "src1", Message: xsql.Message{ "id1" : 2, "f1" : "v2",},},
  484. {Emitter: "src2", Message: xsql.Message{ "id2" : 4, "f2" : "w3",},},
  485. },
  486. },
  487. },
  488. {
  489. &xsql.JoinTuple{
  490. Tuples: []xsql.Tuple{
  491. {Emitter: "src1", Message: xsql.Message{ "id1" : 3, "f1" : "v1",},},
  492. },
  493. },
  494. },
  495. },
  496. result: []map[string]interface{}{{
  497. "id2": float64(2),
  498. }, {
  499. "id2": float64(4),
  500. },{
  501. },},
  502. },
  503. {
  504. sql: "SELECT src1.*, f2 FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
  505. data: xsql.JoinTupleSets{
  506. xsql.JoinTuple{
  507. Tuples: []xsql.Tuple{
  508. {Emitter: "src1", Message: xsql.Message{ "id1" : 1, "f1" : "v1",},},
  509. {Emitter: "src2", Message: xsql.Message{ "id2" : 2, "f2" : "w2",},},
  510. },
  511. },
  512. xsql.JoinTuple{
  513. Tuples: []xsql.Tuple{
  514. {Emitter: "src1", Message: xsql.Message{ "id1" : 2, "f1" : "v2",},},
  515. {Emitter: "src2", Message: xsql.Message{ "id2" : 4, "f2" : "w3",},},
  516. },
  517. },
  518. xsql.JoinTuple{
  519. Tuples: []xsql.Tuple{
  520. {Emitter: "src1", Message: xsql.Message{ "id1" : 3, "f1" : "v1",},},
  521. },
  522. },
  523. },
  524. result: []map[string]interface{}{{
  525. "id1" : float64(1),
  526. "f1" : "v1",
  527. "f2" : "w2",
  528. },{
  529. "id1" : float64(2),
  530. "f1" : "v2",
  531. "f2" : "w3",
  532. },{
  533. "id1" : float64(3),
  534. "f1" : "v1",
  535. }},
  536. },
  537. {
  538. sql: "SELECT * FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
  539. data: xsql.JoinTupleSets{
  540. xsql.JoinTuple{
  541. Tuples: []xsql.Tuple{
  542. {Emitter: "src1", Message: xsql.Message{ "id" : 1, "f1" : "v1",},},
  543. {Emitter: "src2", Message: xsql.Message{ "id" : 2, "f2" : "w2",},},
  544. },
  545. },
  546. xsql.JoinTuple{
  547. Tuples: []xsql.Tuple{
  548. {Emitter: "src1", Message: xsql.Message{ "id" : 2, "f1" : "v2",},},
  549. {Emitter: "src2", Message: xsql.Message{ "id" : 4, "f2" : "w3",},},
  550. },
  551. },
  552. xsql.JoinTuple{
  553. Tuples: []xsql.Tuple{
  554. {Emitter: "src1", Message: xsql.Message{ "id" : 3, "f1" : "v1",},},
  555. },
  556. },
  557. },
  558. result: []map[string]interface{}{{
  559. "id" : float64(1),
  560. "f1" : "v1",
  561. "f2" : "w2",
  562. },{
  563. "id" : float64(2),
  564. "f1" : "v2",
  565. "f2" : "w3",
  566. },{
  567. "id" : float64(3),
  568. "f1" : "v1",
  569. }},
  570. },
  571. {
  572. sql: "SELECT src1.* FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  573. data: xsql.GroupedTuplesSet{
  574. {
  575. &xsql.Tuple{
  576. Emitter: "src1",
  577. Message: xsql.Message{ "id1" : 1, "f1" : "v1", },
  578. },
  579. &xsql.Tuple{
  580. Emitter: "src1",
  581. Message: xsql.Message{ "id1" : 3, "f1" : "v1", },
  582. },
  583. },
  584. {
  585. &xsql.Tuple{
  586. Emitter: "src1",
  587. Message: xsql.Message{ "id1" : 2, "f1" : "v2", },
  588. },
  589. },
  590. },
  591. result: []map[string]interface{}{{
  592. "id1": float64(1),
  593. "f1": "v1",
  594. },{
  595. "id1": float64(2),
  596. "f1": "v2",
  597. },},
  598. },
  599. {
  600. sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  601. data: xsql.GroupedTuplesSet{
  602. {
  603. &xsql.JoinTuple{
  604. Tuples: []xsql.Tuple{
  605. {Emitter: "src1", Message: xsql.Message{ "id1" : 1, "f1" : "v1",},},
  606. {Emitter: "src2", Message: xsql.Message{ "id2" : 2, "f2" : "w2",},},
  607. },
  608. },
  609. },
  610. {
  611. &xsql.JoinTuple{
  612. Tuples: []xsql.Tuple{
  613. {Emitter: "src1", Message: xsql.Message{ "id1" : 2, "f1" : "v2",},},
  614. {Emitter: "src2", Message: xsql.Message{ "id2" : 4, "f2" : "w3",},},
  615. },
  616. },
  617. },
  618. {
  619. &xsql.JoinTuple{
  620. Tuples: []xsql.Tuple{
  621. {Emitter: "src1", Message: xsql.Message{ "id1" : 3, "f1" : "v1",},},
  622. },
  623. },
  624. },
  625. },
  626. result: []map[string]interface{}{{
  627. "id2": float64(2),
  628. "id1": float64(1),
  629. "f1" : "v1",
  630. }, {
  631. "id2": float64(4),
  632. "id1": float64(2),
  633. "f1" : "v2",
  634. },{
  635. "id1": float64(3),
  636. "f1" : "v1",
  637. },},
  638. },
  639. {
  640. sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  641. data: xsql.GroupedTuplesSet{
  642. {
  643. &xsql.JoinTuple{
  644. Tuples: []xsql.Tuple{
  645. {Emitter: "src1", Message: xsql.Message{ "id1" : 1, "f1" : "v1",},},
  646. {Emitter: "src2", Message: xsql.Message{ "id2" : 2, "f2" : "w2",},},
  647. },
  648. },
  649. },
  650. {
  651. &xsql.JoinTuple{
  652. Tuples: []xsql.Tuple{
  653. {Emitter: "src1", Message: xsql.Message{ "id1" : 2, "f1" : "v2",},},
  654. {Emitter: "src2", Message: xsql.Message{ "id2" : 4, "f2" : "w3",},},
  655. },
  656. },
  657. },
  658. {
  659. &xsql.JoinTuple{
  660. Tuples: []xsql.Tuple{
  661. {Emitter: "src1", Message: xsql.Message{ "id1" : 3, "f1" : "v1",},},
  662. },
  663. },
  664. },
  665. },
  666. result: []map[string]interface{}{{
  667. "id2": float64(2),
  668. "id1": float64(1),
  669. "f1" : "v1",
  670. }, {
  671. "id2": float64(4),
  672. "id1": float64(2),
  673. "f1" : "v2",
  674. },{
  675. "id1": float64(3),
  676. "f1" : "v1",
  677. },},
  678. },
  679. }
  680. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  681. for i, tt := range tests {
  682. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  683. pp := &ProjectPlan{Fields:stmt.Fields}
  684. result := pp.Apply(nil, tt.data)
  685. var mapRes []map[string]interface{}
  686. if v, ok := result.([]byte); ok {
  687. err := json.Unmarshal(v, &mapRes)
  688. if err != nil {
  689. t.Errorf("Failed to parse the input into map.\n")
  690. continue
  691. }
  692. //fmt.Printf("%t\n", mapRes["rengine_field_0"])
  693. if !reflect.DeepEqual(tt.result, mapRes) {
  694. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  695. }
  696. } else {
  697. t.Errorf("The returned result is not type of []byte\n")
  698. }
  699. }
  700. }
  701. func TestProjectPlan_Funcs(t *testing.T) {
  702. var tests = []struct {
  703. sql string
  704. data interface{}
  705. result []map[string]interface{}
  706. }{
  707. {
  708. sql: "SELECT round(a) as r FROM test",
  709. data: &xsql.Tuple{
  710. Emitter: "test",
  711. Message: xsql.Message{
  712. "a": 47.5,
  713. },
  714. },
  715. result: []map[string]interface{}{{
  716. "r": float64(48),
  717. }},
  718. }, {
  719. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  720. data: xsql.WindowTuplesSet{
  721. xsql.WindowTuples{
  722. Emitter:"test",
  723. Tuples:[]xsql.Tuple{
  724. {
  725. Emitter: "src1",
  726. Message: xsql.Message{ "a": 53.1 },
  727. },{
  728. Emitter: "src1",
  729. Message: xsql.Message{ "a": 27.4 },
  730. },{
  731. Emitter: "src1",
  732. Message: xsql.Message{ "a": 123123.7 },
  733. },
  734. },
  735. },
  736. },
  737. result: []map[string]interface{}{{
  738. "r": float64(53),
  739. },{
  740. "r": float64(27),
  741. },{
  742. "r": float64(123124),
  743. }},
  744. }, {
  745. sql: "SELECT round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  746. data: xsql.JoinTupleSets{
  747. xsql.JoinTuple{
  748. Tuples: []xsql.Tuple{
  749. {Emitter: "test", Message: xsql.Message{ "id": 1, "a": 65.55},},
  750. {Emitter: "test1", Message: xsql.Message{ "id": 1, "b": 12},},
  751. },
  752. },
  753. xsql.JoinTuple{
  754. Tuples: []xsql.Tuple{
  755. {Emitter: "test", Message: xsql.Message{ "id": 2, "a": 73.499},},
  756. {Emitter: "test1", Message: xsql.Message{ "id": 2, "b": 34},},
  757. },
  758. },
  759. xsql.JoinTuple{
  760. Tuples: []xsql.Tuple{
  761. {Emitter: "test", Message: xsql.Message{ "id": 3, "a": 88.88},},
  762. {Emitter: "test1", Message: xsql.Message{ "id": 3, "b": 6},},
  763. },
  764. },
  765. },
  766. result: []map[string]interface{}{{
  767. "r": float64(66),
  768. },{
  769. "r": float64(73),
  770. },{
  771. "r": float64(89),
  772. }},
  773. }, {
  774. 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)",
  775. data: xsql.JoinTupleSets{
  776. xsql.JoinTuple{
  777. Tuples: []xsql.Tuple{
  778. {Emitter: "test", Message: xsql.Message{ "id": 1, "a": 65.55},},
  779. {Emitter: "test1", Message: xsql.Message{ "id": 1, "b": 12},},
  780. },
  781. },
  782. xsql.JoinTuple{
  783. Tuples: []xsql.Tuple{
  784. {Emitter: "test", Message: xsql.Message{ "id": 2, "a": 73.499},},
  785. {Emitter: "test1", Message: xsql.Message{ "id": 2, "b": 34},},
  786. },
  787. },
  788. xsql.JoinTuple{
  789. Tuples: []xsql.Tuple{
  790. {Emitter: "test", Message: xsql.Message{ "id": 3, "a": 88.88},},
  791. {Emitter: "test1", Message: xsql.Message{ "id": 3, "b": 6},},
  792. },
  793. },
  794. },
  795. result: []map[string]interface{}{{
  796. "concat": "165.5512",
  797. },{
  798. "concat": "273.49934",
  799. },{
  800. "concat": "388.886",
  801. }},
  802. },
  803. }
  804. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  805. for i, tt := range tests {
  806. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  807. if err != nil{
  808. t.Error(err)
  809. }
  810. pp := &ProjectPlan{Fields:stmt.Fields}
  811. result := pp.Apply(nil, tt.data)
  812. var mapRes []map[string]interface{}
  813. if v, ok := result.([]byte); ok {
  814. err := json.Unmarshal(v, &mapRes)
  815. if err != nil {
  816. t.Errorf("Failed to parse the input into map.\n")
  817. continue
  818. }
  819. //fmt.Printf("%t\n", mapRes["rengine_field_0"])
  820. if !reflect.DeepEqual(tt.result, mapRes) {
  821. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  822. }
  823. } else {
  824. t.Errorf("The returned result is not type of []byte\n")
  825. }
  826. }
  827. }
  828. func TestProjectPlan_AggFuncs(t *testing.T) {
  829. var tests = []struct {
  830. sql string
  831. data interface{}
  832. result []map[string]interface{}
  833. }{
  834. {
  835. 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",
  836. data: xsql.GroupedTuplesSet{
  837. {
  838. &xsql.JoinTuple{
  839. Tuples: []xsql.Tuple{
  840. {Emitter: "test", Message: xsql.Message{ "id" : 1, "a": 122.33,},},
  841. {Emitter: "src2", Message: xsql.Message{ "id" : 1, "color" : "w2",},},
  842. },
  843. },
  844. &xsql.JoinTuple{
  845. Tuples: []xsql.Tuple{
  846. {Emitter: "test", Message: xsql.Message{ "id" : 5, "a": 177.51,},},
  847. {Emitter: "src2", Message: xsql.Message{ "id" : 5, "color" : "w2",},},
  848. },
  849. },
  850. },
  851. {
  852. &xsql.JoinTuple{
  853. Tuples: []xsql.Tuple{
  854. {Emitter: "test", Message: xsql.Message{ "id" : 2, "a": 89.03,},},
  855. {Emitter: "src2", Message: xsql.Message{ "id" : 2, "color" : "w1",},},
  856. },
  857. },
  858. &xsql.JoinTuple{
  859. Tuples: []xsql.Tuple{
  860. {Emitter: "test", Message: xsql.Message{ "id" : 4, "a": 14.6,},},
  861. {Emitter: "src2", Message: xsql.Message{ "id" : 4, "color" : "w1",},},
  862. },
  863. },
  864. },
  865. },
  866. result: []map[string]interface{}{{
  867. "c": float64(2),
  868. "r": float64(122),
  869. },{
  870. "c": float64(2),
  871. "r": float64(89),
  872. }},
  873. },
  874. {
  875. sql: "SELECT avg(a) as avg FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  876. data: xsql.GroupedTuplesSet{
  877. {
  878. &xsql.JoinTuple{
  879. Tuples: []xsql.Tuple{
  880. {Emitter: "test", Message: xsql.Message{ "id" : 1, "a": 122.33,},},
  881. {Emitter: "src2", Message: xsql.Message{ "id" : 1, "color" : "w2",},},
  882. },
  883. },
  884. &xsql.JoinTuple{
  885. Tuples: []xsql.Tuple{
  886. {Emitter: "test", Message: xsql.Message{ "id" : 1, "a": 68.54,},},
  887. {Emitter: "src2", Message: xsql.Message{ "id" : 1, "color" : "w2",},},
  888. },
  889. },
  890. &xsql.JoinTuple{
  891. Tuples: []xsql.Tuple{
  892. {Emitter: "test", Message: xsql.Message{ "id" : 4, "a": 98.31,},},
  893. {Emitter: "src2", Message: xsql.Message{ "id" : 4, "color" : "w2",},},
  894. },
  895. },
  896. &xsql.JoinTuple{
  897. Tuples: []xsql.Tuple{
  898. {Emitter: "test", Message: xsql.Message{ "id" : 5, "a": 177.54,},},
  899. {Emitter: "src2", Message: xsql.Message{ "id" : 5, "color" : "w2",},},
  900. },
  901. },
  902. },
  903. {
  904. &xsql.JoinTuple{
  905. Tuples: []xsql.Tuple{
  906. {Emitter: "test", Message: xsql.Message{ "id" : 2, "a": 89.03,},},
  907. {Emitter: "src2", Message: xsql.Message{ "id" : 2, "color" : "w1",},},
  908. },
  909. },
  910. &xsql.JoinTuple{
  911. Tuples: []xsql.Tuple{
  912. {Emitter: "test", Message: xsql.Message{ "id" : 4, "a": 14.6,},},
  913. {Emitter: "src2", Message: xsql.Message{ "id" : 4, "color" : "w1",},},
  914. },
  915. },
  916. },
  917. },
  918. result: []map[string]interface{}{{
  919. "avg": 116.68,
  920. },{
  921. "avg": 51.815,
  922. }},
  923. },
  924. {
  925. sql: "SELECT max(a) as max FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  926. data: xsql.GroupedTuplesSet{
  927. {
  928. &xsql.JoinTuple{
  929. Tuples: []xsql.Tuple{
  930. {Emitter: "test", Message: xsql.Message{ "id" : 1, "a": 122.33,},},
  931. {Emitter: "src2", Message: xsql.Message{ "id" : 1, "color" : "w2",},},
  932. },
  933. },
  934. &xsql.JoinTuple{
  935. Tuples: []xsql.Tuple{
  936. {Emitter: "test", Message: xsql.Message{ "id" : 1, "a": 68.55,},},
  937. {Emitter: "src2", Message: xsql.Message{ "id" : 1, "color" : "w2",},},
  938. },
  939. },
  940. &xsql.JoinTuple{
  941. Tuples: []xsql.Tuple{
  942. {Emitter: "test", Message: xsql.Message{ "id" : 5, "a": 177.51,},},
  943. {Emitter: "src2", Message: xsql.Message{ "id" : 5, "color" : "w2",},},
  944. },
  945. },
  946. },
  947. {
  948. &xsql.JoinTuple{
  949. Tuples: []xsql.Tuple{
  950. {Emitter: "test", Message: xsql.Message{ "id" : 2, "a": 89.03,},},
  951. {Emitter: "src2", Message: xsql.Message{ "id" : 2, "color" : "w1",},},
  952. },
  953. },
  954. &xsql.JoinTuple{
  955. Tuples: []xsql.Tuple{
  956. {Emitter: "test", Message: xsql.Message{ "id" : 4, "a": 14.6,},},
  957. {Emitter: "src2", Message: xsql.Message{ "id" : 4, "color" : "w1",},},
  958. },
  959. },
  960. },
  961. },
  962. result: []map[string]interface{}{{
  963. "max": 177.51,
  964. },{
  965. "max": 89.03,
  966. }},
  967. },
  968. {
  969. sql: "SELECT min(a) as min FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  970. data: xsql.JoinTupleSets{
  971. xsql.JoinTuple{
  972. Tuples: []xsql.Tuple{
  973. {Emitter: "test", Message: xsql.Message{ "id" : 1, "a": 122.33,},},
  974. {Emitter: "src2", Message: xsql.Message{ "id" : 1, "color" : "w2",},},
  975. },
  976. },
  977. xsql.JoinTuple{
  978. Tuples: []xsql.Tuple{
  979. {Emitter: "test", Message: xsql.Message{ "id" : 1, "a": 68.55,},},
  980. {Emitter: "src2", Message: xsql.Message{ "id" : 1, "color" : "w2",},},
  981. },
  982. },
  983. xsql.JoinTuple{
  984. Tuples: []xsql.Tuple{
  985. {Emitter: "test", Message: xsql.Message{ "id" : 5, "a": 177.51,},},
  986. {Emitter: "src2", Message: xsql.Message{ "id" : 5, "color" : "w2",},},
  987. },
  988. },
  989. },
  990. result: []map[string]interface{}{{
  991. "min": 68.55,
  992. }},
  993. },{
  994. sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
  995. data: xsql.WindowTuplesSet{
  996. xsql.WindowTuples{
  997. Emitter:"test",
  998. Tuples:[]xsql.Tuple{
  999. {
  1000. Emitter: "src1",
  1001. Message: xsql.Message{ "a": 53 },
  1002. },{
  1003. Emitter: "src1",
  1004. Message: xsql.Message{ "a": 27 },
  1005. },{
  1006. Emitter: "src1",
  1007. Message: xsql.Message{ "a": 123123 },
  1008. },
  1009. },
  1010. },
  1011. },
  1012. result: []map[string]interface{}{{
  1013. "sum": float64(123203),
  1014. }},
  1015. },
  1016. }
  1017. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1018. for i, tt := range tests {
  1019. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1020. if err != nil{
  1021. t.Error(err)
  1022. }
  1023. pp := &ProjectPlan{Fields:stmt.Fields, IsAggregate: true}
  1024. result := pp.Apply(nil, tt.data)
  1025. var mapRes []map[string]interface{}
  1026. if v, ok := result.([]byte); ok {
  1027. err := json.Unmarshal(v, &mapRes)
  1028. if err != nil {
  1029. t.Errorf("Failed to parse the input into map.\n")
  1030. continue
  1031. }
  1032. //fmt.Printf("%t\n", mapRes["rengine_field_0"])
  1033. if !reflect.DeepEqual(tt.result, mapRes) {
  1034. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  1035. }
  1036. } else {
  1037. t.Errorf("The returned result is not type of []byte\n")
  1038. }
  1039. }
  1040. }