project_test.go 57 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374
  1. // Copyright 2022 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 operator
  15. import (
  16. "errors"
  17. "fmt"
  18. "github.com/lf-edge/ekuiper/internal/conf"
  19. "github.com/lf-edge/ekuiper/internal/topo/context"
  20. "github.com/lf-edge/ekuiper/internal/xsql"
  21. "github.com/lf-edge/ekuiper/pkg/cast"
  22. "reflect"
  23. "strings"
  24. "testing"
  25. )
  26. func TestProjectPlan_Apply1(t *testing.T) {
  27. var tests = []struct {
  28. sql string
  29. data *xsql.Tuple
  30. result []map[string]interface{}
  31. }{
  32. { //0
  33. sql: "SELECT a FROM test",
  34. data: &xsql.Tuple{
  35. Emitter: "test",
  36. Message: xsql.Message{
  37. "a": "val_a",
  38. },
  39. Metadata: xsql.Metadata{
  40. "id": 45,
  41. "other": "mock",
  42. },
  43. },
  44. result: []map[string]interface{}{{
  45. "a": "val_a",
  46. "__meta": xsql.Metadata{
  47. "id": 45,
  48. "other": "mock",
  49. },
  50. }},
  51. },
  52. { //1
  53. sql: "SELECT b FROM test",
  54. data: &xsql.Tuple{
  55. Emitter: "test",
  56. Message: xsql.Message{
  57. "a": "val_a",
  58. },
  59. },
  60. result: []map[string]interface{}{{}},
  61. },
  62. { //2
  63. sql: "SELECT ts FROM test",
  64. data: &xsql.Tuple{
  65. Emitter: "test",
  66. Message: xsql.Message{
  67. "a": "val_a",
  68. "ts": cast.TimeFromUnixMilli(1568854573431),
  69. },
  70. },
  71. result: []map[string]interface{}{{
  72. "ts": cast.TimeFromUnixMilli(1568854573431),
  73. }},
  74. },
  75. //Schemaless may return a message without selecting column
  76. { //3
  77. sql: "SELECT ts FROM test",
  78. data: &xsql.Tuple{
  79. Emitter: "test",
  80. Message: xsql.Message{
  81. "a": "val_a",
  82. "ts2": cast.TimeFromUnixMilli(1568854573431),
  83. },
  84. },
  85. result: []map[string]interface{}{{}},
  86. },
  87. { //4
  88. sql: "SELECT A 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. }},
  98. },
  99. //5
  100. {
  101. sql: `SELECT "value" FROM test`,
  102. data: &xsql.Tuple{
  103. Emitter: "test",
  104. Message: xsql.Message{},
  105. },
  106. result: []map[string]interface{}{{
  107. "kuiper_field_0": "value",
  108. }},
  109. },
  110. //6
  111. {
  112. sql: `SELECT 3.4 FROM test`,
  113. data: &xsql.Tuple{
  114. Emitter: "test",
  115. Message: xsql.Message{},
  116. },
  117. result: []map[string]interface{}{{
  118. "kuiper_field_0": 3.4,
  119. }},
  120. },
  121. //7
  122. {
  123. sql: `SELECT 5 FROM test`,
  124. data: &xsql.Tuple{
  125. Emitter: "test",
  126. Message: xsql.Message{},
  127. },
  128. result: []map[string]interface{}{{
  129. "kuiper_field_0": 5,
  130. }},
  131. },
  132. //8
  133. {
  134. sql: `SELECT a, "value" AS b FROM test`,
  135. data: &xsql.Tuple{
  136. Emitter: "test",
  137. Message: xsql.Message{
  138. "a": "val_a",
  139. },
  140. },
  141. result: []map[string]interface{}{{
  142. "a": "val_a",
  143. "b": "value",
  144. }},
  145. },
  146. //9
  147. {
  148. sql: `SELECT a, "value" AS b, 3.14 as Pi, 0 as Zero FROM test`,
  149. data: &xsql.Tuple{
  150. Emitter: "test",
  151. Message: xsql.Message{
  152. "a": "val_a",
  153. },
  154. },
  155. result: []map[string]interface{}{{
  156. "a": "val_a",
  157. "b": "value",
  158. "Pi": 3.14,
  159. "Zero": 0,
  160. }},
  161. },
  162. //10
  163. {
  164. sql: `SELECT a->b AS ab FROM test`,
  165. data: &xsql.Tuple{
  166. Emitter: "test",
  167. Message: xsql.Message{
  168. "a": map[string]interface{}{"b": "hello"},
  169. },
  170. },
  171. result: []map[string]interface{}{{
  172. "ab": "hello",
  173. }},
  174. },
  175. //11
  176. {
  177. sql: `SELECT a->b AS ab FROM test`,
  178. data: &xsql.Tuple{
  179. Emitter: "test",
  180. Message: xsql.Message{
  181. "a": map[string]interface{}(nil),
  182. },
  183. },
  184. result: []map[string]interface{}{{}},
  185. },
  186. //12
  187. {
  188. sql: `SELECT a->b AS ab FROM test`,
  189. data: &xsql.Tuple{
  190. Emitter: "test",
  191. Message: xsql.Message{
  192. "name": "name",
  193. },
  194. },
  195. result: []map[string]interface{}{{}},
  196. },
  197. //13
  198. {
  199. sql: `SELECT a->b AS ab FROM test`,
  200. data: &xsql.Tuple{
  201. Emitter: "test",
  202. Message: xsql.Message{
  203. "a": "commonstring",
  204. },
  205. },
  206. result: []map[string]interface{}{{}},
  207. },
  208. //14
  209. {
  210. sql: `SELECT a[0]->b AS ab FROM test`,
  211. data: &xsql.Tuple{
  212. Emitter: "test",
  213. Message: xsql.Message{
  214. "a": []interface{}{
  215. map[string]interface{}{"b": "hello1"},
  216. map[string]interface{}{"b": "hello2"},
  217. },
  218. },
  219. },
  220. result: []map[string]interface{}{{
  221. "ab": "hello1",
  222. }},
  223. },
  224. //15
  225. {
  226. sql: `SELECT a[0]->b AS ab FROM test`,
  227. data: &xsql.Tuple{
  228. Emitter: "test",
  229. Message: xsql.Message{
  230. "a": []map[string]interface{}{
  231. {"b": "hello1"},
  232. {"b": "hello2"},
  233. },
  234. },
  235. },
  236. result: []map[string]interface{}{{
  237. "ab": "hello1",
  238. }},
  239. },
  240. //16
  241. {
  242. sql: `SELECT a[2:4] AS ab FROM test`,
  243. data: &xsql.Tuple{
  244. Emitter: "test",
  245. Message: xsql.Message{
  246. "a": []map[string]interface{}{
  247. {"b": "hello1"},
  248. {"b": "hello2"},
  249. {"b": "hello3"},
  250. {"b": "hello4"},
  251. {"b": "hello5"},
  252. },
  253. },
  254. },
  255. result: []map[string]interface{}{{
  256. "ab": []map[string]interface{}{
  257. {"b": "hello3"},
  258. {"b": "hello4"},
  259. },
  260. }},
  261. },
  262. //17
  263. {
  264. sql: `SELECT a[2:] AS ab FROM test`,
  265. data: &xsql.Tuple{
  266. Emitter: "test",
  267. Message: xsql.Message{
  268. "a": []map[string]interface{}{
  269. {"b": "hello1"},
  270. {"b": "hello2"},
  271. {"b": "hello3"},
  272. {"b": "hello4"},
  273. {"b": "hello5"},
  274. },
  275. },
  276. },
  277. result: []map[string]interface{}{{
  278. "ab": []map[string]interface{}{
  279. {"b": "hello3"},
  280. {"b": "hello4"},
  281. {"b": "hello5"},
  282. },
  283. }},
  284. },
  285. //18
  286. {
  287. sql: `SELECT a[2:] AS ab FROM test`,
  288. data: &xsql.Tuple{
  289. Emitter: "test",
  290. Message: xsql.Message{
  291. "a": []interface{}{
  292. true, false, true, false, true, true,
  293. },
  294. },
  295. },
  296. result: []map[string]interface{}{{
  297. "ab": []interface{}{
  298. true, false, true, true,
  299. },
  300. }},
  301. },
  302. //19
  303. {
  304. sql: `SELECT a[:4] AS ab FROM test`,
  305. data: &xsql.Tuple{
  306. Emitter: "test",
  307. Message: xsql.Message{
  308. "a": []interface{}{
  309. true, false, true, false, true, true,
  310. },
  311. },
  312. },
  313. result: []map[string]interface{}{{
  314. "ab": []interface{}{
  315. true, false, true, false,
  316. },
  317. }},
  318. },
  319. //20
  320. {
  321. sql: `SELECT a[:4] AS ab FROM test`,
  322. data: &xsql.Tuple{
  323. Emitter: "test",
  324. Message: xsql.Message{
  325. "a": []interface{}{
  326. 3.14, 3.141, 3.1415, 3.14159, 3.141592, 3.1415926,
  327. },
  328. },
  329. },
  330. result: []map[string]interface{}{{
  331. "ab": []interface{}{
  332. 3.14, 3.141, 3.1415, 3.14159,
  333. },
  334. }},
  335. },
  336. //21
  337. {
  338. sql: `SELECT a->b[:4] AS ab FROM test`,
  339. data: &xsql.Tuple{
  340. Emitter: "test",
  341. Message: xsql.Message{
  342. "a": map[string]interface{}{
  343. "b": []float64{3.14, 3.141, 3.1415, 3.14159, 3.141592, 3.1415926},
  344. },
  345. },
  346. },
  347. result: []map[string]interface{}{{
  348. "ab": []float64{
  349. 3.14, 3.141, 3.1415, 3.14159,
  350. },
  351. }},
  352. },
  353. //22
  354. {
  355. sql: `SELECT a->b[0:1] AS ab FROM test`,
  356. data: &xsql.Tuple{
  357. Emitter: "test",
  358. Message: xsql.Message{
  359. "a": map[string]interface{}{
  360. "b": []float64{3.14, 3.141, 3.1415, 3.14159, 3.141592, 3.1415926},
  361. },
  362. },
  363. },
  364. result: []map[string]interface{}{{
  365. "ab": []float64{
  366. 3.14,
  367. },
  368. }},
  369. },
  370. //23
  371. {
  372. sql: `SELECT a->c->d AS f1 FROM test`,
  373. data: &xsql.Tuple{
  374. Emitter: "test",
  375. Message: xsql.Message{
  376. "a": map[string]interface{}{
  377. "b": "hello",
  378. "c": map[string]interface{}{
  379. "d": 35.2,
  380. },
  381. },
  382. },
  383. },
  384. result: []map[string]interface{}{{
  385. "f1": 35.2,
  386. }},
  387. },
  388. //24
  389. {
  390. sql: `SELECT a->c->d AS f1 FROM test`,
  391. data: &xsql.Tuple{
  392. Emitter: "test",
  393. Message: xsql.Message{
  394. "a": map[string]interface{}{
  395. "b": "hello",
  396. "c": map[string]interface{}{
  397. "e": 35.2,
  398. },
  399. },
  400. },
  401. },
  402. result: []map[string]interface{}{{}},
  403. },
  404. //25
  405. {
  406. sql: `SELECT a->c->d AS f1 FROM test`,
  407. data: &xsql.Tuple{
  408. Emitter: "test",
  409. Message: xsql.Message{
  410. "a": map[string]interface{}{
  411. "b": "hello",
  412. },
  413. },
  414. },
  415. result: []map[string]interface{}{{}},
  416. },
  417. //26
  418. //The int type is not supported yet, the json parser returns float64 for int values
  419. {
  420. sql: `SELECT a->c->d AS f1 FROM test`,
  421. data: &xsql.Tuple{
  422. Emitter: "test",
  423. Message: xsql.Message{
  424. "a": map[string]interface{}{
  425. "b": "hello",
  426. "c": map[string]interface{}{
  427. "d": float64(35),
  428. },
  429. },
  430. },
  431. },
  432. result: []map[string]interface{}{{
  433. "f1": float64(35),
  434. }},
  435. },
  436. //27
  437. {
  438. sql: "SELECT a FROM test",
  439. data: &xsql.Tuple{
  440. Emitter: "test",
  441. Message: xsql.Message{},
  442. },
  443. result: []map[string]interface{}{
  444. {},
  445. },
  446. },
  447. //28
  448. {
  449. sql: "SELECT * FROM test",
  450. data: &xsql.Tuple{
  451. Emitter: "test",
  452. Message: xsql.Message{},
  453. },
  454. result: []map[string]interface{}{
  455. {},
  456. },
  457. },
  458. //29
  459. {
  460. sql: `SELECT * FROM test`,
  461. data: &xsql.Tuple{
  462. Emitter: "test",
  463. Message: xsql.Message{
  464. "a": map[string]interface{}{
  465. "b": "hello",
  466. "c": map[string]interface{}{
  467. "d": 35.2,
  468. },
  469. },
  470. },
  471. },
  472. result: []map[string]interface{}{{
  473. "a": map[string]interface{}{
  474. "b": "hello",
  475. "c": map[string]interface{}{
  476. "d": 35.2,
  477. },
  478. },
  479. }},
  480. },
  481. //30
  482. {
  483. sql: `SELECT * FROM test`,
  484. data: &xsql.Tuple{
  485. Emitter: "test",
  486. Message: xsql.Message{
  487. "a": "val1",
  488. "b": 3.14,
  489. },
  490. },
  491. result: []map[string]interface{}{{
  492. "a": "val1",
  493. "b": 3.14,
  494. }},
  495. },
  496. //31
  497. {
  498. sql: `SELECT 3*4 AS f1 FROM test`,
  499. data: &xsql.Tuple{
  500. Emitter: "test",
  501. Message: xsql.Message{},
  502. },
  503. result: []map[string]interface{}{{
  504. "f1": int64(12),
  505. }},
  506. },
  507. //32
  508. {
  509. sql: `SELECT 4.5*2 AS f1 FROM test`,
  510. data: &xsql.Tuple{
  511. Emitter: "test",
  512. Message: xsql.Message{},
  513. },
  514. result: []map[string]interface{}{{
  515. "f1": float64(9),
  516. }},
  517. },
  518. //33
  519. {
  520. sql: "SELECT `a.b.c` FROM test",
  521. data: &xsql.Tuple{
  522. Emitter: "test",
  523. Message: xsql.Message{
  524. "a.b.c": "val_a",
  525. },
  526. },
  527. result: []map[string]interface{}{{
  528. "a.b.c": "val_a",
  529. }},
  530. },
  531. //34
  532. {
  533. sql: `SELECT CASE a WHEN 10 THEN "true" END AS b FROM test`,
  534. data: &xsql.Tuple{
  535. Emitter: "test",
  536. Message: xsql.Message{
  537. "a": int64(10),
  538. },
  539. },
  540. result: []map[string]interface{}{{
  541. "b": "true",
  542. }},
  543. },
  544. }
  545. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  546. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_Apply1")
  547. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  548. for i, tt := range tests {
  549. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  550. if err != nil {
  551. t.Errorf("parse sql error: %s", err)
  552. continue
  553. }
  554. pp := &ProjectOp{Fields: stmt.Fields, SendMeta: true}
  555. fv, afv := xsql.NewFunctionValuersForOp(nil)
  556. result := pp.Apply(ctx, tt.data, fv, afv)
  557. if !reflect.DeepEqual(tt.result, result) {
  558. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  559. }
  560. }
  561. }
  562. func TestProjectPlan_MultiInput(t *testing.T) {
  563. var tests = []struct {
  564. sql string
  565. data interface{}
  566. result []map[string]interface{}
  567. }{ //0
  568. {
  569. sql: "SELECT * FROM tbl WHERE abc*2+3 > 12 AND abc < 20",
  570. data: &xsql.Tuple{
  571. Emitter: "tbl",
  572. Message: xsql.Message{
  573. "abc": int64(6),
  574. },
  575. },
  576. result: []map[string]interface{}{{
  577. "abc": int64(6),
  578. }},
  579. },
  580. //1
  581. {
  582. sql: "SELECT abc FROM tbl WHERE abc*2+3 > 12 OR def = \"hello\"",
  583. data: &xsql.Tuple{
  584. Emitter: "tbl",
  585. Message: xsql.Message{
  586. "abc": int64(34),
  587. "def": "hello",
  588. },
  589. },
  590. result: []map[string]interface{}{{
  591. "abc": int64(34),
  592. }},
  593. },
  594. //2
  595. {
  596. sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  597. data: xsql.WindowTuplesSet{
  598. Content: []xsql.WindowTuples{
  599. {
  600. Emitter: "src1",
  601. Tuples: []xsql.Tuple{
  602. {
  603. Emitter: "src1",
  604. Message: xsql.Message{"id1": 1, "f1": "v1"},
  605. }, {
  606. Emitter: "src1",
  607. Message: xsql.Message{"id1": 2, "f1": "v2"},
  608. }, {
  609. Emitter: "src1",
  610. Message: xsql.Message{"id1": 3, "f1": "v1"},
  611. },
  612. },
  613. },
  614. },
  615. },
  616. result: []map[string]interface{}{{
  617. "id1": 1,
  618. }, {
  619. "id1": 2,
  620. }, {
  621. "id1": 3,
  622. }},
  623. },
  624. //3
  625. {
  626. sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  627. data: xsql.WindowTuplesSet{
  628. Content: []xsql.WindowTuples{
  629. {
  630. Emitter: "src1",
  631. Tuples: []xsql.Tuple{
  632. {
  633. Emitter: "src1",
  634. Message: xsql.Message{"id1": 1, "f1": "v1"},
  635. }, {
  636. Emitter: "src1",
  637. Message: xsql.Message{"id2": 2, "f1": "v2"},
  638. }, {
  639. Emitter: "src1",
  640. Message: xsql.Message{"id1": 3, "f1": "v1"},
  641. },
  642. },
  643. },
  644. },
  645. },
  646. result: []map[string]interface{}{{
  647. "id1": 1,
  648. }, {}, {
  649. "id1": 3,
  650. }},
  651. },
  652. //4
  653. {
  654. sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  655. data: xsql.WindowTuplesSet{
  656. Content: []xsql.WindowTuples{
  657. {
  658. Emitter: "src1",
  659. Tuples: []xsql.Tuple{
  660. {
  661. Emitter: "src1",
  662. Message: xsql.Message{"id1": 1, "f1": "v1"},
  663. }, {
  664. Emitter: "src1",
  665. Message: xsql.Message{"id1": 2, "f1": "v2"},
  666. }, {
  667. Emitter: "src1",
  668. Message: xsql.Message{"id1": 3, "f1": "v1"},
  669. },
  670. },
  671. },
  672. },
  673. },
  674. result: []map[string]interface{}{{
  675. "id1": 1,
  676. "f1": "v1",
  677. }, {
  678. "id1": 2,
  679. "f1": "v2",
  680. }, {
  681. "id1": 3,
  682. "f1": "v1",
  683. }},
  684. },
  685. //5
  686. {
  687. sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  688. data: xsql.WindowTuplesSet{
  689. Content: []xsql.WindowTuples{
  690. {
  691. Emitter: "src1",
  692. Tuples: []xsql.Tuple{
  693. {
  694. Emitter: "src1",
  695. Message: xsql.Message{"id1": 1, "f1": "v1"},
  696. }, {
  697. Emitter: "src1",
  698. Message: xsql.Message{"id2": 2, "f2": "v2"},
  699. }, {
  700. Emitter: "src1",
  701. Message: xsql.Message{"id1": 3, "f1": "v1"},
  702. },
  703. },
  704. },
  705. },
  706. },
  707. result: []map[string]interface{}{{
  708. "id1": 1,
  709. "f1": "v1",
  710. }, {
  711. "id2": 2,
  712. "f2": "v2",
  713. }, {
  714. "id1": 3,
  715. "f1": "v1",
  716. }},
  717. },
  718. //6
  719. {
  720. sql: "SELECT src1.* FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  721. data: xsql.WindowTuplesSet{
  722. Content: []xsql.WindowTuples{
  723. {
  724. Emitter: "src1",
  725. Tuples: []xsql.Tuple{
  726. {
  727. Emitter: "src1",
  728. Message: xsql.Message{"id1": 1, "f1": "v1"},
  729. }, {
  730. Emitter: "src1",
  731. Message: xsql.Message{"id1": 2, "f1": "v2"},
  732. }, {
  733. Emitter: "src1",
  734. Message: xsql.Message{"id1": 3, "f1": "v1"},
  735. },
  736. },
  737. },
  738. },
  739. },
  740. result: []map[string]interface{}{{
  741. "id1": 1,
  742. "f1": "v1",
  743. }, {
  744. "id1": 2,
  745. "f1": "v2",
  746. }, {
  747. "id1": 3,
  748. "f1": "v1",
  749. }},
  750. },
  751. //7
  752. {
  753. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  754. data: &xsql.JoinTupleSets{
  755. Content: []xsql.JoinTuple{
  756. {
  757. Tuples: []xsql.Tuple{
  758. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  759. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  760. },
  761. },
  762. {
  763. Tuples: []xsql.Tuple{
  764. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  765. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  766. },
  767. },
  768. {
  769. Tuples: []xsql.Tuple{
  770. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  771. },
  772. },
  773. },
  774. },
  775. result: []map[string]interface{}{{
  776. "id1": 1,
  777. }, {
  778. "id1": 2,
  779. }, {
  780. "id1": 3,
  781. }},
  782. },
  783. //8
  784. {
  785. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  786. data: &xsql.JoinTupleSets{
  787. Content: []xsql.JoinTuple{
  788. {
  789. Tuples: []xsql.Tuple{
  790. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  791. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  792. },
  793. },
  794. {
  795. Tuples: []xsql.Tuple{
  796. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  797. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  798. },
  799. },
  800. {
  801. Tuples: []xsql.Tuple{
  802. {Emitter: "src1", Message: xsql.Message{"id2": 3, "f1": "v1"}},
  803. },
  804. },
  805. },
  806. },
  807. result: []map[string]interface{}{{
  808. "id1": 1,
  809. }, {
  810. "id1": 2,
  811. }, {}},
  812. },
  813. //9
  814. {
  815. sql: "SELECT abc FROM tbl group by abc",
  816. data: xsql.GroupedTuplesSet{
  817. {
  818. Content: []xsql.DataValuer{
  819. &xsql.Tuple{
  820. Emitter: "tbl",
  821. Message: xsql.Message{
  822. "abc": int64(6),
  823. "def": "hello",
  824. },
  825. },
  826. },
  827. },
  828. },
  829. result: []map[string]interface{}{{
  830. "abc": int64(6),
  831. }},
  832. },
  833. //10
  834. {
  835. sql: "SELECT abc FROM tbl group by abc",
  836. data: xsql.GroupedTuplesSet{
  837. {
  838. Content: []xsql.DataValuer{
  839. &xsql.Tuple{
  840. Emitter: "tbl",
  841. Message: xsql.Message{
  842. "def": "hello",
  843. },
  844. },
  845. },
  846. },
  847. },
  848. result: []map[string]interface{}{{}},
  849. },
  850. //11
  851. {
  852. sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  853. data: xsql.GroupedTuplesSet{
  854. {
  855. Content: []xsql.DataValuer{
  856. &xsql.Tuple{
  857. Emitter: "src1",
  858. Message: xsql.Message{"id1": 1, "f1": "v1"},
  859. },
  860. &xsql.Tuple{
  861. Emitter: "src1",
  862. Message: xsql.Message{"id1": 3, "f1": "v1"},
  863. },
  864. },
  865. },
  866. {
  867. Content: []xsql.DataValuer{
  868. &xsql.Tuple{
  869. Emitter: "src1",
  870. Message: xsql.Message{"id1": 2, "f1": "v2"},
  871. },
  872. },
  873. },
  874. },
  875. result: []map[string]interface{}{{
  876. "id1": 1,
  877. }, {
  878. "id1": 2,
  879. }},
  880. },
  881. //12
  882. {
  883. sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  884. data: xsql.GroupedTuplesSet{
  885. {
  886. Content: []xsql.DataValuer{
  887. &xsql.Tuple{
  888. Emitter: "src1",
  889. Message: xsql.Message{"id1": 1, "f1": "v1"},
  890. },
  891. &xsql.Tuple{
  892. Emitter: "src1",
  893. Message: xsql.Message{"id1": 3, "f1": "v1"},
  894. },
  895. },
  896. },
  897. {
  898. Content: []xsql.DataValuer{
  899. &xsql.Tuple{
  900. Emitter: "src1",
  901. Message: xsql.Message{"id2": 2, "f1": "v2"},
  902. },
  903. },
  904. },
  905. },
  906. result: []map[string]interface{}{{
  907. "id1": 1,
  908. }, {}},
  909. },
  910. //13
  911. {
  912. sql: "SELECT src2.id2 FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  913. data: xsql.GroupedTuplesSet{
  914. {
  915. Content: []xsql.DataValuer{
  916. &xsql.JoinTuple{
  917. Tuples: []xsql.Tuple{
  918. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  919. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  920. },
  921. },
  922. },
  923. },
  924. {
  925. Content: []xsql.DataValuer{
  926. &xsql.JoinTuple{
  927. Tuples: []xsql.Tuple{
  928. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  929. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  930. },
  931. },
  932. },
  933. },
  934. {
  935. Content: []xsql.DataValuer{
  936. &xsql.JoinTuple{
  937. Tuples: []xsql.Tuple{
  938. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  939. },
  940. },
  941. },
  942. },
  943. },
  944. result: []map[string]interface{}{{
  945. "id2": 2,
  946. }, {
  947. "id2": 4,
  948. }, {}},
  949. },
  950. //14
  951. {
  952. sql: "SELECT src1.*, f2 FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
  953. data: &xsql.JoinTupleSets{
  954. Content: []xsql.JoinTuple{
  955. {
  956. Tuples: []xsql.Tuple{
  957. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  958. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  959. },
  960. },
  961. {
  962. Tuples: []xsql.Tuple{
  963. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  964. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  965. },
  966. },
  967. {
  968. Tuples: []xsql.Tuple{
  969. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  970. },
  971. },
  972. },
  973. },
  974. result: []map[string]interface{}{{
  975. "id1": 1,
  976. "f1": "v1",
  977. "f2": "w2",
  978. }, {
  979. "id1": 2,
  980. "f1": "v2",
  981. "f2": "w3",
  982. }, {
  983. "id1": 3,
  984. "f1": "v1",
  985. }},
  986. },
  987. //15
  988. {
  989. sql: "SELECT * FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
  990. data: &xsql.JoinTupleSets{
  991. Content: []xsql.JoinTuple{
  992. {
  993. Tuples: []xsql.Tuple{
  994. {Emitter: "src1", Message: xsql.Message{"id": 1, "f1": "v1"}},
  995. {Emitter: "src2", Message: xsql.Message{"id": 2, "f2": "w2"}},
  996. },
  997. },
  998. {
  999. Tuples: []xsql.Tuple{
  1000. {Emitter: "src1", Message: xsql.Message{"id": 2, "f1": "v2"}},
  1001. {Emitter: "src2", Message: xsql.Message{"id": 4, "f2": "w3"}},
  1002. },
  1003. },
  1004. {
  1005. Tuples: []xsql.Tuple{
  1006. {Emitter: "src1", Message: xsql.Message{"id": 3, "f1": "v1"}},
  1007. },
  1008. },
  1009. },
  1010. },
  1011. result: []map[string]interface{}{{
  1012. "id": 1,
  1013. "f1": "v1",
  1014. "f2": "w2",
  1015. }, {
  1016. "id": 2,
  1017. "f1": "v2",
  1018. "f2": "w3",
  1019. }, {
  1020. "id": 3,
  1021. "f1": "v1",
  1022. }},
  1023. },
  1024. //16
  1025. {
  1026. sql: "SELECT src1.* FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  1027. data: xsql.GroupedTuplesSet{
  1028. {
  1029. Content: []xsql.DataValuer{
  1030. &xsql.Tuple{
  1031. Emitter: "src1",
  1032. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1033. },
  1034. &xsql.Tuple{
  1035. Emitter: "src1",
  1036. Message: xsql.Message{"id1": 3, "f1": "v1"},
  1037. },
  1038. },
  1039. },
  1040. {
  1041. Content: []xsql.DataValuer{
  1042. &xsql.Tuple{
  1043. Emitter: "src1",
  1044. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1045. },
  1046. },
  1047. },
  1048. },
  1049. result: []map[string]interface{}{{
  1050. "id1": 1,
  1051. "f1": "v1",
  1052. }, {
  1053. "id1": 2,
  1054. "f1": "v2",
  1055. }},
  1056. },
  1057. //17
  1058. {
  1059. sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  1060. data: xsql.GroupedTuplesSet{
  1061. {
  1062. Content: []xsql.DataValuer{
  1063. &xsql.JoinTuple{
  1064. Tuples: []xsql.Tuple{
  1065. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1066. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1067. },
  1068. },
  1069. },
  1070. },
  1071. {
  1072. Content: []xsql.DataValuer{
  1073. &xsql.JoinTuple{
  1074. Tuples: []xsql.Tuple{
  1075. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1076. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1077. },
  1078. },
  1079. },
  1080. },
  1081. {
  1082. Content: []xsql.DataValuer{
  1083. &xsql.JoinTuple{
  1084. Tuples: []xsql.Tuple{
  1085. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  1086. },
  1087. },
  1088. },
  1089. },
  1090. },
  1091. result: []map[string]interface{}{{
  1092. "id2": 2,
  1093. "id1": 1,
  1094. "f1": "v1",
  1095. }, {
  1096. "id2": 4,
  1097. "id1": 2,
  1098. "f1": "v2",
  1099. }, {
  1100. "id1": 3,
  1101. "f1": "v1",
  1102. }},
  1103. },
  1104. //18
  1105. {
  1106. sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  1107. data: xsql.GroupedTuplesSet{
  1108. {
  1109. Content: []xsql.DataValuer{
  1110. &xsql.JoinTuple{
  1111. Tuples: []xsql.Tuple{
  1112. {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1113. {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1114. },
  1115. },
  1116. },
  1117. },
  1118. {
  1119. Content: []xsql.DataValuer{
  1120. &xsql.JoinTuple{
  1121. Tuples: []xsql.Tuple{
  1122. {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1123. {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1124. },
  1125. },
  1126. },
  1127. },
  1128. {
  1129. Content: []xsql.DataValuer{
  1130. &xsql.JoinTuple{
  1131. Tuples: []xsql.Tuple{
  1132. {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  1133. },
  1134. },
  1135. },
  1136. },
  1137. },
  1138. result: []map[string]interface{}{{
  1139. "id2": 2,
  1140. "id1": 1,
  1141. "f1": "v1",
  1142. }, {
  1143. "id2": 4,
  1144. "id1": 2,
  1145. "f1": "v2",
  1146. }, {
  1147. "id1": 3,
  1148. "f1": "v1",
  1149. }},
  1150. },
  1151. }
  1152. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1153. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_MultiInput")
  1154. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1155. for i, tt := range tests {
  1156. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1157. pp := &ProjectOp{Fields: stmt.Fields}
  1158. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1159. result := pp.Apply(ctx, tt.data, fv, afv)
  1160. if !reflect.DeepEqual(tt.result, result) {
  1161. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1162. }
  1163. }
  1164. }
  1165. func TestProjectPlan_Funcs(t *testing.T) {
  1166. var tests = []struct {
  1167. sql string
  1168. data interface{}
  1169. result []map[string]interface{}
  1170. }{
  1171. //0
  1172. {
  1173. sql: "SELECT round(a) as r FROM test",
  1174. data: &xsql.Tuple{
  1175. Emitter: "test",
  1176. Message: xsql.Message{
  1177. "a": 47.5,
  1178. },
  1179. },
  1180. result: []map[string]interface{}{{
  1181. "r": float64(48),
  1182. }},
  1183. },
  1184. //1
  1185. {
  1186. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  1187. data: xsql.WindowTuplesSet{
  1188. Content: []xsql.WindowTuples{
  1189. {
  1190. Emitter: "test",
  1191. Tuples: []xsql.Tuple{
  1192. {
  1193. Emitter: "src1",
  1194. Message: xsql.Message{"a": 53.1},
  1195. }, {
  1196. Emitter: "src1",
  1197. Message: xsql.Message{"a": 27.4},
  1198. }, {
  1199. Emitter: "src1",
  1200. Message: xsql.Message{"a": 123123.7},
  1201. },
  1202. },
  1203. },
  1204. },
  1205. },
  1206. result: []map[string]interface{}{{
  1207. "r": float64(53),
  1208. }, {
  1209. "r": float64(27),
  1210. }, {
  1211. "r": float64(123124),
  1212. }},
  1213. },
  1214. //2
  1215. {
  1216. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  1217. data: xsql.WindowTuplesSet{
  1218. Content: []xsql.WindowTuples{
  1219. {
  1220. Emitter: "test",
  1221. Tuples: []xsql.Tuple{
  1222. {
  1223. Emitter: "src1",
  1224. Message: xsql.Message{"a": 53.1},
  1225. }, {
  1226. Emitter: "src1",
  1227. Message: xsql.Message{"a": 27.4},
  1228. }, {
  1229. Emitter: "src1",
  1230. Message: xsql.Message{"a": 123123.7},
  1231. },
  1232. },
  1233. },
  1234. },
  1235. },
  1236. result: []map[string]interface{}{{
  1237. "r": float64(53),
  1238. }, {
  1239. "r": float64(27),
  1240. }, {
  1241. "r": float64(123124),
  1242. }},
  1243. },
  1244. //3
  1245. {
  1246. sql: "SELECT round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1247. data: &xsql.JoinTupleSets{
  1248. Content: []xsql.JoinTuple{
  1249. {
  1250. Tuples: []xsql.Tuple{
  1251. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
  1252. {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1253. },
  1254. },
  1255. {
  1256. Tuples: []xsql.Tuple{
  1257. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
  1258. {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1259. },
  1260. },
  1261. {
  1262. Tuples: []xsql.Tuple{
  1263. {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
  1264. {Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1265. },
  1266. },
  1267. },
  1268. },
  1269. result: []map[string]interface{}{{
  1270. "r": float64(66),
  1271. }, {
  1272. "r": float64(73),
  1273. }, {
  1274. "r": float64(89),
  1275. }},
  1276. },
  1277. //4
  1278. {
  1279. 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)",
  1280. data: &xsql.JoinTupleSets{
  1281. Content: []xsql.JoinTuple{
  1282. {
  1283. Tuples: []xsql.Tuple{
  1284. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
  1285. {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1286. },
  1287. },
  1288. {
  1289. Tuples: []xsql.Tuple{
  1290. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
  1291. {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1292. },
  1293. },
  1294. {
  1295. Tuples: []xsql.Tuple{
  1296. {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
  1297. {Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1298. },
  1299. },
  1300. },
  1301. },
  1302. result: []map[string]interface{}{{
  1303. "concat": "165.5512",
  1304. }, {
  1305. "concat": "273.49934",
  1306. }, {
  1307. "concat": "388.886",
  1308. }},
  1309. },
  1310. //5
  1311. {
  1312. sql: "SELECT count(a) as r FROM test",
  1313. data: &xsql.Tuple{
  1314. Emitter: "test",
  1315. Message: xsql.Message{
  1316. "a": 47.5,
  1317. },
  1318. },
  1319. result: []map[string]interface{}{{
  1320. "r": 1,
  1321. }},
  1322. },
  1323. //6
  1324. {
  1325. sql: "SELECT meta(test.device) as d FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1326. data: &xsql.JoinTupleSets{
  1327. Content: []xsql.JoinTuple{
  1328. {
  1329. Tuples: []xsql.Tuple{
  1330. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}, Metadata: xsql.Metadata{"device": "devicea"}},
  1331. {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1332. },
  1333. },
  1334. {
  1335. Tuples: []xsql.Tuple{
  1336. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1337. {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1338. },
  1339. },
  1340. {
  1341. Tuples: []xsql.Tuple{
  1342. {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}, Metadata: xsql.Metadata{"device": "devicec"}},
  1343. {Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1344. },
  1345. },
  1346. },
  1347. },
  1348. result: []map[string]interface{}{{
  1349. "d": "devicea",
  1350. }, {
  1351. "d": "deviceb",
  1352. }, {
  1353. "d": "devicec",
  1354. }},
  1355. },
  1356. }
  1357. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1358. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_Funcs")
  1359. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1360. for i, tt := range tests {
  1361. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1362. if err != nil {
  1363. t.Error(err)
  1364. }
  1365. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
  1366. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1367. result := pp.Apply(ctx, tt.data, fv, afv)
  1368. if !reflect.DeepEqual(tt.result, result) {
  1369. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1370. }
  1371. }
  1372. }
  1373. func TestProjectPlan_AggFuncs(t *testing.T) {
  1374. var tests = []struct {
  1375. sql string
  1376. data interface{}
  1377. result []map[string]interface{}
  1378. }{
  1379. //0
  1380. {
  1381. sql: "SELECT count(*) as c, round(a) as r, window_start() as ws, window_end() as we FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1382. data: xsql.GroupedTuplesSet{
  1383. {
  1384. Content: []xsql.DataValuer{
  1385. &xsql.JoinTuple{
  1386. Tuples: []xsql.Tuple{
  1387. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1388. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1389. },
  1390. },
  1391. &xsql.JoinTuple{
  1392. Tuples: []xsql.Tuple{
  1393. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1394. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1395. },
  1396. },
  1397. },
  1398. WindowRange: &xsql.WindowRange{
  1399. WindowStart: 1541152486013,
  1400. WindowEnd: 1541152487013,
  1401. },
  1402. },
  1403. {
  1404. Content: []xsql.DataValuer{
  1405. &xsql.JoinTuple{
  1406. Tuples: []xsql.Tuple{
  1407. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1408. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1409. },
  1410. },
  1411. &xsql.JoinTuple{
  1412. Tuples: []xsql.Tuple{
  1413. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1414. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1415. },
  1416. },
  1417. },
  1418. WindowRange: &xsql.WindowRange{
  1419. WindowStart: 1541152486013,
  1420. WindowEnd: 1541152487013,
  1421. },
  1422. },
  1423. },
  1424. result: []map[string]interface{}{{
  1425. "c": 2,
  1426. "r": float64(122),
  1427. "ws": int64(1541152486013),
  1428. "we": int64(1541152487013),
  1429. }, {
  1430. "c": 2,
  1431. "r": float64(89),
  1432. "ws": int64(1541152486013),
  1433. "we": int64(1541152487013),
  1434. }},
  1435. },
  1436. //1
  1437. {
  1438. sql: "SELECT count(a) as c, avg(a) as a, sum(a) as s, min(a) as min, max(a) as max FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1439. data: xsql.GroupedTuplesSet{
  1440. {
  1441. Content: []xsql.DataValuer{
  1442. &xsql.JoinTuple{
  1443. Tuples: []xsql.Tuple{
  1444. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1445. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1446. },
  1447. },
  1448. &xsql.JoinTuple{
  1449. Tuples: []xsql.Tuple{
  1450. {Emitter: "test", Message: xsql.Message{"id": 5}},
  1451. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1452. },
  1453. },
  1454. },
  1455. },
  1456. {
  1457. Content: []xsql.DataValuer{
  1458. &xsql.JoinTuple{
  1459. Tuples: []xsql.Tuple{
  1460. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1461. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1462. },
  1463. },
  1464. &xsql.JoinTuple{
  1465. Tuples: []xsql.Tuple{
  1466. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1467. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1468. },
  1469. },
  1470. },
  1471. },
  1472. },
  1473. result: []map[string]interface{}{{
  1474. "c": 1,
  1475. "a": 122.33,
  1476. "s": 122.33,
  1477. "min": 122.33,
  1478. "max": 122.33,
  1479. }, {
  1480. "c": 2,
  1481. "s": 103.63,
  1482. "a": 51.815,
  1483. "min": 14.6,
  1484. "max": 89.03,
  1485. }},
  1486. },
  1487. //2
  1488. {
  1489. sql: "SELECT avg(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1490. data: xsql.GroupedTuplesSet{
  1491. {
  1492. Content: []xsql.DataValuer{
  1493. &xsql.JoinTuple{
  1494. Tuples: []xsql.Tuple{
  1495. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1496. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1497. },
  1498. },
  1499. &xsql.JoinTuple{
  1500. Tuples: []xsql.Tuple{
  1501. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
  1502. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1503. },
  1504. },
  1505. &xsql.JoinTuple{
  1506. Tuples: []xsql.Tuple{
  1507. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 98.31}},
  1508. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
  1509. },
  1510. },
  1511. &xsql.JoinTuple{
  1512. Tuples: []xsql.Tuple{
  1513. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
  1514. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1515. },
  1516. },
  1517. },
  1518. },
  1519. {
  1520. Content: []xsql.DataValuer{
  1521. &xsql.JoinTuple{
  1522. Tuples: []xsql.Tuple{
  1523. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1524. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1525. },
  1526. },
  1527. &xsql.JoinTuple{
  1528. Tuples: []xsql.Tuple{
  1529. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1530. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1531. },
  1532. },
  1533. },
  1534. },
  1535. },
  1536. result: []map[string]interface{}{{
  1537. "avg": 116.68,
  1538. }, {
  1539. "avg": 51.815,
  1540. }},
  1541. },
  1542. //3
  1543. {
  1544. sql: "SELECT max(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1545. data: xsql.GroupedTuplesSet{
  1546. {
  1547. Content: []xsql.DataValuer{
  1548. &xsql.JoinTuple{
  1549. Tuples: []xsql.Tuple{
  1550. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1551. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1552. },
  1553. },
  1554. &xsql.JoinTuple{
  1555. Tuples: []xsql.Tuple{
  1556. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1557. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1558. },
  1559. },
  1560. &xsql.JoinTuple{
  1561. Tuples: []xsql.Tuple{
  1562. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1563. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1564. },
  1565. },
  1566. },
  1567. },
  1568. {
  1569. Content: []xsql.DataValuer{
  1570. &xsql.JoinTuple{
  1571. Tuples: []xsql.Tuple{
  1572. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1573. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1574. },
  1575. },
  1576. &xsql.JoinTuple{
  1577. Tuples: []xsql.Tuple{
  1578. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1579. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1580. },
  1581. },
  1582. },
  1583. },
  1584. },
  1585. result: []map[string]interface{}{{
  1586. "max": 177.51,
  1587. }, {
  1588. "max": 89.03,
  1589. }},
  1590. },
  1591. //4
  1592. {
  1593. sql: "SELECT min(a), window_start(), window_end() FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1594. data: &xsql.JoinTupleSets{
  1595. Content: []xsql.JoinTuple{
  1596. {
  1597. Tuples: []xsql.Tuple{
  1598. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1599. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1600. },
  1601. },
  1602. {
  1603. Tuples: []xsql.Tuple{
  1604. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1605. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1606. },
  1607. },
  1608. {
  1609. Tuples: []xsql.Tuple{
  1610. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1611. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1612. },
  1613. },
  1614. },
  1615. WindowRange: &xsql.WindowRange{
  1616. WindowStart: 1541152486013,
  1617. WindowEnd: 1541152487013,
  1618. },
  1619. },
  1620. result: []map[string]interface{}{{
  1621. "min": 68.55,
  1622. "window_start": int64(1541152486013),
  1623. "window_end": int64(1541152487013),
  1624. }},
  1625. },
  1626. //5
  1627. {
  1628. sql: "SELECT count(*) as all, count(a) as c, avg(a) as a, sum(a) as s, min(a) as min, max(a) as max FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1629. data: &xsql.JoinTupleSets{
  1630. Content: []xsql.JoinTuple{
  1631. {
  1632. Tuples: []xsql.Tuple{
  1633. {Emitter: "test", Message: xsql.Message{"id": 1}},
  1634. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1635. },
  1636. },
  1637. {
  1638. Tuples: []xsql.Tuple{
  1639. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1640. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1641. },
  1642. },
  1643. {
  1644. Tuples: []xsql.Tuple{
  1645. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1646. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1647. },
  1648. },
  1649. },
  1650. },
  1651. result: []map[string]interface{}{{
  1652. "all": 3,
  1653. "c": 2,
  1654. "a": 123.03,
  1655. "s": 246.06,
  1656. "min": 68.55,
  1657. "max": 177.51,
  1658. }},
  1659. },
  1660. //6
  1661. {
  1662. sql: "SELECT sum(a), window_start() as ws, window_end() FROM test GROUP BY TumblingWindow(ss, 10)",
  1663. data: xsql.WindowTuplesSet{
  1664. Content: []xsql.WindowTuples{{
  1665. Emitter: "test",
  1666. Tuples: []xsql.Tuple{
  1667. {
  1668. Emitter: "src1",
  1669. Message: xsql.Message{"a": 53},
  1670. }, {
  1671. Emitter: "src1",
  1672. Message: xsql.Message{"a": 27},
  1673. }, {
  1674. Emitter: "src1",
  1675. Message: xsql.Message{"a": 123123},
  1676. },
  1677. },
  1678. },
  1679. },
  1680. WindowRange: &xsql.WindowRange{
  1681. WindowStart: 1541152486013,
  1682. WindowEnd: 1541152487013,
  1683. },
  1684. },
  1685. result: []map[string]interface{}{{
  1686. "sum": 123203,
  1687. "ws": int64(1541152486013),
  1688. "window_end": int64(1541152487013),
  1689. }},
  1690. },
  1691. //7
  1692. {
  1693. sql: "SELECT sum(a) as s FROM test GROUP BY TumblingWindow(ss, 10)",
  1694. data: xsql.WindowTuplesSet{
  1695. Content: []xsql.WindowTuples{{
  1696. Emitter: "test",
  1697. Tuples: []xsql.Tuple{
  1698. {
  1699. Emitter: "src1",
  1700. Message: xsql.Message{"a": 53, "s": 123203},
  1701. }, {
  1702. Emitter: "src1",
  1703. Message: xsql.Message{"a": 27},
  1704. }, {
  1705. Emitter: "src1",
  1706. Message: xsql.Message{"a": 123123},
  1707. },
  1708. },
  1709. },
  1710. },
  1711. },
  1712. result: []map[string]interface{}{{
  1713. "s": 123203,
  1714. }},
  1715. },
  1716. //8
  1717. {
  1718. sql: "SELECT sum(a) FROM test GROUP BY TumblingWindow(ss, 10)",
  1719. data: xsql.WindowTuplesSet{
  1720. Content: []xsql.WindowTuples{{
  1721. Emitter: "test",
  1722. Tuples: []xsql.Tuple{
  1723. {
  1724. Emitter: "src1",
  1725. Message: xsql.Message{"a": 53},
  1726. }, {
  1727. Emitter: "src1",
  1728. Message: xsql.Message{"a": 27},
  1729. }, {
  1730. Emitter: "src1",
  1731. Message: xsql.Message{"a": 123123},
  1732. },
  1733. },
  1734. },
  1735. },
  1736. },
  1737. result: []map[string]interface{}{{
  1738. "sum": 123203,
  1739. }},
  1740. },
  1741. //9
  1742. {
  1743. sql: "SELECT count(*) as all, count(a) as c, avg(a) as a, sum(a) as s, min(a) as min, max(a) as max FROM test GROUP BY TumblingWindow(ss, 10)",
  1744. data: xsql.WindowTuplesSet{
  1745. Content: []xsql.WindowTuples{{
  1746. Emitter: "test",
  1747. Tuples: []xsql.Tuple{
  1748. {
  1749. Emitter: "src1",
  1750. Message: xsql.Message{"a": 53},
  1751. }, {
  1752. Emitter: "src1",
  1753. Message: xsql.Message{"a": 27},
  1754. }, {
  1755. Emitter: "src1",
  1756. Message: xsql.Message{"s": 123123},
  1757. },
  1758. },
  1759. },
  1760. },
  1761. },
  1762. result: []map[string]interface{}{{
  1763. "all": 3,
  1764. "c": 2,
  1765. "a": 40,
  1766. "s": 80,
  1767. "min": 27,
  1768. "max": 53,
  1769. }},
  1770. },
  1771. //10
  1772. {
  1773. sql: "SELECT count(*), meta(test1.device) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1774. data: xsql.GroupedTuplesSet{
  1775. {
  1776. Content: []xsql.DataValuer{
  1777. &xsql.JoinTuple{
  1778. Tuples: []xsql.Tuple{
  1779. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1780. {Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
  1781. },
  1782. },
  1783. &xsql.JoinTuple{
  1784. Tuples: []xsql.Tuple{
  1785. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1786. {Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1787. },
  1788. },
  1789. },
  1790. },
  1791. {
  1792. Content: []xsql.DataValuer{
  1793. &xsql.JoinTuple{
  1794. Tuples: []xsql.Tuple{
  1795. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1796. {Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
  1797. },
  1798. },
  1799. &xsql.JoinTuple{
  1800. Tuples: []xsql.Tuple{
  1801. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1802. {Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
  1803. },
  1804. },
  1805. },
  1806. },
  1807. },
  1808. result: []map[string]interface{}{{
  1809. "count": 2,
  1810. "meta": "devicea",
  1811. }, {
  1812. "count": 2,
  1813. "meta": "devicec",
  1814. }},
  1815. },
  1816. //11
  1817. {
  1818. 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",
  1819. data: xsql.GroupedTuplesSet{
  1820. {
  1821. Content: []xsql.DataValuer{
  1822. &xsql.JoinTuple{
  1823. Tuples: []xsql.Tuple{
  1824. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "d": "devicea"}},
  1825. {Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
  1826. },
  1827. },
  1828. &xsql.JoinTuple{
  1829. Tuples: []xsql.Tuple{
  1830. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1831. {Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1832. },
  1833. },
  1834. },
  1835. },
  1836. {
  1837. Content: []xsql.DataValuer{
  1838. &xsql.JoinTuple{
  1839. Tuples: []xsql.Tuple{
  1840. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "d": "devicec"}},
  1841. {Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
  1842. },
  1843. },
  1844. &xsql.JoinTuple{
  1845. Tuples: []xsql.Tuple{
  1846. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1847. {Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
  1848. },
  1849. },
  1850. },
  1851. },
  1852. },
  1853. result: []map[string]interface{}{{
  1854. "c": 2,
  1855. "d": "devicea",
  1856. }, {
  1857. "c": 2,
  1858. "d": "devicec",
  1859. }},
  1860. },
  1861. //12
  1862. {
  1863. sql: "SELECT * FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1864. data: xsql.GroupedTuplesSet{
  1865. {
  1866. Content: []xsql.DataValuer{
  1867. &xsql.JoinTuple{
  1868. Tuples: []xsql.Tuple{
  1869. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1870. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1871. },
  1872. },
  1873. &xsql.JoinTuple{
  1874. Tuples: []xsql.Tuple{
  1875. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1876. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1877. },
  1878. },
  1879. },
  1880. WindowRange: &xsql.WindowRange{
  1881. WindowStart: 1541152486013,
  1882. WindowEnd: 1541152487013,
  1883. },
  1884. },
  1885. {
  1886. Content: []xsql.DataValuer{
  1887. &xsql.JoinTuple{
  1888. Tuples: []xsql.Tuple{
  1889. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1890. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1891. },
  1892. },
  1893. &xsql.JoinTuple{
  1894. Tuples: []xsql.Tuple{
  1895. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1896. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1897. },
  1898. },
  1899. },
  1900. WindowRange: &xsql.WindowRange{
  1901. WindowStart: 1541152486013,
  1902. WindowEnd: 1541152487013,
  1903. },
  1904. },
  1905. },
  1906. result: []map[string]interface{}{{
  1907. "a": 122.33,
  1908. "c": 2,
  1909. "color": "w2",
  1910. "id": 1,
  1911. "r": 122,
  1912. }, {
  1913. "a": 89.03,
  1914. "c": 2,
  1915. "color": "w1",
  1916. "id": 2,
  1917. "r": 89,
  1918. }},
  1919. },
  1920. //13
  1921. {
  1922. sql: "SELECT collect(a) as r1 FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1923. data: xsql.GroupedTuplesSet{
  1924. {
  1925. Content: []xsql.DataValuer{
  1926. &xsql.JoinTuple{
  1927. Tuples: []xsql.Tuple{
  1928. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1929. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1930. },
  1931. },
  1932. &xsql.JoinTuple{
  1933. Tuples: []xsql.Tuple{
  1934. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1935. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1936. },
  1937. },
  1938. },
  1939. },
  1940. {
  1941. Content: []xsql.DataValuer{
  1942. &xsql.JoinTuple{
  1943. Tuples: []xsql.Tuple{
  1944. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1945. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1946. },
  1947. },
  1948. &xsql.JoinTuple{
  1949. Tuples: []xsql.Tuple{
  1950. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1951. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1952. },
  1953. },
  1954. },
  1955. },
  1956. },
  1957. result: []map[string]interface{}{{
  1958. "r1": []interface{}{122.33, 177.51},
  1959. }, {"r1": []interface{}{89.03, 14.6}}},
  1960. },
  1961. //14
  1962. {
  1963. sql: "SELECT collect(*)[1] as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1964. data: xsql.WindowTuplesSet{
  1965. Content: []xsql.WindowTuples{
  1966. {
  1967. Emitter: "test",
  1968. Tuples: []xsql.Tuple{
  1969. {
  1970. Emitter: "src1",
  1971. Message: xsql.Message{"a": 53, "s": 123203},
  1972. }, {
  1973. Emitter: "src1",
  1974. Message: xsql.Message{"a": 27},
  1975. }, {
  1976. Emitter: "src1",
  1977. Message: xsql.Message{"a": 123123},
  1978. },
  1979. },
  1980. },
  1981. },
  1982. WindowRange: &xsql.WindowRange{
  1983. WindowStart: 1541152486013,
  1984. WindowEnd: 1541152487013,
  1985. },
  1986. },
  1987. result: []map[string]interface{}{{
  1988. "c1": xsql.Message{
  1989. "a": 27,
  1990. },
  1991. }},
  1992. },
  1993. //15
  1994. {
  1995. sql: "SELECT collect(*)[1]->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1996. data: xsql.WindowTuplesSet{
  1997. Content: []xsql.WindowTuples{{
  1998. Emitter: "test",
  1999. Tuples: []xsql.Tuple{
  2000. {
  2001. Emitter: "src1",
  2002. Message: xsql.Message{"a": 53, "s": 123203},
  2003. }, {
  2004. Emitter: "src1",
  2005. Message: xsql.Message{"a": 27},
  2006. }, {
  2007. Emitter: "src1",
  2008. Message: xsql.Message{"a": 123123},
  2009. },
  2010. },
  2011. },
  2012. },
  2013. },
  2014. result: []map[string]interface{}{{
  2015. "c1": 27,
  2016. }},
  2017. },
  2018. //16
  2019. {
  2020. sql: "SELECT collect(*)[1]->sl[0] as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  2021. data: xsql.WindowTuplesSet{
  2022. Content: []xsql.WindowTuples{{
  2023. Emitter: "test",
  2024. Tuples: []xsql.Tuple{
  2025. {
  2026. Emitter: "src1",
  2027. Message: xsql.Message{"a": 53, "sl": []string{"hello", "world"}},
  2028. }, {
  2029. Emitter: "src1",
  2030. Message: xsql.Message{"a": 27, "sl": []string{"new", "horizon"}},
  2031. }, {
  2032. Emitter: "src1",
  2033. Message: xsql.Message{"a": 123123, "sl": []string{"south", "africa"}},
  2034. },
  2035. },
  2036. },
  2037. },
  2038. },
  2039. result: []map[string]interface{}{{
  2040. "c1": "new",
  2041. }},
  2042. },
  2043. //17
  2044. {
  2045. sql: "SELECT deduplicate(id, true) as r1 FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  2046. data: xsql.GroupedTuplesSet{
  2047. {
  2048. Content: []xsql.DataValuer{
  2049. &xsql.JoinTuple{
  2050. Tuples: []xsql.Tuple{
  2051. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  2052. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  2053. },
  2054. },
  2055. &xsql.JoinTuple{
  2056. Tuples: []xsql.Tuple{
  2057. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  2058. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  2059. },
  2060. },
  2061. },
  2062. },
  2063. {
  2064. Content: []xsql.DataValuer{
  2065. &xsql.JoinTuple{
  2066. Tuples: []xsql.Tuple{
  2067. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  2068. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  2069. },
  2070. },
  2071. &xsql.JoinTuple{
  2072. Tuples: []xsql.Tuple{
  2073. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  2074. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  2075. },
  2076. },
  2077. },
  2078. },
  2079. },
  2080. result: []map[string]interface{}{
  2081. {
  2082. "r1": []interface{}{
  2083. xsql.Message{"a": 122.33, "c": 2, "color": "w2", "id": 1, "r": 122},
  2084. xsql.Message{"a": 177.51, "color": "w2", "id": 5}},
  2085. }, {
  2086. "r1": []interface{}{
  2087. xsql.Message{"a": 89.03, "c": 2, "color": "w1", "id": 2, "r": 89},
  2088. xsql.Message{"a": 14.6, "color": "w1", "id": 4}},
  2089. },
  2090. },
  2091. },
  2092. //18
  2093. {
  2094. sql: "SELECT deduplicate(a, false)->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  2095. data: xsql.WindowTuplesSet{
  2096. Content: []xsql.WindowTuples{{
  2097. Emitter: "test",
  2098. Tuples: []xsql.Tuple{
  2099. {
  2100. Emitter: "src1",
  2101. Message: xsql.Message{"a": 53, "s": 123203},
  2102. }, {
  2103. Emitter: "src1",
  2104. Message: xsql.Message{"a": 27},
  2105. }, {
  2106. Emitter: "src1",
  2107. Message: xsql.Message{"a": 123123},
  2108. },
  2109. },
  2110. },
  2111. },
  2112. },
  2113. result: []map[string]interface{}{{
  2114. "c1": 123123,
  2115. }},
  2116. },
  2117. //19
  2118. {
  2119. sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  2120. data: xsql.WindowTuplesSet{
  2121. Content: []xsql.WindowTuples{{
  2122. Emitter: "test",
  2123. Tuples: []xsql.Tuple{
  2124. {
  2125. Emitter: "src1",
  2126. Message: xsql.Message{"a": 53, "s": 123203},
  2127. }, {
  2128. Emitter: "src1",
  2129. Message: xsql.Message{"a": 27},
  2130. }, {
  2131. Emitter: "src1",
  2132. Message: xsql.Message{"a": 53},
  2133. },
  2134. },
  2135. },
  2136. },
  2137. },
  2138. result: []map[string]interface{}{{}},
  2139. },
  2140. //20
  2141. {
  2142. sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  2143. data: xsql.WindowTuplesSet{
  2144. Content: []xsql.WindowTuples{{
  2145. Emitter: "test",
  2146. Tuples: []xsql.Tuple{
  2147. {
  2148. Emitter: "src1",
  2149. Message: xsql.Message{"a": 53, "s": 123203},
  2150. }, {
  2151. Emitter: "src1",
  2152. Message: xsql.Message{"a": 27},
  2153. }, {
  2154. Emitter: "src1",
  2155. Message: xsql.Message{"a": 53},
  2156. },
  2157. },
  2158. },
  2159. },
  2160. },
  2161. result: []map[string]interface{}{{}},
  2162. },
  2163. //21 when got column after group by operation, return the first tuple's column
  2164. {
  2165. sql: "SELECT A.module, A.topic , max(A.value), B.topic as var2, max(B.value) as max2, C.topic as var3, max(C.value) as max3 FROM A FULL JOIN B on A.module=B.module FULL JOIN C on A.module=C.module GROUP BY A.module, TUMBLINGWINDOW(ss, 10)",
  2166. data: xsql.GroupedTuplesSet{
  2167. {
  2168. Content: []xsql.DataValuer{
  2169. &xsql.JoinTuple{
  2170. Tuples: []xsql.Tuple{
  2171. {Emitter: "B", Message: xsql.Message{"module": 1, "topic": "moduleB topic", "value": 1}},
  2172. },
  2173. },
  2174. &xsql.JoinTuple{
  2175. Tuples: []xsql.Tuple{
  2176. {Emitter: "C", Message: xsql.Message{"module": 1, "topic": "moduleC topic", "value": 100}},
  2177. },
  2178. },
  2179. },
  2180. },
  2181. },
  2182. result: []map[string]interface{}{{
  2183. "var2": "moduleB topic",
  2184. "max2": 1,
  2185. "max3": 100,
  2186. }},
  2187. },
  2188. }
  2189. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  2190. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_AggFuncs")
  2191. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  2192. for i, tt := range tests {
  2193. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  2194. if err != nil {
  2195. t.Error(err)
  2196. }
  2197. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: true}
  2198. fv, afv := xsql.NewFunctionValuersForOp(nil)
  2199. result := pp.Apply(ctx, tt.data, fv, afv)
  2200. if !reflect.DeepEqual(tt.result, result) {
  2201. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  2202. }
  2203. }
  2204. }
  2205. func TestProjectPlanError(t *testing.T) {
  2206. var tests = []struct {
  2207. sql string
  2208. data interface{}
  2209. result interface{}
  2210. }{
  2211. //0
  2212. {
  2213. sql: "SELECT a FROM test",
  2214. data: errors.New("an error from upstream"),
  2215. result: errors.New("an error from upstream"),
  2216. },
  2217. //1
  2218. {
  2219. sql: "SELECT a * 5 FROM test",
  2220. data: &xsql.Tuple{
  2221. Emitter: "test",
  2222. Message: xsql.Message{
  2223. "a": "val_a",
  2224. },
  2225. },
  2226. result: errors.New("run Select error: invalid operation string(val_a) * int64(5)"),
  2227. },
  2228. //2
  2229. {
  2230. sql: `SELECT a[0]->b AS ab FROM test`,
  2231. data: &xsql.Tuple{
  2232. Emitter: "test",
  2233. Message: xsql.Message{
  2234. "a": "common string",
  2235. },
  2236. },
  2237. result: errors.New("run Select error: invalid operation string(common string) [] *xsql.BracketEvalResult(&{0 0})"),
  2238. },
  2239. //3
  2240. {
  2241. sql: `SELECT round(a) as r FROM test`,
  2242. data: &xsql.Tuple{
  2243. Emitter: "test",
  2244. Message: xsql.Message{
  2245. "a": "common string",
  2246. },
  2247. },
  2248. result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
  2249. },
  2250. //4
  2251. {
  2252. sql: `SELECT round(a) as r FROM test`,
  2253. data: &xsql.Tuple{
  2254. Emitter: "test",
  2255. Message: xsql.Message{
  2256. "abc": "common string",
  2257. },
  2258. },
  2259. result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
  2260. },
  2261. //5
  2262. {
  2263. sql: "SELECT avg(a) as avg FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  2264. data: xsql.GroupedTuplesSet{
  2265. {
  2266. Content: []xsql.DataValuer{
  2267. &xsql.JoinTuple{
  2268. Tuples: []xsql.Tuple{
  2269. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  2270. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  2271. },
  2272. },
  2273. &xsql.JoinTuple{
  2274. Tuples: []xsql.Tuple{
  2275. {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
  2276. {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  2277. },
  2278. },
  2279. &xsql.JoinTuple{
  2280. Tuples: []xsql.Tuple{
  2281. {Emitter: "test", Message: xsql.Message{"id": 4, "a": "dde"}},
  2282. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
  2283. },
  2284. },
  2285. &xsql.JoinTuple{
  2286. Tuples: []xsql.Tuple{
  2287. {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
  2288. {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  2289. },
  2290. },
  2291. },
  2292. },
  2293. {
  2294. Content: []xsql.DataValuer{
  2295. &xsql.JoinTuple{
  2296. Tuples: []xsql.Tuple{
  2297. {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  2298. {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  2299. },
  2300. },
  2301. &xsql.JoinTuple{
  2302. Tuples: []xsql.Tuple{
  2303. {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  2304. {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  2305. },
  2306. },
  2307. },
  2308. },
  2309. },
  2310. result: errors.New("run Select error: call func avg error: requires float64 but found string(dde)"),
  2311. },
  2312. //6
  2313. {
  2314. sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
  2315. data: xsql.WindowTuplesSet{
  2316. Content: []xsql.WindowTuples{{
  2317. Emitter: "test",
  2318. Tuples: []xsql.Tuple{
  2319. {
  2320. Emitter: "src1",
  2321. Message: xsql.Message{"a": 53},
  2322. }, {
  2323. Emitter: "src1",
  2324. Message: xsql.Message{"a": "ddd"},
  2325. }, {
  2326. Emitter: "src1",
  2327. Message: xsql.Message{"a": 123123},
  2328. },
  2329. },
  2330. },
  2331. },
  2332. },
  2333. result: errors.New("run Select error: call func sum error: requires int but found string(ddd)"),
  2334. },
  2335. //7
  2336. {
  2337. sql: `SELECT a[0]->b AS ab FROM test`,
  2338. data: &xsql.Tuple{
  2339. Emitter: "test",
  2340. Message: xsql.Message{
  2341. "a": []map[string]interface{}(nil),
  2342. },
  2343. },
  2344. result: errors.New("run Select error: out of index: 0 of 0"),
  2345. },
  2346. }
  2347. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  2348. contextLogger := conf.Log.WithField("rule", "TestProjectPlanError")
  2349. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  2350. for i, tt := range tests {
  2351. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  2352. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
  2353. fv, afv := xsql.NewFunctionValuersForOp(nil)
  2354. result := pp.Apply(ctx, tt.data, fv, afv)
  2355. if !reflect.DeepEqual(tt.result, result) {
  2356. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  2357. }
  2358. }
  2359. }