join_test.go 55 KB

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