project_test.go 59 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259
  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. "f2": "w2",
  1059. }, {
  1060. "id2": 4,
  1061. "id1": 2,
  1062. "f1": "v2",
  1063. "f2": "w3",
  1064. }, {
  1065. "id1": 3,
  1066. "f1": "v1",
  1067. }},
  1068. },
  1069. //18
  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. Groups: []*xsql.GroupedTuples{
  1074. {
  1075. Content: []xsql.TupleRow{
  1076. &xsql.JoinTuple{
  1077. Tuples: []xsql.TupleRow{
  1078. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1079. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1080. },
  1081. },
  1082. },
  1083. },
  1084. {
  1085. Content: []xsql.TupleRow{
  1086. &xsql.JoinTuple{
  1087. Tuples: []xsql.TupleRow{
  1088. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1089. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1090. },
  1091. },
  1092. },
  1093. },
  1094. {
  1095. Content: []xsql.TupleRow{
  1096. &xsql.JoinTuple{
  1097. Tuples: []xsql.TupleRow{
  1098. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
  1099. },
  1100. },
  1101. },
  1102. },
  1103. },
  1104. },
  1105. result: []map[string]interface{}{{
  1106. "id2": 2,
  1107. "id1": 1,
  1108. "f1": "v1",
  1109. "f2": "w2",
  1110. }, {
  1111. "id2": 4,
  1112. "id1": 2,
  1113. "f1": "v2",
  1114. "f2": "w3",
  1115. }, {
  1116. "id1": 3,
  1117. "f1": "v1",
  1118. }},
  1119. },
  1120. }
  1121. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1122. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_MultiInput")
  1123. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1124. for i, tt := range tests {
  1125. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1126. pp := &ProjectOp{Fields: stmt.Fields}
  1127. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1128. result := pp.Apply(ctx, tt.data, fv, afv)
  1129. if !reflect.DeepEqual(tt.result, result) {
  1130. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1131. }
  1132. }
  1133. }
  1134. func TestProjectPlan_Funcs(t *testing.T) {
  1135. var tests = []struct {
  1136. sql string
  1137. data interface{}
  1138. result []map[string]interface{}
  1139. }{
  1140. //0
  1141. {
  1142. sql: "SELECT round(a) as r FROM test",
  1143. data: &xsql.Tuple{
  1144. Emitter: "test",
  1145. Message: xsql.Message{
  1146. "a": 47.5,
  1147. },
  1148. },
  1149. result: []map[string]interface{}{{
  1150. "r": float64(48),
  1151. }},
  1152. },
  1153. //1
  1154. {
  1155. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  1156. data: &xsql.WindowTuples{
  1157. Content: []xsql.TupleRow{
  1158. &xsql.Tuple{Emitter: "src1",
  1159. Message: xsql.Message{"a": 53.1},
  1160. }, &xsql.Tuple{Emitter: "src1",
  1161. Message: xsql.Message{"a": 27.4},
  1162. }, &xsql.Tuple{Emitter: "src1",
  1163. Message: xsql.Message{"a": 123123.7},
  1164. },
  1165. },
  1166. },
  1167. result: []map[string]interface{}{{
  1168. "r": float64(53),
  1169. }, {
  1170. "r": float64(27),
  1171. }, {
  1172. "r": float64(123124),
  1173. }},
  1174. },
  1175. //2
  1176. {
  1177. sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
  1178. data: &xsql.WindowTuples{
  1179. Content: []xsql.TupleRow{
  1180. &xsql.Tuple{Emitter: "src1",
  1181. Message: xsql.Message{"a": 53.1},
  1182. }, &xsql.Tuple{Emitter: "src1",
  1183. Message: xsql.Message{"a": 27.4},
  1184. }, &xsql.Tuple{Emitter: "src1",
  1185. Message: xsql.Message{"a": 123123.7},
  1186. },
  1187. },
  1188. },
  1189. result: []map[string]interface{}{{
  1190. "r": float64(53),
  1191. }, {
  1192. "r": float64(27),
  1193. }, {
  1194. "r": float64(123124),
  1195. }},
  1196. },
  1197. //3
  1198. {
  1199. sql: "SELECT round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1200. data: &xsql.JoinTuples{
  1201. Content: []*xsql.JoinTuple{
  1202. {
  1203. Tuples: []xsql.TupleRow{
  1204. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
  1205. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1206. },
  1207. },
  1208. {
  1209. Tuples: []xsql.TupleRow{
  1210. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
  1211. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1212. },
  1213. },
  1214. {
  1215. Tuples: []xsql.TupleRow{
  1216. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
  1217. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1218. },
  1219. },
  1220. },
  1221. },
  1222. result: []map[string]interface{}{{
  1223. "r": float64(66),
  1224. }, {
  1225. "r": float64(73),
  1226. }, {
  1227. "r": float64(89),
  1228. }},
  1229. },
  1230. //4
  1231. {
  1232. 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)",
  1233. data: &xsql.JoinTuples{
  1234. Content: []*xsql.JoinTuple{
  1235. {
  1236. Tuples: []xsql.TupleRow{
  1237. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
  1238. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1239. },
  1240. },
  1241. {
  1242. Tuples: []xsql.TupleRow{
  1243. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
  1244. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1245. },
  1246. },
  1247. {
  1248. Tuples: []xsql.TupleRow{
  1249. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
  1250. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1251. },
  1252. },
  1253. },
  1254. },
  1255. result: []map[string]interface{}{{
  1256. "concat": "165.5512",
  1257. }, {
  1258. "concat": "273.49934",
  1259. }, {
  1260. "concat": "388.886",
  1261. }},
  1262. },
  1263. //5
  1264. {
  1265. sql: "SELECT count(a) as r FROM test",
  1266. data: &xsql.Tuple{
  1267. Emitter: "test",
  1268. Message: xsql.Message{
  1269. "a": 47.5,
  1270. },
  1271. },
  1272. result: []map[string]interface{}{{
  1273. "r": 1,
  1274. }},
  1275. },
  1276. //6
  1277. {
  1278. sql: "SELECT meta(test.device) as d FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1279. data: &xsql.JoinTuples{
  1280. Content: []*xsql.JoinTuple{
  1281. {
  1282. Tuples: []xsql.TupleRow{
  1283. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}, Metadata: xsql.Metadata{"device": "devicea"}},
  1284. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
  1285. },
  1286. },
  1287. {
  1288. Tuples: []xsql.TupleRow{
  1289. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1290. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
  1291. },
  1292. },
  1293. {
  1294. Tuples: []xsql.TupleRow{
  1295. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}, Metadata: xsql.Metadata{"device": "devicec"}},
  1296. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
  1297. },
  1298. },
  1299. },
  1300. },
  1301. result: []map[string]interface{}{{
  1302. "d": "devicea",
  1303. }, {
  1304. "d": "deviceb",
  1305. }, {
  1306. "d": "devicec",
  1307. }},
  1308. },
  1309. }
  1310. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1311. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_Funcs")
  1312. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1313. for i, tt := range tests {
  1314. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1315. if err != nil {
  1316. t.Error(err)
  1317. }
  1318. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
  1319. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1320. result := pp.Apply(ctx, tt.data, fv, afv)
  1321. if !reflect.DeepEqual(tt.result, result) {
  1322. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1323. }
  1324. }
  1325. }
  1326. func TestProjectPlan_AggFuncs(t *testing.T) {
  1327. var tests = []struct {
  1328. sql string
  1329. data interface{}
  1330. result []map[string]interface{}
  1331. }{
  1332. { //0
  1333. 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",
  1334. data: &xsql.GroupedTuplesSet{
  1335. Groups: []*xsql.GroupedTuples{
  1336. {
  1337. Content: []xsql.TupleRow{
  1338. &xsql.JoinTuple{
  1339. Tuples: []xsql.TupleRow{
  1340. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1341. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1342. },
  1343. },
  1344. &xsql.JoinTuple{
  1345. Tuples: []xsql.TupleRow{
  1346. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1347. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1348. },
  1349. },
  1350. },
  1351. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1352. },
  1353. {
  1354. Content: []xsql.TupleRow{
  1355. &xsql.JoinTuple{
  1356. Tuples: []xsql.TupleRow{
  1357. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1358. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1359. },
  1360. },
  1361. &xsql.JoinTuple{
  1362. Tuples: []xsql.TupleRow{
  1363. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1364. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1365. },
  1366. },
  1367. },
  1368. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1369. },
  1370. },
  1371. },
  1372. result: []map[string]interface{}{{
  1373. "c": 2,
  1374. "r": float64(122),
  1375. "ws": int64(1541152486013),
  1376. "we": int64(1541152487013),
  1377. }, {
  1378. "c": 2,
  1379. "r": float64(89),
  1380. "ws": int64(1541152486013),
  1381. "we": int64(1541152487013),
  1382. }},
  1383. },
  1384. //1
  1385. {
  1386. 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",
  1387. data: &xsql.GroupedTuplesSet{
  1388. Groups: []*xsql.GroupedTuples{
  1389. {
  1390. Content: []xsql.TupleRow{
  1391. &xsql.JoinTuple{
  1392. Tuples: []xsql.TupleRow{
  1393. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1394. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1395. },
  1396. },
  1397. &xsql.JoinTuple{
  1398. Tuples: []xsql.TupleRow{
  1399. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5}},
  1400. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1401. },
  1402. },
  1403. },
  1404. },
  1405. {
  1406. Content: []xsql.TupleRow{
  1407. &xsql.JoinTuple{
  1408. Tuples: []xsql.TupleRow{
  1409. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1410. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1411. },
  1412. },
  1413. &xsql.JoinTuple{
  1414. Tuples: []xsql.TupleRow{
  1415. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1416. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1417. },
  1418. },
  1419. },
  1420. },
  1421. },
  1422. },
  1423. result: []map[string]interface{}{{
  1424. "c": 1,
  1425. "a": 122.33,
  1426. "s": 122.33,
  1427. "min": 122.33,
  1428. "max": 122.33,
  1429. }, {
  1430. "c": 2,
  1431. "s": 103.63,
  1432. "a": 51.815,
  1433. "min": 14.6,
  1434. "max": 89.03,
  1435. }},
  1436. },
  1437. //2
  1438. {
  1439. sql: "SELECT avg(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1440. data: &xsql.GroupedTuplesSet{
  1441. Groups: []*xsql.GroupedTuples{
  1442. {
  1443. Content: []xsql.TupleRow{
  1444. &xsql.JoinTuple{
  1445. Tuples: []xsql.TupleRow{
  1446. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1447. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1448. },
  1449. },
  1450. &xsql.JoinTuple{
  1451. Tuples: []xsql.TupleRow{
  1452. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
  1453. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1454. },
  1455. },
  1456. &xsql.JoinTuple{
  1457. Tuples: []xsql.TupleRow{
  1458. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 98.31}},
  1459. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
  1460. },
  1461. },
  1462. &xsql.JoinTuple{
  1463. Tuples: []xsql.TupleRow{
  1464. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
  1465. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1466. },
  1467. },
  1468. },
  1469. },
  1470. {
  1471. Content: []xsql.TupleRow{
  1472. &xsql.JoinTuple{
  1473. Tuples: []xsql.TupleRow{
  1474. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1475. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1476. },
  1477. },
  1478. &xsql.JoinTuple{
  1479. Tuples: []xsql.TupleRow{
  1480. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1481. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1482. },
  1483. },
  1484. },
  1485. },
  1486. },
  1487. },
  1488. result: []map[string]interface{}{{
  1489. "avg": 116.68,
  1490. }, {
  1491. "avg": 51.815,
  1492. }},
  1493. },
  1494. //3
  1495. {
  1496. sql: "SELECT max(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1497. data: &xsql.GroupedTuplesSet{
  1498. Groups: []*xsql.GroupedTuples{
  1499. {
  1500. Content: []xsql.TupleRow{
  1501. &xsql.JoinTuple{
  1502. Tuples: []xsql.TupleRow{
  1503. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1504. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1505. },
  1506. },
  1507. &xsql.JoinTuple{
  1508. Tuples: []xsql.TupleRow{
  1509. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1510. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1511. },
  1512. },
  1513. &xsql.JoinTuple{
  1514. Tuples: []xsql.TupleRow{
  1515. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1516. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1517. },
  1518. },
  1519. },
  1520. },
  1521. {
  1522. Content: []xsql.TupleRow{
  1523. &xsql.JoinTuple{
  1524. Tuples: []xsql.TupleRow{
  1525. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1526. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1527. },
  1528. },
  1529. &xsql.JoinTuple{
  1530. Tuples: []xsql.TupleRow{
  1531. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1532. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1533. },
  1534. },
  1535. },
  1536. },
  1537. },
  1538. },
  1539. result: []map[string]interface{}{{
  1540. "max": 177.51,
  1541. }, {
  1542. "max": 89.03,
  1543. }},
  1544. },
  1545. //4
  1546. {
  1547. sql: "SELECT min(a), window_start(), window_end() FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
  1548. data: &xsql.JoinTuples{
  1549. Content: []*xsql.JoinTuple{
  1550. {
  1551. Tuples: []xsql.TupleRow{
  1552. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1553. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1554. },
  1555. },
  1556. {
  1557. Tuples: []xsql.TupleRow{
  1558. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1559. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1560. },
  1561. },
  1562. {
  1563. Tuples: []xsql.TupleRow{
  1564. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1565. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1566. },
  1567. },
  1568. },
  1569. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1570. },
  1571. result: []map[string]interface{}{{
  1572. "min": 68.55,
  1573. "window_start": int64(1541152486013),
  1574. "window_end": int64(1541152487013),
  1575. }},
  1576. },
  1577. //5
  1578. {
  1579. 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)",
  1580. data: &xsql.JoinTuples{
  1581. Content: []*xsql.JoinTuple{
  1582. {
  1583. Tuples: []xsql.TupleRow{
  1584. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1}},
  1585. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1586. },
  1587. },
  1588. {
  1589. Tuples: []xsql.TupleRow{
  1590. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
  1591. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1592. },
  1593. },
  1594. {
  1595. Tuples: []xsql.TupleRow{
  1596. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1597. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1598. },
  1599. },
  1600. },
  1601. },
  1602. result: []map[string]interface{}{{
  1603. "all": 3,
  1604. "c": 2,
  1605. "a": 123.03,
  1606. "s": 246.06,
  1607. "min": 68.55,
  1608. "max": 177.51,
  1609. }},
  1610. },
  1611. //6
  1612. {
  1613. sql: "SELECT sum(a), window_start() as ws, window_end() FROM test GROUP BY TumblingWindow(ss, 10)",
  1614. data: &xsql.WindowTuples{
  1615. Content: []xsql.TupleRow{
  1616. &xsql.Tuple{Emitter: "src1",
  1617. Message: xsql.Message{"a": 53},
  1618. }, &xsql.Tuple{Emitter: "src1",
  1619. Message: xsql.Message{"a": 27},
  1620. }, &xsql.Tuple{Emitter: "src1",
  1621. Message: xsql.Message{"a": 123123},
  1622. },
  1623. },
  1624. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1625. },
  1626. result: []map[string]interface{}{{
  1627. "sum": 123203,
  1628. "ws": int64(1541152486013),
  1629. "window_end": int64(1541152487013),
  1630. }},
  1631. },
  1632. //7
  1633. {
  1634. sql: "SELECT sum(a) as s FROM test GROUP BY TumblingWindow(ss, 10)",
  1635. data: &xsql.WindowTuples{
  1636. Content: []xsql.TupleRow{
  1637. &xsql.Tuple{Emitter: "src1",
  1638. Message: xsql.Message{"a": 53, "s": 123203},
  1639. }, &xsql.Tuple{Emitter: "src1",
  1640. Message: xsql.Message{"a": 27},
  1641. }, &xsql.Tuple{Emitter: "src1",
  1642. Message: xsql.Message{"a": 123123},
  1643. },
  1644. },
  1645. },
  1646. result: []map[string]interface{}{{
  1647. "s": 123203,
  1648. }},
  1649. },
  1650. //8
  1651. {
  1652. sql: "SELECT sum(a) FROM test GROUP BY TumblingWindow(ss, 10)",
  1653. data: &xsql.WindowTuples{
  1654. Content: []xsql.TupleRow{
  1655. &xsql.Tuple{Emitter: "src1",
  1656. Message: xsql.Message{"a": 53},
  1657. }, &xsql.Tuple{Emitter: "src1",
  1658. Message: xsql.Message{"a": 27},
  1659. }, &xsql.Tuple{Emitter: "src1",
  1660. Message: xsql.Message{"a": 123123},
  1661. },
  1662. },
  1663. },
  1664. result: []map[string]interface{}{{
  1665. "sum": 123203,
  1666. }},
  1667. },
  1668. //9
  1669. {
  1670. 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)",
  1671. data: &xsql.WindowTuples{
  1672. Content: []xsql.TupleRow{
  1673. &xsql.Tuple{Emitter: "src1",
  1674. Message: xsql.Message{"a": 53},
  1675. }, &xsql.Tuple{Emitter: "src1",
  1676. Message: xsql.Message{"a": 27},
  1677. }, &xsql.Tuple{Emitter: "src1",
  1678. Message: xsql.Message{"s": 123123},
  1679. },
  1680. },
  1681. },
  1682. result: []map[string]interface{}{{
  1683. "all": 3,
  1684. "c": 2,
  1685. "a": 40,
  1686. "s": 80,
  1687. "min": 27,
  1688. "max": 53,
  1689. }},
  1690. },
  1691. //10
  1692. {
  1693. sql: "SELECT count(*), meta(test1.device) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1694. data: &xsql.GroupedTuplesSet{
  1695. Groups: []*xsql.GroupedTuples{
  1696. {
  1697. Content: []xsql.TupleRow{
  1698. &xsql.JoinTuple{
  1699. Tuples: []xsql.TupleRow{
  1700. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  1701. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
  1702. },
  1703. },
  1704. &xsql.JoinTuple{
  1705. Tuples: []xsql.TupleRow{
  1706. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1707. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1708. },
  1709. },
  1710. },
  1711. },
  1712. {
  1713. Content: []xsql.TupleRow{
  1714. &xsql.JoinTuple{
  1715. Tuples: []xsql.TupleRow{
  1716. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  1717. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
  1718. },
  1719. },
  1720. &xsql.JoinTuple{
  1721. Tuples: []xsql.TupleRow{
  1722. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1723. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
  1724. },
  1725. },
  1726. },
  1727. },
  1728. },
  1729. },
  1730. result: []map[string]interface{}{{
  1731. "count": 2,
  1732. "meta": "devicea",
  1733. }, {
  1734. "count": 2,
  1735. "meta": "devicec",
  1736. }},
  1737. },
  1738. //11
  1739. {
  1740. 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",
  1741. data: &xsql.GroupedTuplesSet{
  1742. Groups: []*xsql.GroupedTuples{
  1743. {
  1744. Content: []xsql.TupleRow{
  1745. &xsql.JoinTuple{
  1746. Tuples: []xsql.TupleRow{
  1747. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "d": "devicea"}},
  1748. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
  1749. },
  1750. },
  1751. &xsql.JoinTuple{
  1752. Tuples: []xsql.TupleRow{
  1753. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1754. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
  1755. },
  1756. },
  1757. },
  1758. },
  1759. {
  1760. Content: []xsql.TupleRow{
  1761. &xsql.JoinTuple{
  1762. Tuples: []xsql.TupleRow{
  1763. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "d": "devicec"}},
  1764. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
  1765. },
  1766. },
  1767. &xsql.JoinTuple{
  1768. Tuples: []xsql.TupleRow{
  1769. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1770. &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
  1771. },
  1772. },
  1773. },
  1774. },
  1775. },
  1776. },
  1777. result: []map[string]interface{}{{
  1778. "c": 2,
  1779. "d": "devicea",
  1780. }, {
  1781. "c": 2,
  1782. "d": "devicec",
  1783. }},
  1784. },
  1785. //12
  1786. {
  1787. sql: "SELECT * FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1788. data: &xsql.GroupedTuplesSet{
  1789. Groups: []*xsql.GroupedTuples{
  1790. {
  1791. Content: []xsql.TupleRow{
  1792. &xsql.JoinTuple{
  1793. Tuples: []xsql.TupleRow{
  1794. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1795. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1796. },
  1797. },
  1798. &xsql.JoinTuple{
  1799. Tuples: []xsql.TupleRow{
  1800. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1801. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1802. },
  1803. },
  1804. },
  1805. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1806. },
  1807. {
  1808. Content: []xsql.TupleRow{
  1809. &xsql.JoinTuple{
  1810. Tuples: []xsql.TupleRow{
  1811. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1812. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1813. },
  1814. },
  1815. &xsql.JoinTuple{
  1816. Tuples: []xsql.TupleRow{
  1817. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1818. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1819. },
  1820. },
  1821. },
  1822. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1823. },
  1824. },
  1825. },
  1826. result: []map[string]interface{}{{
  1827. "a": 122.33,
  1828. "c": 2,
  1829. "color": "w2",
  1830. "id": 1,
  1831. "r": 122,
  1832. }, {
  1833. "a": 89.03,
  1834. "c": 2,
  1835. "color": "w1",
  1836. "id": 2,
  1837. "r": 89,
  1838. }},
  1839. },
  1840. //13
  1841. {
  1842. sql: "SELECT collect(a) as r1 FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1843. data: &xsql.GroupedTuplesSet{
  1844. Groups: []*xsql.GroupedTuples{
  1845. {
  1846. Content: []xsql.TupleRow{
  1847. &xsql.JoinTuple{
  1848. Tuples: []xsql.TupleRow{
  1849. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1850. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1851. },
  1852. },
  1853. &xsql.JoinTuple{
  1854. Tuples: []xsql.TupleRow{
  1855. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1856. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1857. },
  1858. },
  1859. },
  1860. },
  1861. {
  1862. Content: []xsql.TupleRow{
  1863. &xsql.JoinTuple{
  1864. Tuples: []xsql.TupleRow{
  1865. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1866. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1867. },
  1868. },
  1869. &xsql.JoinTuple{
  1870. Tuples: []xsql.TupleRow{
  1871. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1872. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1873. },
  1874. },
  1875. },
  1876. },
  1877. },
  1878. },
  1879. result: []map[string]interface{}{{
  1880. "r1": []interface{}{122.33, 177.51},
  1881. }, {"r1": []interface{}{89.03, 14.6}}},
  1882. },
  1883. //14
  1884. {
  1885. sql: "SELECT collect(*)[1] as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1886. data: &xsql.WindowTuples{
  1887. Content: []xsql.TupleRow{
  1888. &xsql.Tuple{Emitter: "src1",
  1889. Message: xsql.Message{"a": 53, "s": 123203},
  1890. }, &xsql.Tuple{Emitter: "src1",
  1891. Message: xsql.Message{"a": 27},
  1892. }, &xsql.Tuple{Emitter: "src1",
  1893. Message: xsql.Message{"a": 123123},
  1894. },
  1895. },
  1896. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  1897. },
  1898. result: []map[string]interface{}{{
  1899. "c1": xsql.Message{
  1900. "a": 27,
  1901. },
  1902. }},
  1903. },
  1904. //15
  1905. {
  1906. sql: "SELECT collect(*)[1]->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1907. data: &xsql.WindowTuples{
  1908. Content: []xsql.TupleRow{
  1909. &xsql.Tuple{Emitter: "src1",
  1910. Message: xsql.Message{"a": 53, "s": 123203},
  1911. }, &xsql.Tuple{Emitter: "src1",
  1912. Message: xsql.Message{"a": 27},
  1913. }, &xsql.Tuple{Emitter: "src1",
  1914. Message: xsql.Message{"a": 123123},
  1915. },
  1916. },
  1917. },
  1918. result: []map[string]interface{}{{
  1919. "c1": 27,
  1920. }},
  1921. },
  1922. //16
  1923. {
  1924. sql: "SELECT collect(*)[1]->sl[0] as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1925. data: &xsql.WindowTuples{
  1926. Content: []xsql.TupleRow{
  1927. &xsql.Tuple{Emitter: "src1",
  1928. Message: xsql.Message{"a": 53, "sl": []string{"hello", "world"}},
  1929. }, &xsql.Tuple{Emitter: "src1",
  1930. Message: xsql.Message{"a": 27, "sl": []string{"new", "horizon"}},
  1931. }, &xsql.Tuple{Emitter: "src1",
  1932. Message: xsql.Message{"a": 123123, "sl": []string{"south", "africa"}},
  1933. },
  1934. },
  1935. },
  1936. result: []map[string]interface{}{{
  1937. "c1": "new",
  1938. }},
  1939. },
  1940. //17
  1941. {
  1942. sql: "SELECT deduplicate(id, true) as r1 FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  1943. data: &xsql.GroupedTuplesSet{
  1944. Groups: []*xsql.GroupedTuples{
  1945. {
  1946. Content: []xsql.TupleRow{
  1947. &xsql.JoinTuple{
  1948. Tuples: []xsql.TupleRow{
  1949. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
  1950. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  1951. },
  1952. },
  1953. &xsql.JoinTuple{
  1954. Tuples: []xsql.TupleRow{
  1955. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
  1956. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  1957. },
  1958. },
  1959. },
  1960. },
  1961. {
  1962. Content: []xsql.TupleRow{
  1963. &xsql.JoinTuple{
  1964. Tuples: []xsql.TupleRow{
  1965. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
  1966. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  1967. },
  1968. },
  1969. &xsql.JoinTuple{
  1970. Tuples: []xsql.TupleRow{
  1971. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  1972. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  1973. },
  1974. },
  1975. },
  1976. },
  1977. },
  1978. },
  1979. result: []map[string]interface{}{
  1980. {
  1981. "r1": []interface{}{
  1982. xsql.Message{"a": 122.33, "c": 2, "color": "w2", "id": 1, "r": 122},
  1983. xsql.Message{"a": 177.51, "color": "w2", "id": 5}},
  1984. }, {
  1985. "r1": []interface{}{
  1986. xsql.Message{"a": 89.03, "c": 2, "color": "w1", "id": 2, "r": 89},
  1987. xsql.Message{"a": 14.6, "color": "w1", "id": 4}},
  1988. },
  1989. },
  1990. },
  1991. //18
  1992. {
  1993. sql: "SELECT deduplicate(a, false)->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  1994. data: &xsql.WindowTuples{
  1995. Content: []xsql.TupleRow{
  1996. &xsql.Tuple{Emitter: "src1",
  1997. Message: xsql.Message{"a": 53, "s": 123203},
  1998. }, &xsql.Tuple{Emitter: "src1",
  1999. Message: xsql.Message{"a": 27},
  2000. }, &xsql.Tuple{Emitter: "src1",
  2001. Message: xsql.Message{"a": 123123},
  2002. },
  2003. },
  2004. },
  2005. result: []map[string]interface{}{{
  2006. "c1": 123123,
  2007. }},
  2008. },
  2009. //19
  2010. {
  2011. sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  2012. data: &xsql.WindowTuples{
  2013. Content: []xsql.TupleRow{
  2014. &xsql.Tuple{Emitter: "src1",
  2015. Message: xsql.Message{"a": 53, "s": 123203},
  2016. }, &xsql.Tuple{Emitter: "src1",
  2017. Message: xsql.Message{"a": 27},
  2018. }, &xsql.Tuple{Emitter: "src1",
  2019. Message: xsql.Message{"a": 53},
  2020. },
  2021. },
  2022. },
  2023. result: []map[string]interface{}{{}},
  2024. },
  2025. //20
  2026. {
  2027. sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
  2028. data: &xsql.WindowTuples{
  2029. Content: []xsql.TupleRow{
  2030. &xsql.Tuple{Emitter: "src1",
  2031. Message: xsql.Message{"a": 53, "s": 123203},
  2032. }, &xsql.Tuple{Emitter: "src1",
  2033. Message: xsql.Message{"a": 27},
  2034. }, &xsql.Tuple{Emitter: "src1",
  2035. Message: xsql.Message{"a": 53},
  2036. },
  2037. },
  2038. },
  2039. result: []map[string]interface{}{{}},
  2040. },
  2041. //21 when got column after group by operation, return the first tuple's column
  2042. {
  2043. 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)",
  2044. data: &xsql.GroupedTuplesSet{
  2045. Groups: []*xsql.GroupedTuples{
  2046. {
  2047. Content: []xsql.TupleRow{
  2048. &xsql.JoinTuple{
  2049. Tuples: []xsql.TupleRow{
  2050. &xsql.Tuple{Emitter: "B", Message: xsql.Message{"module": 1, "topic": "moduleB topic", "value": 1}},
  2051. },
  2052. },
  2053. &xsql.JoinTuple{
  2054. Tuples: []xsql.TupleRow{
  2055. &xsql.Tuple{Emitter: "C", Message: xsql.Message{"module": 1, "topic": "moduleC topic", "value": 100}},
  2056. },
  2057. },
  2058. },
  2059. },
  2060. },
  2061. },
  2062. result: []map[string]interface{}{{
  2063. "var2": "moduleB topic",
  2064. "max2": 1,
  2065. "max3": 100,
  2066. }},
  2067. },
  2068. }
  2069. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  2070. contextLogger := conf.Log.WithField("rule", "TestProjectPlan_AggFuncs")
  2071. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  2072. for i, tt := range tests {
  2073. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  2074. if err != nil {
  2075. t.Error(err)
  2076. }
  2077. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: true}
  2078. fv, afv := xsql.NewFunctionValuersForOp(nil)
  2079. result := pp.Apply(ctx, tt.data, fv, afv)
  2080. if !reflect.DeepEqual(tt.result, result) {
  2081. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  2082. }
  2083. }
  2084. }
  2085. func TestProjectPlanError(t *testing.T) {
  2086. var tests = []struct {
  2087. sql string
  2088. data interface{}
  2089. result interface{}
  2090. }{
  2091. //0
  2092. {
  2093. sql: "SELECT a FROM test",
  2094. data: errors.New("an error from upstream"),
  2095. result: errors.New("an error from upstream"),
  2096. },
  2097. //1
  2098. {
  2099. sql: "SELECT a * 5 FROM test",
  2100. data: &xsql.Tuple{
  2101. Emitter: "test",
  2102. Message: xsql.Message{
  2103. "a": "val_a",
  2104. },
  2105. },
  2106. result: errors.New("run Select error: invalid operation string(val_a) * int64(5)"),
  2107. },
  2108. //2
  2109. {
  2110. sql: `SELECT a[0]->b AS ab FROM test`,
  2111. data: &xsql.Tuple{
  2112. Emitter: "test",
  2113. Message: xsql.Message{
  2114. "a": "common string",
  2115. },
  2116. },
  2117. result: errors.New("run Select error: invalid operation string(common string) [] *xsql.BracketEvalResult(&{0 0})"),
  2118. },
  2119. //3
  2120. {
  2121. sql: `SELECT round(a) as r FROM test`,
  2122. data: &xsql.Tuple{
  2123. Emitter: "test",
  2124. Message: xsql.Message{
  2125. "a": "common string",
  2126. },
  2127. },
  2128. result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
  2129. },
  2130. //4
  2131. {
  2132. sql: `SELECT round(a) as r FROM test`,
  2133. data: &xsql.Tuple{
  2134. Emitter: "test",
  2135. Message: xsql.Message{
  2136. "abc": "common string",
  2137. },
  2138. },
  2139. result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
  2140. },
  2141. //5
  2142. {
  2143. sql: "SELECT avg(a) as avg FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
  2144. data: &xsql.GroupedTuplesSet{
  2145. Groups: []*xsql.GroupedTuples{
  2146. {
  2147. Content: []xsql.TupleRow{
  2148. &xsql.JoinTuple{
  2149. Tuples: []xsql.TupleRow{
  2150. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
  2151. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  2152. },
  2153. },
  2154. &xsql.JoinTuple{
  2155. Tuples: []xsql.TupleRow{
  2156. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
  2157. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
  2158. },
  2159. },
  2160. &xsql.JoinTuple{
  2161. Tuples: []xsql.TupleRow{
  2162. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": "dde"}},
  2163. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
  2164. },
  2165. },
  2166. &xsql.JoinTuple{
  2167. Tuples: []xsql.TupleRow{
  2168. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
  2169. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
  2170. },
  2171. },
  2172. },
  2173. },
  2174. {
  2175. Content: []xsql.TupleRow{
  2176. &xsql.JoinTuple{
  2177. Tuples: []xsql.TupleRow{
  2178. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
  2179. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
  2180. },
  2181. },
  2182. &xsql.JoinTuple{
  2183. Tuples: []xsql.TupleRow{
  2184. &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
  2185. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
  2186. },
  2187. },
  2188. },
  2189. },
  2190. },
  2191. },
  2192. result: errors.New("run Select error: call func avg error: requires float64 but found string(dde)"),
  2193. },
  2194. //6
  2195. {
  2196. sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
  2197. data: &xsql.WindowTuples{
  2198. Content: []xsql.TupleRow{
  2199. &xsql.Tuple{Emitter: "src1",
  2200. Message: xsql.Message{"a": 53},
  2201. }, &xsql.Tuple{Emitter: "src1",
  2202. Message: xsql.Message{"a": "ddd"},
  2203. }, &xsql.Tuple{Emitter: "src1",
  2204. Message: xsql.Message{"a": 123123},
  2205. },
  2206. },
  2207. },
  2208. result: errors.New("run Select error: call func sum error: requires int but found string(ddd)"),
  2209. },
  2210. //7
  2211. {
  2212. sql: `SELECT a[0]->b AS ab FROM test`,
  2213. data: &xsql.Tuple{
  2214. Emitter: "test",
  2215. Message: xsql.Message{
  2216. "a": []map[string]interface{}(nil),
  2217. },
  2218. },
  2219. result: errors.New("run Select error: out of index: 0 of 0"),
  2220. },
  2221. }
  2222. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  2223. contextLogger := conf.Log.WithField("rule", "TestProjectPlanError")
  2224. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  2225. for i, tt := range tests {
  2226. stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  2227. pp := &ProjectOp{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
  2228. fv, afv := xsql.NewFunctionValuersForOp(nil)
  2229. result := pp.Apply(ctx, tt.data, fv, afv)
  2230. if !reflect.DeepEqual(tt.result, result) {
  2231. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  2232. }
  2233. }
  2234. }