project_test.go 59 KB

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