window_rule_test.go 46 KB

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