window_rule_test.go 45 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441
  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/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][]string{
  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: invalid data type for color, expect string but found int(3)",
  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: invalid data type for color, expect string but found int(7)",
  412. }},
  413. {{
  414. "color": "blue",
  415. "size": float64(2),
  416. "ts": float64(1541152487632),
  417. }},
  418. {{
  419. "error": "error in preprocessor: invalid data type for size, expect bigint but found string(blue)",
  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. "window_start": float64(1541152488000),
  707. "window_end": float64(1541152489000),
  708. "color": "yellow",
  709. "ts": float64(1541152488442),
  710. }},
  711. },
  712. M: map[string]interface{}{
  713. "op_4_project_0_exceptions_total": int64(0),
  714. "op_4_project_0_process_latency_us": int64(0),
  715. "op_4_project_0_records_in_total": int64(2),
  716. "op_4_project_0_records_out_total": int64(2),
  717. "sink_mockSink_0_exceptions_total": int64(0),
  718. "sink_mockSink_0_records_in_total": int64(2),
  719. "sink_mockSink_0_records_out_total": int64(2),
  720. "source_demoE_0_exceptions_total": int64(0),
  721. "source_demoE_0_records_in_total": int64(6),
  722. "source_demoE_0_records_out_total": int64(6),
  723. "op_2_window_0_exceptions_total": int64(0),
  724. "op_2_window_0_process_latency_us": int64(0),
  725. "op_2_window_0_records_in_total": int64(6),
  726. "op_2_window_0_records_out_total": int64(5),
  727. "op_3_filter_0_exceptions_total": int64(0),
  728. "op_3_filter_0_process_latency_us": int64(0),
  729. "op_3_filter_0_records_in_total": int64(5),
  730. "op_3_filter_0_records_out_total": int64(2),
  731. },
  732. }, {
  733. Name: `TestEventWindowRule3`,
  734. Sql: `SELECT color, temp, demoE.ts FROM demoE INNER JOIN demo1E ON demoE.ts = demo1E.ts GROUP BY SlidingWindow(ss, 1)`,
  735. R: [][]map[string]interface{}{
  736. {{
  737. "color": "red",
  738. "temp": 25.5,
  739. "ts": float64(1541152486013),
  740. }}, {{
  741. "color": "red",
  742. "temp": 25.5,
  743. "ts": float64(1541152486013),
  744. }}, {{
  745. "color": "blue",
  746. "temp": 28.1,
  747. "ts": float64(1541152487632),
  748. }}, {{
  749. "color": "blue",
  750. "temp": 28.1,
  751. "ts": float64(1541152487632),
  752. }, {
  753. "color": "yellow",
  754. "temp": 27.4,
  755. "ts": float64(1541152488442),
  756. }}, {{
  757. "color": "yellow",
  758. "temp": 27.4,
  759. "ts": float64(1541152488442),
  760. }, {
  761. "color": "red",
  762. "temp": 25.5,
  763. "ts": float64(1541152489252),
  764. }},
  765. },
  766. M: map[string]interface{}{
  767. "op_5_project_0_exceptions_total": int64(0),
  768. "op_5_project_0_process_latency_us": int64(0),
  769. "op_5_project_0_records_in_total": int64(5),
  770. "op_5_project_0_records_out_total": int64(5),
  771. "sink_mockSink_0_exceptions_total": int64(0),
  772. "sink_mockSink_0_records_in_total": int64(5),
  773. "sink_mockSink_0_records_out_total": int64(5),
  774. "source_demoE_0_exceptions_total": int64(0),
  775. "source_demoE_0_records_in_total": int64(6),
  776. "source_demoE_0_records_out_total": int64(6),
  777. "source_demo1E_0_exceptions_total": int64(0),
  778. "source_demo1E_0_records_in_total": int64(6),
  779. "source_demo1E_0_records_out_total": int64(6),
  780. "op_3_window_0_exceptions_total": int64(0),
  781. "op_3_window_0_process_latency_us": int64(0),
  782. "op_3_window_0_records_in_total": int64(12),
  783. "op_3_window_0_records_out_total": int64(5),
  784. "op_4_join_0_exceptions_total": int64(0),
  785. "op_4_join_0_process_latency_us": int64(0),
  786. "op_4_join_0_records_in_total": int64(5),
  787. "op_4_join_0_records_out_total": int64(5),
  788. },
  789. }, {
  790. Name: `TestEventWindowRule4`,
  791. Sql: `SELECT window_start() as ws, color, window_end() as we FROM demoE GROUP BY SlidingWindow(ss, 2), color ORDER BY color`,
  792. R: [][]map[string]interface{}{
  793. {{
  794. "color": "red",
  795. "ws": float64(1541152484013),
  796. "we": float64(1541152486013),
  797. }}, {{
  798. "color": "blue",
  799. "ws": float64(1541152485632),
  800. "we": float64(1541152487632),
  801. }, {
  802. "color": "red",
  803. "ws": float64(1541152485632),
  804. "we": float64(1541152487632),
  805. }}, {{
  806. "color": "blue",
  807. "ws": float64(1541152486442),
  808. "we": float64(1541152488442),
  809. }, {
  810. "color": "yellow",
  811. "ws": float64(1541152486442),
  812. "we": float64(1541152488442),
  813. }}, {{
  814. "color": "blue",
  815. "ws": float64(1541152487252),
  816. "we": float64(1541152489252),
  817. }, {
  818. "color": "red",
  819. "ws": float64(1541152487252),
  820. "we": float64(1541152489252),
  821. }, {
  822. "color": "yellow",
  823. "ws": float64(1541152487252),
  824. "we": float64(1541152489252),
  825. }},
  826. },
  827. M: map[string]interface{}{
  828. "op_5_project_0_exceptions_total": int64(0),
  829. "op_5_project_0_process_latency_us": int64(0),
  830. "op_5_project_0_records_in_total": int64(4),
  831. "op_5_project_0_records_out_total": int64(4),
  832. "sink_mockSink_0_exceptions_total": int64(0),
  833. "sink_mockSink_0_records_in_total": int64(4),
  834. "sink_mockSink_0_records_out_total": int64(4),
  835. "source_demoE_0_exceptions_total": int64(0),
  836. "source_demoE_0_records_in_total": int64(6),
  837. "source_demoE_0_records_out_total": int64(6),
  838. "op_2_window_0_exceptions_total": int64(0),
  839. "op_2_window_0_process_latency_us": int64(0),
  840. "op_2_window_0_records_in_total": int64(6),
  841. "op_2_window_0_records_out_total": int64(4),
  842. "op_3_aggregate_0_exceptions_total": int64(0),
  843. "op_3_aggregate_0_process_latency_us": int64(0),
  844. "op_3_aggregate_0_records_in_total": int64(4),
  845. "op_3_aggregate_0_records_out_total": int64(4),
  846. "op_4_order_0_exceptions_total": int64(0),
  847. "op_4_order_0_process_latency_us": int64(0),
  848. "op_4_order_0_records_in_total": int64(4),
  849. "op_4_order_0_records_out_total": int64(4),
  850. },
  851. }, {
  852. Name: `TestEventWindowRule5`,
  853. Sql: `SELECT temp FROM sessionDemoE GROUP BY SessionWindow(ss, 2, 1) `,
  854. R: [][]map[string]interface{}{
  855. {{
  856. "temp": 25.5,
  857. }}, {{
  858. "temp": 28.1,
  859. }, {
  860. "temp": 27.4,
  861. }, {
  862. "temp": 25.5,
  863. }}, {{
  864. "temp": 26.2,
  865. }, {
  866. "temp": 26.8,
  867. }, {
  868. "temp": 28.9,
  869. }, {
  870. "temp": 29.1,
  871. }, {
  872. "temp": 32.2,
  873. }}, {{
  874. "temp": 30.9,
  875. }},
  876. },
  877. M: map[string]interface{}{
  878. "op_3_project_0_exceptions_total": int64(0),
  879. "op_3_project_0_process_latency_us": int64(0),
  880. "op_3_project_0_records_in_total": int64(4),
  881. "op_3_project_0_records_out_total": int64(4),
  882. "sink_mockSink_0_exceptions_total": int64(0),
  883. "sink_mockSink_0_records_in_total": int64(4),
  884. "sink_mockSink_0_records_out_total": int64(4),
  885. "source_sessionDemoE_0_exceptions_total": int64(0),
  886. "source_sessionDemoE_0_records_in_total": int64(12),
  887. "source_sessionDemoE_0_records_out_total": int64(12),
  888. "op_2_window_0_exceptions_total": int64(0),
  889. "op_2_window_0_process_latency_us": int64(0),
  890. "op_2_window_0_records_in_total": int64(12),
  891. "op_2_window_0_records_out_total": int64(4),
  892. },
  893. }, {
  894. Name: `TestEventWindowRule6`,
  895. 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)`,
  896. R: [][]map[string]interface{}{
  897. {{
  898. "m": 25.5,
  899. "c": float64(1),
  900. }}, {{
  901. "m": 25.5,
  902. "c": float64(1),
  903. }}, {{
  904. "m": 28.1,
  905. "c": float64(1),
  906. }}, {{
  907. "m": 28.1,
  908. "c": float64(2),
  909. }}, {{
  910. "m": 27.4,
  911. "c": float64(2),
  912. }},
  913. },
  914. M: map[string]interface{}{
  915. "op_5_project_0_exceptions_total": int64(0),
  916. "op_5_project_0_process_latency_us": int64(0),
  917. "op_5_project_0_records_in_total": int64(5),
  918. "op_5_project_0_records_out_total": int64(5),
  919. "sink_mockSink_0_exceptions_total": int64(0),
  920. "sink_mockSink_0_records_in_total": int64(5),
  921. "sink_mockSink_0_records_out_total": int64(5),
  922. "source_demoE_0_exceptions_total": int64(0),
  923. "source_demoE_0_records_in_total": int64(6),
  924. "source_demoE_0_records_out_total": int64(6),
  925. "source_demo1E_0_exceptions_total": int64(0),
  926. "source_demo1E_0_records_in_total": int64(6),
  927. "source_demo1E_0_records_out_total": int64(6),
  928. "op_3_window_0_exceptions_total": int64(0),
  929. "op_3_window_0_records_in_total": int64(12),
  930. "op_3_window_0_records_out_total": int64(5),
  931. "op_4_join_0_exceptions_total": int64(0),
  932. "op_4_join_0_process_latency_us": int64(0),
  933. "op_4_join_0_records_in_total": int64(5),
  934. "op_4_join_0_records_out_total": int64(5),
  935. },
  936. }, {
  937. Name: `TestEventWindowRule7`,
  938. Sql: `SELECT * FROM demoErr GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
  939. R: [][]map[string]interface{}{
  940. {{
  941. "error": "error in preprocessor: invalid data type for color, expect string but found int(2)",
  942. }},
  943. {{
  944. "color": "red",
  945. "size": float64(3),
  946. "ts": float64(1541152486013),
  947. }},
  948. {{
  949. "color": "red",
  950. "size": float64(3),
  951. "ts": float64(1541152486013),
  952. }},
  953. {{
  954. "color": "yellow",
  955. "size": float64(4),
  956. "ts": float64(1541152488442),
  957. }}, {{
  958. "color": "yellow",
  959. "size": float64(4),
  960. "ts": float64(1541152488442),
  961. }, {
  962. "color": "red",
  963. "size": float64(1),
  964. "ts": float64(1541152489252),
  965. }}, {{
  966. "color": "red",
  967. "size": float64(1),
  968. "ts": float64(1541152489252),
  969. }},
  970. },
  971. M: map[string]interface{}{
  972. "op_3_project_0_exceptions_total": int64(1),
  973. "op_3_project_0_process_latency_us": int64(0),
  974. "op_3_project_0_records_in_total": int64(6),
  975. "op_3_project_0_records_out_total": int64(5),
  976. "sink_mockSink_0_exceptions_total": int64(0),
  977. "sink_mockSink_0_records_in_total": int64(6),
  978. "sink_mockSink_0_records_out_total": int64(6),
  979. "source_demoErr_0_exceptions_total": int64(1),
  980. "source_demoErr_0_records_in_total": int64(6),
  981. "source_demoErr_0_records_out_total": int64(6),
  982. "op_2_window_0_exceptions_total": int64(1),
  983. "op_2_window_0_process_latency_us": int64(0),
  984. "op_2_window_0_records_in_total": int64(6),
  985. "op_2_window_0_records_out_total": int64(5),
  986. },
  987. }, {
  988. Name: `TestEventWindowRule8`,
  989. Sql: `SELECT temp, window_start(), window_end() FROM sessionDemoE GROUP BY SessionWindow(ss, 2, 1) `,
  990. R: [][]map[string]interface{}{
  991. {{
  992. "temp": 25.5,
  993. "window_start": float64(1541152486013),
  994. "window_end": float64(1541152487013),
  995. }}, {{
  996. "temp": 28.1,
  997. "window_start": float64(1541152487932),
  998. "window_end": float64(1541152490000),
  999. }, {
  1000. "temp": 27.4,
  1001. "window_start": float64(1541152487932),
  1002. "window_end": float64(1541152490000),
  1003. }, {
  1004. "temp": 25.5,
  1005. "window_start": float64(1541152487932),
  1006. "window_end": float64(1541152490000),
  1007. }}, {{
  1008. "temp": 26.2,
  1009. "window_start": float64(1541152490000),
  1010. "window_end": float64(1541152494000),
  1011. }, {
  1012. "temp": 26.8,
  1013. "window_start": float64(1541152490000),
  1014. "window_end": float64(1541152494000),
  1015. }, {
  1016. "temp": 28.9,
  1017. "window_start": float64(1541152490000),
  1018. "window_end": float64(1541152494000),
  1019. }, {
  1020. "temp": 29.1,
  1021. "window_start": float64(1541152490000),
  1022. "window_end": float64(1541152494000),
  1023. }, {
  1024. "temp": 32.2,
  1025. "window_start": float64(1541152490000),
  1026. "window_end": float64(1541152494000),
  1027. }}, {{
  1028. "temp": 30.9,
  1029. "window_start": float64(1541152494000),
  1030. "window_end": float64(1541152495112),
  1031. }},
  1032. },
  1033. M: map[string]interface{}{
  1034. "op_3_project_0_exceptions_total": int64(0),
  1035. "op_3_project_0_process_latency_us": int64(0),
  1036. "op_3_project_0_records_in_total": int64(4),
  1037. "op_3_project_0_records_out_total": int64(4),
  1038. "sink_mockSink_0_exceptions_total": int64(0),
  1039. "sink_mockSink_0_records_in_total": int64(4),
  1040. "sink_mockSink_0_records_out_total": int64(4),
  1041. "source_sessionDemoE_0_exceptions_total": int64(0),
  1042. "source_sessionDemoE_0_records_in_total": int64(12),
  1043. "source_sessionDemoE_0_records_out_total": int64(12),
  1044. "op_2_window_0_exceptions_total": int64(0),
  1045. "op_2_window_0_process_latency_us": int64(0),
  1046. "op_2_window_0_records_in_total": int64(12),
  1047. "op_2_window_0_records_out_total": int64(4),
  1048. },
  1049. }, {
  1050. Name: `TestEventWindowRule9`,
  1051. Sql: `SELECT window_end(), color, window_start() FROM demoE GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
  1052. R: [][]map[string]interface{}{
  1053. {{
  1054. "color": "red",
  1055. "window_start": float64(1541152485000),
  1056. "window_end": float64(1541152487000),
  1057. }},
  1058. {{
  1059. "color": "red",
  1060. "window_start": float64(1541152486000),
  1061. "window_end": float64(1541152488000),
  1062. }, {
  1063. "color": "blue",
  1064. "window_start": float64(1541152486000),
  1065. "window_end": float64(1541152488000),
  1066. }},
  1067. {{
  1068. "color": "blue",
  1069. "window_start": float64(1541152487000),
  1070. "window_end": float64(1541152489000),
  1071. }, {
  1072. "color": "yellow",
  1073. "window_start": float64(1541152487000),
  1074. "window_end": float64(1541152489000),
  1075. }}, {{
  1076. "color": "yellow",
  1077. "window_start": float64(1541152488000),
  1078. "window_end": float64(1541152490000),
  1079. }, {
  1080. "color": "red",
  1081. "window_start": float64(1541152488000),
  1082. "window_end": float64(1541152490000),
  1083. }}, {{
  1084. "color": "red",
  1085. "window_start": float64(1541152489000),
  1086. "window_end": float64(1541152491000),
  1087. }},
  1088. },
  1089. M: map[string]interface{}{
  1090. "op_3_project_0_exceptions_total": int64(0),
  1091. "op_3_project_0_process_latency_us": int64(0),
  1092. "op_3_project_0_records_in_total": int64(5),
  1093. "op_3_project_0_records_out_total": int64(5),
  1094. "sink_mockSink_0_exceptions_total": int64(0),
  1095. "sink_mockSink_0_records_in_total": int64(5),
  1096. "sink_mockSink_0_records_out_total": int64(5),
  1097. "source_demoE_0_exceptions_total": int64(0),
  1098. "source_demoE_0_records_in_total": int64(6),
  1099. "source_demoE_0_records_out_total": int64(6),
  1100. "op_2_window_0_exceptions_total": int64(0),
  1101. "op_2_window_0_process_latency_us": int64(0),
  1102. "op_2_window_0_records_in_total": int64(6),
  1103. "op_2_window_0_records_out_total": int64(5),
  1104. },
  1105. },
  1106. }
  1107. HandleStream(true, streamList, t)
  1108. options := []*api.RuleOption{
  1109. {
  1110. BufferLength: 100,
  1111. SendError: true,
  1112. IsEventTime: true,
  1113. LateTol: 1000,
  1114. }, {
  1115. BufferLength: 100,
  1116. SendError: true,
  1117. Qos: api.AtLeastOnce,
  1118. CheckpointInterval: 5000,
  1119. IsEventTime: true,
  1120. LateTol: 1000,
  1121. }, {
  1122. BufferLength: 100,
  1123. SendError: true,
  1124. Qos: api.ExactlyOnce,
  1125. CheckpointInterval: 5000,
  1126. IsEventTime: true,
  1127. LateTol: 1000,
  1128. },
  1129. }
  1130. for j, opt := range options {
  1131. DoRuleTest(t, tests, j, opt, 10)
  1132. }
  1133. }
  1134. func TestWindowError(t *testing.T) {
  1135. //Reset
  1136. streamList := []string{"ldemo", "ldemo1"}
  1137. HandleStream(false, streamList, t)
  1138. var tests = []RuleTest{
  1139. {
  1140. Name: `TestWindowErrorRule1`,
  1141. Sql: `SELECT size * 3 FROM ldemo GROUP BY TUMBLINGWINDOW(ss, 2)`,
  1142. R: [][]map[string]interface{}{
  1143. {{
  1144. "error": "run Select error: invalid operation string(string) * int64(3)",
  1145. }}, {{
  1146. "kuiper_field_0": float64(6),
  1147. }, {}},
  1148. },
  1149. M: map[string]interface{}{
  1150. "op_3_project_0_exceptions_total": int64(1),
  1151. "op_3_project_0_process_latency_us": int64(0),
  1152. "op_3_project_0_records_in_total": int64(2),
  1153. "op_3_project_0_records_out_total": int64(1),
  1154. "sink_mockSink_0_exceptions_total": int64(0),
  1155. "sink_mockSink_0_records_in_total": int64(2),
  1156. "sink_mockSink_0_records_out_total": int64(2),
  1157. "source_ldemo_0_exceptions_total": int64(0),
  1158. "source_ldemo_0_records_in_total": int64(5),
  1159. "source_ldemo_0_records_out_total": int64(5),
  1160. "op_2_window_0_exceptions_total": int64(0),
  1161. "op_2_window_0_process_latency_us": int64(0),
  1162. "op_2_window_0_records_in_total": int64(5),
  1163. "op_2_window_0_records_out_total": int64(2),
  1164. },
  1165. }, {
  1166. Name: `TestWindowErrorRule2`,
  1167. Sql: `SELECT color, ts FROM ldemo where size > 2 GROUP BY tumblingwindow(ss, 1)`,
  1168. R: [][]map[string]interface{}{
  1169. {{
  1170. "error": "run Where error: invalid operation string(string) > int64(2)",
  1171. }}, {{
  1172. "color": "red",
  1173. "ts": float64(1541152486013),
  1174. }}, {{
  1175. "ts": float64(1541152487632),
  1176. }}, {}, {},
  1177. },
  1178. M: map[string]interface{}{
  1179. "op_4_project_0_exceptions_total": int64(1),
  1180. "op_4_project_0_process_latency_us": int64(0),
  1181. "op_4_project_0_records_in_total": int64(5),
  1182. "op_4_project_0_records_out_total": int64(4),
  1183. "sink_mockSink_0_exceptions_total": int64(0),
  1184. "sink_mockSink_0_records_in_total": int64(5),
  1185. "sink_mockSink_0_records_out_total": int64(5),
  1186. "source_ldemo_0_exceptions_total": int64(0),
  1187. "source_ldemo_0_records_in_total": int64(5),
  1188. "source_ldemo_0_records_out_total": int64(5),
  1189. "op_3_window_0_exceptions_total": int64(1),
  1190. "op_3_window_0_process_latency_us": int64(0),
  1191. "op_3_window_0_records_in_total": int64(3),
  1192. "op_3_window_0_records_out_total": int64(4),
  1193. "op_2_filter_0_exceptions_total": int64(1),
  1194. "op_2_filter_0_process_latency_us": int64(0),
  1195. "op_2_filter_0_records_in_total": int64(5),
  1196. "op_2_filter_0_records_out_total": int64(2),
  1197. },
  1198. }, {
  1199. Name: `TestWindowErrorRule3`,
  1200. Sql: `SELECT color, temp, ts FROM ldemo INNER JOIN ldemo1 ON ldemo.ts = ldemo1.ts GROUP BY SlidingWindow(ss, 1)`,
  1201. R: [][]map[string]interface{}{
  1202. {{
  1203. "color": "red",
  1204. "temp": 25.5,
  1205. "ts": float64(1541152486013),
  1206. }}, {{
  1207. "color": "red",
  1208. "temp": 25.5,
  1209. "ts": float64(1541152486013),
  1210. }}, {{
  1211. "color": "red",
  1212. "temp": 25.5,
  1213. "ts": float64(1541152486013),
  1214. }}, {{
  1215. "temp": 28.1,
  1216. "ts": float64(1541152487632),
  1217. }}, {{
  1218. "temp": 28.1,
  1219. "ts": float64(1541152487632),
  1220. }}, {{
  1221. "error": "run Join error: invalid operation int64(1541152487632) = string(1541152488442)",
  1222. }}, {{
  1223. "error": "run Join error: invalid operation int64(1541152488442) = string(1541152488442)",
  1224. }}, {{
  1225. "error": "run Join error: invalid operation int64(1541152488442) = string(1541152488442)",
  1226. }},
  1227. },
  1228. M: map[string]interface{}{
  1229. "op_5_project_0_exceptions_total": int64(3),
  1230. "op_5_project_0_process_latency_us": int64(0),
  1231. "op_5_project_0_records_in_total": int64(8),
  1232. "op_5_project_0_records_out_total": int64(5),
  1233. "sink_mockSink_0_exceptions_total": int64(0),
  1234. "sink_mockSink_0_records_in_total": int64(8),
  1235. "sink_mockSink_0_records_out_total": int64(8),
  1236. "source_ldemo_0_exceptions_total": int64(0),
  1237. "source_ldemo_0_records_in_total": int64(5),
  1238. "source_ldemo_0_records_out_total": int64(5),
  1239. "source_ldemo1_0_exceptions_total": int64(0),
  1240. "source_ldemo1_0_records_in_total": int64(5),
  1241. "source_ldemo1_0_records_out_total": int64(5),
  1242. "op_3_window_0_exceptions_total": int64(0),
  1243. "op_3_window_0_process_latency_us": int64(0),
  1244. "op_3_window_0_records_in_total": int64(10),
  1245. "op_3_window_0_records_out_total": int64(10),
  1246. "op_4_join_0_exceptions_total": int64(3),
  1247. "op_4_join_0_process_latency_us": int64(0),
  1248. "op_4_join_0_records_in_total": int64(10),
  1249. "op_4_join_0_records_out_total": int64(5),
  1250. },
  1251. }, {
  1252. Name: `TestWindowErrorRule4`,
  1253. Sql: `SELECT color FROM ldemo GROUP BY SlidingWindow(ss, 2), color having collect(size)[0] >= 2 order by color`,
  1254. R: [][]map[string]interface{}{
  1255. {{
  1256. "color": "red",
  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. "error": "run Having error: invalid operation string(string) >= int64(2)",
  1263. }}, {{
  1264. "color": float64(49),
  1265. }, {}},
  1266. },
  1267. M: map[string]interface{}{
  1268. "op_6_project_0_exceptions_total": int64(3),
  1269. "op_6_project_0_process_latency_us": int64(0),
  1270. "op_6_project_0_records_in_total": int64(5),
  1271. "op_6_project_0_records_out_total": int64(2),
  1272. "sink_mockSink_0_exceptions_total": int64(0),
  1273. "sink_mockSink_0_records_in_total": int64(5),
  1274. "sink_mockSink_0_records_out_total": int64(5),
  1275. "source_ldemo_0_exceptions_total": int64(0),
  1276. "source_ldemo_0_records_in_total": int64(5),
  1277. "source_ldemo_0_records_out_total": int64(5),
  1278. "op_2_window_0_exceptions_total": int64(0),
  1279. "op_2_window_0_process_latency_us": int64(0),
  1280. "op_2_window_0_records_in_total": int64(5),
  1281. "op_2_window_0_records_out_total": int64(5),
  1282. "op_3_aggregate_0_exceptions_total": int64(0),
  1283. "op_3_aggregate_0_process_latency_us": int64(0),
  1284. "op_3_aggregate_0_records_in_total": int64(5),
  1285. "op_3_aggregate_0_records_out_total": int64(5),
  1286. "op_4_having_0_exceptions_total": int64(3),
  1287. "op_4_having_0_process_latency_us": int64(0),
  1288. "op_4_having_0_records_in_total": int64(5),
  1289. "op_4_having_0_records_out_total": int64(2),
  1290. },
  1291. }, {
  1292. Name: `TestWindowErrorRule5`,
  1293. Sql: `SELECT color, size FROM ldemo GROUP BY tumblingwindow(ss, 1) ORDER BY size`,
  1294. R: [][]map[string]interface{}{
  1295. {{
  1296. "error": "run Order By error: incompatible types for comparison: int and string",
  1297. }}, {{
  1298. "size": float64(3),
  1299. }}, {{
  1300. "color": float64(49),
  1301. "size": float64(2),
  1302. }}, {{
  1303. "color": "red",
  1304. }},
  1305. },
  1306. M: map[string]interface{}{
  1307. "op_4_project_0_exceptions_total": int64(1),
  1308. "op_4_project_0_process_latency_us": int64(0),
  1309. "op_4_project_0_records_in_total": int64(4),
  1310. "op_4_project_0_records_out_total": int64(3),
  1311. "sink_mockSink_0_exceptions_total": int64(0),
  1312. "sink_mockSink_0_records_in_total": int64(4),
  1313. "sink_mockSink_0_records_out_total": int64(4),
  1314. "source_ldemo_0_exceptions_total": int64(0),
  1315. "source_ldemo_0_records_in_total": int64(5),
  1316. "source_ldemo_0_records_out_total": int64(5),
  1317. "op_2_window_0_exceptions_total": int64(0),
  1318. "op_2_window_0_process_latency_us": int64(0),
  1319. "op_2_window_0_records_in_total": int64(5),
  1320. "op_2_window_0_records_out_total": int64(4),
  1321. "op_3_order_0_exceptions_total": int64(1),
  1322. "op_3_order_0_process_latency_us": int64(0),
  1323. "op_3_order_0_records_in_total": int64(4),
  1324. "op_3_order_0_records_out_total": int64(3),
  1325. },
  1326. },
  1327. }
  1328. HandleStream(true, streamList, t)
  1329. DoRuleTest(t, tests, 0, &api.RuleOption{
  1330. BufferLength: 100,
  1331. SendError: true,
  1332. }, 0)
  1333. }