project_test.go 59 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255
  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.WindowTuples{
  598. Content: []xsql.TupleRow{
  599. &xsql.Tuple{Emitter: "src1",
  600. Message: xsql.Message{"id1": 1, "f1": "v1"},
  601. }, &xsql.Tuple{Emitter: "src1",
  602. Message: xsql.Message{"id1": 2, "f1": "v2"},
  603. }, &xsql.Tuple{Emitter: "src1",
  604. Message: xsql.Message{"id1": 3, "f1": "v1"},
  605. },
  606. },
  607. },
  608. result: []map[string]interface{}{{
  609. "id1": 1,
  610. }, {
  611. "id1": 2,
  612. }, {
  613. "id1": 3,
  614. }},
  615. },
  616. //3
  617. {
  618. sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  619. data: &xsql.WindowTuples{
  620. Content: []xsql.TupleRow{
  621. &xsql.Tuple{Emitter: "src1",
  622. Message: xsql.Message{"id1": 1, "f1": "v1"},
  623. }, &xsql.Tuple{Emitter: "src1",
  624. Message: xsql.Message{"id2": 2, "f1": "v2"},
  625. }, &xsql.Tuple{Emitter: "src1",
  626. Message: xsql.Message{"id1": 3, "f1": "v1"},
  627. },
  628. },
  629. },
  630. result: []map[string]interface{}{{
  631. "id1": 1,
  632. }, {}, {
  633. "id1": 3,
  634. }},
  635. },
  636. //4
  637. {
  638. sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  639. data: &xsql.WindowTuples{
  640. Content: []xsql.TupleRow{
  641. &xsql.Tuple{Emitter: "src1",
  642. Message: xsql.Message{"id1": 1, "f1": "v1"},
  643. }, &xsql.Tuple{Emitter: "src1",
  644. Message: xsql.Message{"id1": 2, "f1": "v2"},
  645. }, &xsql.Tuple{Emitter: "src1",
  646. Message: xsql.Message{"id1": 3, "f1": "v1"},
  647. },
  648. },
  649. },
  650. result: []map[string]interface{}{{
  651. "id1": 1,
  652. "f1": "v1",
  653. }, {
  654. "id1": 2,
  655. "f1": "v2",
  656. }, {
  657. "id1": 3,
  658. "f1": "v1",
  659. }},
  660. },
  661. //5
  662. {
  663. sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  664. data: &xsql.WindowTuples{
  665. Content: []xsql.TupleRow{
  666. &xsql.Tuple{Emitter: "src1",
  667. Message: xsql.Message{"id1": 1, "f1": "v1"},
  668. }, &xsql.Tuple{Emitter: "src1",
  669. Message: xsql.Message{"id2": 2, "f2": "v2"},
  670. }, &xsql.Tuple{Emitter: "src1",
  671. Message: xsql.Message{"id1": 3, "f1": "v1"},
  672. },
  673. },
  674. },
  675. result: []map[string]interface{}{{
  676. "id1": 1,
  677. "f1": "v1",
  678. }, {
  679. "id2": 2,
  680. "f2": "v2",
  681. }, {
  682. "id1": 3,
  683. "f1": "v1",
  684. }},
  685. },
  686. //6
  687. {
  688. sql: "SELECT src1.* FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  689. data: &xsql.WindowTuples{
  690. Content: []xsql.TupleRow{
  691. &xsql.Tuple{Emitter: "src1",
  692. Message: xsql.Message{"id1": 1, "f1": "v1"},
  693. }, &xsql.Tuple{Emitter: "src1",
  694. Message: xsql.Message{"id1": 2, "f1": "v2"},
  695. }, &xsql.Tuple{Emitter: "src1",
  696. Message: xsql.Message{"id1": 3, "f1": "v1"},
  697. },
  698. },
  699. },
  700. result: []map[string]interface{}{{
  701. "id1": 1,
  702. "f1": "v1",
  703. }, {
  704. "id1": 2,
  705. "f1": "v2",
  706. }, {
  707. "id1": 3,
  708. "f1": "v1",
  709. }},
  710. },
  711. //7
  712. {
  713. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  714. data: &xsql.JoinTuples{
  715. Content: []*xsql.JoinTuple{
  716. {
  717. Tuples: []xsql.TupleRow{
  718. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  719. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  720. },
  721. },
  722. {
  723. Tuples: []xsql.TupleRow{
  724. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  725. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  726. },
  727. },
  728. {
  729. Tuples: []xsql.TupleRow{
  730. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  731. },
  732. },
  733. },
  734. },
  735. result: []map[string]interface{}{{
  736. "id1": 1,
  737. }, {
  738. "id1": 2,
  739. }, {
  740. "id1": 3,
  741. }},
  742. },
  743. //8
  744. {
  745. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
  746. data: &xsql.JoinTuples{
  747. Content: []*xsql.JoinTuple{
  748. {
  749. Tuples: []xsql.TupleRow{
  750. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  751. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  752. },
  753. },
  754. {
  755. Tuples: []xsql.TupleRow{
  756. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  757. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  758. },
  759. },
  760. {
  761. Tuples: []xsql.TupleRow{
  762. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id2": 3, "f1": "v1"}},
  763. },
  764. },
  765. },
  766. },
  767. result: []map[string]interface{}{{
  768. "id1": 1,
  769. }, {
  770. "id1": 2,
  771. }, {}},
  772. },
  773. //9
  774. {
  775. sql: "SELECT abc FROM tbl group by abc",
  776. data: &xsql.GroupedTuplesSet{
  777. Groups: []*xsql.GroupedTuples{
  778. {
  779. Content: []xsql.TupleRow{
  780. &xsql.Tuple{Emitter: "tbl",
  781. Message: xsql.Message{
  782. "abc": int64(6),
  783. "def": "hello",
  784. },
  785. },
  786. },
  787. },
  788. },
  789. },
  790. result: []map[string]interface{}{{
  791. "abc": int64(6),
  792. }},
  793. },
  794. //10
  795. {
  796. sql: "SELECT abc FROM tbl group by abc",
  797. data: &xsql.GroupedTuplesSet{
  798. Groups: []*xsql.GroupedTuples{
  799. {
  800. Content: []xsql.TupleRow{
  801. &xsql.Tuple{Emitter: "tbl",
  802. Message: xsql.Message{
  803. "def": "hello",
  804. },
  805. },
  806. },
  807. },
  808. },
  809. },
  810. result: []map[string]interface{}{{}},
  811. },
  812. //11
  813. {
  814. sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  815. data: &xsql.GroupedTuplesSet{
  816. Groups: []*xsql.GroupedTuples{
  817. {
  818. Content: []xsql.TupleRow{
  819. &xsql.Tuple{Emitter: "src1",
  820. Message: xsql.Message{"id1": 1, "f1": "v1"},
  821. },
  822. &xsql.Tuple{Emitter: "src1",
  823. Message: xsql.Message{"id1": 3, "f1": "v1"},
  824. },
  825. },
  826. },
  827. {
  828. Content: []xsql.TupleRow{
  829. &xsql.Tuple{Emitter: "src1",
  830. Message: xsql.Message{"id1": 2, "f1": "v2"},
  831. },
  832. },
  833. },
  834. },
  835. },
  836. result: []map[string]interface{}{{
  837. "id1": 1,
  838. }, {
  839. "id1": 2,
  840. }},
  841. },
  842. //12
  843. {
  844. sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  845. data: &xsql.GroupedTuplesSet{
  846. Groups: []*xsql.GroupedTuples{
  847. {
  848. Content: []xsql.TupleRow{
  849. &xsql.Tuple{Emitter: "src1",
  850. Message: xsql.Message{"id1": 1, "f1": "v1"},
  851. },
  852. &xsql.Tuple{Emitter: "src1",
  853. Message: xsql.Message{"id1": 3, "f1": "v1"},
  854. },
  855. },
  856. },
  857. {
  858. Content: []xsql.TupleRow{
  859. &xsql.Tuple{Emitter: "src1",
  860. Message: xsql.Message{"id2": 2, "f1": "v2"},
  861. },
  862. },
  863. },
  864. },
  865. },
  866. result: []map[string]interface{}{{
  867. "id1": 1,
  868. }, {}},
  869. },
  870. //13
  871. {
  872. sql: "SELECT src2.id2 FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  873. data: &xsql.GroupedTuplesSet{
  874. Groups: []*xsql.GroupedTuples{
  875. {
  876. Content: []xsql.TupleRow{
  877. &xsql.JoinTuple{
  878. Tuples: []xsql.TupleRow{
  879. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  880. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  881. },
  882. },
  883. },
  884. },
  885. {
  886. Content: []xsql.TupleRow{
  887. &xsql.JoinTuple{
  888. Tuples: []xsql.TupleRow{
  889. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  890. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  891. },
  892. },
  893. },
  894. },
  895. {
  896. Content: []xsql.TupleRow{
  897. &xsql.JoinTuple{
  898. Tuples: []xsql.TupleRow{
  899. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  900. },
  901. },
  902. },
  903. },
  904. },
  905. },
  906. result: []map[string]interface{}{{
  907. "id2": 2,
  908. }, {
  909. "id2": 4,
  910. }, {}},
  911. },
  912. //14
  913. {
  914. sql: "SELECT src1.*, f2 FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
  915. data: &xsql.JoinTuples{
  916. Content: []*xsql.JoinTuple{
  917. {
  918. Tuples: []xsql.TupleRow{
  919. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  920. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  921. },
  922. },
  923. {
  924. Tuples: []xsql.TupleRow{
  925. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  926. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  927. },
  928. },
  929. {
  930. Tuples: []xsql.TupleRow{
  931. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  932. },
  933. },
  934. },
  935. },
  936. result: []map[string]interface{}{{
  937. "id1": 1,
  938. "f1": "v1",
  939. "f2": "w2",
  940. }, {
  941. "id1": 2,
  942. "f1": "v2",
  943. "f2": "w3",
  944. }, {
  945. "id1": 3,
  946. "f1": "v1",
  947. }},
  948. },
  949. //15
  950. {
  951. sql: "SELECT * FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
  952. data: &xsql.JoinTuples{
  953. Content: []*xsql.JoinTuple{
  954. {
  955. Tuples: []xsql.TupleRow{
  956. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id": 1, "f1": "v1"}},
  957. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "f2": "w2"}},
  958. },
  959. },
  960. {
  961. Tuples: []xsql.TupleRow{
  962. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id": 2, "f1": "v2"}},
  963. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "f2": "w3"}},
  964. },
  965. },
  966. {
  967. Tuples: []xsql.TupleRow{
  968. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id": 3, "f1": "v1"}},
  969. },
  970. },
  971. },
  972. },
  973. result: []map[string]interface{}{{
  974. "id": 1,
  975. "f1": "v1",
  976. "f2": "w2",
  977. }, {
  978. "id": 2,
  979. "f1": "v2",
  980. "f2": "w3",
  981. }, {
  982. "id": 3,
  983. "f1": "v1",
  984. }},
  985. },
  986. //16
  987. {
  988. sql: "SELECT src1.* FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
  989. data: &xsql.GroupedTuplesSet{
  990. Groups: []*xsql.GroupedTuples{
  991. {
  992. Content: []xsql.TupleRow{
  993. &xsql.Tuple{Emitter: "src1",
  994. Message: xsql.Message{"id1": 1, "f1": "v1"},
  995. },
  996. &xsql.Tuple{Emitter: "src1",
  997. Message: xsql.Message{"id1": 3, "f1": "v1"},
  998. },
  999. },
  1000. },
  1001. {
  1002. Content: []xsql.TupleRow{
  1003. &xsql.Tuple{Emitter: "src1",
  1004. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1005. },
  1006. },
  1007. },
  1008. },
  1009. },
  1010. result: []map[string]interface{}{{
  1011. "id1": 1,
  1012. "f1": "v1",
  1013. }, {
  1014. "id1": 2,
  1015. "f1": "v2",
  1016. }},
  1017. },
  1018. //17
  1019. {
  1020. sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  1021. data: &xsql.GroupedTuplesSet{
  1022. Groups: []*xsql.GroupedTuples{
  1023. {
  1024. Content: []xsql.TupleRow{
  1025. &xsql.JoinTuple{
  1026. Tuples: []xsql.TupleRow{
  1027. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1028. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1029. },
  1030. },
  1031. },
  1032. },
  1033. {
  1034. Content: []xsql.TupleRow{
  1035. &xsql.JoinTuple{
  1036. Tuples: []xsql.TupleRow{
  1037. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1038. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1039. },
  1040. },
  1041. },
  1042. },
  1043. {
  1044. Content: []xsql.TupleRow{
  1045. &xsql.JoinTuple{
  1046. Tuples: []xsql.TupleRow{
  1047. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  1048. },
  1049. },
  1050. },
  1051. },
  1052. },
  1053. },
  1054. result: []map[string]interface{}{{
  1055. "id2": 2,
  1056. "id1": 1,
  1057. "f1": "v1",
  1058. }, {
  1059. "id2": 4,
  1060. "id1": 2,
  1061. "f1": "v2",
  1062. }, {
  1063. "id1": 3,
  1064. "f1": "v1",
  1065. }},
  1066. },
  1067. //18
  1068. {
  1069. sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
  1070. data: &xsql.GroupedTuplesSet{
  1071. Groups: []*xsql.GroupedTuples{
  1072. {
  1073. Content: []xsql.TupleRow{
  1074. &xsql.JoinTuple{
  1075. Tuples: []xsql.TupleRow{
  1076. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1077. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1078. },
  1079. },
  1080. },
  1081. },
  1082. {
  1083. Content: []xsql.TupleRow{
  1084. &xsql.JoinTuple{
  1085. Tuples: []xsql.TupleRow{
  1086. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1087. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1088. },
  1089. },
  1090. },
  1091. },
  1092. {
  1093. Content: []xsql.TupleRow{
  1094. &xsql.JoinTuple{
  1095. Tuples: []xsql.TupleRow{
  1096. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  1097. },
  1098. },
  1099. },
  1100. },
  1101. },
  1102. },
  1103. result: []map[string]interface{}{{
  1104. "id2": 2,
  1105. "id1": 1,
  1106. "f1": "v1",
  1107. }, {
  1108. "id2": 4,
  1109. "id1": 2,
  1110. "f1": "v2",
  1111. }, {
  1112. "id1": 3,
  1113. "f1": "v1",
  1114. }},
  1115. },
  1116. }
  1117. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1118. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_MultiInput")
  1119. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1120. for i, tt := range tests {
  1121. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1122. pp := &ProjectOp{Fields: stmt.Fields}
  1123. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1124. result := pp.Apply(ctx, tt.data, fv, afv)
  1125. if !reflect.DeepEqual(tt.result, result) {
  1126. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1127. }
  1128. }
  1129. }
  1130. func TestProjectPlan_Funcs(t *testing.T) {
  1131. var tests = []struct {
  1132. sql string
  1133. data interface{}
  1134. result []map[string]interface{}
  1135. }{
  1136. //0
  1137. {
  1138. sql: "SELECT round(a) as r FROM test",
  1139. data: &xsql.Tuple{
  1140. Emitter: "test",
  1141. Message: xsql.Message{
  1142. "a": 47.5,
  1143. },
  1144. },
  1145. result: []map[string]interface{}{{
  1146. "r": float64(48),
  1147. }},
  1148. },
  1149. //1
  1150. {
  1151. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  1152. data: &xsql.WindowTuples{
  1153. Content: []xsql.TupleRow{
  1154. &xsql.Tuple{Emitter: "src1",
  1155. Message: xsql.Message{"a": 53.1},
  1156. }, &xsql.Tuple{Emitter: "src1",
  1157. Message: xsql.Message{"a": 27.4},
  1158. }, &xsql.Tuple{Emitter: "src1",
  1159. Message: xsql.Message{"a": 123123.7},
  1160. },
  1161. },
  1162. },
  1163. result: []map[string]interface{}{{
  1164. "r": float64(53),
  1165. }, {
  1166. "r": float64(27),
  1167. }, {
  1168. "r": float64(123124),
  1169. }},
  1170. },
  1171. //2
  1172. {
  1173. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  1174. data: &xsql.WindowTuples{
  1175. Content: []xsql.TupleRow{
  1176. &xsql.Tuple{Emitter: "src1",
  1177. Message: xsql.Message{"a": 53.1},
  1178. }, &xsql.Tuple{Emitter: "src1",
  1179. Message: xsql.Message{"a": 27.4},
  1180. }, &xsql.Tuple{Emitter: "src1",
  1181. Message: xsql.Message{"a": 123123.7},
  1182. },
  1183. },
  1184. },
  1185. result: []map[string]interface{}{{
  1186. "r": float64(53),
  1187. }, {
  1188. "r": float64(27),
  1189. }, {
  1190. "r": float64(123124),
  1191. }},
  1192. },
  1193. //3
  1194. {
  1195. sql: "SELECT round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1196. data: &xsql.JoinTuples{
  1197. Content: []*xsql.JoinTuple{
  1198. {
  1199. Tuples: []xsql.TupleRow{
  1200. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
  1201. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1202. },
  1203. },
  1204. {
  1205. Tuples: []xsql.TupleRow{
  1206. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
  1207. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1208. },
  1209. },
  1210. {
  1211. Tuples: []xsql.TupleRow{
  1212. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
  1213. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1214. },
  1215. },
  1216. },
  1217. },
  1218. result: []map[string]interface{}{{
  1219. "r": float64(66),
  1220. }, {
  1221. "r": float64(73),
  1222. }, {
  1223. "r": float64(89),
  1224. }},
  1225. },
  1226. //4
  1227. {
  1228. 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)",
  1229. data: &xsql.JoinTuples{
  1230. Content: []*xsql.JoinTuple{
  1231. {
  1232. Tuples: []xsql.TupleRow{
  1233. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
  1234. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1235. },
  1236. },
  1237. {
  1238. Tuples: []xsql.TupleRow{
  1239. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
  1240. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1241. },
  1242. },
  1243. {
  1244. Tuples: []xsql.TupleRow{
  1245. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
  1246. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1247. },
  1248. },
  1249. },
  1250. },
  1251. result: []map[string]interface{}{{
  1252. "concat": "165.5512",
  1253. }, {
  1254. "concat": "273.49934",
  1255. }, {
  1256. "concat": "388.886",
  1257. }},
  1258. },
  1259. //5
  1260. {
  1261. sql: "SELECT count(a) as r FROM test",
  1262. data: &xsql.Tuple{
  1263. Emitter: "test",
  1264. Message: xsql.Message{
  1265. "a": 47.5,
  1266. },
  1267. },
  1268. result: []map[string]interface{}{{
  1269. "r": 1,
  1270. }},
  1271. },
  1272. //6
  1273. {
  1274. sql: "SELECT meta(test.device) as d FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1275. data: &xsql.JoinTuples{
  1276. Content: []*xsql.JoinTuple{
  1277. {
  1278. Tuples: []xsql.TupleRow{
  1279. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}, Metadata: xsql.Metadata{"device": "devicea"}},
  1280. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1281. },
  1282. },
  1283. {
  1284. Tuples: []xsql.TupleRow{
  1285. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1286. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1287. },
  1288. },
  1289. {
  1290. Tuples: []xsql.TupleRow{
  1291. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}, Metadata: xsql.Metadata{"device": "devicec"}},
  1292. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1293. },
  1294. },
  1295. },
  1296. },
  1297. result: []map[string]interface{}{{
  1298. "d": "devicea",
  1299. }, {
  1300. "d": "deviceb",
  1301. }, {
  1302. "d": "devicec",
  1303. }},
  1304. },
  1305. }
  1306. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1307. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_Funcs")
  1308. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1309. for i, tt := range tests {
  1310. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1311. if err != nil {
  1312. t.Error(err)
  1313. }
  1314. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
  1315. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1316. result := pp.Apply(ctx, tt.data, fv, afv)
  1317. if !reflect.DeepEqual(tt.result, result) {
  1318. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1319. }
  1320. }
  1321. }
  1322. func TestProjectPlan_AggFuncs(t *testing.T) {
  1323. var tests = []struct {
  1324. sql string
  1325. data interface{}
  1326. result []map[string]interface{}
  1327. }{
  1328. { //0
  1329. 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",
  1330. data: &xsql.GroupedTuplesSet{
  1331. Groups: []*xsql.GroupedTuples{
  1332. {
  1333. Content: []xsql.TupleRow{
  1334. &xsql.JoinTuple{
  1335. Tuples: []xsql.TupleRow{
  1336. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1337. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1338. },
  1339. },
  1340. &xsql.JoinTuple{
  1341. Tuples: []xsql.TupleRow{
  1342. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1343. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1344. },
  1345. },
  1346. },
  1347. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1348. },
  1349. {
  1350. Content: []xsql.TupleRow{
  1351. &xsql.JoinTuple{
  1352. Tuples: []xsql.TupleRow{
  1353. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1354. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1355. },
  1356. },
  1357. &xsql.JoinTuple{
  1358. Tuples: []xsql.TupleRow{
  1359. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1360. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1361. },
  1362. },
  1363. },
  1364. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1365. },
  1366. },
  1367. },
  1368. result: []map[string]interface{}{{
  1369. "c": 2,
  1370. "r": float64(122),
  1371. "ws": int64(1541152486013),
  1372. "we": int64(1541152487013),
  1373. }, {
  1374. "c": 2,
  1375. "r": float64(89),
  1376. "ws": int64(1541152486013),
  1377. "we": int64(1541152487013),
  1378. }},
  1379. },
  1380. //1
  1381. {
  1382. 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",
  1383. data: &xsql.GroupedTuplesSet{
  1384. Groups: []*xsql.GroupedTuples{
  1385. {
  1386. Content: []xsql.TupleRow{
  1387. &xsql.JoinTuple{
  1388. Tuples: []xsql.TupleRow{
  1389. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1390. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1391. },
  1392. },
  1393. &xsql.JoinTuple{
  1394. Tuples: []xsql.TupleRow{
  1395. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5}},
  1396. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1397. },
  1398. },
  1399. },
  1400. },
  1401. {
  1402. Content: []xsql.TupleRow{
  1403. &xsql.JoinTuple{
  1404. Tuples: []xsql.TupleRow{
  1405. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1406. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1407. },
  1408. },
  1409. &xsql.JoinTuple{
  1410. Tuples: []xsql.TupleRow{
  1411. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1412. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1413. },
  1414. },
  1415. },
  1416. },
  1417. },
  1418. },
  1419. result: []map[string]interface{}{{
  1420. "c": 1,
  1421. "a": 122.33,
  1422. "s": 122.33,
  1423. "min": 122.33,
  1424. "max": 122.33,
  1425. }, {
  1426. "c": 2,
  1427. "s": 103.63,
  1428. "a": 51.815,
  1429. "min": 14.6,
  1430. "max": 89.03,
  1431. }},
  1432. },
  1433. //2
  1434. {
  1435. sql: "SELECT avg(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1436. data: &xsql.GroupedTuplesSet{
  1437. Groups: []*xsql.GroupedTuples{
  1438. {
  1439. Content: []xsql.TupleRow{
  1440. &xsql.JoinTuple{
  1441. Tuples: []xsql.TupleRow{
  1442. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1443. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1444. },
  1445. },
  1446. &xsql.JoinTuple{
  1447. Tuples: []xsql.TupleRow{
  1448. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
  1449. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1450. },
  1451. },
  1452. &xsql.JoinTuple{
  1453. Tuples: []xsql.TupleRow{
  1454. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 98.31}},
  1455. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
  1456. },
  1457. },
  1458. &xsql.JoinTuple{
  1459. Tuples: []xsql.TupleRow{
  1460. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
  1461. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1462. },
  1463. },
  1464. },
  1465. },
  1466. {
  1467. Content: []xsql.TupleRow{
  1468. &xsql.JoinTuple{
  1469. Tuples: []xsql.TupleRow{
  1470. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1471. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1472. },
  1473. },
  1474. &xsql.JoinTuple{
  1475. Tuples: []xsql.TupleRow{
  1476. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1477. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1478. },
  1479. },
  1480. },
  1481. },
  1482. },
  1483. },
  1484. result: []map[string]interface{}{{
  1485. "avg": 116.68,
  1486. }, {
  1487. "avg": 51.815,
  1488. }},
  1489. },
  1490. //3
  1491. {
  1492. sql: "SELECT max(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1493. data: &xsql.GroupedTuplesSet{
  1494. Groups: []*xsql.GroupedTuples{
  1495. {
  1496. Content: []xsql.TupleRow{
  1497. &xsql.JoinTuple{
  1498. Tuples: []xsql.TupleRow{
  1499. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1500. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1501. },
  1502. },
  1503. &xsql.JoinTuple{
  1504. Tuples: []xsql.TupleRow{
  1505. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1506. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1507. },
  1508. },
  1509. &xsql.JoinTuple{
  1510. Tuples: []xsql.TupleRow{
  1511. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1512. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1513. },
  1514. },
  1515. },
  1516. },
  1517. {
  1518. Content: []xsql.TupleRow{
  1519. &xsql.JoinTuple{
  1520. Tuples: []xsql.TupleRow{
  1521. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1522. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1523. },
  1524. },
  1525. &xsql.JoinTuple{
  1526. Tuples: []xsql.TupleRow{
  1527. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1528. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1529. },
  1530. },
  1531. },
  1532. },
  1533. },
  1534. },
  1535. result: []map[string]interface{}{{
  1536. "max": 177.51,
  1537. }, {
  1538. "max": 89.03,
  1539. }},
  1540. },
  1541. //4
  1542. {
  1543. sql: "SELECT min(a), window_start(), window_end() FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1544. data: &xsql.JoinTuples{
  1545. Content: []*xsql.JoinTuple{
  1546. {
  1547. Tuples: []xsql.TupleRow{
  1548. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1549. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1550. },
  1551. },
  1552. {
  1553. Tuples: []xsql.TupleRow{
  1554. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1555. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1556. },
  1557. },
  1558. {
  1559. Tuples: []xsql.TupleRow{
  1560. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1561. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1562. },
  1563. },
  1564. },
  1565. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1566. },
  1567. result: []map[string]interface{}{{
  1568. "min": 68.55,
  1569. "window_start": int64(1541152486013),
  1570. "window_end": int64(1541152487013),
  1571. }},
  1572. },
  1573. //5
  1574. {
  1575. 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)",
  1576. data: &xsql.JoinTuples{
  1577. Content: []*xsql.JoinTuple{
  1578. {
  1579. Tuples: []xsql.TupleRow{
  1580. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1}},
  1581. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1582. },
  1583. },
  1584. {
  1585. Tuples: []xsql.TupleRow{
  1586. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1587. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1588. },
  1589. },
  1590. {
  1591. Tuples: []xsql.TupleRow{
  1592. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1593. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1594. },
  1595. },
  1596. },
  1597. },
  1598. result: []map[string]interface{}{{
  1599. "all": 3,
  1600. "c": 2,
  1601. "a": 123.03,
  1602. "s": 246.06,
  1603. "min": 68.55,
  1604. "max": 177.51,
  1605. }},
  1606. },
  1607. //6
  1608. {
  1609. sql: "SELECT sum(a), window_start() as ws, window_end() FROM test GROUP BY TumblingWindow(ss, 10)",
  1610. data: &xsql.WindowTuples{
  1611. Content: []xsql.TupleRow{
  1612. &xsql.Tuple{Emitter: "src1",
  1613. Message: xsql.Message{"a": 53},
  1614. }, &xsql.Tuple{Emitter: "src1",
  1615. Message: xsql.Message{"a": 27},
  1616. }, &xsql.Tuple{Emitter: "src1",
  1617. Message: xsql.Message{"a": 123123},
  1618. },
  1619. },
  1620. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1621. },
  1622. result: []map[string]interface{}{{
  1623. "sum": 123203,
  1624. "ws": int64(1541152486013),
  1625. "window_end": int64(1541152487013),
  1626. }},
  1627. },
  1628. //7
  1629. {
  1630. sql: "SELECT sum(a) as s FROM test GROUP BY TumblingWindow(ss, 10)",
  1631. data: &xsql.WindowTuples{
  1632. Content: []xsql.TupleRow{
  1633. &xsql.Tuple{Emitter: "src1",
  1634. Message: xsql.Message{"a": 53, "s": 123203},
  1635. }, &xsql.Tuple{Emitter: "src1",
  1636. Message: xsql.Message{"a": 27},
  1637. }, &xsql.Tuple{Emitter: "src1",
  1638. Message: xsql.Message{"a": 123123},
  1639. },
  1640. },
  1641. },
  1642. result: []map[string]interface{}{{
  1643. "s": 123203,
  1644. }},
  1645. },
  1646. //8
  1647. {
  1648. sql: "SELECT sum(a) FROM test GROUP BY TumblingWindow(ss, 10)",
  1649. data: &xsql.WindowTuples{
  1650. Content: []xsql.TupleRow{
  1651. &xsql.Tuple{Emitter: "src1",
  1652. Message: xsql.Message{"a": 53},
  1653. }, &xsql.Tuple{Emitter: "src1",
  1654. Message: xsql.Message{"a": 27},
  1655. }, &xsql.Tuple{Emitter: "src1",
  1656. Message: xsql.Message{"a": 123123},
  1657. },
  1658. },
  1659. },
  1660. result: []map[string]interface{}{{
  1661. "sum": 123203,
  1662. }},
  1663. },
  1664. //9
  1665. {
  1666. 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)",
  1667. data: &xsql.WindowTuples{
  1668. Content: []xsql.TupleRow{
  1669. &xsql.Tuple{Emitter: "src1",
  1670. Message: xsql.Message{"a": 53},
  1671. }, &xsql.Tuple{Emitter: "src1",
  1672. Message: xsql.Message{"a": 27},
  1673. }, &xsql.Tuple{Emitter: "src1",
  1674. Message: xsql.Message{"s": 123123},
  1675. },
  1676. },
  1677. },
  1678. result: []map[string]interface{}{{
  1679. "all": 3,
  1680. "c": 2,
  1681. "a": 40,
  1682. "s": 80,
  1683. "min": 27,
  1684. "max": 53,
  1685. }},
  1686. },
  1687. //10
  1688. {
  1689. sql: "SELECT count(*), meta(test1.device) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1690. data: &xsql.GroupedTuplesSet{
  1691. Groups: []*xsql.GroupedTuples{
  1692. {
  1693. Content: []xsql.TupleRow{
  1694. &xsql.JoinTuple{
  1695. Tuples: []xsql.TupleRow{
  1696. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1697. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
  1698. },
  1699. },
  1700. &xsql.JoinTuple{
  1701. Tuples: []xsql.TupleRow{
  1702. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1703. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1704. },
  1705. },
  1706. },
  1707. },
  1708. {
  1709. Content: []xsql.TupleRow{
  1710. &xsql.JoinTuple{
  1711. Tuples: []xsql.TupleRow{
  1712. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1713. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
  1714. },
  1715. },
  1716. &xsql.JoinTuple{
  1717. Tuples: []xsql.TupleRow{
  1718. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1719. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
  1720. },
  1721. },
  1722. },
  1723. },
  1724. },
  1725. },
  1726. result: []map[string]interface{}{{
  1727. "count": 2,
  1728. "meta": "devicea",
  1729. }, {
  1730. "count": 2,
  1731. "meta": "devicec",
  1732. }},
  1733. },
  1734. //11
  1735. {
  1736. 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",
  1737. data: &xsql.GroupedTuplesSet{
  1738. Groups: []*xsql.GroupedTuples{
  1739. {
  1740. Content: []xsql.TupleRow{
  1741. &xsql.JoinTuple{
  1742. Tuples: []xsql.TupleRow{
  1743. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "d": "devicea"}},
  1744. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
  1745. },
  1746. },
  1747. &xsql.JoinTuple{
  1748. Tuples: []xsql.TupleRow{
  1749. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1750. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1751. },
  1752. },
  1753. },
  1754. },
  1755. {
  1756. Content: []xsql.TupleRow{
  1757. &xsql.JoinTuple{
  1758. Tuples: []xsql.TupleRow{
  1759. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "d": "devicec"}},
  1760. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
  1761. },
  1762. },
  1763. &xsql.JoinTuple{
  1764. Tuples: []xsql.TupleRow{
  1765. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1766. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
  1767. },
  1768. },
  1769. },
  1770. },
  1771. },
  1772. },
  1773. result: []map[string]interface{}{{
  1774. "c": 2,
  1775. "d": "devicea",
  1776. }, {
  1777. "c": 2,
  1778. "d": "devicec",
  1779. }},
  1780. },
  1781. //12
  1782. {
  1783. sql: "SELECT * FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1784. data: &xsql.GroupedTuplesSet{
  1785. Groups: []*xsql.GroupedTuples{
  1786. {
  1787. Content: []xsql.TupleRow{
  1788. &xsql.JoinTuple{
  1789. Tuples: []xsql.TupleRow{
  1790. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1791. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1792. },
  1793. },
  1794. &xsql.JoinTuple{
  1795. Tuples: []xsql.TupleRow{
  1796. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1797. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1798. },
  1799. },
  1800. },
  1801. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1802. },
  1803. {
  1804. Content: []xsql.TupleRow{
  1805. &xsql.JoinTuple{
  1806. Tuples: []xsql.TupleRow{
  1807. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1808. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1809. },
  1810. },
  1811. &xsql.JoinTuple{
  1812. Tuples: []xsql.TupleRow{
  1813. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1814. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1815. },
  1816. },
  1817. },
  1818. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1819. },
  1820. },
  1821. },
  1822. result: []map[string]interface{}{{
  1823. "a": 122.33,
  1824. "c": 2,
  1825. "color": "w2",
  1826. "id": 1,
  1827. "r": 122,
  1828. }, {
  1829. "a": 89.03,
  1830. "c": 2,
  1831. "color": "w1",
  1832. "id": 2,
  1833. "r": 89,
  1834. }},
  1835. },
  1836. //13
  1837. {
  1838. sql: "SELECT collect(a) as r1 FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1839. data: &xsql.GroupedTuplesSet{
  1840. Groups: []*xsql.GroupedTuples{
  1841. {
  1842. Content: []xsql.TupleRow{
  1843. &xsql.JoinTuple{
  1844. Tuples: []xsql.TupleRow{
  1845. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1846. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1847. },
  1848. },
  1849. &xsql.JoinTuple{
  1850. Tuples: []xsql.TupleRow{
  1851. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1852. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1853. },
  1854. },
  1855. },
  1856. },
  1857. {
  1858. Content: []xsql.TupleRow{
  1859. &xsql.JoinTuple{
  1860. Tuples: []xsql.TupleRow{
  1861. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1862. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1863. },
  1864. },
  1865. &xsql.JoinTuple{
  1866. Tuples: []xsql.TupleRow{
  1867. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1868. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1869. },
  1870. },
  1871. },
  1872. },
  1873. },
  1874. },
  1875. result: []map[string]interface{}{{
  1876. "r1": []interface{}{122.33, 177.51},
  1877. }, {"r1": []interface{}{89.03, 14.6}}},
  1878. },
  1879. //14
  1880. {
  1881. sql: "SELECT collect(*)[1] as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1882. data: &xsql.WindowTuples{
  1883. Content: []xsql.TupleRow{
  1884. &xsql.Tuple{Emitter: "src1",
  1885. Message: xsql.Message{"a": 53, "s": 123203},
  1886. }, &xsql.Tuple{Emitter: "src1",
  1887. Message: xsql.Message{"a": 27},
  1888. }, &xsql.Tuple{Emitter: "src1",
  1889. Message: xsql.Message{"a": 123123},
  1890. },
  1891. },
  1892. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1893. },
  1894. result: []map[string]interface{}{{
  1895. "c1": xsql.Message{
  1896. "a": 27,
  1897. },
  1898. }},
  1899. },
  1900. //15
  1901. {
  1902. sql: "SELECT collect(*)[1]->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1903. data: &xsql.WindowTuples{
  1904. Content: []xsql.TupleRow{
  1905. &xsql.Tuple{Emitter: "src1",
  1906. Message: xsql.Message{"a": 53, "s": 123203},
  1907. }, &xsql.Tuple{Emitter: "src1",
  1908. Message: xsql.Message{"a": 27},
  1909. }, &xsql.Tuple{Emitter: "src1",
  1910. Message: xsql.Message{"a": 123123},
  1911. },
  1912. },
  1913. },
  1914. result: []map[string]interface{}{{
  1915. "c1": 27,
  1916. }},
  1917. },
  1918. //16
  1919. {
  1920. sql: "SELECT collect(*)[1]->sl[0] as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1921. data: &xsql.WindowTuples{
  1922. Content: []xsql.TupleRow{
  1923. &xsql.Tuple{Emitter: "src1",
  1924. Message: xsql.Message{"a": 53, "sl": []string{"hello", "world"}},
  1925. }, &xsql.Tuple{Emitter: "src1",
  1926. Message: xsql.Message{"a": 27, "sl": []string{"new", "horizon"}},
  1927. }, &xsql.Tuple{Emitter: "src1",
  1928. Message: xsql.Message{"a": 123123, "sl": []string{"south", "africa"}},
  1929. },
  1930. },
  1931. },
  1932. result: []map[string]interface{}{{
  1933. "c1": "new",
  1934. }},
  1935. },
  1936. //17
  1937. {
  1938. sql: "SELECT deduplicate(id, true) as r1 FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1939. data: &xsql.GroupedTuplesSet{
  1940. Groups: []*xsql.GroupedTuples{
  1941. {
  1942. Content: []xsql.TupleRow{
  1943. &xsql.JoinTuple{
  1944. Tuples: []xsql.TupleRow{
  1945. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1946. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1947. },
  1948. },
  1949. &xsql.JoinTuple{
  1950. Tuples: []xsql.TupleRow{
  1951. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1952. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1953. },
  1954. },
  1955. },
  1956. },
  1957. {
  1958. Content: []xsql.TupleRow{
  1959. &xsql.JoinTuple{
  1960. Tuples: []xsql.TupleRow{
  1961. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1962. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1963. },
  1964. },
  1965. &xsql.JoinTuple{
  1966. Tuples: []xsql.TupleRow{
  1967. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1968. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1969. },
  1970. },
  1971. },
  1972. },
  1973. },
  1974. },
  1975. result: []map[string]interface{}{
  1976. {
  1977. "r1": []interface{}{
  1978. xsql.Message{"a": 122.33, "c": 2, "color": "w2", "id": 1, "r": 122},
  1979. xsql.Message{"a": 177.51, "color": "w2", "id": 5}},
  1980. }, {
  1981. "r1": []interface{}{
  1982. xsql.Message{"a": 89.03, "c": 2, "color": "w1", "id": 2, "r": 89},
  1983. xsql.Message{"a": 14.6, "color": "w1", "id": 4}},
  1984. },
  1985. },
  1986. },
  1987. //18
  1988. {
  1989. sql: "SELECT deduplicate(a, false)->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1990. data: &xsql.WindowTuples{
  1991. Content: []xsql.TupleRow{
  1992. &xsql.Tuple{Emitter: "src1",
  1993. Message: xsql.Message{"a": 53, "s": 123203},
  1994. }, &xsql.Tuple{Emitter: "src1",
  1995. Message: xsql.Message{"a": 27},
  1996. }, &xsql.Tuple{Emitter: "src1",
  1997. Message: xsql.Message{"a": 123123},
  1998. },
  1999. },
  2000. },
  2001. result: []map[string]interface{}{{
  2002. "c1": 123123,
  2003. }},
  2004. },
  2005. //19
  2006. {
  2007. sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  2008. data: &xsql.WindowTuples{
  2009. Content: []xsql.TupleRow{
  2010. &xsql.Tuple{Emitter: "src1",
  2011. Message: xsql.Message{"a": 53, "s": 123203},
  2012. }, &xsql.Tuple{Emitter: "src1",
  2013. Message: xsql.Message{"a": 27},
  2014. }, &xsql.Tuple{Emitter: "src1",
  2015. Message: xsql.Message{"a": 53},
  2016. },
  2017. },
  2018. },
  2019. result: []map[string]interface{}{{}},
  2020. },
  2021. //20
  2022. {
  2023. sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  2024. data: &xsql.WindowTuples{
  2025. Content: []xsql.TupleRow{
  2026. &xsql.Tuple{Emitter: "src1",
  2027. Message: xsql.Message{"a": 53, "s": 123203},
  2028. }, &xsql.Tuple{Emitter: "src1",
  2029. Message: xsql.Message{"a": 27},
  2030. }, &xsql.Tuple{Emitter: "src1",
  2031. Message: xsql.Message{"a": 53},
  2032. },
  2033. },
  2034. },
  2035. result: []map[string]interface{}{{}},
  2036. },
  2037. //21 when got column after group by operation, return the first tuple's column
  2038. {
  2039. 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)",
  2040. data: &xsql.GroupedTuplesSet{
  2041. Groups: []*xsql.GroupedTuples{
  2042. {
  2043. Content: []xsql.TupleRow{
  2044. &xsql.JoinTuple{
  2045. Tuples: []xsql.TupleRow{
  2046. &xsql.Tuple{Emitter: "B", Message: xsql.Message{"module": 1, "topic": "moduleB topic", "value": 1}},
  2047. },
  2048. },
  2049. &xsql.JoinTuple{
  2050. Tuples: []xsql.TupleRow{
  2051. &xsql.Tuple{Emitter: "C", Message: xsql.Message{"module": 1, "topic": "moduleC topic", "value": 100}},
  2052. },
  2053. },
  2054. },
  2055. },
  2056. },
  2057. },
  2058. result: []map[string]interface{}{{
  2059. "var2": "moduleB topic",
  2060. "max2": 1,
  2061. "max3": 100,
  2062. }},
  2063. },
  2064. }
  2065. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  2066. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_AggFuncs")
  2067. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  2068. for i, tt := range tests {
  2069. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  2070. if err != nil {
  2071. t.Error(err)
  2072. }
  2073. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: true}
  2074. fv, afv := xsql.NewFunctionValuersForOp(nil)
  2075. result := pp.Apply(ctx, tt.data, fv, afv)
  2076. if !reflect.DeepEqual(tt.result, result) {
  2077. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  2078. }
  2079. }
  2080. }
  2081. func TestProjectPlanError(t *testing.T) {
  2082. var tests = []struct {
  2083. sql string
  2084. data interface{}
  2085. result interface{}
  2086. }{
  2087. //0
  2088. {
  2089. sql: "SELECT a FROM test",
  2090. data: errors.New("an error from upstream"),
  2091. result: errors.New("an error from upstream"),
  2092. },
  2093. //1
  2094. {
  2095. sql: "SELECT a * 5 FROM test",
  2096. data: &xsql.Tuple{
  2097. Emitter: "test",
  2098. Message: xsql.Message{
  2099. "a": "val_a",
  2100. },
  2101. },
  2102. result: errors.New("run Select error: invalid operation string(val_a) * int64(5)"),
  2103. },
  2104. //2
  2105. {
  2106. sql: `SELECT a[0]->b AS ab FROM test`,
  2107. data: &xsql.Tuple{
  2108. Emitter: "test",
  2109. Message: xsql.Message{
  2110. "a": "common string",
  2111. },
  2112. },
  2113. result: errors.New("run Select error: invalid operation string(common string) [] *xsql.BracketEvalResult(&{0 0})"),
  2114. },
  2115. //3
  2116. {
  2117. sql: `SELECT round(a) as r FROM test`,
  2118. data: &xsql.Tuple{
  2119. Emitter: "test",
  2120. Message: xsql.Message{
  2121. "a": "common string",
  2122. },
  2123. },
  2124. result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
  2125. },
  2126. //4
  2127. {
  2128. sql: `SELECT round(a) as r FROM test`,
  2129. data: &xsql.Tuple{
  2130. Emitter: "test",
  2131. Message: xsql.Message{
  2132. "abc": "common string",
  2133. },
  2134. },
  2135. result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
  2136. },
  2137. //5
  2138. {
  2139. sql: "SELECT avg(a) as avg FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  2140. data: &xsql.GroupedTuplesSet{
  2141. Groups: []*xsql.GroupedTuples{
  2142. {
  2143. Content: []xsql.TupleRow{
  2144. &xsql.JoinTuple{
  2145. Tuples: []xsql.TupleRow{
  2146. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  2147. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  2148. },
  2149. },
  2150. &xsql.JoinTuple{
  2151. Tuples: []xsql.TupleRow{
  2152. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
  2153. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  2154. },
  2155. },
  2156. &xsql.JoinTuple{
  2157. Tuples: []xsql.TupleRow{
  2158. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": "dde"}},
  2159. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
  2160. },
  2161. },
  2162. &xsql.JoinTuple{
  2163. Tuples: []xsql.TupleRow{
  2164. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
  2165. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  2166. },
  2167. },
  2168. },
  2169. },
  2170. {
  2171. Content: []xsql.TupleRow{
  2172. &xsql.JoinTuple{
  2173. Tuples: []xsql.TupleRow{
  2174. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  2175. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  2176. },
  2177. },
  2178. &xsql.JoinTuple{
  2179. Tuples: []xsql.TupleRow{
  2180. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  2181. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  2182. },
  2183. },
  2184. },
  2185. },
  2186. },
  2187. },
  2188. result: errors.New("run Select error: call func avg error: requires float64 but found string(dde)"),
  2189. },
  2190. //6
  2191. {
  2192. sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
  2193. data: &xsql.WindowTuples{
  2194. Content: []xsql.TupleRow{
  2195. &xsql.Tuple{Emitter: "src1",
  2196. Message: xsql.Message{"a": 53},
  2197. }, &xsql.Tuple{Emitter: "src1",
  2198. Message: xsql.Message{"a": "ddd"},
  2199. }, &xsql.Tuple{Emitter: "src1",
  2200. Message: xsql.Message{"a": 123123},
  2201. },
  2202. },
  2203. },
  2204. result: errors.New("run Select error: call func sum error: requires int but found string(ddd)"),
  2205. },
  2206. //7
  2207. {
  2208. sql: `SELECT a[0]->b AS ab FROM test`,
  2209. data: &xsql.Tuple{
  2210. Emitter: "test",
  2211. Message: xsql.Message{
  2212. "a": []map[string]interface{}(nil),
  2213. },
  2214. },
  2215. result: errors.New("run Select error: out of index: 0 of 0"),
  2216. },
  2217. }
  2218. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  2219. contextLogger := conf.Log.WithField("rule", "TestProjectPlanError")
  2220. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  2221. for i, tt := range tests {
  2222. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  2223. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
  2224. fv, afv := xsql.NewFunctionValuersForOp(nil)
  2225. result := pp.Apply(ctx, tt.data, fv, afv)
  2226. if !reflect.DeepEqual(tt.result, result) {
  2227. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  2228. }
  2229. }
  2230. }