project_test.go 60 KB

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