123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397 |
- // Copyright 2022-2023 EMQ Technologies Co., Ltd.
- //
- // Licensed under the Apache License, Version 2.0 (the "License");
- // you may not use this file except in compliance with the License.
- // You may obtain a copy of the License at
- //
- // http://www.apache.org/licenses/LICENSE-2.0
- //
- // Unless required by applicable law or agreed to in writing, software
- // distributed under the License is distributed on an "AS IS" BASIS,
- // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- // See the License for the specific language governing permissions and
- // limitations under the License.
- package operator
- import (
- "errors"
- "fmt"
- "reflect"
- "strings"
- "testing"
- "github.com/lf-edge/ekuiper/internal/conf"
- "github.com/lf-edge/ekuiper/internal/topo/context"
- "github.com/lf-edge/ekuiper/internal/xsql"
- "github.com/lf-edge/ekuiper/pkg/ast"
- "github.com/lf-edge/ekuiper/pkg/cast"
- )
- func parseStmt(p *ProjectOp, fields ast.Fields) {
- p.AllWildcard = false
- p.WildcardEmitters = make(map[string]bool)
- for _, field := range fields {
- if field.AName != "" {
- p.AliasFields = append(p.AliasFields, field)
- p.AliasNames = append(p.AliasNames, field.AName)
- } else {
- switch ft := field.Expr.(type) {
- case *ast.Wildcard:
- p.AllWildcard = true
- case *ast.FieldRef:
- if ft.Name == "*" {
- p.WildcardEmitters[string(ft.StreamName)] = true
- } else {
- p.ColNames = append(p.ColNames, []string{ft.Name, string(ft.StreamName)})
- }
- default:
- p.ExprFields = append(p.ExprFields, field)
- p.ExprNames = append(p.ExprNames, field.Name)
- }
- }
- }
- }
- func parseResult(opResult interface{}, aggregate bool) (result []map[string]interface{}, err error) {
- switch rt := opResult.(type) {
- case xsql.TupleRow:
- result = append(result, rt.ToMap())
- case xsql.Collection:
- result = rt.ToMaps()
- default:
- err = errors.New("unexpected result type")
- }
- return
- }
- func TestProjectPlan_Apply1(t *testing.T) {
- tests := []struct {
- sql string
- data *xsql.Tuple
- result []map[string]interface{}
- }{
- { // 0
- sql: "SELECT a FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- },
- Metadata: xsql.Metadata{
- "id": 45,
- "other": "mock",
- },
- },
- result: []map[string]interface{}{{
- "a": "val_a",
- "__meta": xsql.Metadata{
- "id": 45,
- "other": "mock",
- },
- }},
- },
- { // 1
- sql: "SELECT b FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- },
- },
- result: []map[string]interface{}{{}},
- },
- { // 2
- sql: "SELECT ts FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- "ts": cast.TimeFromUnixMilli(1568854573431),
- },
- },
- result: []map[string]interface{}{{
- "ts": cast.TimeFromUnixMilli(1568854573431),
- }},
- },
- // Schemaless may return a message without selecting column
- { // 3
- sql: "SELECT ts FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- "ts2": cast.TimeFromUnixMilli(1568854573431),
- },
- },
- result: []map[string]interface{}{{}},
- },
- { // 4
- sql: "SELECT A FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- },
- },
- result: []map[string]interface{}{{
- "A": "val_a",
- }},
- },
- // 5
- {
- sql: `SELECT "value" FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- "kuiper_field_0": "value",
- }},
- },
- // 6
- {
- sql: `SELECT 3.4 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- "kuiper_field_0": 3.4,
- }},
- },
- // 7
- {
- sql: `SELECT 5 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- "kuiper_field_0": 5,
- }},
- },
- // 8
- {
- sql: `SELECT a, "value" AS b FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- },
- },
- result: []map[string]interface{}{{
- "a": "val_a",
- "b": "value",
- }},
- },
- // 9
- {
- sql: `SELECT a, "value" AS b, 3.14 as Pi, 0 as Zero FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- },
- },
- result: []map[string]interface{}{{
- "a": "val_a",
- "b": "value",
- "Pi": 3.14,
- "Zero": 0,
- }},
- },
- // 10
- {
- sql: `SELECT a->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{"b": "hello"},
- },
- },
- result: []map[string]interface{}{{
- "ab": "hello",
- }},
- },
- // 11
- {
- sql: `SELECT a->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}(nil),
- },
- },
- result: []map[string]interface{}{{}},
- },
- // 12
- {
- sql: `SELECT a->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "name": "name",
- },
- },
- result: []map[string]interface{}{{}},
- },
- // 13
- {
- sql: `SELECT a->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "commonstring",
- },
- },
- result: []map[string]interface{}{{}},
- },
- // 14
- {
- sql: `SELECT a[0]->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": []interface{}{
- map[string]interface{}{"b": "hello1"},
- map[string]interface{}{"b": "hello2"},
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": "hello1",
- }},
- },
- // 15
- {
- sql: `SELECT a[0]->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": []map[string]interface{}{
- {"b": "hello1"},
- {"b": "hello2"},
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": "hello1",
- }},
- },
- // 16
- {
- sql: `SELECT a[2:4] AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": []map[string]interface{}{
- {"b": "hello1"},
- {"b": "hello2"},
- {"b": "hello3"},
- {"b": "hello4"},
- {"b": "hello5"},
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": []map[string]interface{}{
- {"b": "hello3"},
- {"b": "hello4"},
- },
- }},
- },
- // 17
- {
- sql: `SELECT a[2:] AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": []map[string]interface{}{
- {"b": "hello1"},
- {"b": "hello2"},
- {"b": "hello3"},
- {"b": "hello4"},
- {"b": "hello5"},
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": []map[string]interface{}{
- {"b": "hello3"},
- {"b": "hello4"},
- {"b": "hello5"},
- },
- }},
- },
- // 18
- {
- sql: `SELECT a[2:] AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": []interface{}{
- true, false, true, false, true, true,
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": []interface{}{
- true, false, true, true,
- },
- }},
- },
- // 19
- {
- sql: `SELECT a[:4] AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": []interface{}{
- true, false, true, false, true, true,
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": []interface{}{
- true, false, true, false,
- },
- }},
- },
- // 20
- {
- sql: `SELECT a[:4] AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": []interface{}{
- 3.14, 3.141, 3.1415, 3.14159, 3.141592, 3.1415926,
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": []interface{}{
- 3.14, 3.141, 3.1415, 3.14159,
- },
- }},
- },
- // 21
- {
- sql: `SELECT a->b[:4] AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{
- "b": []float64{3.14, 3.141, 3.1415, 3.14159, 3.141592, 3.1415926},
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": []float64{
- 3.14, 3.141, 3.1415, 3.14159,
- },
- }},
- },
- // 22
- {
- sql: `SELECT a->b[0:1] AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{
- "b": []float64{3.14, 3.141, 3.1415, 3.14159, 3.141592, 3.1415926},
- },
- },
- },
- result: []map[string]interface{}{{
- "ab": []float64{
- 3.14,
- },
- }},
- },
- // 23
- {
- sql: `SELECT a->c->d AS f1 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{
- "b": "hello",
- "c": map[string]interface{}{
- "d": 35.2,
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "f1": 35.2,
- }},
- },
- // 24
- {
- sql: `SELECT a->c->d AS f1 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{
- "b": "hello",
- "c": map[string]interface{}{
- "e": 35.2,
- },
- },
- },
- },
- result: []map[string]interface{}{{}},
- },
- // 25
- {
- sql: `SELECT a->c->d AS f1 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{
- "b": "hello",
- },
- },
- },
- result: []map[string]interface{}{{}},
- },
- // 26
- // The int type is not supported yet, the json parser returns float64 for int values
- {
- sql: `SELECT a->c->d AS f1 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{
- "b": "hello",
- "c": map[string]interface{}{
- "d": float64(35),
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "f1": float64(35),
- }},
- },
- // 27
- {
- sql: "SELECT a FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{
- {},
- },
- },
- // 28
- {
- sql: "SELECT * FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{
- {},
- },
- },
- // 29
- {
- sql: `SELECT * FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{
- "b": "hello",
- "c": map[string]interface{}{
- "d": 35.2,
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "a": map[string]interface{}{
- "b": "hello",
- "c": map[string]interface{}{
- "d": 35.2,
- },
- },
- }},
- },
- // 30
- {
- sql: `SELECT * FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val1",
- "b": 3.14,
- },
- },
- result: []map[string]interface{}{{
- "a": "val1",
- "b": 3.14,
- }},
- },
- // 31
- {
- sql: `SELECT 3*4 AS f1 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- "f1": int64(12),
- }},
- },
- // 32
- {
- sql: `SELECT 4.5*2 AS f1 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- "f1": float64(9),
- }},
- },
- // 33
- {
- sql: "SELECT `a.b.c` FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a.b.c": "val_a",
- },
- },
- result: []map[string]interface{}{{
- "a.b.c": "val_a",
- }},
- },
- // 34
- {
- sql: `SELECT CASE a WHEN 10 THEN "true" END AS b FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": int64(10),
- },
- },
- result: []map[string]interface{}{{
- "b": "true",
- }},
- },
- // 35
- {
- sql: `SELECT a->b AS ab, *, abs(f1) FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": map[string]interface{}{
- "b": "test",
- },
- "b": "b",
- "f1": -12,
- },
- },
- result: []map[string]interface{}{{
- "a": map[string]interface{}{
- "b": "test",
- },
- "ab": "test",
- "abs": 12,
- "b": "b",
- "f1": -12,
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- contextLogger := conf.Log.WithField("rule", "TestProjectPlan_Apply1")
- ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- if err != nil {
- t.Errorf("parse sql error: %s", err)
- continue
- }
- pp := &ProjectOp{SendMeta: true, IsAggregate: xsql.WithAggFields(stmt)}
- parseStmt(pp, stmt.Fields)
- fv, afv := xsql.NewFunctionValuersForOp(nil)
- opResult := pp.Apply(ctx, tt.data, fv, afv)
- result, err := parseResult(opResult, pp.IsAggregate)
- if err != nil {
- t.Errorf("parse result error: %s", err)
- continue
- }
- if !reflect.DeepEqual(tt.result, result) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
- }
- }
- }
- func TestProjectPlan_MultiInput(t *testing.T) {
- tests := []struct {
- sql string
- data interface{}
- result []map[string]interface{}
- }{ // 0
- {
- sql: "SELECT * FROM tbl WHERE abc*2+3 > 12 AND abc < 20",
- data: &xsql.Tuple{
- Emitter: "tbl",
- Message: xsql.Message{
- "abc": int64(6),
- },
- },
- result: []map[string]interface{}{{
- "abc": int64(6),
- }},
- },
- // 1
- {
- sql: "SELECT abc FROM tbl WHERE abc*2+3 > 12 OR def = \"hello\"",
- data: &xsql.Tuple{
- Emitter: "tbl",
- Message: xsql.Message{
- "abc": int64(34),
- "def": "hello",
- },
- },
- result: []map[string]interface{}{{
- "abc": int64(34),
- }},
- },
- // 2
- {
- sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- }, {
- "id1": 2,
- }, {
- "id1": 3,
- }},
- },
- // 3
- {
- sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id2": 2, "f1": "v2"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- }, {}, {
- "id1": 3,
- }},
- },
- // 4
- {
- sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- "f1": "v1",
- }, {
- "id1": 2,
- "f1": "v2",
- }, {
- "id1": 3,
- "f1": "v1",
- }},
- },
- // 5
- {
- sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id2": 2, "f2": "v2"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- "f1": "v1",
- }, {
- "id2": 2,
- "f2": "v2",
- }, {
- "id1": 3,
- "f1": "v1",
- }},
- },
- // 6
- {
- sql: "SELECT src1.* FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- "f1": "v1",
- }, {
- "id1": 2,
- "f1": "v2",
- }, {
- "id1": 3,
- "f1": "v1",
- }},
- },
- // 7
- {
- sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- }, {
- "id1": 2,
- }, {
- "id1": 3,
- }},
- },
- // 8
- {
- sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id2": 3, "f1": "v1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- }, {
- "id1": 2,
- }, {}},
- },
- // 9
- {
- sql: "SELECT abc FROM tbl group by abc",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "tbl",
- Message: xsql.Message{
- "abc": int64(6),
- "def": "hello",
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "abc": int64(6),
- }},
- },
- // 10
- {
- sql: "SELECT abc FROM tbl group by abc",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "tbl",
- Message: xsql.Message{
- "def": "hello",
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{}},
- },
- // 11
- {
- sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- },
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- }, {
- "id1": 2,
- }},
- },
- // 12
- {
- sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- },
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id2": 2, "f1": "v2"},
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- }, {}},
- },
- // 13
- {
- sql: "SELECT src2.id2 FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id2": 2,
- }, {
- "id2": 4,
- }, {}},
- },
- // 14
- {
- sql: "SELECT src1.*, f2 FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- "f1": "v1",
- "f2": "w2",
- }, {
- "id1": 2,
- "f1": "v2",
- "f2": "w3",
- }, {
- "id1": 3,
- "f1": "v1",
- }},
- },
- // 15
- {
- sql: "SELECT * FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id": 1, "f1": "v1"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "f2": "w2"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id": 2, "f1": "v2"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "f2": "w3"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id": 3, "f1": "v1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id": 1,
- "f1": "v1",
- "f2": "w2",
- }, {
- "id": 2,
- "f1": "v2",
- "f2": "w3",
- }, {
- "id": 3,
- "f1": "v1",
- }},
- },
- // 16
- {
- sql: "SELECT src1.* FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- },
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": 1,
- "f1": "v1",
- }, {
- "id1": 2,
- "f1": "v2",
- }},
- },
- // 17
- {
- sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id2": 2,
- "id1": 1,
- "f1": "v1",
- }, {
- "id2": 4,
- "id1": 2,
- "f1": "v2",
- }, {
- "id1": 3,
- "f1": "v1",
- }},
- },
- // 18
- {
- sql: "SELECT src2.id2, src1.* FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id2": 2,
- "id1": 1,
- "f1": "v1",
- }, {
- "id2": 4,
- "id1": 2,
- "f1": "v2",
- }, {
- "id1": 3,
- "f1": "v1",
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- contextLogger := conf.Log.WithField("rule", "TestProjectPlan_MultiInput")
- ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- pp := &ProjectOp{SendMeta: true, IsAggregate: xsql.WithAggFields(stmt)}
- parseStmt(pp, stmt.Fields)
- fv, afv := xsql.NewFunctionValuersForOp(nil)
- opResult := pp.Apply(ctx, tt.data, fv, afv)
- result, err := parseResult(opResult, pp.IsAggregate)
- if err != nil {
- t.Errorf("parse result error: %s", err)
- continue
- }
- if !reflect.DeepEqual(tt.result, result) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
- }
- }
- }
- func TestProjectPlan_Funcs(t *testing.T) {
- tests := []struct {
- sql string
- data interface{}
- result []map[string]interface{}
- }{
- // 0
- {
- sql: "SELECT round(a) as r FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": 47.5,
- },
- },
- result: []map[string]interface{}{{
- "r": float64(48),
- }},
- },
- // 1
- {
- sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53.1},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27.4},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123.7},
- },
- },
- },
- result: []map[string]interface{}{{
- "r": float64(53),
- }, {
- "r": float64(27),
- }, {
- "r": float64(123124),
- }},
- },
- // 2
- {
- sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53.1},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27.4},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123.7},
- },
- },
- },
- result: []map[string]interface{}{{
- "r": float64(53),
- }, {
- "r": float64(27),
- }, {
- "r": float64(123124),
- }},
- },
- // 3
- {
- sql: "SELECT round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "r": float64(66),
- }, {
- "r": float64(73),
- }, {
- "r": float64(89),
- }},
- },
- // 4
- {
- sql: "SELECT CONCAT(test.id, test.a, test1.b) as concat FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "concat": "165.5512",
- }, {
- "concat": "273.49934",
- }, {
- "concat": "388.886",
- }},
- },
- // 5
- {
- sql: "SELECT count(a) as r FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": 47.5,
- },
- },
- result: []map[string]interface{}{{
- "r": 1,
- }},
- },
- // 6
- {
- sql: "SELECT meta(test.device) as d FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}, Metadata: xsql.Metadata{"device": "devicea"}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}, Metadata: xsql.Metadata{"device": "deviceb"}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}, Metadata: xsql.Metadata{"device": "devicec"}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "d": "devicea",
- }, {
- "d": "deviceb",
- }, {
- "d": "devicec",
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- contextLogger := conf.Log.WithField("rule", "TestProjectPlan_Funcs")
- ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- if err != nil {
- t.Error(err)
- }
- pp := &ProjectOp{SendMeta: true, IsAggregate: xsql.WithAggFields(stmt)}
- parseStmt(pp, stmt.Fields)
- fv, afv := xsql.NewFunctionValuersForOp(nil)
- opResult := pp.Apply(ctx, tt.data, fv, afv)
- result, err := parseResult(opResult, pp.IsAggregate)
- if err != nil {
- t.Errorf("parse result error: %s", err)
- continue
- }
- if !reflect.DeepEqual(tt.result, result) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
- }
- }
- }
- func TestProjectPlan_AggFuncs(t *testing.T) {
- tests := []struct {
- sql string
- data interface{}
- result []map[string]interface{}
- }{
- { // 0
- sql: "SELECT count(*) as c, round(a) as r, window_start() as ws, window_end() as we FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
- },
- },
- },
- result: []map[string]interface{}{{
- "c": 2,
- "r": float64(122),
- "ws": int64(1541152486013),
- "we": int64(1541152487013),
- }, {
- "c": 2,
- "r": float64(89),
- "ws": int64(1541152486013),
- "we": int64(1541152487013),
- }},
- },
- // 1
- {
- sql: "SELECT count(a) as c, avg(a) as a, sum(a) as s, min(a) as min, max(a) as max FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "c": 1,
- "a": 122.33,
- "s": 122.33,
- "min": 122.33,
- "max": 122.33,
- }, {
- "c": 2,
- "s": 103.63,
- "a": 51.815,
- "min": 14.6,
- "max": 89.03,
- }},
- },
- // 2
- {
- sql: "SELECT avg(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 98.31}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "avg": 116.68,
- }, {
- "avg": 51.815,
- }},
- },
- // 3
- {
- sql: "SELECT max(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "max": 177.51,
- }, {
- "max": 89.03,
- }},
- },
- // 4
- {
- sql: "SELECT min(a), window_start(), window_end() FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
- },
- result: []map[string]interface{}{{
- "min": 68.55,
- "window_start": int64(1541152486013),
- "window_end": int64(1541152487013),
- }},
- },
- // 5
- {
- sql: "SELECT count(*) as all, count(a) as c, avg(a) as a, sum(a) as s, min(a) as min, max(a) as max FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.JoinTuples{
- Content: []*xsql.JoinTuple{
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- {
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "all": 3,
- "c": 2,
- "a": 123.03,
- "s": 246.06,
- "min": 68.55,
- "max": 177.51,
- }},
- },
- // 6
- {
- sql: "SELECT sum(a), window_start() as ws, window_end() FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
- },
- result: []map[string]interface{}{{
- "sum": int64(123203),
- "ws": int64(1541152486013),
- "window_end": int64(1541152487013),
- }},
- },
- // 7
- {
- sql: "SELECT sum(a) as s FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53, "s": 123203},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- result: []map[string]interface{}{{
- "s": int64(123203),
- }},
- },
- // 8
- {
- sql: "SELECT sum(a) FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- result: []map[string]interface{}{{
- "sum": int64(123203),
- }},
- },
- // 9
- {
- sql: "SELECT count(*) as all, count(a) as c, avg(a) as a, sum(a) as s, min(a) as min, max(a) as max FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"s": 123123},
- },
- },
- },
- result: []map[string]interface{}{{
- "all": 3,
- "c": 2,
- "a": int64(40),
- "s": int64(80),
- "min": int64(27),
- "max": int64(53),
- }},
- },
- // 10
- {
- sql: "SELECT count(*), meta(test1.device) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "count": 2,
- "meta": "devicea",
- }, {
- "count": 2,
- "meta": "devicec",
- }},
- },
- // 11
- {
- sql: "SELECT count(*) as c, meta(test1.device) as d FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "d": "devicea"}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "d": "devicec"}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "c": 2,
- "d": "devicea",
- }, {
- "c": 2,
- "d": "devicec",
- }},
- },
- // 12
- {
- sql: "SELECT * FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
- },
- },
- },
- result: []map[string]interface{}{{
- "a": 122.33,
- "c": 2,
- "color": "w2",
- "id": 1,
- "r": 122,
- }, {
- "a": 89.03,
- "c": 2,
- "color": "w1",
- "id": 2,
- "r": 89,
- }},
- },
- // 13
- {
- sql: "SELECT collect(a) as r1 FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "r1": []interface{}{122.33, 177.51},
- }, {"r1": []interface{}{89.03, 14.6}}},
- },
- // 14
- {
- sql: "SELECT collect(*)[1] as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53, "s": 123203},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
- },
- result: []map[string]interface{}{{
- "c1": map[string]interface{}{
- "a": 27,
- },
- }},
- },
- // 15
- {
- sql: "SELECT collect(*)[1]->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53, "s": 123203},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- result: []map[string]interface{}{{
- "c1": 27,
- }},
- },
- // 16
- {
- sql: "SELECT collect(*)[1]->sl[0] as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53, "sl": []string{"hello", "world"}},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27, "sl": []string{"new", "horizon"}},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123, "sl": []string{"south", "africa"}},
- },
- },
- },
- result: []map[string]interface{}{{
- "c1": "new",
- }},
- },
- // 17
- {
- sql: "SELECT deduplicate(id, true) as r1 FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{
- {
- "r1": []interface{}{
- map[string]interface{}{"a": 122.33, "c": 2, "color": "w2", "id": 1, "r": 122},
- map[string]interface{}{"a": 177.51, "color": "w2", "id": 5},
- },
- }, {
- "r1": []interface{}{
- map[string]interface{}{"a": 89.03, "c": 2, "color": "w1", "id": 2, "r": 89},
- map[string]interface{}{"a": 14.6, "color": "w1", "id": 4},
- },
- },
- },
- },
- // 18
- {
- sql: "SELECT deduplicate(a, false)->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53, "s": 123203},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- result: []map[string]interface{}{{
- "c1": 123123,
- }},
- },
- // 19
- {
- sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53, "s": 123203},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- },
- },
- },
- result: []map[string]interface{}{{}},
- },
- // 20
- {
- sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53, "s": 123203},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- },
- },
- },
- result: []map[string]interface{}{{}},
- },
- // 21 when got column after group by operation, return the first tuple's column
- {
- sql: "SELECT A.module, A.topic , max(A.value), B.topic as var2, max(B.value) as max2, C.topic as var3, max(C.value) as max3 FROM A FULL JOIN B on A.module=B.module FULL JOIN C on A.module=C.module GROUP BY A.module, TUMBLINGWINDOW(ss, 10)",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "B", Message: xsql.Message{"module": 1, "topic": "moduleB topic", "value": 1}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "C", Message: xsql.Message{"module": 1, "topic": "moduleC topic", "value": 100}},
- },
- },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "var2": "moduleB topic",
- "max2": int64(1),
- "max3": int64(100),
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- contextLogger := conf.Log.WithField("rule", "TestProjectPlan_AggFuncs")
- ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- if err != nil {
- t.Error(err)
- }
- pp := &ProjectOp{SendMeta: true, IsAggregate: true}
- parseStmt(pp, stmt.Fields)
- fv, afv := xsql.NewFunctionValuersForOp(nil)
- opResult := pp.Apply(ctx, tt.data, fv, afv)
- result, err := parseResult(opResult, pp.IsAggregate)
- if err != nil {
- t.Errorf("parse result error: %s", err)
- continue
- }
- if !reflect.DeepEqual(tt.result, result) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, result)
- }
- }
- }
- func TestProjectPlanError(t *testing.T) {
- tests := []struct {
- sql string
- data interface{}
- result interface{}
- }{
- // 0
- {
- sql: "SELECT a FROM test",
- data: errors.New("an error from upstream"),
- result: errors.New("an error from upstream"),
- },
- // 1
- {
- sql: "SELECT a * 5 FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- },
- },
- result: errors.New("run Select error: invalid operation string(val_a) * int64(5)"),
- },
- // 2
- {
- sql: `SELECT a[0]->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "common string",
- },
- },
- result: errors.New("run Select error: invalid operation string(common string) [] *xsql.BracketEvalResult(&{0 0})"),
- },
- // 3
- {
- sql: `SELECT round(a) as r FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "common string",
- },
- },
- result: errors.New("run Select error: call func round error: cannot convert string(common string) to float64"),
- },
- // 5
- {
- sql: "SELECT avg(a) as avg FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: &xsql.GroupedTuplesSet{
- Groups: []*xsql.GroupedTuples{
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": "dde"}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- },
- {
- Content: []xsql.TupleRow{
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.TupleRow{
- &xsql.Tuple{Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- &xsql.Tuple{Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- },
- },
- result: errors.New("run Select error: call func avg error: requires float64 but found string(dde)"),
- },
- // 6
- {
- sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
- data: &xsql.WindowTuples{
- Content: []xsql.TupleRow{
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": "ddd"},
- }, &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- result: errors.New("run Select error: call func sum error: requires int but found string(ddd)"),
- },
- // 7
- {
- sql: `SELECT a[0]->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": []map[string]interface{}(nil),
- },
- },
- result: errors.New("run Select error: out of index: 0 of 0"),
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- contextLogger := conf.Log.WithField("rule", "TestProjectPlanError")
- ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- pp := &ProjectOp{SendMeta: true, IsAggregate: xsql.WithAggFields(stmt)}
- parseStmt(pp, stmt.Fields)
- fv, afv := xsql.NewFunctionValuersForOp(nil)
- opResult := pp.Apply(ctx, tt.data, fv, afv)
- if !reflect.DeepEqual(tt.result, opResult) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, opResult)
- }
- }
- }
|