window_rule_test.go 46 KB

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