project_test.go 61 KB

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