join_test.go 57 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047
  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 operator
  15. import (
  16. "encoding/json"
  17. "errors"
  18. "fmt"
  19. "reflect"
  20. "strings"
  21. "testing"
  22. "github.com/lf-edge/ekuiper/internal/conf"
  23. "github.com/lf-edge/ekuiper/internal/topo/context"
  24. "github.com/lf-edge/ekuiper/internal/xsql"
  25. "github.com/lf-edge/ekuiper/pkg/ast"
  26. "github.com/lf-edge/ekuiper/pkg/cast"
  27. )
  28. func str2Map(s string) map[string]interface{} {
  29. var input map[string]interface{}
  30. if err := json.Unmarshal([]byte(s), &input); err != nil {
  31. fmt.Printf("Failed to parse the JSON data.\n")
  32. return nil
  33. }
  34. return input
  35. }
  36. func TestLeftJoinPlan_Apply(t *testing.T) {
  37. tests := []struct {
  38. sql string
  39. data *xsql.WindowTuples
  40. result interface{}
  41. }{
  42. { // 0
  43. sql: "SELECT id1 FROM src1 left join src2 on id1 = id2",
  44. data: &xsql.WindowTuples{
  45. Content: []xsql.TupleRow{
  46. &xsql.Tuple{
  47. Emitter: "src1",
  48. Message: xsql.Message{"id1": 1, "f1": "v1"},
  49. }, &xsql.Tuple{
  50. Emitter: "src1",
  51. Message: xsql.Message{"id1": 2, "f1": "v2"},
  52. }, &xsql.Tuple{
  53. Emitter: "src1",
  54. Message: xsql.Message{"id1": 3, "f1": "v3"},
  55. },
  56. &xsql.Tuple{
  57. Emitter: "src2",
  58. Message: xsql.Message{"id2": 1, "f2": "w1"},
  59. }, &xsql.Tuple{
  60. Emitter: "src2",
  61. Message: xsql.Message{"id2": 2, "f2": "w2"},
  62. }, &xsql.Tuple{
  63. Emitter: "src2",
  64. Message: xsql.Message{"id2": 4, "f2": "w3"},
  65. },
  66. },
  67. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  68. },
  69. result: &xsql.JoinTuples{
  70. Content: []*xsql.JoinTuple{
  71. {
  72. Tuples: []xsql.TupleRow{
  73. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  74. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  75. },
  76. },
  77. {
  78. Tuples: []xsql.TupleRow{
  79. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  80. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  81. },
  82. },
  83. {
  84. Tuples: []xsql.TupleRow{
  85. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  86. },
  87. },
  88. },
  89. WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
  90. },
  91. },
  92. { // 1
  93. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
  94. data: &xsql.WindowTuples{
  95. Content: []xsql.TupleRow{
  96. &xsql.Tuple{
  97. Emitter: "src1",
  98. Message: xsql.Message{"id1": 1, "f1": "v1"},
  99. }, &xsql.Tuple{
  100. Emitter: "src1",
  101. Message: xsql.Message{"id1": 2, "f1": "v2"},
  102. }, &xsql.Tuple{
  103. Emitter: "src1",
  104. Message: xsql.Message{"id1": 3, "f1": "v3"},
  105. },
  106. &xsql.Tuple{
  107. Emitter: "src2",
  108. Message: xsql.Message{"id2": 1, "f2": "w1"},
  109. }, &xsql.Tuple{
  110. Emitter: "src2",
  111. Message: xsql.Message{"f2": "w2"},
  112. }, &xsql.Tuple{
  113. Emitter: "src2",
  114. Message: xsql.Message{"id2": 4, "f2": "w3"},
  115. },
  116. },
  117. },
  118. result: &xsql.JoinTuples{
  119. Content: []*xsql.JoinTuple{
  120. {
  121. Tuples: []xsql.TupleRow{
  122. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  123. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  124. },
  125. },
  126. {
  127. Tuples: []xsql.TupleRow{
  128. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  129. },
  130. },
  131. {
  132. Tuples: []xsql.TupleRow{
  133. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  134. },
  135. },
  136. },
  137. },
  138. },
  139. { // 2
  140. sql: "SELECT id1 FROM src1 left join src2 on src1.ts = src2.ts",
  141. data: &xsql.WindowTuples{
  142. Content: []xsql.TupleRow{
  143. &xsql.Tuple{
  144. Emitter: "src1",
  145. Message: xsql.Message{"id1": 1, "f1": "v1", "ts": cast.TimeFromUnixMilli(1568854515000)},
  146. }, &xsql.Tuple{
  147. Emitter: "src1",
  148. Message: xsql.Message{"id1": 2, "f1": "v2", "ts": cast.TimeFromUnixMilli(1568854525000)},
  149. }, &xsql.Tuple{
  150. Emitter: "src1",
  151. Message: xsql.Message{"id1": 3, "f1": "v3", "ts": cast.TimeFromUnixMilli(1568854535000)},
  152. },
  153. &xsql.Tuple{
  154. Emitter: "src2",
  155. Message: xsql.Message{"id2": 1, "f2": "w1", "ts": cast.TimeFromUnixMilli(1568854515000)},
  156. }, &xsql.Tuple{
  157. Emitter: "src2",
  158. Message: xsql.Message{"id2": 2, "f2": "w2", "ts": cast.TimeFromUnixMilli(1568854525000)},
  159. }, &xsql.Tuple{
  160. Emitter: "src2",
  161. Message: xsql.Message{"id2": 4, "f2": "w3", "ts": cast.TimeFromUnixMilli(1568854545000)},
  162. },
  163. },
  164. },
  165. result: &xsql.JoinTuples{
  166. Content: []*xsql.JoinTuple{
  167. {
  168. Tuples: []xsql.TupleRow{
  169. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1", "ts": cast.TimeFromUnixMilli(1568854515000)}},
  170. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1", "ts": cast.TimeFromUnixMilli(1568854515000)}},
  171. },
  172. },
  173. {
  174. Tuples: []xsql.TupleRow{
  175. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2", "ts": cast.TimeFromUnixMilli(1568854525000)}},
  176. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2", "ts": cast.TimeFromUnixMilli(1568854525000)}},
  177. },
  178. },
  179. {
  180. Tuples: []xsql.TupleRow{
  181. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3", "ts": cast.TimeFromUnixMilli(1568854535000)}},
  182. },
  183. },
  184. },
  185. },
  186. },
  187. { // 3
  188. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  189. data: &xsql.WindowTuples{
  190. Content: []xsql.TupleRow{
  191. &xsql.Tuple{
  192. Emitter: "src1",
  193. Message: xsql.Message{"id1": 1, "f1": "v1"},
  194. }, &xsql.Tuple{
  195. Emitter: "src1",
  196. Message: xsql.Message{"id1": 2, "f1": "v2"},
  197. }, &xsql.Tuple{
  198. Emitter: "src1",
  199. Message: xsql.Message{"id1": 3, "f1": "v3"},
  200. },
  201. &xsql.Tuple{
  202. Emitter: "src2",
  203. Message: xsql.Message{"id2": 4, "f2": "w1"},
  204. }, &xsql.Tuple{
  205. Emitter: "src2",
  206. Message: xsql.Message{"id2": 5, "f2": "w2"},
  207. }, &xsql.Tuple{
  208. Emitter: "src2",
  209. Message: xsql.Message{"id2": 6, "f2": "w3"},
  210. },
  211. },
  212. },
  213. result: nil,
  214. },
  215. { // 4
  216. sql: "SELECT id1 FROM src1 As s1 left join src2 as s2 on s1.id1 = s2.id2",
  217. data: &xsql.WindowTuples{
  218. Content: []xsql.TupleRow{
  219. &xsql.Tuple{
  220. Emitter: "s1",
  221. Message: xsql.Message{"id1": 1, "f1": "v1"},
  222. }, &xsql.Tuple{
  223. Emitter: "s1",
  224. Message: xsql.Message{"id1": 2, "f1": "v2"},
  225. }, &xsql.Tuple{
  226. Emitter: "s1",
  227. Message: xsql.Message{"id1": 3, "f1": "v3"},
  228. },
  229. &xsql.Tuple{
  230. Emitter: "s2",
  231. Message: xsql.Message{"id2": 1, "f2": "w1"},
  232. }, &xsql.Tuple{
  233. Emitter: "s2",
  234. Message: xsql.Message{"id2": 2, "f2": "w2"},
  235. }, &xsql.Tuple{
  236. Emitter: "s2",
  237. Message: xsql.Message{"id2": 4, "f2": "w3"},
  238. },
  239. },
  240. },
  241. result: &xsql.JoinTuples{
  242. Content: []*xsql.JoinTuple{
  243. {
  244. Tuples: []xsql.TupleRow{
  245. &xsql.Tuple{Emitter: "s1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  246. &xsql.Tuple{Emitter: "s2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  247. },
  248. },
  249. {
  250. Tuples: []xsql.TupleRow{
  251. &xsql.Tuple{Emitter: "s1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  252. &xsql.Tuple{Emitter: "s2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  253. },
  254. },
  255. {
  256. Tuples: []xsql.TupleRow{
  257. &xsql.Tuple{Emitter: "s1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  258. },
  259. },
  260. },
  261. },
  262. },
  263. { // 5
  264. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
  265. data: &xsql.WindowTuples{
  266. Content: []xsql.TupleRow{
  267. &xsql.Tuple{
  268. Emitter: "src1",
  269. Message: xsql.Message{"id1": 1, "f1": "v1"},
  270. },
  271. &xsql.Tuple{
  272. Emitter: "src2",
  273. Message: xsql.Message{"id2": 1, "f2": "w1"},
  274. }, &xsql.Tuple{
  275. Emitter: "src2",
  276. Message: xsql.Message{"id2": 1, "f2": "w2"},
  277. },
  278. },
  279. },
  280. result: &xsql.JoinTuples{
  281. Content: []*xsql.JoinTuple{
  282. {
  283. Tuples: []xsql.TupleRow{
  284. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  285. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  286. },
  287. },
  288. {
  289. Tuples: []xsql.TupleRow{
  290. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}}, &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w2"}},
  291. },
  292. },
  293. },
  294. },
  295. },
  296. { // 6
  297. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
  298. data: &xsql.WindowTuples{
  299. Content: []xsql.TupleRow{
  300. &xsql.Tuple{
  301. Emitter: "src1",
  302. Message: xsql.Message{"id1": 1, "f1": "v1"},
  303. }, &xsql.Tuple{
  304. Emitter: "src1",
  305. Message: xsql.Message{"id1": 2, "f1": "v2"},
  306. }, &xsql.Tuple{
  307. Emitter: "src1",
  308. Message: xsql.Message{"id1": 3, "f1": "v3"},
  309. },
  310. },
  311. },
  312. result: &xsql.JoinTuples{
  313. Content: []*xsql.JoinTuple{
  314. {
  315. Tuples: []xsql.TupleRow{
  316. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  317. },
  318. },
  319. {
  320. Tuples: []xsql.TupleRow{
  321. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  322. },
  323. },
  324. {
  325. Tuples: []xsql.TupleRow{
  326. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  327. },
  328. },
  329. },
  330. },
  331. },
  332. {
  333. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
  334. data: &xsql.WindowTuples{
  335. Content: []xsql.TupleRow{
  336. &xsql.Tuple{
  337. Emitter: "src1",
  338. Message: xsql.Message{"id1": 1, "f1": "v1"},
  339. }, &xsql.Tuple{
  340. Emitter: "src1",
  341. Message: xsql.Message{"id1": 2, "f1": "v2"},
  342. }, &xsql.Tuple{
  343. Emitter: "src1",
  344. Message: xsql.Message{"id1": 3, "f1": "v3"},
  345. },
  346. },
  347. },
  348. result: &xsql.JoinTuples{
  349. Content: []*xsql.JoinTuple{
  350. {
  351. Tuples: []xsql.TupleRow{
  352. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  353. },
  354. },
  355. {
  356. Tuples: []xsql.TupleRow{
  357. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  358. },
  359. },
  360. {
  361. Tuples: []xsql.TupleRow{
  362. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  363. },
  364. },
  365. },
  366. },
  367. },
  368. {
  369. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
  370. data: &xsql.WindowTuples{
  371. Content: []xsql.TupleRow{
  372. &xsql.Tuple{
  373. Emitter: "src2",
  374. Message: xsql.Message{"id2": 1, "f2": "w1"},
  375. }, &xsql.Tuple{
  376. Emitter: "src2",
  377. Message: xsql.Message{"id2": 1, "f2": "w2"},
  378. },
  379. },
  380. },
  381. result: nil,
  382. },
  383. {
  384. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
  385. data: &xsql.WindowTuples{
  386. Content: []xsql.TupleRow{
  387. &xsql.Tuple{
  388. Emitter: "src2",
  389. Message: xsql.Message{"id2": 1, "f2": "w1"},
  390. }, &xsql.Tuple{
  391. Emitter: "src2",
  392. Message: xsql.Message{"id2": 1, "f2": "w2"},
  393. },
  394. },
  395. },
  396. result: nil,
  397. },
  398. { // 10 select id1 FROM src1 left join src2 on null = null
  399. sql: "SELECT id1 FROM src1 left join src2 on src1.id2 = src2.id1",
  400. data: &xsql.WindowTuples{
  401. Content: []xsql.TupleRow{
  402. &xsql.Tuple{
  403. Emitter: "src1",
  404. Message: xsql.Message{"id1": 1, "f2": "w1"},
  405. },
  406. &xsql.Tuple{
  407. Emitter: "src2",
  408. Message: xsql.Message{"id2": 2, "f2": "w1"},
  409. }, &xsql.Tuple{
  410. Emitter: "src2",
  411. Message: xsql.Message{"id2": 2, "f2": "w2"},
  412. },
  413. },
  414. },
  415. result: &xsql.JoinTuples{
  416. Content: []*xsql.JoinTuple{
  417. {
  418. Tuples: []xsql.TupleRow{
  419. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f2": "w1"}},
  420. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w1"}},
  421. },
  422. },
  423. {
  424. Tuples: []xsql.TupleRow{
  425. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f2": "w1"}},
  426. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  427. },
  428. },
  429. },
  430. },
  431. },
  432. {
  433. sql: "SELECT id1 FROM src1 left join src2 on src1.id1*2 = src2.id2",
  434. data: &xsql.WindowTuples{
  435. Content: []xsql.TupleRow{
  436. &xsql.Tuple{
  437. Emitter: "src1",
  438. Message: xsql.Message{"id1": 1, "f1": "v1"},
  439. }, &xsql.Tuple{
  440. Emitter: "src1",
  441. Message: xsql.Message{"id1": 2, "f1": "v2"},
  442. }, &xsql.Tuple{
  443. Emitter: "src1",
  444. Message: xsql.Message{"id1": 3, "f1": "v3"},
  445. },
  446. &xsql.Tuple{
  447. Emitter: "src2",
  448. Message: xsql.Message{"id2": 1, "f2": "w1"},
  449. }, &xsql.Tuple{
  450. Emitter: "src2",
  451. Message: xsql.Message{"id2": 2, "f2": "w2"},
  452. }, &xsql.Tuple{
  453. Emitter: "src2",
  454. Message: xsql.Message{"id2": 4, "f2": "w3"},
  455. },
  456. },
  457. },
  458. result: &xsql.JoinTuples{
  459. Content: []*xsql.JoinTuple{
  460. {
  461. Tuples: []xsql.TupleRow{
  462. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  463. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  464. },
  465. },
  466. {
  467. Tuples: []xsql.TupleRow{
  468. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  469. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  470. },
  471. },
  472. {
  473. Tuples: []xsql.TupleRow{
  474. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  475. },
  476. },
  477. },
  478. },
  479. },
  480. {
  481. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2*2",
  482. data: &xsql.WindowTuples{
  483. Content: []xsql.TupleRow{
  484. &xsql.Tuple{
  485. Emitter: "src1",
  486. Message: xsql.Message{"id1": 1, "f1": "v1"},
  487. }, &xsql.Tuple{
  488. Emitter: "src1",
  489. Message: xsql.Message{"id1": 2, "f1": "v2"},
  490. }, &xsql.Tuple{
  491. Emitter: "src1",
  492. Message: xsql.Message{"id1": 3, "f1": "v3"},
  493. },
  494. &xsql.Tuple{
  495. Emitter: "src2",
  496. Message: xsql.Message{"id2": 1, "f2": "w1"},
  497. }, &xsql.Tuple{
  498. Emitter: "src2",
  499. Message: xsql.Message{"id2": 2, "f2": "w2"},
  500. }, &xsql.Tuple{
  501. Emitter: "src2",
  502. Message: xsql.Message{"id2": 4, "f2": "w3"},
  503. },
  504. },
  505. },
  506. result: &xsql.JoinTuples{
  507. Content: []*xsql.JoinTuple{
  508. {
  509. Tuples: []xsql.TupleRow{
  510. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  511. },
  512. },
  513. {
  514. Tuples: []xsql.TupleRow{
  515. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  516. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  517. },
  518. },
  519. {
  520. Tuples: []xsql.TupleRow{
  521. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  522. },
  523. },
  524. },
  525. },
  526. },
  527. {
  528. sql: "SELECT id1 FROM src1 left join src2 on src1.f1->cid = src2.f2->cid",
  529. data: &xsql.WindowTuples{
  530. Content: []xsql.TupleRow{
  531. &xsql.Tuple{
  532. Emitter: "src1",
  533. Message: xsql.Message{"id1": 1, "f1": str2Map(`{"cid" : 1, "name" : "tom1"}`)},
  534. }, &xsql.Tuple{
  535. Emitter: "src1",
  536. Message: xsql.Message{"id1": 2, "f1": str2Map(`{"cid" : 2, "name" : "mike1"}`)},
  537. }, &xsql.Tuple{
  538. Emitter: "src1",
  539. Message: xsql.Message{"id1": 3, "f1": str2Map(`{"cid" : 3, "name" : "alice1"}`)},
  540. },
  541. &xsql.Tuple{
  542. Emitter: "src2",
  543. Message: xsql.Message{"id2": 1, "f2": str2Map(`{"cid" : 1, "name" : "tom2"}`)},
  544. }, &xsql.Tuple{
  545. Emitter: "src2",
  546. Message: xsql.Message{"id2": 2, "f2": str2Map(`{"cid" : 2, "name" : "mike2"}`)},
  547. }, &xsql.Tuple{
  548. Emitter: "src2",
  549. Message: xsql.Message{"id2": 4, "f2": str2Map(`{"cid" : 4, "name" : "alice2"}`)},
  550. },
  551. },
  552. },
  553. result: &xsql.JoinTuples{
  554. Content: []*xsql.JoinTuple{
  555. {
  556. Tuples: []xsql.TupleRow{
  557. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": str2Map(`{"cid" : 1, "name" : "tom1"}`)}},
  558. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": str2Map(`{"cid" : 1, "name" : "tom2"}`)}},
  559. },
  560. },
  561. {
  562. Tuples: []xsql.TupleRow{
  563. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": str2Map(`{"cid" : 2, "name" : "mike1"}`)}},
  564. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": str2Map(`{"cid" : 2, "name" : "mike2"}`)}},
  565. },
  566. },
  567. {
  568. Tuples: []xsql.TupleRow{
  569. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": str2Map(`{"cid" : 3, "name" : "alice1"}`)}},
  570. },
  571. },
  572. },
  573. },
  574. },
  575. {
  576. sql: "SELECT id1, mqtt(src1.topic) AS a, mqtt(src2.topic) as b FROM src1 left join src2 on src1.id1 = src2.id2",
  577. data: &xsql.WindowTuples{
  578. Content: []xsql.TupleRow{
  579. &xsql.Tuple{
  580. Emitter: "src1",
  581. Message: xsql.Message{"id1": 1, "f1": "v1"},
  582. Metadata: xsql.Metadata{"topic": "devices/type1/device001"},
  583. },
  584. &xsql.Tuple{
  585. Emitter: "src2",
  586. Message: xsql.Message{"id2": 1, "f2": "w1"},
  587. Metadata: xsql.Metadata{"topic": "devices/type2/device001"},
  588. },
  589. },
  590. },
  591. result: &xsql.JoinTuples{
  592. Content: []*xsql.JoinTuple{
  593. {
  594. Tuples: []xsql.TupleRow{
  595. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}, Metadata: xsql.Metadata{"topic": "devices/type1/device001"}},
  596. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}, Metadata: xsql.Metadata{"topic": "devices/type2/device001"}},
  597. },
  598. },
  599. },
  600. },
  601. },
  602. {
  603. sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
  604. data: &xsql.WindowTuples{
  605. Content: []xsql.TupleRow{
  606. &xsql.Tuple{
  607. Emitter: "src1",
  608. Message: xsql.Message{"id1": 1, "f1": "v1"},
  609. }, &xsql.Tuple{
  610. Emitter: "src1",
  611. Message: xsql.Message{"id1": 1, "f1": "v2"},
  612. }, &xsql.Tuple{
  613. Emitter: "src1",
  614. Message: xsql.Message{"id1": 3, "f1": "v3"},
  615. }, &xsql.Tuple{
  616. Emitter: "src1",
  617. Message: xsql.Message{"id1": 3, "f1": "v4"},
  618. }, &xsql.Tuple{
  619. Emitter: "src1",
  620. Message: xsql.Message{"id1": 4, "f1": "v5"},
  621. },
  622. &xsql.Tuple{
  623. Emitter: "src2",
  624. Message: xsql.Message{"id2": 1, "f2": "w1"},
  625. }, &xsql.Tuple{
  626. Emitter: "src2",
  627. Message: xsql.Message{"id2": 3, "f2": "w2"},
  628. }, &xsql.Tuple{
  629. Emitter: "src2",
  630. Message: xsql.Message{"id2": 3, "f2": "w3"},
  631. },
  632. },
  633. },
  634. result: &xsql.JoinTuples{
  635. Content: []*xsql.JoinTuple{
  636. {
  637. Tuples: []xsql.TupleRow{
  638. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  639. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  640. },
  641. },
  642. {
  643. Tuples: []xsql.TupleRow{
  644. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v2"}},
  645. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  646. },
  647. },
  648. {
  649. Tuples: []xsql.TupleRow{
  650. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  651. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 3, "f2": "w2"}},
  652. },
  653. },
  654. {
  655. Tuples: []xsql.TupleRow{
  656. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  657. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 3, "f2": "w3"}},
  658. },
  659. },
  660. {
  661. Tuples: []xsql.TupleRow{
  662. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v4"}},
  663. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 3, "f2": "w2"}},
  664. },
  665. },
  666. {
  667. Tuples: []xsql.TupleRow{
  668. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v4"}},
  669. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 3, "f2": "w3"}},
  670. },
  671. },
  672. {
  673. Tuples: []xsql.TupleRow{
  674. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 4, "f1": "v5"}},
  675. },
  676. },
  677. },
  678. },
  679. },
  680. {
  681. sql: "SELECT id1 FROM src1 full join src2 on src1.id1 = src2.id2",
  682. data: &xsql.WindowTuples{
  683. Content: []xsql.TupleRow{
  684. &xsql.Tuple{
  685. Emitter: "src1",
  686. Message: xsql.Message{"id1": 1, "f1": "v1"},
  687. }, &xsql.Tuple{
  688. Emitter: "src1",
  689. Message: xsql.Message{"id1": 2, "f1": "v2"},
  690. }, &xsql.Tuple{
  691. Emitter: "src1",
  692. Message: xsql.Message{"id1": 3, "f1": "v3"},
  693. },
  694. &xsql.Tuple{
  695. Emitter: "src2",
  696. Message: xsql.Message{"id2": 1, "f2": "w1"},
  697. }, &xsql.Tuple{
  698. Emitter: "src2",
  699. Message: xsql.Message{"id2": 2, "f2": "w2"},
  700. }, &xsql.Tuple{
  701. Emitter: "src2",
  702. Message: xsql.Message{"id2": 2, "f2": "w3"},
  703. },
  704. },
  705. },
  706. result: &xsql.JoinTuples{
  707. Content: []*xsql.JoinTuple{
  708. {
  709. Tuples: []xsql.TupleRow{
  710. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  711. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  712. },
  713. },
  714. {
  715. Tuples: []xsql.TupleRow{
  716. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  717. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  718. },
  719. },
  720. {
  721. Tuples: []xsql.TupleRow{
  722. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  723. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w3"}},
  724. },
  725. },
  726. {
  727. Tuples: []xsql.TupleRow{
  728. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  729. },
  730. },
  731. },
  732. },
  733. },
  734. }
  735. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  736. contextLogger := conf.Log.WithField("rule", "TestLeftJoinPlan_Apply")
  737. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  738. for i, tt := range tests {
  739. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  740. if err != nil {
  741. t.Errorf("statement parse error %s", err)
  742. break
  743. }
  744. if table, ok := stmt.Sources[0].(*ast.Table); !ok {
  745. t.Errorf("statement source is not a table")
  746. } else {
  747. fv, afv := xsql.NewFunctionValuersForOp(nil)
  748. pp := &JoinOp{Joins: stmt.Joins, From: table}
  749. result := pp.Apply(ctx, tt.data, fv, afv)
  750. if !reflect.DeepEqual(tt.result, result) {
  751. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  752. }
  753. }
  754. }
  755. }
  756. func TestInnerJoinPlan_Apply(t *testing.T) {
  757. tests := []struct {
  758. sql string
  759. data *xsql.WindowTuples
  760. result interface{}
  761. }{
  762. {
  763. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  764. data: &xsql.WindowTuples{
  765. Content: []xsql.TupleRow{
  766. &xsql.Tuple{
  767. Emitter: "src1",
  768. Message: xsql.Message{"id1": 1, "f1": "v1"},
  769. }, &xsql.Tuple{
  770. Emitter: "src1",
  771. Message: xsql.Message{"id1": 2, "f1": "v2"},
  772. }, &xsql.Tuple{
  773. Emitter: "src1",
  774. Message: xsql.Message{"id1": 3, "f1": "v3"},
  775. },
  776. &xsql.Tuple{
  777. Emitter: "src2",
  778. Message: xsql.Message{"id2": 1, "f2": "w1"},
  779. }, &xsql.Tuple{
  780. Emitter: "src2",
  781. Message: xsql.Message{"id2": 2, "f2": "w2"},
  782. }, &xsql.Tuple{
  783. Emitter: "src2",
  784. Message: xsql.Message{"id2": 4, "f2": "w3"},
  785. },
  786. },
  787. },
  788. result: &xsql.JoinTuples{
  789. Content: []*xsql.JoinTuple{
  790. {
  791. Tuples: []xsql.TupleRow{
  792. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  793. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  794. },
  795. },
  796. {
  797. Tuples: []xsql.TupleRow{
  798. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  799. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  800. },
  801. },
  802. },
  803. },
  804. },
  805. {
  806. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  807. data: &xsql.WindowTuples{
  808. Content: []xsql.TupleRow{
  809. &xsql.Tuple{
  810. Emitter: "src1",
  811. Message: xsql.Message{"id1": 1, "f1": "v1"},
  812. }, &xsql.Tuple{
  813. Emitter: "src1",
  814. Message: xsql.Message{"id1": 2, "f1": "v2"},
  815. }, &xsql.Tuple{
  816. Emitter: "src1",
  817. Message: xsql.Message{"id1": 3, "f1": "v3"},
  818. },
  819. &xsql.Tuple{
  820. Emitter: "src2",
  821. Message: xsql.Message{"id2": 1, "f2": "w1"},
  822. }, &xsql.Tuple{
  823. Emitter: "src2",
  824. Message: xsql.Message{"f2": "w2"},
  825. }, &xsql.Tuple{
  826. Emitter: "src2",
  827. Message: xsql.Message{"id2": 4, "f2": "w3"},
  828. },
  829. },
  830. },
  831. result: &xsql.JoinTuples{
  832. Content: []*xsql.JoinTuple{
  833. {
  834. Tuples: []xsql.TupleRow{
  835. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  836. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  837. },
  838. },
  839. },
  840. },
  841. },
  842. {
  843. sql: "SELECT id1 FROM src1 As s1 inner join src2 as s2 on s1.id1 = s2.id2",
  844. data: &xsql.WindowTuples{
  845. Content: []xsql.TupleRow{
  846. &xsql.Tuple{
  847. Emitter: "s1",
  848. Message: xsql.Message{"id1": 1, "f1": "v1"},
  849. }, &xsql.Tuple{
  850. Emitter: "s1",
  851. Message: xsql.Message{"id1": 2, "f1": "v2"},
  852. }, &xsql.Tuple{
  853. Emitter: "s1",
  854. Message: xsql.Message{"id1": 3, "f1": "v3"},
  855. },
  856. &xsql.Tuple{
  857. Emitter: "s2",
  858. Message: xsql.Message{"id2": 1, "f2": "w1"},
  859. }, &xsql.Tuple{
  860. Emitter: "s2",
  861. Message: xsql.Message{"id2": 2, "f2": "w2"},
  862. }, &xsql.Tuple{
  863. Emitter: "s2",
  864. Message: xsql.Message{"id2": 4, "f2": "w3"},
  865. },
  866. },
  867. },
  868. result: &xsql.JoinTuples{
  869. Content: []*xsql.JoinTuple{
  870. {
  871. Tuples: []xsql.TupleRow{
  872. &xsql.Tuple{Emitter: "s1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  873. &xsql.Tuple{Emitter: "s2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  874. },
  875. },
  876. {
  877. Tuples: []xsql.TupleRow{
  878. &xsql.Tuple{Emitter: "s1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  879. &xsql.Tuple{Emitter: "s2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  880. },
  881. },
  882. },
  883. },
  884. },
  885. {
  886. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  887. data: &xsql.WindowTuples{
  888. Content: []xsql.TupleRow{
  889. &xsql.Tuple{
  890. Emitter: "src1",
  891. Message: xsql.Message{"id1": 1, "f1": "v1"},
  892. },
  893. &xsql.Tuple{
  894. Emitter: "src2",
  895. Message: xsql.Message{"id2": 1, "f2": "w1"},
  896. }, &xsql.Tuple{
  897. Emitter: "src2",
  898. Message: xsql.Message{"id2": 1, "f2": "w2"},
  899. },
  900. },
  901. },
  902. result: &xsql.JoinTuples{
  903. Content: []*xsql.JoinTuple{
  904. {
  905. Tuples: []xsql.TupleRow{
  906. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  907. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  908. },
  909. },
  910. {
  911. Tuples: []xsql.TupleRow{
  912. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  913. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w2"}},
  914. },
  915. },
  916. },
  917. },
  918. },
  919. {
  920. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  921. data: &xsql.WindowTuples{
  922. Content: []xsql.TupleRow{
  923. &xsql.Tuple{
  924. Emitter: "src1",
  925. Message: xsql.Message{"id1": 1, "f1": "v1"},
  926. },
  927. &xsql.Tuple{
  928. Emitter: "src1",
  929. Message: xsql.Message{"id1": 1, "f1": "v2"},
  930. },
  931. &xsql.Tuple{
  932. Emitter: "src2",
  933. Message: xsql.Message{"id2": 1, "f2": "w1"},
  934. }, &xsql.Tuple{
  935. Emitter: "src2",
  936. Message: xsql.Message{"id2": 1, "f2": "w2"},
  937. },
  938. },
  939. },
  940. result: &xsql.JoinTuples{
  941. Content: []*xsql.JoinTuple{
  942. {
  943. Tuples: []xsql.TupleRow{
  944. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  945. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  946. },
  947. },
  948. {
  949. Tuples: []xsql.TupleRow{
  950. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  951. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w2"}},
  952. },
  953. },
  954. {
  955. Tuples: []xsql.TupleRow{
  956. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v2"}},
  957. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  958. },
  959. },
  960. {
  961. Tuples: []xsql.TupleRow{
  962. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v2"}},
  963. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w2"}},
  964. },
  965. },
  966. },
  967. },
  968. },
  969. {
  970. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  971. data: &xsql.WindowTuples{
  972. Content: []xsql.TupleRow{
  973. &xsql.Tuple{
  974. Emitter: "src1",
  975. Message: xsql.Message{"id1": 1, "f1": "v1"},
  976. }, &xsql.Tuple{
  977. Emitter: "src1",
  978. Message: xsql.Message{"id1": 2, "f1": "v2"},
  979. }, &xsql.Tuple{
  980. Emitter: "src1",
  981. Message: xsql.Message{"id1": 3, "f1": "v3"},
  982. },
  983. },
  984. },
  985. result: nil,
  986. },
  987. {
  988. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  989. data: &xsql.WindowTuples{
  990. Content: []xsql.TupleRow{
  991. &xsql.Tuple{
  992. Emitter: "src1",
  993. Message: xsql.Message{"id1": 1, "f1": "v1"},
  994. }, &xsql.Tuple{
  995. Emitter: "src1",
  996. Message: xsql.Message{"id1": 2, "f1": "v2"},
  997. }, &xsql.Tuple{
  998. Emitter: "src1",
  999. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1000. },
  1001. },
  1002. },
  1003. result: nil,
  1004. },
  1005. {
  1006. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  1007. data: &xsql.WindowTuples{
  1008. Content: []xsql.TupleRow{
  1009. &xsql.Tuple{
  1010. Emitter: "src2",
  1011. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1012. }, &xsql.Tuple{
  1013. Emitter: "src2",
  1014. Message: xsql.Message{"id2": 1, "f2": "w2"},
  1015. },
  1016. },
  1017. },
  1018. result: nil,
  1019. },
  1020. {
  1021. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2",
  1022. data: &xsql.WindowTuples{
  1023. Content: []xsql.TupleRow{
  1024. &xsql.Tuple{
  1025. Emitter: "src2",
  1026. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1027. }, &xsql.Tuple{
  1028. Emitter: "src2",
  1029. Message: xsql.Message{"id2": 1, "f2": "w2"},
  1030. },
  1031. },
  1032. },
  1033. result: nil,
  1034. },
  1035. {
  1036. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1*2 = src2.id2",
  1037. data: &xsql.WindowTuples{
  1038. Content: []xsql.TupleRow{
  1039. &xsql.Tuple{
  1040. Emitter: "src1",
  1041. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1042. }, &xsql.Tuple{
  1043. Emitter: "src1",
  1044. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1045. }, &xsql.Tuple{
  1046. Emitter: "src1",
  1047. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1048. },
  1049. &xsql.Tuple{
  1050. Emitter: "src2",
  1051. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1052. }, &xsql.Tuple{
  1053. Emitter: "src2",
  1054. Message: xsql.Message{"id2": 2, "f2": "w2"},
  1055. }, &xsql.Tuple{
  1056. Emitter: "src2",
  1057. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1058. },
  1059. },
  1060. },
  1061. result: &xsql.JoinTuples{
  1062. Content: []*xsql.JoinTuple{
  1063. {
  1064. Tuples: []xsql.TupleRow{
  1065. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1066. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1067. },
  1068. },
  1069. {
  1070. Tuples: []xsql.TupleRow{
  1071. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1072. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1073. },
  1074. },
  1075. },
  1076. },
  1077. },
  1078. {
  1079. sql: "SELECT id1 FROM src1 inner join src2 on src1.id1 = src2.id2*2",
  1080. data: &xsql.WindowTuples{
  1081. Content: []xsql.TupleRow{
  1082. &xsql.Tuple{
  1083. Emitter: "src1",
  1084. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1085. }, &xsql.Tuple{
  1086. Emitter: "src1",
  1087. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1088. }, &xsql.Tuple{
  1089. Emitter: "src1",
  1090. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1091. },
  1092. &xsql.Tuple{
  1093. Emitter: "src2",
  1094. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1095. }, &xsql.Tuple{
  1096. Emitter: "src2",
  1097. Message: xsql.Message{"id2": 2, "f2": "w2"},
  1098. }, &xsql.Tuple{
  1099. Emitter: "src2",
  1100. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1101. },
  1102. },
  1103. },
  1104. result: &xsql.JoinTuples{
  1105. Content: []*xsql.JoinTuple{
  1106. {
  1107. Tuples: []xsql.TupleRow{
  1108. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1109. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1110. },
  1111. },
  1112. },
  1113. },
  1114. },
  1115. {
  1116. sql: "SELECT id1 FROM src1 inner join src2 on src1.f1->cid = src2.f2->cid",
  1117. data: &xsql.WindowTuples{
  1118. Content: []xsql.TupleRow{
  1119. &xsql.Tuple{
  1120. Emitter: "src1",
  1121. Message: xsql.Message{"id1": 1, "f1": str2Map(`{"cid" : 1, "name" : "tom1"}`)},
  1122. }, &xsql.Tuple{
  1123. Emitter: "src1",
  1124. Message: xsql.Message{"id1": 2, "f1": str2Map(`{"cid" : 2, "name" : "mike1"}`)},
  1125. }, &xsql.Tuple{
  1126. Emitter: "src1",
  1127. Message: xsql.Message{"id1": 3, "f1": str2Map(`{"cid" : 3, "name" : "alice1"}`)},
  1128. },
  1129. &xsql.Tuple{
  1130. Emitter: "src2",
  1131. Message: xsql.Message{"id2": 1, "f2": str2Map(`{"cid" : 1, "name" : "tom2"}`)},
  1132. }, &xsql.Tuple{
  1133. Emitter: "src2",
  1134. Message: xsql.Message{"id2": 2, "f2": str2Map(`{"cid" : 2, "name" : "mike2"}`)},
  1135. }, &xsql.Tuple{
  1136. Emitter: "src2",
  1137. Message: xsql.Message{"id2": 4, "f2": str2Map(`{"cid" : 4, "name" : "alice2"}`)},
  1138. },
  1139. },
  1140. },
  1141. result: &xsql.JoinTuples{
  1142. Content: []*xsql.JoinTuple{
  1143. {
  1144. Tuples: []xsql.TupleRow{
  1145. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": str2Map(`{"cid" : 1, "name" : "tom1"}`)}},
  1146. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": str2Map(`{"cid" : 1, "name" : "tom2"}`)}},
  1147. },
  1148. },
  1149. {
  1150. Tuples: []xsql.TupleRow{
  1151. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": str2Map(`{"cid" : 2, "name" : "mike1"}`)}},
  1152. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": str2Map(`{"cid" : 2, "name" : "mike2"}`)}},
  1153. },
  1154. },
  1155. },
  1156. },
  1157. },
  1158. {
  1159. sql: "SELECT id1 FROM src1 As s1 inner join src2 as s2 on s1.id1 = s2.id2",
  1160. data: &xsql.WindowTuples{
  1161. Content: []xsql.TupleRow{
  1162. &xsql.Tuple{
  1163. Emitter: "s1",
  1164. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1165. }, &xsql.Tuple{
  1166. Emitter: "s1",
  1167. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1168. }, &xsql.Tuple{
  1169. Emitter: "s1",
  1170. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1171. },
  1172. &xsql.Tuple{
  1173. Emitter: "s2",
  1174. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1175. }, &xsql.Tuple{
  1176. Emitter: "s2",
  1177. Message: xsql.Message{"id2": 2, "f2": "w2"},
  1178. }, &xsql.Tuple{
  1179. Emitter: "s2",
  1180. Message: xsql.Message{"id2": 2, "f2": "w3"},
  1181. },
  1182. },
  1183. },
  1184. result: &xsql.JoinTuples{
  1185. Content: []*xsql.JoinTuple{
  1186. {
  1187. Tuples: []xsql.TupleRow{
  1188. &xsql.Tuple{Emitter: "s1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1189. &xsql.Tuple{Emitter: "s2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1190. },
  1191. },
  1192. {
  1193. Tuples: []xsql.TupleRow{
  1194. &xsql.Tuple{Emitter: "s1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1195. &xsql.Tuple{Emitter: "s2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1196. },
  1197. },
  1198. {
  1199. Tuples: []xsql.TupleRow{
  1200. &xsql.Tuple{Emitter: "s1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1201. &xsql.Tuple{Emitter: "s2", Message: xsql.Message{"id2": 2, "f2": "w3"}},
  1202. },
  1203. },
  1204. },
  1205. },
  1206. },
  1207. }
  1208. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1209. contextLogger := conf.Log.WithField("rule", "TestInnerJoinPlan_Apply")
  1210. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1211. for i, tt := range tests {
  1212. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1213. if err != nil {
  1214. t.Errorf("statement parse error %s", err)
  1215. break
  1216. }
  1217. if table, ok := stmt.Sources[0].(*ast.Table); !ok {
  1218. t.Errorf("statement source is not a table")
  1219. } else {
  1220. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1221. pp := &JoinOp{Joins: stmt.Joins, From: table}
  1222. result := pp.Apply(ctx, tt.data, fv, afv)
  1223. if !reflect.DeepEqual(tt.result, result) {
  1224. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1225. }
  1226. }
  1227. }
  1228. }
  1229. func TestRightJoinPlan_Apply(t *testing.T) {
  1230. tests := []struct {
  1231. sql string
  1232. data *xsql.WindowTuples
  1233. result interface{}
  1234. }{
  1235. { // 0
  1236. sql: "SELECT id1 FROM src1 right join src2 on src1.id1 = src2.id2",
  1237. data: &xsql.WindowTuples{
  1238. Content: []xsql.TupleRow{
  1239. &xsql.Tuple{
  1240. Emitter: "src1",
  1241. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1242. }, &xsql.Tuple{
  1243. Emitter: "src1",
  1244. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1245. }, &xsql.Tuple{
  1246. Emitter: "src1",
  1247. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1248. },
  1249. &xsql.Tuple{
  1250. Emitter: "src2",
  1251. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1252. }, &xsql.Tuple{
  1253. Emitter: "src2",
  1254. Message: xsql.Message{"id2": 2, "f2": "w2"},
  1255. }, &xsql.Tuple{
  1256. Emitter: "src2",
  1257. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1258. },
  1259. },
  1260. },
  1261. result: &xsql.JoinTuples{
  1262. Content: []*xsql.JoinTuple{
  1263. {
  1264. Tuples: []xsql.TupleRow{
  1265. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1266. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1267. },
  1268. },
  1269. {
  1270. Tuples: []xsql.TupleRow{
  1271. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1272. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1273. },
  1274. },
  1275. {
  1276. Tuples: []xsql.TupleRow{
  1277. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1278. },
  1279. },
  1280. },
  1281. },
  1282. },
  1283. { // 1
  1284. sql: "SELECT id1 FROM src1 right join src2 on src1.id1 = src2.id2",
  1285. data: &xsql.WindowTuples{
  1286. Content: []xsql.TupleRow{
  1287. &xsql.Tuple{
  1288. Emitter: "src1",
  1289. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1290. }, &xsql.Tuple{
  1291. Emitter: "src1",
  1292. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1293. }, &xsql.Tuple{
  1294. Emitter: "src1",
  1295. Message: xsql.Message{"id1": 1, "f1": "v3"},
  1296. },
  1297. &xsql.Tuple{
  1298. Emitter: "src2",
  1299. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1300. }, &xsql.Tuple{
  1301. Emitter: "src2",
  1302. Message: xsql.Message{"f2": "w2"},
  1303. }, &xsql.Tuple{
  1304. Emitter: "src2",
  1305. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1306. },
  1307. },
  1308. },
  1309. result: &xsql.JoinTuples{
  1310. Content: []*xsql.JoinTuple{
  1311. {
  1312. Tuples: []xsql.TupleRow{
  1313. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1314. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1315. },
  1316. },
  1317. {
  1318. Tuples: []xsql.TupleRow{
  1319. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1320. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v3"}},
  1321. },
  1322. },
  1323. {
  1324. Tuples: []xsql.TupleRow{
  1325. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"f2": "w2"}},
  1326. },
  1327. },
  1328. {
  1329. Tuples: []xsql.TupleRow{
  1330. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1331. },
  1332. },
  1333. },
  1334. },
  1335. },
  1336. { // 2
  1337. sql: "SELECT id1 FROM src1 right join src2 on src1.id1 = src2.id2",
  1338. data: &xsql.WindowTuples{
  1339. Content: []xsql.TupleRow{
  1340. &xsql.Tuple{
  1341. Emitter: "src1",
  1342. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1343. },
  1344. &xsql.Tuple{
  1345. Emitter: "src2",
  1346. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1347. }, &xsql.Tuple{
  1348. Emitter: "src2",
  1349. Message: xsql.Message{"id2": 1, "f2": "w2"},
  1350. },
  1351. },
  1352. },
  1353. result: &xsql.JoinTuples{
  1354. Content: []*xsql.JoinTuple{
  1355. {
  1356. Tuples: []xsql.TupleRow{
  1357. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1358. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1359. },
  1360. },
  1361. {
  1362. Tuples: []xsql.TupleRow{
  1363. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w2"}},
  1364. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1365. },
  1366. },
  1367. },
  1368. },
  1369. },
  1370. { // 3
  1371. sql: "SELECT id1 FROM src1 right join src2 on src1.id1 = src2.id2",
  1372. data: &xsql.WindowTuples{
  1373. Content: []xsql.TupleRow{
  1374. &xsql.Tuple{
  1375. Emitter: "src1",
  1376. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1377. }, &xsql.Tuple{
  1378. Emitter: "src1",
  1379. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1380. }, &xsql.Tuple{
  1381. Emitter: "src1",
  1382. Message: xsql.Message{"id1": 1, "f1": "v3"},
  1383. },
  1384. },
  1385. },
  1386. result: nil,
  1387. },
  1388. { // 4
  1389. sql: "SELECT id1 FROM src1 right join src2 on src1.id1 = src2.id2",
  1390. data: &xsql.WindowTuples{
  1391. Content: []xsql.TupleRow{
  1392. &xsql.Tuple{
  1393. Emitter: "src2",
  1394. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1395. }, &xsql.Tuple{
  1396. Emitter: "src2",
  1397. Message: xsql.Message{"f2": "w2"},
  1398. }, &xsql.Tuple{
  1399. Emitter: "src2",
  1400. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1401. },
  1402. },
  1403. },
  1404. result: &xsql.JoinTuples{
  1405. Content: []*xsql.JoinTuple{
  1406. {
  1407. Tuples: []xsql.TupleRow{
  1408. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1409. },
  1410. },
  1411. {
  1412. Tuples: []xsql.TupleRow{
  1413. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"f2": "w2"}},
  1414. },
  1415. },
  1416. {
  1417. Tuples: []xsql.TupleRow{
  1418. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1419. },
  1420. },
  1421. },
  1422. },
  1423. },
  1424. }
  1425. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1426. contextLogger := conf.Log.WithField("rule", "TestRightJoinPlan_Apply")
  1427. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1428. for i, tt := range tests {
  1429. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1430. if err != nil {
  1431. t.Errorf("statement parse error %s", err)
  1432. break
  1433. }
  1434. if table, ok := stmt.Sources[0].(*ast.Table); !ok {
  1435. t.Errorf("statement source is not a table")
  1436. } else {
  1437. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1438. pp := &JoinOp{Joins: stmt.Joins, From: table}
  1439. result := pp.Apply(ctx, tt.data, fv, afv)
  1440. if !reflect.DeepEqual(tt.result, result) {
  1441. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1442. }
  1443. }
  1444. }
  1445. }
  1446. func TestFullJoinPlan_Apply(t *testing.T) {
  1447. tests := []struct {
  1448. sql string
  1449. data *xsql.WindowTuples
  1450. result interface{}
  1451. }{
  1452. {
  1453. sql: "SELECT id1 FROM src1 full join src2 on src1.id1 = src2.id2",
  1454. data: &xsql.WindowTuples{
  1455. Content: []xsql.TupleRow{
  1456. &xsql.Tuple{
  1457. Emitter: "src1",
  1458. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1459. }, &xsql.Tuple{
  1460. Emitter: "src1",
  1461. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1462. }, &xsql.Tuple{
  1463. Emitter: "src1",
  1464. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1465. },
  1466. &xsql.Tuple{
  1467. Emitter: "src2",
  1468. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1469. }, &xsql.Tuple{
  1470. Emitter: "src2",
  1471. Message: xsql.Message{"id2": 2, "f2": "w2"},
  1472. }, &xsql.Tuple{
  1473. Emitter: "src2",
  1474. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1475. }, &xsql.Tuple{
  1476. Emitter: "src2",
  1477. Message: xsql.Message{"id2": 2, "f2": "w4"},
  1478. },
  1479. },
  1480. },
  1481. result: &xsql.JoinTuples{
  1482. Content: []*xsql.JoinTuple{
  1483. {
  1484. Tuples: []xsql.TupleRow{
  1485. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1486. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1487. },
  1488. },
  1489. {
  1490. Tuples: []xsql.TupleRow{
  1491. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1492. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1493. },
  1494. },
  1495. {
  1496. Tuples: []xsql.TupleRow{
  1497. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}}, &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w4"}},
  1498. },
  1499. },
  1500. {
  1501. Tuples: []xsql.TupleRow{
  1502. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  1503. },
  1504. },
  1505. {
  1506. Tuples: []xsql.TupleRow{
  1507. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1508. },
  1509. },
  1510. },
  1511. },
  1512. },
  1513. {
  1514. sql: "SELECT id1 FROM src1 full join src2 on src1.id1 = src2.id2",
  1515. data: &xsql.WindowTuples{
  1516. Content: []xsql.TupleRow{
  1517. &xsql.Tuple{
  1518. Emitter: "src1",
  1519. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1520. }, &xsql.Tuple{
  1521. Emitter: "src1",
  1522. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1523. }, &xsql.Tuple{
  1524. Emitter: "src1",
  1525. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1526. },
  1527. &xsql.Tuple{
  1528. Emitter: "src2",
  1529. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1530. }, &xsql.Tuple{
  1531. Emitter: "src2",
  1532. Message: xsql.Message{"id2": 2, "f2": "w2"},
  1533. }, &xsql.Tuple{
  1534. Emitter: "src2",
  1535. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1536. },
  1537. },
  1538. },
  1539. result: &xsql.JoinTuples{
  1540. Content: []*xsql.JoinTuple{
  1541. {
  1542. Tuples: []xsql.TupleRow{
  1543. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1544. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1545. },
  1546. },
  1547. {
  1548. Tuples: []xsql.TupleRow{
  1549. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1550. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1551. },
  1552. },
  1553. {
  1554. Tuples: []xsql.TupleRow{
  1555. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  1556. },
  1557. },
  1558. {
  1559. Tuples: []xsql.TupleRow{
  1560. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1561. },
  1562. },
  1563. },
  1564. },
  1565. },
  1566. {
  1567. sql: "SELECT id1 FROM src1 full join src2 on src1.id1 = src2.id2",
  1568. data: &xsql.WindowTuples{
  1569. Content: []xsql.TupleRow{
  1570. &xsql.Tuple{
  1571. Emitter: "src1",
  1572. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1573. }, &xsql.Tuple{
  1574. Emitter: "src1",
  1575. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1576. }, &xsql.Tuple{
  1577. Emitter: "src1",
  1578. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1579. },
  1580. },
  1581. },
  1582. result: &xsql.JoinTuples{
  1583. Content: []*xsql.JoinTuple{
  1584. {
  1585. Tuples: []xsql.TupleRow{
  1586. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1587. },
  1588. },
  1589. {
  1590. Tuples: []xsql.TupleRow{
  1591. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1592. },
  1593. },
  1594. {
  1595. Tuples: []xsql.TupleRow{
  1596. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  1597. },
  1598. },
  1599. },
  1600. },
  1601. },
  1602. {
  1603. sql: "SELECT id1 FROM src1 full join src2 on src1.id1 = src2.id2",
  1604. data: &xsql.WindowTuples{
  1605. Content: []xsql.TupleRow{
  1606. &xsql.Tuple{
  1607. Emitter: "src1",
  1608. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1609. }, &xsql.Tuple{
  1610. Emitter: "src1",
  1611. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1612. }, &xsql.Tuple{
  1613. Emitter: "src1",
  1614. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1615. },
  1616. &xsql.Tuple{
  1617. Emitter: "src2",
  1618. Message: xsql.Message{"id2": 4, "f2": "w1"},
  1619. }, &xsql.Tuple{
  1620. Emitter: "src2",
  1621. Message: xsql.Message{"id2": 5, "f2": "w2"},
  1622. }, &xsql.Tuple{
  1623. Emitter: "src2",
  1624. Message: xsql.Message{"id2": 6, "f2": "w3"},
  1625. },
  1626. },
  1627. },
  1628. result: &xsql.JoinTuples{
  1629. Content: []*xsql.JoinTuple{
  1630. {
  1631. Tuples: []xsql.TupleRow{
  1632. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1633. },
  1634. },
  1635. {
  1636. Tuples: []xsql.TupleRow{
  1637. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1638. },
  1639. },
  1640. {
  1641. Tuples: []xsql.TupleRow{
  1642. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  1643. },
  1644. },
  1645. {
  1646. Tuples: []xsql.TupleRow{
  1647. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w1"}},
  1648. },
  1649. },
  1650. {
  1651. Tuples: []xsql.TupleRow{
  1652. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 5, "f2": "w2"}},
  1653. },
  1654. },
  1655. {
  1656. Tuples: []xsql.TupleRow{
  1657. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 6, "f2": "w3"}},
  1658. },
  1659. },
  1660. },
  1661. },
  1662. },
  1663. {
  1664. sql: "SELECT id1 FROM src1 full join src2 on src1.id1 = src2.id2",
  1665. data: &xsql.WindowTuples{
  1666. Content: []xsql.TupleRow{
  1667. &xsql.Tuple{
  1668. Emitter: "src2",
  1669. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1670. }, &xsql.Tuple{
  1671. Emitter: "src2",
  1672. Message: xsql.Message{"id2": 2, "f2": "w2"},
  1673. }, &xsql.Tuple{
  1674. Emitter: "src2",
  1675. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1676. },
  1677. },
  1678. },
  1679. result: &xsql.JoinTuples{
  1680. Content: []*xsql.JoinTuple{
  1681. {
  1682. Tuples: []xsql.TupleRow{
  1683. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1684. },
  1685. },
  1686. {
  1687. Tuples: []xsql.TupleRow{
  1688. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1689. },
  1690. },
  1691. {
  1692. Tuples: []xsql.TupleRow{
  1693. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1694. },
  1695. },
  1696. },
  1697. },
  1698. },
  1699. }
  1700. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1701. contextLogger := conf.Log.WithField("rule", "TestFullJoinPlan_Apply")
  1702. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1703. for i, tt := range tests {
  1704. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1705. if err != nil {
  1706. t.Errorf("statement parse error %s", err)
  1707. break
  1708. }
  1709. if table, ok := stmt.Sources[0].(*ast.Table); !ok {
  1710. t.Errorf("statement source is not a table")
  1711. } else {
  1712. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1713. pp := &JoinOp{Joins: stmt.Joins, From: table}
  1714. result := pp.Apply(ctx, tt.data, fv, afv)
  1715. if !reflect.DeepEqual(tt.result, result) {
  1716. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1717. }
  1718. }
  1719. }
  1720. }
  1721. func TestCrossJoinPlan_Apply(t *testing.T) {
  1722. tests := []struct {
  1723. sql string
  1724. data *xsql.WindowTuples
  1725. result interface{}
  1726. }{
  1727. {
  1728. sql: "SELECT id1 FROM src1 cross join src2",
  1729. data: &xsql.WindowTuples{
  1730. Content: []xsql.TupleRow{
  1731. &xsql.Tuple{
  1732. Emitter: "src1",
  1733. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1734. }, &xsql.Tuple{
  1735. Emitter: "src1",
  1736. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1737. }, &xsql.Tuple{
  1738. Emitter: "src1",
  1739. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1740. },
  1741. &xsql.Tuple{
  1742. Emitter: "src2",
  1743. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1744. }, &xsql.Tuple{
  1745. Emitter: "src2",
  1746. Message: xsql.Message{"id2": 2, "f2": "w2"},
  1747. }, &xsql.Tuple{
  1748. Emitter: "src2",
  1749. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1750. },
  1751. },
  1752. },
  1753. result: &xsql.JoinTuples{
  1754. Content: []*xsql.JoinTuple{
  1755. {
  1756. Tuples: []xsql.TupleRow{
  1757. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1758. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1759. },
  1760. },
  1761. {
  1762. Tuples: []xsql.TupleRow{
  1763. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1764. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1765. },
  1766. },
  1767. {
  1768. Tuples: []xsql.TupleRow{
  1769. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1770. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1771. },
  1772. },
  1773. {
  1774. Tuples: []xsql.TupleRow{
  1775. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1776. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1777. },
  1778. },
  1779. {
  1780. Tuples: []xsql.TupleRow{
  1781. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1782. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1783. },
  1784. },
  1785. {
  1786. Tuples: []xsql.TupleRow{
  1787. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
  1788. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1789. },
  1790. },
  1791. {
  1792. Tuples: []xsql.TupleRow{
  1793. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  1794. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1795. },
  1796. },
  1797. {
  1798. Tuples: []xsql.TupleRow{
  1799. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  1800. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
  1801. },
  1802. },
  1803. {
  1804. Tuples: []xsql.TupleRow{
  1805. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v3"}},
  1806. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
  1807. },
  1808. },
  1809. },
  1810. },
  1811. },
  1812. {
  1813. sql: "SELECT id1 FROM src1 cross join src2",
  1814. data: &xsql.WindowTuples{
  1815. Content: []xsql.TupleRow{
  1816. &xsql.Tuple{
  1817. Emitter: "src1",
  1818. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1819. },
  1820. &xsql.Tuple{
  1821. Emitter: "src2",
  1822. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1823. }, &xsql.Tuple{
  1824. Emitter: "src2",
  1825. Message: xsql.Message{"id2": 1, "f2": "w2"},
  1826. },
  1827. },
  1828. },
  1829. result: &xsql.JoinTuples{
  1830. Content: []*xsql.JoinTuple{
  1831. {
  1832. Tuples: []xsql.TupleRow{
  1833. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
  1834. &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w1"}},
  1835. },
  1836. },
  1837. {
  1838. Tuples: []xsql.TupleRow{
  1839. &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}}, &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 1, "f2": "w2"}},
  1840. },
  1841. },
  1842. },
  1843. },
  1844. },
  1845. {
  1846. sql: "SELECT id1 FROM src1 cross join src2",
  1847. data: &xsql.WindowTuples{
  1848. Content: []xsql.TupleRow{
  1849. &xsql.Tuple{
  1850. Emitter: "src2",
  1851. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1852. }, &xsql.Tuple{
  1853. Emitter: "src2",
  1854. Message: xsql.Message{"id2": 1, "f2": "w2"},
  1855. },
  1856. },
  1857. },
  1858. result: nil,
  1859. },
  1860. {
  1861. sql: "SELECT id1 FROM src2 cross join src1",
  1862. data: &xsql.WindowTuples{
  1863. Content: []xsql.TupleRow{
  1864. &xsql.Tuple{
  1865. Emitter: "src2",
  1866. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1867. }, &xsql.Tuple{
  1868. Emitter: "src2",
  1869. Message: xsql.Message{"id2": 1, "f2": "w2"},
  1870. },
  1871. },
  1872. },
  1873. result: nil,
  1874. },
  1875. }
  1876. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1877. contextLogger := conf.Log.WithField("rule", "TestCrossJoinPlan_Apply")
  1878. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1879. for i, tt := range tests {
  1880. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1881. if err != nil {
  1882. t.Errorf("statement parse error %s", err)
  1883. break
  1884. }
  1885. if table, ok := stmt.Sources[0].(*ast.Table); !ok {
  1886. t.Errorf("statement source is not a table")
  1887. } else {
  1888. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1889. pp := &JoinOp{Joins: stmt.Joins, From: table}
  1890. result := pp.Apply(ctx, tt.data, fv, afv)
  1891. if !reflect.DeepEqual(tt.result, result) {
  1892. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1893. }
  1894. }
  1895. }
  1896. }
  1897. func TestCrossJoinPlanError(t *testing.T) {
  1898. tests := []struct {
  1899. sql string
  1900. data interface{}
  1901. result interface{}
  1902. }{
  1903. {
  1904. sql: "SELECT id1 FROM src1 cross join src2",
  1905. data: errors.New("an error from upstream"),
  1906. result: errors.New("an error from upstream"),
  1907. }, {
  1908. sql: "SELECT id1 FROM src1 full join src2 on src1.id1 = src2.id2",
  1909. data: &xsql.WindowTuples{
  1910. Content: []xsql.TupleRow{
  1911. &xsql.Tuple{
  1912. Emitter: "src1",
  1913. Message: xsql.Message{"id1": 1, "f1": "v1"},
  1914. }, &xsql.Tuple{
  1915. Emitter: "src1",
  1916. Message: xsql.Message{"id1": 2, "f1": "v2"},
  1917. }, &xsql.Tuple{
  1918. Emitter: "src1",
  1919. Message: xsql.Message{"id1": 3, "f1": "v3"},
  1920. },
  1921. &xsql.Tuple{
  1922. Emitter: "src2",
  1923. Message: xsql.Message{"id2": 1, "f2": "w1"},
  1924. }, &xsql.Tuple{
  1925. Emitter: "src2",
  1926. Message: xsql.Message{"id2": "3", "f2": "w2"},
  1927. }, &xsql.Tuple{
  1928. Emitter: "src2",
  1929. Message: xsql.Message{"id2": 4, "f2": "w3"},
  1930. }, &xsql.Tuple{
  1931. Emitter: "src2",
  1932. Message: xsql.Message{"id2": 2, "f2": "w4"},
  1933. },
  1934. },
  1935. },
  1936. result: errors.New("run Join error: invalid operation int64(1) = string(3)"),
  1937. },
  1938. }
  1939. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  1940. contextLogger := conf.Log.WithField("rule", "TestCrossJoinPlan_Apply")
  1941. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  1942. for i, tt := range tests {
  1943. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  1944. if err != nil {
  1945. t.Errorf("statement parse error %s", err)
  1946. break
  1947. }
  1948. if table, ok := stmt.Sources[0].(*ast.Table); !ok {
  1949. t.Errorf("statement source is not a table")
  1950. } else {
  1951. fv, afv := xsql.NewFunctionValuersForOp(nil)
  1952. pp := &JoinOp{Joins: stmt.Joins, From: table}
  1953. result := pp.Apply(ctx, tt.data, fv, afv)
  1954. if !reflect.DeepEqual(tt.result, result) {
  1955. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
  1956. }
  1957. }
  1958. }
  1959. }