window_rule_test.go 46 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439
  1. // Copyright 2021-2022 EMQ Technologies Co., Ltd.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package topotest
  15. import (
  16. "github.com/lf-edge/ekuiper/internal/topo"
  17. "github.com/lf-edge/ekuiper/pkg/api"
  18. "testing"
  19. )
  20. func TestWindow(t *testing.T) {
  21. //Reset
  22. streamList := []string{"demo", "demoError", "demo1", "sessionDemo", "table1"}
  23. HandleStream(false, streamList, t)
  24. var tests = []RuleTest{
  25. {
  26. Name: `TestWindowRule1`,
  27. Sql: `SELECT * FROM demo GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
  28. R: [][]map[string]interface{}{
  29. {{
  30. "color": "red",
  31. "size": float64(3),
  32. "ts": float64(1541152486013),
  33. }, {
  34. "color": "blue",
  35. "size": float64(6),
  36. "ts": float64(1541152486822),
  37. }},
  38. {{
  39. "color": "red",
  40. "size": float64(3),
  41. "ts": float64(1541152486013),
  42. }, {
  43. "color": "blue",
  44. "size": float64(6),
  45. "ts": float64(1541152486822),
  46. }, {
  47. "color": "blue",
  48. "size": float64(2),
  49. "ts": float64(1541152487632),
  50. }},
  51. {{
  52. "color": "blue",
  53. "size": float64(2),
  54. "ts": float64(1541152487632),
  55. }, {
  56. "color": "yellow",
  57. "size": float64(4),
  58. "ts": float64(1541152488442),
  59. }},
  60. {{
  61. "color": "yellow",
  62. "size": float64(4),
  63. "ts": float64(1541152488442),
  64. }, {
  65. "color": "red",
  66. "size": float64(1),
  67. "ts": float64(1541152489252),
  68. }},
  69. },
  70. M: map[string]interface{}{
  71. "op_3_project_0_exceptions_total": int64(0),
  72. "op_3_project_0_process_latency_us": int64(0),
  73. "op_3_project_0_records_in_total": int64(4),
  74. "op_3_project_0_records_out_total": int64(4),
  75. "sink_mockSink_0_exceptions_total": int64(0),
  76. "sink_mockSink_0_records_in_total": int64(4),
  77. "sink_mockSink_0_records_out_total": int64(4),
  78. "source_demo_0_exceptions_total": int64(0),
  79. "source_demo_0_records_in_total": int64(5),
  80. "source_demo_0_records_out_total": int64(5),
  81. "op_2_window_0_exceptions_total": int64(0),
  82. "op_2_window_0_process_latency_us": int64(0),
  83. "op_2_window_0_records_in_total": int64(5),
  84. "op_2_window_0_records_out_total": int64(4),
  85. },
  86. }, {
  87. Name: `TestWindowRule2`,
  88. Sql: `SELECT color, ts FROM demo where size > 2 GROUP BY tumblingwindow(ss, 1)`,
  89. R: [][]map[string]interface{}{
  90. {{
  91. "color": "red",
  92. "ts": float64(1541152486013),
  93. }, {
  94. "color": "blue",
  95. "ts": float64(1541152486822),
  96. }},
  97. {{
  98. "color": "yellow",
  99. "ts": float64(1541152488442),
  100. }},
  101. },
  102. M: map[string]interface{}{
  103. "op_4_project_0_exceptions_total": int64(0),
  104. "op_4_project_0_process_latency_us": int64(0),
  105. "op_4_project_0_records_in_total": int64(2),
  106. "op_4_project_0_records_out_total": int64(2),
  107. "sink_mockSink_0_exceptions_total": int64(0),
  108. "sink_mockSink_0_records_in_total": int64(2),
  109. "sink_mockSink_0_records_out_total": int64(2),
  110. "source_demo_0_exceptions_total": int64(0),
  111. "source_demo_0_records_in_total": int64(5),
  112. "source_demo_0_records_out_total": int64(5),
  113. "op_3_window_0_exceptions_total": int64(0),
  114. "op_3_window_0_process_latency_us": int64(0),
  115. "op_3_window_0_records_in_total": int64(3),
  116. "op_3_window_0_records_out_total": int64(2),
  117. "op_2_filter_0_exceptions_total": int64(0),
  118. "op_2_filter_0_process_latency_us": int64(0),
  119. "op_2_filter_0_records_in_total": int64(5),
  120. "op_2_filter_0_records_out_total": int64(3),
  121. },
  122. }, {
  123. Name: `TestWindowRule3`,
  124. Sql: `SELECT color, temp, demo.ts as ts1, demo1.ts as ts2, demo.ts - demo1.ts as diff FROM demo INNER JOIN demo1 ON ts1 = ts2 GROUP BY SlidingWindow(ss, 1)`,
  125. R: [][]map[string]interface{}{
  126. {{
  127. "color": "red",
  128. "temp": 25.5,
  129. "ts1": float64(1541152486013),
  130. "ts2": float64(1541152486013),
  131. "diff": float64(0),
  132. }}, {{
  133. "color": "red",
  134. "temp": 25.5,
  135. "ts1": float64(1541152486013),
  136. "ts2": float64(1541152486013),
  137. "diff": float64(0),
  138. }}, {{
  139. "color": "red",
  140. "temp": 25.5,
  141. "ts1": float64(1541152486013),
  142. "ts2": float64(1541152486013),
  143. "diff": float64(0),
  144. }}, {{
  145. "color": "blue",
  146. "temp": 28.1,
  147. "ts1": float64(1541152487632),
  148. "ts2": float64(1541152487632),
  149. "diff": float64(0),
  150. }}, {{
  151. "color": "blue",
  152. "temp": 28.1,
  153. "ts1": float64(1541152487632),
  154. "ts2": float64(1541152487632),
  155. "diff": float64(0),
  156. }}, {{
  157. "color": "blue",
  158. "temp": 28.1,
  159. "ts1": float64(1541152487632),
  160. "ts2": float64(1541152487632),
  161. "diff": float64(0),
  162. }, {
  163. "color": "yellow",
  164. "temp": 27.4,
  165. "ts1": float64(1541152488442),
  166. "ts2": float64(1541152488442),
  167. "diff": float64(0),
  168. }}, {{
  169. "color": "yellow",
  170. "temp": 27.4,
  171. "ts1": float64(1541152488442),
  172. "ts2": float64(1541152488442),
  173. "diff": float64(0),
  174. }}, {{
  175. "color": "yellow",
  176. "temp": 27.4,
  177. "ts1": float64(1541152488442),
  178. "ts2": float64(1541152488442),
  179. "diff": float64(0),
  180. }, {
  181. "color": "red",
  182. "temp": 25.5,
  183. "ts1": float64(1541152489252),
  184. "ts2": float64(1541152489252),
  185. "diff": float64(0),
  186. }},
  187. },
  188. M: map[string]interface{}{
  189. "op_5_project_0_exceptions_total": int64(0),
  190. "op_5_project_0_process_latency_us": int64(0),
  191. "op_5_project_0_records_in_total": int64(8),
  192. "op_5_project_0_records_out_total": int64(8),
  193. "sink_mockSink_0_exceptions_total": int64(0),
  194. "sink_mockSink_0_records_in_total": int64(8),
  195. "sink_mockSink_0_records_out_total": int64(8),
  196. "source_demo_0_exceptions_total": int64(0),
  197. "source_demo_0_records_in_total": int64(5),
  198. "source_demo_0_records_out_total": int64(5),
  199. "source_demo1_0_exceptions_total": int64(0),
  200. "source_demo1_0_records_in_total": int64(5),
  201. "source_demo1_0_records_out_total": int64(5),
  202. "op_3_window_0_exceptions_total": int64(0),
  203. "op_3_window_0_process_latency_us": int64(0),
  204. "op_3_window_0_records_in_total": int64(10),
  205. "op_3_window_0_records_out_total": int64(10),
  206. "op_4_join_0_exceptions_total": int64(0),
  207. "op_4_join_0_process_latency_us": int64(0),
  208. "op_4_join_0_records_in_total": int64(10),
  209. "op_4_join_0_records_out_total": int64(8),
  210. },
  211. T: &topo.PrintableTopo{
  212. Sources: []string{"source_demo", "source_demo1"},
  213. Edges: map[string][]string{
  214. "source_demo": {"op_3_window"},
  215. "source_demo1": {"op_3_window"},
  216. "op_3_window": {"op_4_join"},
  217. "op_4_join": {"op_5_project"},
  218. "op_5_project": {"sink_mockSink"},
  219. },
  220. },
  221. }, {
  222. Name: `TestWindowRule4`,
  223. Sql: `SELECT color, count(*) as c FROM demo GROUP BY SlidingWindow(ss, 2), color ORDER BY color`,
  224. R: [][]map[string]interface{}{
  225. {{
  226. "color": "red",
  227. "c": float64(1),
  228. }}, {{
  229. "color": "blue",
  230. "c": float64(1),
  231. }, {
  232. "color": "red",
  233. "c": float64(1),
  234. }}, {{
  235. "color": "blue",
  236. "c": float64(2),
  237. }, {
  238. "color": "red",
  239. "c": float64(1),
  240. }}, {{
  241. "color": "blue",
  242. "c": float64(2),
  243. }, {
  244. "color": "yellow",
  245. "c": float64(1),
  246. }}, {{
  247. "color": "blue",
  248. "c": float64(1),
  249. }, {
  250. "color": "red",
  251. "c": float64(1),
  252. }, {
  253. "color": "yellow",
  254. "c": float64(1),
  255. }},
  256. },
  257. M: map[string]interface{}{
  258. "op_5_project_0_exceptions_total": int64(0),
  259. "op_5_project_0_process_latency_us": int64(0),
  260. "op_5_project_0_records_in_total": int64(5),
  261. "op_5_project_0_records_out_total": int64(5),
  262. "sink_mockSink_0_exceptions_total": int64(0),
  263. "sink_mockSink_0_records_in_total": int64(5),
  264. "sink_mockSink_0_records_out_total": int64(5),
  265. "source_demo_0_exceptions_total": int64(0),
  266. "source_demo_0_records_in_total": int64(5),
  267. "source_demo_0_records_out_total": int64(5),
  268. "op_2_window_0_exceptions_total": int64(0),
  269. "op_2_window_0_process_latency_us": int64(0),
  270. "op_2_window_0_records_in_total": int64(5),
  271. "op_2_window_0_records_out_total": int64(5),
  272. "op_3_aggregate_0_exceptions_total": int64(0),
  273. "op_3_aggregate_0_process_latency_us": int64(0),
  274. "op_3_aggregate_0_records_in_total": int64(5),
  275. "op_3_aggregate_0_records_out_total": int64(5),
  276. "op_4_order_0_exceptions_total": int64(0),
  277. "op_4_order_0_process_latency_us": int64(0),
  278. "op_4_order_0_records_in_total": int64(5),
  279. "op_4_order_0_records_out_total": int64(5),
  280. },
  281. }, {
  282. Name: `TestWindowRule5`,
  283. Sql: `SELECT count(temp), window_start() as ws, window_end() FROM sessionDemo GROUP BY SessionWindow(ss, 2, 1) `,
  284. R: [][]map[string]interface{}{
  285. {{
  286. "count": float64(2),
  287. "ws": float64(1541152486013),
  288. "window_end": float64(1541152487823), // timeout
  289. }}, {{
  290. "count": float64(3),
  291. "ws": float64(1541152487932),
  292. "window_end": float64(1541152490000), // tick
  293. }}, {{
  294. "count": float64(5),
  295. "ws": float64(1541152490000),
  296. "window_end": float64(1541152494000), // tick
  297. }}, {{
  298. "count": float64(1),
  299. "ws": float64(1541152494000),
  300. "window_end": float64(1541152495112), // timeout
  301. }},
  302. },
  303. M: map[string]interface{}{
  304. "op_3_project_0_exceptions_total": int64(0),
  305. "op_3_project_0_process_latency_us": int64(0),
  306. "op_3_project_0_records_in_total": int64(4),
  307. "op_3_project_0_records_out_total": int64(4),
  308. "sink_mockSink_0_exceptions_total": int64(0),
  309. "sink_mockSink_0_records_in_total": int64(4),
  310. "sink_mockSink_0_records_out_total": int64(4),
  311. "source_sessionDemo_0_exceptions_total": int64(0),
  312. "source_sessionDemo_0_records_in_total": int64(11),
  313. "source_sessionDemo_0_records_out_total": int64(11),
  314. "op_2_window_0_exceptions_total": int64(0),
  315. "op_2_window_0_process_latency_us": int64(0),
  316. "op_2_window_0_records_in_total": int64(11),
  317. "op_2_window_0_records_out_total": int64(4),
  318. },
  319. }, {
  320. Name: `TestWindowRule6`,
  321. Sql: `SELECT window_end(), sum(temp) as temp, count(color) as c, window_start() FROM demo INNER JOIN demo1 ON demo.ts = demo1.ts GROUP BY SlidingWindow(ss, 1)`,
  322. R: [][]map[string]interface{}{
  323. {{
  324. "temp": 25.5,
  325. "c": float64(1),
  326. "window_start": float64(1541152485115),
  327. "window_end": float64(1541152486115),
  328. }}, {{
  329. "temp": 25.5,
  330. "c": float64(1),
  331. "window_start": float64(1541152485822),
  332. "window_end": float64(1541152486822),
  333. }}, {{
  334. "temp": 25.5,
  335. "c": float64(1),
  336. "window_start": float64(1541152485903),
  337. "window_end": float64(1541152486903),
  338. }}, {{
  339. "temp": 28.1,
  340. "c": float64(1),
  341. "window_start": float64(1541152486702),
  342. "window_end": float64(1541152487702),
  343. }}, {{
  344. "temp": 28.1,
  345. "c": float64(1),
  346. "window_start": float64(1541152487442),
  347. "window_end": float64(1541152488442),
  348. }}, {{
  349. "temp": 55.5,
  350. "c": float64(2),
  351. "window_start": float64(1541152487605),
  352. "window_end": float64(1541152488605),
  353. }}, {{
  354. "temp": 27.4,
  355. "c": float64(1),
  356. "window_start": float64(1541152488252),
  357. "window_end": float64(1541152489252),
  358. }}, {{
  359. "temp": 52.9,
  360. "c": float64(2),
  361. "window_start": float64(1541152488305),
  362. "window_end": float64(1541152489305),
  363. }},
  364. },
  365. M: map[string]interface{}{
  366. "op_5_project_0_exceptions_total": int64(0),
  367. "op_5_project_0_process_latency_us": int64(0),
  368. "op_5_project_0_records_in_total": int64(8),
  369. "op_5_project_0_records_out_total": int64(8),
  370. "sink_mockSink_0_exceptions_total": int64(0),
  371. "sink_mockSink_0_records_in_total": int64(8),
  372. "sink_mockSink_0_records_out_total": int64(8),
  373. "source_demo_0_exceptions_total": int64(0),
  374. "source_demo_0_records_in_total": int64(5),
  375. "source_demo_0_records_out_total": int64(5),
  376. "source_demo1_0_exceptions_total": int64(0),
  377. "source_demo1_0_records_in_total": int64(5),
  378. "source_demo1_0_records_out_total": int64(5),
  379. "op_3_window_0_exceptions_total": int64(0),
  380. "op_3_window_0_process_latency_us": int64(0),
  381. "op_3_window_0_records_in_total": int64(10),
  382. "op_3_window_0_records_out_total": int64(10),
  383. "op_4_join_0_exceptions_total": int64(0),
  384. "op_4_join_0_process_latency_us": int64(0),
  385. "op_4_join_0_records_in_total": int64(10),
  386. "op_4_join_0_records_out_total": int64(8),
  387. },
  388. }, {
  389. Name: `TestWindowRule7`,
  390. Sql: `SELECT * FROM demoError GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
  391. R: [][]map[string]interface{}{
  392. {{
  393. "error": "error in preprocessor: invalid data type for color, expect string but found int(3)",
  394. }},
  395. {{
  396. "color": "blue",
  397. "size": float64(6),
  398. "ts": float64(1541152486822),
  399. }},
  400. {{
  401. "color": "blue",
  402. "size": float64(6),
  403. "ts": float64(1541152486822),
  404. }, {
  405. "color": "blue",
  406. "size": float64(2),
  407. "ts": float64(1541152487632),
  408. }},
  409. {{
  410. "error": "error in preprocessor: invalid data type for color, expect string but found int(7)",
  411. }},
  412. {{
  413. "color": "blue",
  414. "size": float64(2),
  415. "ts": float64(1541152487632),
  416. }},
  417. {{
  418. "error": "error in preprocessor: invalid data type for size, expect bigint but found string(blue)",
  419. }},
  420. },
  421. M: map[string]interface{}{
  422. "op_3_project_0_exceptions_total": int64(3),
  423. "op_3_project_0_process_latency_us": int64(0),
  424. "op_3_project_0_records_in_total": int64(6),
  425. "op_3_project_0_records_out_total": int64(3),
  426. "sink_mockSink_0_exceptions_total": int64(0),
  427. "sink_mockSink_0_records_in_total": int64(6),
  428. "sink_mockSink_0_records_out_total": int64(6),
  429. "source_demoError_0_exceptions_total": int64(3),
  430. "source_demoError_0_records_in_total": int64(5),
  431. "source_demoError_0_records_out_total": int64(5),
  432. "op_2_window_0_exceptions_total": int64(3),
  433. "op_2_window_0_process_latency_us": int64(0),
  434. "op_2_window_0_records_in_total": int64(5),
  435. "op_2_window_0_records_out_total": int64(3),
  436. },
  437. }, {
  438. Name: `TestWindowRule8`,
  439. Sql: `SELECT color, window_end(), ts, count(*) as c, window_start() FROM demo where size > 2 GROUP BY tumblingwindow(ss, 1) having c > 1`,
  440. R: [][]map[string]interface{}{
  441. {{
  442. "color": "red",
  443. "ts": float64(1541152486013),
  444. "c": float64(2),
  445. "window_start": float64(1541152486000),
  446. "window_end": float64(1541152487000),
  447. }},
  448. },
  449. M: map[string]interface{}{
  450. "op_5_project_0_exceptions_total": int64(0),
  451. "op_5_project_0_process_latency_us": int64(0),
  452. "op_5_project_0_records_in_total": int64(1),
  453. "op_5_project_0_records_out_total": int64(1),
  454. "sink_mockSink_0_exceptions_total": int64(0),
  455. "sink_mockSink_0_records_in_total": int64(1),
  456. "sink_mockSink_0_records_out_total": int64(1),
  457. "source_demo_0_exceptions_total": int64(0),
  458. "source_demo_0_records_in_total": int64(5),
  459. "source_demo_0_records_out_total": int64(5),
  460. "op_3_window_0_exceptions_total": int64(0),
  461. "op_3_window_0_process_latency_us": int64(0),
  462. "op_3_window_0_records_in_total": int64(3),
  463. "op_3_window_0_records_out_total": int64(2),
  464. "op_2_filter_0_exceptions_total": int64(0),
  465. "op_2_filter_0_process_latency_us": int64(0),
  466. "op_2_filter_0_records_in_total": int64(5),
  467. "op_2_filter_0_records_out_total": int64(3),
  468. "op_4_having_0_exceptions_total": int64(0),
  469. "op_4_having_0_process_latency_us": int64(0),
  470. "op_4_having_0_records_in_total": int64(2),
  471. "op_4_having_0_records_out_total": int64(1),
  472. },
  473. }, {
  474. Name: `TestWindowRule9`,
  475. Sql: `SELECT color, window_start(), window_end() FROM demo GROUP BY HOPPINGWINDOW(ss, 2, 1) FILTER( WHERE size > 2)`,
  476. R: [][]map[string]interface{}{
  477. {{
  478. "color": "red",
  479. "window_start": float64(1541152485000),
  480. "window_end": float64(1541152487000),
  481. }, {
  482. "color": "blue",
  483. "window_start": float64(1541152485000),
  484. "window_end": float64(1541152487000),
  485. }},
  486. {{
  487. "color": "red",
  488. "window_start": float64(1541152486000),
  489. "window_end": float64(1541152488000),
  490. }, {
  491. "color": "blue",
  492. "window_start": float64(1541152486000),
  493. "window_end": float64(1541152488000),
  494. }},
  495. {{
  496. "color": "yellow",
  497. "window_start": float64(1541152487000),
  498. "window_end": float64(1541152489000),
  499. }},
  500. {{
  501. "color": "yellow",
  502. "window_start": float64(1541152488000),
  503. "window_end": float64(1541152490000),
  504. }},
  505. },
  506. M: map[string]interface{}{
  507. "op_4_project_0_exceptions_total": int64(0),
  508. "op_4_project_0_process_latency_us": int64(0),
  509. "op_4_project_0_records_in_total": int64(4),
  510. "op_4_project_0_records_out_total": int64(4),
  511. "sink_mockSink_0_exceptions_total": int64(0),
  512. "sink_mockSink_0_records_in_total": int64(4),
  513. "sink_mockSink_0_records_out_total": int64(4),
  514. "source_demo_0_exceptions_total": int64(0),
  515. "source_demo_0_records_in_total": int64(5),
  516. "source_demo_0_records_out_total": int64(5),
  517. "op_3_window_0_exceptions_total": int64(0),
  518. "op_3_window_0_process_latency_us": int64(0),
  519. "op_3_window_0_records_in_total": int64(3),
  520. "op_3_window_0_records_out_total": int64(4),
  521. },
  522. }, {
  523. Name: `TestCountWindowRule1`,
  524. Sql: `SELECT collect(*)[0]->color as c, window_end() as we FROM demo GROUP BY COUNTWINDOW(3)`,
  525. R: [][]map[string]interface{}{
  526. {{
  527. "c": "red",
  528. "we": 1.541152487632e+12,
  529. }},
  530. },
  531. M: map[string]interface{}{
  532. "op_3_project_0_exceptions_total": int64(0),
  533. "op_3_project_0_process_latency_us": int64(0),
  534. "op_3_project_0_records_in_total": int64(1),
  535. "op_3_project_0_records_out_total": int64(1),
  536. "sink_mockSink_0_exceptions_total": int64(0),
  537. "sink_mockSink_0_records_in_total": int64(1),
  538. "sink_mockSink_0_records_out_total": int64(1),
  539. "source_demo_0_exceptions_total": int64(0),
  540. "source_demo_0_records_in_total": int64(5),
  541. "source_demo_0_records_out_total": int64(5),
  542. "op_2_window_0_exceptions_total": int64(0),
  543. "op_2_window_0_process_latency_us": int64(0),
  544. "op_2_window_0_records_in_total": int64(5),
  545. "op_2_window_0_records_out_total": int64(1),
  546. },
  547. }, {
  548. Name: `TestWindowRule10`,
  549. Sql: `SELECT deduplicate(color, false)->color as c FROM demo GROUP BY SlidingWindow(hh, 1)`,
  550. R: [][]map[string]interface{}{
  551. {{
  552. "c": "red",
  553. }}, {{
  554. "c": "blue",
  555. }}, {{}}, {{
  556. "c": "yellow",
  557. }}, {{}},
  558. },
  559. M: map[string]interface{}{
  560. "op_3_project_0_exceptions_total": int64(0),
  561. "op_3_project_0_process_latency_us": int64(0),
  562. "op_3_project_0_records_in_total": int64(5),
  563. "op_3_project_0_records_out_total": int64(5),
  564. "sink_mockSink_0_exceptions_total": int64(0),
  565. "sink_mockSink_0_records_in_total": int64(5),
  566. "sink_mockSink_0_records_out_total": int64(5),
  567. "source_demo_0_exceptions_total": int64(0),
  568. "source_demo_0_records_in_total": int64(5),
  569. "source_demo_0_records_out_total": int64(5),
  570. "op_2_window_0_exceptions_total": int64(0),
  571. "op_2_window_0_process_latency_us": int64(0),
  572. "op_2_window_0_records_in_total": int64(5),
  573. "op_2_window_0_records_out_total": int64(5),
  574. },
  575. }, {
  576. Name: `TestWindowRule11`,
  577. Sql: `SELECT color, name, window_start(), window_end() FROM demo INNER JOIN table1 on demo.ts = table1.id where demo.size > 2 and table1.size > 1 GROUP BY tumblingwindow(ss, 1)`,
  578. R: [][]map[string]interface{}{
  579. {{
  580. "color": "red",
  581. "name": "name1",
  582. "window_start": float64(1541152486000),
  583. "window_end": float64(1541152487000),
  584. }},
  585. },
  586. M: map[string]interface{}{
  587. "op_2_window_0_exceptions_total": int64(0),
  588. "op_2_window_0_process_latency_us": int64(0),
  589. "op_2_window_0_records_in_total": int64(5),
  590. "op_2_window_0_records_out_total": int64(4),
  591. "op_4_join_aligner_0_records_in_total": int64(5),
  592. "op_4_join_aligner_0_records_out_total": int64(4),
  593. "op_5_join_0_exceptions_total": int64(0),
  594. "op_5_join_0_records_in_total": int64(4),
  595. "op_5_join_0_records_out_total": int64(1),
  596. "op_6_project_0_exceptions_total": int64(0),
  597. "op_6_project_0_records_in_total": int64(1),
  598. "op_6_project_0_records_out_total": int64(1),
  599. "sink_mockSink_0_exceptions_total": int64(0),
  600. "sink_mockSink_0_records_in_total": int64(1),
  601. "sink_mockSink_0_records_out_total": int64(1),
  602. "source_demo_0_exceptions_total": int64(0),
  603. "source_demo_0_records_in_total": int64(5),
  604. "source_demo_0_records_out_total": int64(5),
  605. "source_table1_0_exceptions_total": int64(0),
  606. "source_table1_0_records_in_total": int64(4),
  607. "source_table1_0_records_out_total": int64(1),
  608. },
  609. },
  610. }
  611. HandleStream(true, streamList, t)
  612. options := []*api.RuleOption{
  613. {
  614. BufferLength: 100,
  615. SendError: true,
  616. }, {
  617. BufferLength: 100,
  618. SendError: true,
  619. Qos: api.AtLeastOnce,
  620. CheckpointInterval: 5000,
  621. }, {
  622. BufferLength: 100,
  623. SendError: true,
  624. Qos: api.ExactlyOnce,
  625. CheckpointInterval: 5000,
  626. },
  627. }
  628. for j, opt := range options {
  629. DoRuleTest(t, tests, j, opt, 15)
  630. }
  631. }
  632. func TestEventWindow(t *testing.T) {
  633. //Reset
  634. streamList := []string{"demoE", "demoErr", "demo1E", "sessionDemoE"}
  635. HandleStream(false, streamList, t)
  636. var tests = []RuleTest{
  637. {
  638. Name: `TestEventWindowRule1`,
  639. Sql: `SELECT * FROM demoE GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
  640. R: [][]map[string]interface{}{
  641. {{
  642. "color": "red",
  643. "size": float64(3),
  644. "ts": float64(1541152486013),
  645. }},
  646. {{
  647. "color": "red",
  648. "size": float64(3),
  649. "ts": float64(1541152486013),
  650. }, {
  651. "color": "blue",
  652. "size": float64(2),
  653. "ts": float64(1541152487632),
  654. }},
  655. {{
  656. "color": "blue",
  657. "size": float64(2),
  658. "ts": float64(1541152487632),
  659. }, {
  660. "color": "yellow",
  661. "size": float64(4),
  662. "ts": float64(1541152488442),
  663. }}, {{
  664. "color": "yellow",
  665. "size": float64(4),
  666. "ts": float64(1541152488442),
  667. }, {
  668. "color": "red",
  669. "size": float64(1),
  670. "ts": float64(1541152489252),
  671. }}, {{
  672. "color": "red",
  673. "size": float64(1),
  674. "ts": float64(1541152489252),
  675. }},
  676. },
  677. M: map[string]interface{}{
  678. "op_3_project_0_exceptions_total": int64(0),
  679. "op_3_project_0_process_latency_us": int64(0),
  680. "op_3_project_0_records_in_total": int64(5),
  681. "op_3_project_0_records_out_total": int64(5),
  682. "sink_mockSink_0_exceptions_total": int64(0),
  683. "sink_mockSink_0_records_in_total": int64(5),
  684. "sink_mockSink_0_records_out_total": int64(5),
  685. "source_demoE_0_exceptions_total": int64(0),
  686. "source_demoE_0_records_in_total": int64(6),
  687. "source_demoE_0_records_out_total": int64(6),
  688. "op_2_window_0_exceptions_total": int64(0),
  689. "op_2_window_0_process_latency_us": int64(0),
  690. "op_2_window_0_records_in_total": int64(6),
  691. "op_2_window_0_records_out_total": int64(5),
  692. },
  693. }, {
  694. Name: `TestEventWindowRule2`,
  695. Sql: `SELECT window_start(), window_end(), color, ts FROM demoE where size > 2 GROUP BY tumblingwindow(ss, 1)`,
  696. R: [][]map[string]interface{}{
  697. {{
  698. "window_start": float64(1541152486000),
  699. "window_end": float64(1541152487000),
  700. "color": "red",
  701. "ts": float64(1541152486013),
  702. }},
  703. {{
  704. "window_start": float64(1541152488000),
  705. "window_end": float64(1541152489000),
  706. "color": "yellow",
  707. "ts": float64(1541152488442),
  708. }},
  709. },
  710. M: map[string]interface{}{
  711. "op_4_project_0_exceptions_total": int64(0),
  712. "op_4_project_0_process_latency_us": int64(0),
  713. "op_4_project_0_records_in_total": int64(2),
  714. "op_4_project_0_records_out_total": int64(2),
  715. "sink_mockSink_0_exceptions_total": int64(0),
  716. "sink_mockSink_0_records_in_total": int64(2),
  717. "sink_mockSink_0_records_out_total": int64(2),
  718. "source_demoE_0_exceptions_total": int64(0),
  719. "source_demoE_0_records_in_total": int64(6),
  720. "source_demoE_0_records_out_total": int64(6),
  721. "op_2_window_0_exceptions_total": int64(0),
  722. "op_2_window_0_process_latency_us": int64(0),
  723. "op_2_window_0_records_in_total": int64(6),
  724. "op_2_window_0_records_out_total": int64(4),
  725. "op_3_filter_0_exceptions_total": int64(0),
  726. "op_3_filter_0_process_latency_us": int64(0),
  727. "op_3_filter_0_records_in_total": int64(4),
  728. "op_3_filter_0_records_out_total": int64(2),
  729. },
  730. }, {
  731. Name: `TestEventWindowRule3`,
  732. Sql: `SELECT color, temp, demoE.ts FROM demoE INNER JOIN demo1E ON demoE.ts = demo1E.ts GROUP BY SlidingWindow(ss, 1)`,
  733. R: [][]map[string]interface{}{
  734. {{
  735. "color": "red",
  736. "temp": 25.5,
  737. "ts": float64(1541152486013),
  738. }}, {{
  739. "color": "red",
  740. "temp": 25.5,
  741. "ts": float64(1541152486013),
  742. }}, {{
  743. "color": "blue",
  744. "temp": 28.1,
  745. "ts": float64(1541152487632),
  746. }}, {{
  747. "color": "blue",
  748. "temp": 28.1,
  749. "ts": float64(1541152487632),
  750. }, {
  751. "color": "yellow",
  752. "temp": 27.4,
  753. "ts": float64(1541152488442),
  754. }}, {{
  755. "color": "yellow",
  756. "temp": 27.4,
  757. "ts": float64(1541152488442),
  758. }, {
  759. "color": "red",
  760. "temp": 25.5,
  761. "ts": float64(1541152489252),
  762. }},
  763. },
  764. M: map[string]interface{}{
  765. "op_5_project_0_exceptions_total": int64(0),
  766. "op_5_project_0_process_latency_us": int64(0),
  767. "op_5_project_0_records_in_total": int64(5),
  768. "op_5_project_0_records_out_total": int64(5),
  769. "sink_mockSink_0_exceptions_total": int64(0),
  770. "sink_mockSink_0_records_in_total": int64(5),
  771. "sink_mockSink_0_records_out_total": int64(5),
  772. "source_demoE_0_exceptions_total": int64(0),
  773. "source_demoE_0_records_in_total": int64(6),
  774. "source_demoE_0_records_out_total": int64(6),
  775. "source_demo1E_0_exceptions_total": int64(0),
  776. "source_demo1E_0_records_in_total": int64(6),
  777. "source_demo1E_0_records_out_total": int64(6),
  778. "op_3_window_0_exceptions_total": int64(0),
  779. "op_3_window_0_process_latency_us": int64(0),
  780. "op_3_window_0_records_in_total": int64(12),
  781. "op_3_window_0_records_out_total": int64(5),
  782. "op_4_join_0_exceptions_total": int64(0),
  783. "op_4_join_0_process_latency_us": int64(0),
  784. "op_4_join_0_records_in_total": int64(5),
  785. "op_4_join_0_records_out_total": int64(5),
  786. },
  787. }, {
  788. Name: `TestEventWindowRule4`,
  789. Sql: `SELECT window_start() as ws, color, window_end() as we FROM demoE GROUP BY SlidingWindow(ss, 2), color ORDER BY color`,
  790. R: [][]map[string]interface{}{
  791. {{
  792. "color": "red",
  793. "ws": float64(1541152484013),
  794. "we": float64(1541152486013),
  795. }}, {{
  796. "color": "blue",
  797. "ws": float64(1541152485632),
  798. "we": float64(1541152487632),
  799. }, {
  800. "color": "red",
  801. "ws": float64(1541152485632),
  802. "we": float64(1541152487632),
  803. }}, {{
  804. "color": "blue",
  805. "ws": float64(1541152486442),
  806. "we": float64(1541152488442),
  807. }, {
  808. "color": "yellow",
  809. "ws": float64(1541152486442),
  810. "we": float64(1541152488442),
  811. }}, {{
  812. "color": "blue",
  813. "ws": float64(1541152487252),
  814. "we": float64(1541152489252),
  815. }, {
  816. "color": "red",
  817. "ws": float64(1541152487252),
  818. "we": float64(1541152489252),
  819. }, {
  820. "color": "yellow",
  821. "ws": float64(1541152487252),
  822. "we": float64(1541152489252),
  823. }},
  824. },
  825. M: map[string]interface{}{
  826. "op_5_project_0_exceptions_total": int64(0),
  827. "op_5_project_0_process_latency_us": int64(0),
  828. "op_5_project_0_records_in_total": int64(4),
  829. "op_5_project_0_records_out_total": int64(4),
  830. "sink_mockSink_0_exceptions_total": int64(0),
  831. "sink_mockSink_0_records_in_total": int64(4),
  832. "sink_mockSink_0_records_out_total": int64(4),
  833. "source_demoE_0_exceptions_total": int64(0),
  834. "source_demoE_0_records_in_total": int64(6),
  835. "source_demoE_0_records_out_total": int64(6),
  836. "op_2_window_0_exceptions_total": int64(0),
  837. "op_2_window_0_process_latency_us": int64(0),
  838. "op_2_window_0_records_in_total": int64(6),
  839. "op_2_window_0_records_out_total": int64(4),
  840. "op_3_aggregate_0_exceptions_total": int64(0),
  841. "op_3_aggregate_0_process_latency_us": int64(0),
  842. "op_3_aggregate_0_records_in_total": int64(4),
  843. "op_3_aggregate_0_records_out_total": int64(4),
  844. "op_4_order_0_exceptions_total": int64(0),
  845. "op_4_order_0_process_latency_us": int64(0),
  846. "op_4_order_0_records_in_total": int64(4),
  847. "op_4_order_0_records_out_total": int64(4),
  848. },
  849. }, {
  850. Name: `TestEventWindowRule5`,
  851. Sql: `SELECT temp FROM sessionDemoE GROUP BY SessionWindow(ss, 2, 1) `,
  852. R: [][]map[string]interface{}{
  853. {{
  854. "temp": 25.5,
  855. }}, {{
  856. "temp": 28.1,
  857. }, {
  858. "temp": 27.4,
  859. }, {
  860. "temp": 25.5,
  861. }}, {{
  862. "temp": 26.2,
  863. }, {
  864. "temp": 26.8,
  865. }, {
  866. "temp": 28.9,
  867. }, {
  868. "temp": 29.1,
  869. }, {
  870. "temp": 32.2,
  871. }}, {{
  872. "temp": 30.9,
  873. }},
  874. },
  875. M: map[string]interface{}{
  876. "op_3_project_0_exceptions_total": int64(0),
  877. "op_3_project_0_process_latency_us": int64(0),
  878. "op_3_project_0_records_in_total": int64(4),
  879. "op_3_project_0_records_out_total": int64(4),
  880. "sink_mockSink_0_exceptions_total": int64(0),
  881. "sink_mockSink_0_records_in_total": int64(4),
  882. "sink_mockSink_0_records_out_total": int64(4),
  883. "source_sessionDemoE_0_exceptions_total": int64(0),
  884. "source_sessionDemoE_0_records_in_total": int64(12),
  885. "source_sessionDemoE_0_records_out_total": int64(12),
  886. "op_2_window_0_exceptions_total": int64(0),
  887. "op_2_window_0_process_latency_us": int64(0),
  888. "op_2_window_0_records_in_total": int64(12),
  889. "op_2_window_0_records_out_total": int64(4),
  890. },
  891. }, {
  892. Name: `TestEventWindowRule6`,
  893. Sql: `SELECT max(temp) as m, count(color) as c FROM demoE INNER JOIN demo1E ON demoE.ts = demo1E.ts GROUP BY SlidingWindow(ss, 1)`,
  894. R: [][]map[string]interface{}{
  895. {{
  896. "m": 25.5,
  897. "c": float64(1),
  898. }}, {{
  899. "m": 25.5,
  900. "c": float64(1),
  901. }}, {{
  902. "m": 28.1,
  903. "c": float64(1),
  904. }}, {{
  905. "m": 28.1,
  906. "c": float64(2),
  907. }}, {{
  908. "m": 27.4,
  909. "c": float64(2),
  910. }},
  911. },
  912. M: map[string]interface{}{
  913. "op_5_project_0_exceptions_total": int64(0),
  914. "op_5_project_0_process_latency_us": int64(0),
  915. "op_5_project_0_records_in_total": int64(5),
  916. "op_5_project_0_records_out_total": int64(5),
  917. "sink_mockSink_0_exceptions_total": int64(0),
  918. "sink_mockSink_0_records_in_total": int64(5),
  919. "sink_mockSink_0_records_out_total": int64(5),
  920. "source_demoE_0_exceptions_total": int64(0),
  921. "source_demoE_0_records_in_total": int64(6),
  922. "source_demoE_0_records_out_total": int64(6),
  923. "source_demo1E_0_exceptions_total": int64(0),
  924. "source_demo1E_0_records_in_total": int64(6),
  925. "source_demo1E_0_records_out_total": int64(6),
  926. "op_3_window_0_exceptions_total": int64(0),
  927. "op_3_window_0_records_in_total": int64(12),
  928. "op_3_window_0_records_out_total": int64(5),
  929. "op_4_join_0_exceptions_total": int64(0),
  930. "op_4_join_0_process_latency_us": int64(0),
  931. "op_4_join_0_records_in_total": int64(5),
  932. "op_4_join_0_records_out_total": int64(5),
  933. },
  934. }, {
  935. Name: `TestEventWindowRule7`,
  936. Sql: `SELECT * FROM demoErr GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
  937. R: [][]map[string]interface{}{
  938. {{
  939. "error": "error in preprocessor: invalid data type for color, expect string but found int(2)",
  940. }},
  941. {{
  942. "color": "red",
  943. "size": float64(3),
  944. "ts": float64(1541152486013),
  945. }},
  946. {{
  947. "color": "red",
  948. "size": float64(3),
  949. "ts": float64(1541152486013),
  950. }},
  951. {{
  952. "color": "yellow",
  953. "size": float64(4),
  954. "ts": float64(1541152488442),
  955. }}, {{
  956. "color": "yellow",
  957. "size": float64(4),
  958. "ts": float64(1541152488442),
  959. }, {
  960. "color": "red",
  961. "size": float64(1),
  962. "ts": float64(1541152489252),
  963. }}, {{
  964. "color": "red",
  965. "size": float64(1),
  966. "ts": float64(1541152489252),
  967. }},
  968. },
  969. M: map[string]interface{}{
  970. "op_3_project_0_exceptions_total": int64(1),
  971. "op_3_project_0_process_latency_us": int64(0),
  972. "op_3_project_0_records_in_total": int64(6),
  973. "op_3_project_0_records_out_total": int64(5),
  974. "sink_mockSink_0_exceptions_total": int64(0),
  975. "sink_mockSink_0_records_in_total": int64(6),
  976. "sink_mockSink_0_records_out_total": int64(6),
  977. "source_demoErr_0_exceptions_total": int64(1),
  978. "source_demoErr_0_records_in_total": int64(6),
  979. "source_demoErr_0_records_out_total": int64(6),
  980. "op_2_window_0_exceptions_total": int64(1),
  981. "op_2_window_0_process_latency_us": int64(0),
  982. "op_2_window_0_records_in_total": int64(6),
  983. "op_2_window_0_records_out_total": int64(5),
  984. },
  985. }, {
  986. Name: `TestEventWindowRule8`,
  987. Sql: `SELECT temp, window_start(), window_end() FROM sessionDemoE GROUP BY SessionWindow(ss, 2, 1) `,
  988. R: [][]map[string]interface{}{
  989. {{
  990. "temp": 25.5,
  991. "window_start": float64(1541152486013),
  992. "window_end": float64(1541152487013),
  993. }}, {{
  994. "temp": 28.1,
  995. "window_start": float64(1541152487932),
  996. "window_end": float64(1541152490000),
  997. }, {
  998. "temp": 27.4,
  999. "window_start": float64(1541152487932),
  1000. "window_end": float64(1541152490000),
  1001. }, {
  1002. "temp": 25.5,
  1003. "window_start": float64(1541152487932),
  1004. "window_end": float64(1541152490000),
  1005. }}, {{
  1006. "temp": 26.2,
  1007. "window_start": float64(1541152490000),
  1008. "window_end": float64(1541152494000),
  1009. }, {
  1010. "temp": 26.8,
  1011. "window_start": float64(1541152490000),
  1012. "window_end": float64(1541152494000),
  1013. }, {
  1014. "temp": 28.9,
  1015. "window_start": float64(1541152490000),
  1016. "window_end": float64(1541152494000),
  1017. }, {
  1018. "temp": 29.1,
  1019. "window_start": float64(1541152490000),
  1020. "window_end": float64(1541152494000),
  1021. }, {
  1022. "temp": 32.2,
  1023. "window_start": float64(1541152490000),
  1024. "window_end": float64(1541152494000),
  1025. }}, {{
  1026. "temp": 30.9,
  1027. "window_start": float64(1541152494000),
  1028. "window_end": float64(1541152495112),
  1029. }},
  1030. },
  1031. M: map[string]interface{}{
  1032. "op_3_project_0_exceptions_total": int64(0),
  1033. "op_3_project_0_process_latency_us": int64(0),
  1034. "op_3_project_0_records_in_total": int64(4),
  1035. "op_3_project_0_records_out_total": int64(4),
  1036. "sink_mockSink_0_exceptions_total": int64(0),
  1037. "sink_mockSink_0_records_in_total": int64(4),
  1038. "sink_mockSink_0_records_out_total": int64(4),
  1039. "source_sessionDemoE_0_exceptions_total": int64(0),
  1040. "source_sessionDemoE_0_records_in_total": int64(12),
  1041. "source_sessionDemoE_0_records_out_total": int64(12),
  1042. "op_2_window_0_exceptions_total": int64(0),
  1043. "op_2_window_0_process_latency_us": int64(0),
  1044. "op_2_window_0_records_in_total": int64(12),
  1045. "op_2_window_0_records_out_total": int64(4),
  1046. },
  1047. }, {
  1048. Name: `TestEventWindowRule9`,
  1049. Sql: `SELECT window_end(), color, window_start() FROM demoE GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
  1050. R: [][]map[string]interface{}{
  1051. {{
  1052. "color": "red",
  1053. "window_start": float64(1541152485000),
  1054. "window_end": float64(1541152487000),
  1055. }},
  1056. {{
  1057. "color": "red",
  1058. "window_start": float64(1541152486000),
  1059. "window_end": float64(1541152488000),
  1060. }, {
  1061. "color": "blue",
  1062. "window_start": float64(1541152486000),
  1063. "window_end": float64(1541152488000),
  1064. }},
  1065. {{
  1066. "color": "blue",
  1067. "window_start": float64(1541152487000),
  1068. "window_end": float64(1541152489000),
  1069. }, {
  1070. "color": "yellow",
  1071. "window_start": float64(1541152487000),
  1072. "window_end": float64(1541152489000),
  1073. }}, {{
  1074. "color": "yellow",
  1075. "window_start": float64(1541152488000),
  1076. "window_end": float64(1541152490000),
  1077. }, {
  1078. "color": "red",
  1079. "window_start": float64(1541152488000),
  1080. "window_end": float64(1541152490000),
  1081. }}, {{
  1082. "color": "red",
  1083. "window_start": float64(1541152489000),
  1084. "window_end": float64(1541152491000),
  1085. }},
  1086. },
  1087. M: map[string]interface{}{
  1088. "op_3_project_0_exceptions_total": int64(0),
  1089. "op_3_project_0_process_latency_us": int64(0),
  1090. "op_3_project_0_records_in_total": int64(5),
  1091. "op_3_project_0_records_out_total": int64(5),
  1092. "sink_mockSink_0_exceptions_total": int64(0),
  1093. "sink_mockSink_0_records_in_total": int64(5),
  1094. "sink_mockSink_0_records_out_total": int64(5),
  1095. "source_demoE_0_exceptions_total": int64(0),
  1096. "source_demoE_0_records_in_total": int64(6),
  1097. "source_demoE_0_records_out_total": int64(6),
  1098. "op_2_window_0_exceptions_total": int64(0),
  1099. "op_2_window_0_process_latency_us": int64(0),
  1100. "op_2_window_0_records_in_total": int64(6),
  1101. "op_2_window_0_records_out_total": int64(5),
  1102. },
  1103. },
  1104. }
  1105. HandleStream(true, streamList, t)
  1106. options := []*api.RuleOption{
  1107. {
  1108. BufferLength: 100,
  1109. SendError: true,
  1110. IsEventTime: true,
  1111. LateTol: 1000,
  1112. }, {
  1113. BufferLength: 100,
  1114. SendError: true,
  1115. Qos: api.AtLeastOnce,
  1116. CheckpointInterval: 5000,
  1117. IsEventTime: true,
  1118. LateTol: 1000,
  1119. }, {
  1120. BufferLength: 100,
  1121. SendError: true,
  1122. Qos: api.ExactlyOnce,
  1123. CheckpointInterval: 5000,
  1124. IsEventTime: true,
  1125. LateTol: 1000,
  1126. },
  1127. }
  1128. for j, opt := range options {
  1129. DoRuleTest(t, tests, j, opt, 10)
  1130. }
  1131. }
  1132. func TestWindowError(t *testing.T) {
  1133. //Reset
  1134. streamList := []string{"ldemo", "ldemo1"}
  1135. HandleStream(false, streamList, t)
  1136. var tests = []RuleTest{
  1137. {
  1138. Name: `TestWindowErrorRule1`,
  1139. Sql: `SELECT size * 3 FROM ldemo GROUP BY TUMBLINGWINDOW(ss, 2)`,
  1140. R: [][]map[string]interface{}{
  1141. {{
  1142. "error": "run Select error: invalid operation string(string) * int64(3)",
  1143. }}, {{
  1144. "kuiper_field_0": float64(6),
  1145. }, {}},
  1146. },
  1147. M: map[string]interface{}{
  1148. "op_3_project_0_exceptions_total": int64(1),
  1149. "op_3_project_0_process_latency_us": int64(0),
  1150. "op_3_project_0_records_in_total": int64(2),
  1151. "op_3_project_0_records_out_total": int64(1),
  1152. "sink_mockSink_0_exceptions_total": int64(0),
  1153. "sink_mockSink_0_records_in_total": int64(2),
  1154. "sink_mockSink_0_records_out_total": int64(2),
  1155. "source_ldemo_0_exceptions_total": int64(0),
  1156. "source_ldemo_0_records_in_total": int64(5),
  1157. "source_ldemo_0_records_out_total": int64(5),
  1158. "op_2_window_0_exceptions_total": int64(0),
  1159. "op_2_window_0_process_latency_us": int64(0),
  1160. "op_2_window_0_records_in_total": int64(5),
  1161. "op_2_window_0_records_out_total": int64(2),
  1162. },
  1163. }, {
  1164. Name: `TestWindowErrorRule2`,
  1165. Sql: `SELECT color, ts FROM ldemo where size > 2 GROUP BY tumblingwindow(ss, 1)`,
  1166. R: [][]map[string]interface{}{
  1167. {{
  1168. "error": "run Where error: invalid operation string(string) > int64(2)",
  1169. }}, {{
  1170. "color": "red",
  1171. "ts": float64(1541152486013),
  1172. }}, {{
  1173. "ts": float64(1541152487632),
  1174. }},
  1175. },
  1176. M: map[string]interface{}{
  1177. "op_4_project_0_exceptions_total": int64(1),
  1178. "op_4_project_0_process_latency_us": int64(0),
  1179. "op_4_project_0_records_in_total": int64(3),
  1180. "op_4_project_0_records_out_total": int64(2),
  1181. "sink_mockSink_0_exceptions_total": int64(0),
  1182. "sink_mockSink_0_records_in_total": int64(3),
  1183. "sink_mockSink_0_records_out_total": int64(3),
  1184. "source_ldemo_0_exceptions_total": int64(0),
  1185. "source_ldemo_0_records_in_total": int64(5),
  1186. "source_ldemo_0_records_out_total": int64(5),
  1187. "op_3_window_0_exceptions_total": int64(1),
  1188. "op_3_window_0_process_latency_us": int64(0),
  1189. "op_3_window_0_records_in_total": int64(3),
  1190. "op_3_window_0_records_out_total": int64(2),
  1191. "op_2_filter_0_exceptions_total": int64(1),
  1192. "op_2_filter_0_process_latency_us": int64(0),
  1193. "op_2_filter_0_records_in_total": int64(5),
  1194. "op_2_filter_0_records_out_total": int64(2),
  1195. },
  1196. }, {
  1197. Name: `TestWindowErrorRule3`,
  1198. Sql: `SELECT color, temp, ts FROM ldemo INNER JOIN ldemo1 ON ldemo.ts = ldemo1.ts GROUP BY SlidingWindow(ss, 1)`,
  1199. R: [][]map[string]interface{}{
  1200. {{
  1201. "color": "red",
  1202. "temp": 25.5,
  1203. "ts": float64(1541152486013),
  1204. }}, {{
  1205. "color": "red",
  1206. "temp": 25.5,
  1207. "ts": float64(1541152486013),
  1208. }}, {{
  1209. "color": "red",
  1210. "temp": 25.5,
  1211. "ts": float64(1541152486013),
  1212. }}, {{
  1213. "temp": 28.1,
  1214. "ts": float64(1541152487632),
  1215. }}, {{
  1216. "temp": 28.1,
  1217. "ts": float64(1541152487632),
  1218. }}, {{
  1219. "error": "run Join error: invalid operation int64(1541152487632) = string(1541152488442)",
  1220. }}, {{
  1221. "error": "run Join error: invalid operation int64(1541152488442) = string(1541152488442)",
  1222. }}, {{
  1223. "error": "run Join error: invalid operation int64(1541152488442) = string(1541152488442)",
  1224. }},
  1225. },
  1226. M: map[string]interface{}{
  1227. "op_5_project_0_exceptions_total": int64(3),
  1228. "op_5_project_0_process_latency_us": int64(0),
  1229. "op_5_project_0_records_in_total": int64(8),
  1230. "op_5_project_0_records_out_total": int64(5),
  1231. "sink_mockSink_0_exceptions_total": int64(0),
  1232. "sink_mockSink_0_records_in_total": int64(8),
  1233. "sink_mockSink_0_records_out_total": int64(8),
  1234. "source_ldemo_0_exceptions_total": int64(0),
  1235. "source_ldemo_0_records_in_total": int64(5),
  1236. "source_ldemo_0_records_out_total": int64(5),
  1237. "source_ldemo1_0_exceptions_total": int64(0),
  1238. "source_ldemo1_0_records_in_total": int64(5),
  1239. "source_ldemo1_0_records_out_total": int64(5),
  1240. "op_3_window_0_exceptions_total": int64(0),
  1241. "op_3_window_0_process_latency_us": int64(0),
  1242. "op_3_window_0_records_in_total": int64(10),
  1243. "op_3_window_0_records_out_total": int64(10),
  1244. "op_4_join_0_exceptions_total": int64(3),
  1245. "op_4_join_0_process_latency_us": int64(0),
  1246. "op_4_join_0_records_in_total": int64(10),
  1247. "op_4_join_0_records_out_total": int64(5),
  1248. },
  1249. }, {
  1250. Name: `TestWindowErrorRule4`,
  1251. Sql: `SELECT color FROM ldemo GROUP BY SlidingWindow(ss, 2), color having collect(size)[0] >= 2 order by color`,
  1252. R: [][]map[string]interface{}{
  1253. {{
  1254. "color": "red",
  1255. }}, {{
  1256. "error": "run Having error: invalid operation string(string) >= int64(2)",
  1257. }}, {{
  1258. "error": "run Having error: invalid operation string(string) >= int64(2)",
  1259. }}, {{
  1260. "error": "run Having error: invalid operation string(string) >= int64(2)",
  1261. }}, {{
  1262. "color": float64(49),
  1263. }, {}},
  1264. },
  1265. M: map[string]interface{}{
  1266. "op_6_project_0_exceptions_total": int64(3),
  1267. "op_6_project_0_process_latency_us": int64(0),
  1268. "op_6_project_0_records_in_total": int64(5),
  1269. "op_6_project_0_records_out_total": int64(2),
  1270. "sink_mockSink_0_exceptions_total": int64(0),
  1271. "sink_mockSink_0_records_in_total": int64(5),
  1272. "sink_mockSink_0_records_out_total": int64(5),
  1273. "source_ldemo_0_exceptions_total": int64(0),
  1274. "source_ldemo_0_records_in_total": int64(5),
  1275. "source_ldemo_0_records_out_total": int64(5),
  1276. "op_2_window_0_exceptions_total": int64(0),
  1277. "op_2_window_0_process_latency_us": int64(0),
  1278. "op_2_window_0_records_in_total": int64(5),
  1279. "op_2_window_0_records_out_total": int64(5),
  1280. "op_3_aggregate_0_exceptions_total": int64(0),
  1281. "op_3_aggregate_0_process_latency_us": int64(0),
  1282. "op_3_aggregate_0_records_in_total": int64(5),
  1283. "op_3_aggregate_0_records_out_total": int64(5),
  1284. "op_4_having_0_exceptions_total": int64(3),
  1285. "op_4_having_0_process_latency_us": int64(0),
  1286. "op_4_having_0_records_in_total": int64(5),
  1287. "op_4_having_0_records_out_total": int64(2),
  1288. },
  1289. }, {
  1290. Name: `TestWindowErrorRule5`,
  1291. Sql: `SELECT color, size FROM ldemo GROUP BY tumblingwindow(ss, 1) ORDER BY size`,
  1292. R: [][]map[string]interface{}{
  1293. {{
  1294. "error": "run Order By error: incompatible types for comparison: int and string",
  1295. }}, {{
  1296. "size": float64(3),
  1297. }}, {{
  1298. "color": float64(49),
  1299. "size": float64(2),
  1300. }}, {{
  1301. "color": "red",
  1302. }},
  1303. },
  1304. M: map[string]interface{}{
  1305. "op_4_project_0_exceptions_total": int64(1),
  1306. "op_4_project_0_process_latency_us": int64(0),
  1307. "op_4_project_0_records_in_total": int64(4),
  1308. "op_4_project_0_records_out_total": int64(3),
  1309. "sink_mockSink_0_exceptions_total": int64(0),
  1310. "sink_mockSink_0_records_in_total": int64(4),
  1311. "sink_mockSink_0_records_out_total": int64(4),
  1312. "source_ldemo_0_exceptions_total": int64(0),
  1313. "source_ldemo_0_records_in_total": int64(5),
  1314. "source_ldemo_0_records_out_total": int64(5),
  1315. "op_2_window_0_exceptions_total": int64(0),
  1316. "op_2_window_0_process_latency_us": int64(0),
  1317. "op_2_window_0_records_in_total": int64(5),
  1318. "op_2_window_0_records_out_total": int64(4),
  1319. "op_3_order_0_exceptions_total": int64(1),
  1320. "op_3_order_0_process_latency_us": int64(0),
  1321. "op_3_order_0_records_in_total": int64(4),
  1322. "op_3_order_0_records_out_total": int64(3),
  1323. },
  1324. },
  1325. }
  1326. HandleStream(true, streamList, t)
  1327. DoRuleTest(t, tests, 0, &api.RuleOption{
  1328. BufferLength: 100,
  1329. SendError: true,
  1330. }, 0)
  1331. }