project_test.go 66 KB

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