123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692 |
- package plans
- import (
- "encoding/json"
- "errors"
- "fmt"
- "github.com/emqx/kuiper/common"
- "github.com/emqx/kuiper/xsql"
- "github.com/emqx/kuiper/xstream/contexts"
- "reflect"
- "strings"
- "testing"
- )
- func TestProjectPlan_Apply1(t *testing.T) {
- var tests = []struct {
- sql string
- data *xsql.Tuple
- result []map[string]interface{}
- }{
- {
- 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": map[string]interface{}{
- "id": float64(45),
- "other": "mock",
- },
- }},
- },
- {
- sql: "SELECT b FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- },
- },
- result: []map[string]interface{}{{}},
- },
- {
- sql: "SELECT ts FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- "ts": common.TimeFromUnixMilli(1568854573431),
- },
- },
- result: []map[string]interface{}{{
- "ts": "2019-09-19T00:56:13.431Z",
- }},
- },
- //Schemaless may return a message without selecting column
- {
- sql: "SELECT ts FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- "ts2": common.TimeFromUnixMilli(1568854573431),
- },
- },
- result: []map[string]interface{}{{}},
- },
- {
- sql: "SELECT A FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "val_a",
- },
- },
- result: []map[string]interface{}{{
- "A": "val_a",
- }},
- },
- {
- sql: `SELECT "value" FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- DEFAULT_FIELD_NAME_PREFIX + "0": "value",
- }},
- },
- {
- sql: `SELECT 3.4 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- DEFAULT_FIELD_NAME_PREFIX + "0": 3.4,
- }},
- },
- {
- sql: `SELECT 5 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- DEFAULT_FIELD_NAME_PREFIX + "0": 5.0,
- }},
- },
- {
- 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",
- }},
- },
- {
- 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.0,
- }},
- },
- {
- 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",
- }},
- },
- {
- 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{}{{}},
- },
- {
- sql: `SELECT a->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "name": "name",
- },
- },
- result: []map[string]interface{}{{}},
- },
- {
- sql: `SELECT a->b AS ab FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": "commonstring",
- },
- },
- result: []map[string]interface{}{{}},
- },
- {
- 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",
- }},
- },
- {
- 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",
- }},
- },
- {
- 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": []interface{}{
- map[string]interface{}{"b": "hello3"},
- map[string]interface{}{"b": "hello4"},
- },
- }},
- },
- {
- 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,
- }},
- },
- {
- 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{}{{}},
- },
- {
- 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{}{{}},
- },
- //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),
- }},
- },
- {
- sql: "SELECT a FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{
- {},
- },
- },
- {
- sql: "SELECT * FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{
- {},
- },
- },
- {
- 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,
- },
- },
- }},
- },
- {
- 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,
- }},
- },
- {
- sql: `SELECT 3*4 AS f1 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- "f1": float64(12),
- }},
- },
- {
- sql: `SELECT 4.5*2 AS f1 FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{},
- },
- result: []map[string]interface{}{{
- "f1": float64(9),
- }},
- },
- {
- 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",
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- contextLogger := common.Log.WithField("rule", "TestProjectPlan_Apply1")
- ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- pp := &ProjectPlan{Fields: stmt.Fields, SendMeta: true}
- pp.isTest = true
- fv, afv := xsql.NewAggregateFunctionValuers()
- result := pp.Apply(ctx, tt.data, fv, afv)
- var mapRes []map[string]interface{}
- if v, ok := result.([]byte); ok {
- err := json.Unmarshal(v, &mapRes)
- if err != nil {
- t.Errorf("Failed to parse the input into map.\n")
- continue
- }
- //fmt.Printf("%t\n", mapRes["rengine_field_0"])
- if !reflect.DeepEqual(tt.result, mapRes) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
- }
- } else {
- t.Errorf("%d. The returned result is not type of []byte\n", i)
- }
- }
- }
- func TestProjectPlan_MultiInput(t *testing.T) {
- var tests = []struct {
- sql string
- data interface{}
- result []map[string]interface{}
- }{
- {
- 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": float64(6), //json marshall problem
- }},
- },
- {
- 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": float64(34),
- }},
- },
- {
- sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "src1",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- }, {
- "id1": float64(2),
- }, {
- "id1": float64(3),
- }},
- },
- {
- sql: "SELECT id1 FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "src1",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id2": 2, "f1": "v2"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- }, {}, {
- "id1": float64(3),
- }},
- },
- {
- sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "src1",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- "f1": "v1",
- }, {
- "id1": float64(2),
- "f1": "v2",
- }, {
- "id1": float64(3),
- "f1": "v1",
- }},
- },
- {
- sql: "SELECT * FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "src1",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id2": 2, "f2": "v2"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- "f1": "v1",
- }, {
- "id2": float64(2),
- "f2": "v2",
- }, {
- "id1": float64(3),
- "f1": "v1",
- }},
- },
- {
- sql: "SELECT src1.* FROM src1 WHERE f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "src1",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- "f1": "v1",
- }, {
- "id1": float64(2),
- "f1": "v2",
- }, {
- "id1": float64(3),
- "f1": "v1",
- }},
- },
- {
- sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- }, {
- "id1": float64(2),
- }, {
- "id1": float64(3),
- }},
- },
- {
- sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2 WHERE src1.f1 = \"v1\" GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id2": 3, "f1": "v1"}},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- }, {
- "id1": float64(2),
- }, {}},
- },
- {
- sql: "SELECT abc FROM tbl group by abc",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.Tuple{
- Emitter: "tbl",
- Message: xsql.Message{
- "abc": int64(6),
- "def": "hello",
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "abc": float64(6),
- }},
- },
- {
- sql: "SELECT abc FROM tbl group by abc",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.Tuple{
- Emitter: "tbl",
- Message: xsql.Message{
- "def": "hello",
- },
- },
- },
- },
- result: []map[string]interface{}{{}},
- },
- {
- sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- },
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- {
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- }, {
- "id1": float64(2),
- }},
- },
- {
- sql: "SELECT id1 FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- },
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- {
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id2": 2, "f1": "v2"},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- }, {}},
- },
- {
- sql: "SELECT src2.id2 FROM src1 left join src2 on src1.id1 = src2.id2 GROUP BY src2.f2, TUMBLINGWINDOW(ss, 10)",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id2": float64(2),
- }, {
- "id2": float64(4),
- }, {}},
- },
- {
- sql: "SELECT src1.*, f2 FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- "f1": "v1",
- "f2": "w2",
- }, {
- "id1": float64(2),
- "f1": "v2",
- "f2": "w3",
- }, {
- "id1": float64(3),
- "f1": "v1",
- }},
- },
- {
- sql: "SELECT * FROM src1 left join src2 GROUP BY TUMBLINGWINDOW(ss, 10)",
- data: xsql.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id": 1, "f1": "v1"}},
- {Emitter: "src2", Message: xsql.Message{"id": 2, "f2": "w2"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id": 2, "f1": "v2"}},
- {Emitter: "src2", Message: xsql.Message{"id": 4, "f2": "w3"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id": 3, "f1": "v1"}},
- },
- },
- },
- result: []map[string]interface{}{{
- "id": float64(1),
- "f1": "v1",
- "f2": "w2",
- }, {
- "id": float64(2),
- "f1": "v2",
- "f2": "w3",
- }, {
- "id": float64(3),
- "f1": "v1",
- }},
- },
- {
- sql: "SELECT src1.* FROM src1 GROUP BY TUMBLINGWINDOW(ss, 10), f1",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 1, "f1": "v1"},
- },
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 3, "f1": "v1"},
- },
- },
- {
- &xsql.Tuple{
- Emitter: "src1",
- Message: xsql.Message{"id1": 2, "f1": "v2"},
- },
- },
- },
- result: []map[string]interface{}{{
- "id1": float64(1),
- "f1": "v1",
- }, {
- "id1": float64(2),
- "f1": "v2",
- }},
- },
- {
- 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{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id2": float64(2),
- "id1": float64(1),
- "f1": "v1",
- }, {
- "id2": float64(4),
- "id1": float64(2),
- "f1": "v2",
- }, {
- "id1": float64(3),
- "f1": "v1",
- }},
- },
- {
- 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{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}},
- {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "id2": float64(2),
- "id1": float64(1),
- "f1": "v1",
- }, {
- "id2": float64(4),
- "id1": float64(2),
- "f1": "v2",
- }, {
- "id1": float64(3),
- "f1": "v1",
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- contextLogger := common.Log.WithField("rule", "TestProjectPlan_MultiInput")
- ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- pp := &ProjectPlan{Fields: stmt.Fields}
- pp.isTest = true
- fv, afv := xsql.NewAggregateFunctionValuers()
- result := pp.Apply(ctx, tt.data, fv, afv)
- var mapRes []map[string]interface{}
- if v, ok := result.([]byte); ok {
- err := json.Unmarshal(v, &mapRes)
- if err != nil {
- t.Errorf("Failed to parse the input into map.\n")
- continue
- }
- //fmt.Printf("%t\n", mapRes["rengine_field_0"])
- if !reflect.DeepEqual(tt.result, mapRes) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
- }
- } else {
- t.Errorf("The returned result is not type of []byte\n")
- }
- }
- }
- func TestProjectPlan_Funcs(t *testing.T) {
- var tests = []struct {
- sql string
- data interface{}
- result []map[string]interface{}
- }{
- {
- 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),
- }},
- }, {
- sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "test",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"a": 53.1},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 27.4},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 123123.7},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "r": float64(53),
- }, {
- "r": float64(27),
- }, {
- "r": float64(123124),
- }},
- }, {
- sql: "SELECT round(a) as r FROM test GROUP BY TumblingWindow(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "test",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"a": 53.1},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 27.4},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 123123.7},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "r": float64(53),
- }, {
- "r": float64(27),
- }, {
- "r": float64(123124),
- }},
- }, {
- sql: "SELECT round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: xsql.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
- {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
- {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
- {Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
- },
- },
- },
- result: []map[string]interface{}{{
- "r": float64(66),
- }, {
- "r": float64(73),
- }, {
- "r": float64(89),
- }},
- }, {
- 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.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}},
- {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}},
- {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}},
- {Emitter: "test1", Message: xsql.Message{"id": 3, "b": 6}},
- },
- },
- },
- result: []map[string]interface{}{{
- "concat": "165.5512",
- }, {
- "concat": "273.49934",
- }, {
- "concat": "388.886",
- }},
- }, {
- sql: "SELECT count(a) as r FROM test",
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "a": 47.5,
- },
- },
- result: []map[string]interface{}{{
- "r": float64(1),
- }},
- }, {
- sql: "SELECT meta(test.device) as d FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: xsql.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}, Metadata: xsql.Metadata{"device": "devicea"}},
- {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}, Metadata: xsql.Metadata{"device": "deviceb"}},
- {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 3, "a": 88.88}, Metadata: xsql.Metadata{"device": "devicec"}},
- {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 := common.Log.WithField("rule", "TestProjectPlan_Funcs")
- ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- if err != nil {
- t.Error(err)
- }
- pp := &ProjectPlan{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
- pp.isTest = true
- fv, afv := xsql.NewAggregateFunctionValuers()
- result := pp.Apply(ctx, tt.data, fv, afv)
- var mapRes []map[string]interface{}
- if v, ok := result.([]byte); ok {
- err := json.Unmarshal(v, &mapRes)
- if err != nil {
- t.Errorf("Failed to parse the input into map.\n")
- continue
- }
- //fmt.Printf("%t\n", mapRes["rengine_field_0"])
- if !reflect.DeepEqual(tt.result, mapRes) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
- }
- } else {
- t.Errorf("%d. The returned result is not type of []byte\n", i)
- }
- }
- }
- func TestProjectPlan_AggFuncs(t *testing.T) {
- var tests = []struct {
- sql string
- data interface{}
- result []map[string]interface{}
- }{
- {
- sql: "SELECT count(*) as c, round(a) as r FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "r": 122}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "r": 89}},
- {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "c": float64(2),
- "r": float64(122),
- }, {
- "c": float64(2),
- "r": float64(89),
- }},
- },
- {
- sql: "SELECT avg(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 4, "a": 98.31}},
- {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
- {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
- {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "avg": 116.68,
- }, {
- "avg": 51.815,
- }},
- },
- {
- sql: "SELECT max(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color",
- data: xsql.GroupedTuplesSet{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
- {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w1"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "max": 177.51,
- }, {
- "max": 89.03,
- }},
- },
- {
- sql: "SELECT min(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: xsql.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- result: []map[string]interface{}{{
- "min": 68.55,
- }},
- }, {
- sql: "SELECT min(a) as m FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)",
- data: xsql.JoinTupleSets{
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "m": 68.55}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- result: []map[string]interface{}{{
- "m": 68.55,
- }},
- }, {
- sql: "SELECT sum(a) FROM test GROUP BY TumblingWindow(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "test",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "sum": float64(123203),
- }},
- }, {
- sql: "SELECT sum(a) as s FROM test GROUP BY TumblingWindow(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "test",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"a": 53, "s": 123203},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "s": float64(123203),
- }},
- }, {
- sql: "SELECT sum(a) FROM test GROUP BY TumblingWindow(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "test",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 27},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "sum": float64(123203),
- }},
- },
- {
- 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{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- {Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- {Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
- {Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- {Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "count": float64(2),
- "meta": "devicea",
- }, {
- "count": float64(2),
- "meta": "devicec",
- }},
- },
- {
- 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{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33, "c": 2, "d": "devicea"}},
- {Emitter: "test1", Message: xsql.Message{"id": 1, "color": "w2"}, Metadata: xsql.Metadata{"device": "devicea"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}},
- {Emitter: "test1", Message: xsql.Message{"id": 5, "color": "w2"}, Metadata: xsql.Metadata{"device": "deviceb"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03, "c": 2, "d": "devicec"}},
- {Emitter: "test1", Message: xsql.Message{"id": 2, "color": "w1"}, Metadata: xsql.Metadata{"device": "devicec"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- {Emitter: "test1", Message: xsql.Message{"id": 4, "color": "w1"}, Metadata: xsql.Metadata{"device": "deviced"}},
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "c": float64(2),
- "d": "devicea",
- }, {
- "c": float64(2),
- "d": "devicec",
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- contextLogger := common.Log.WithField("rule", "TestProjectPlan_AggFuncs")
- ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- if err != nil {
- t.Error(err)
- }
- pp := &ProjectPlan{Fields: stmt.Fields, IsAggregate: true}
- fv, afv := xsql.NewAggregateFunctionValuers()
- result := pp.Apply(ctx, tt.data, fv, afv)
- var mapRes []map[string]interface{}
- if v, ok := result.([]byte); ok {
- err := json.Unmarshal(v, &mapRes)
- if err != nil {
- t.Errorf("Failed to parse the input into map.\n")
- continue
- }
- //fmt.Printf("%t\n", mapRes["rengine_field_0"])
- if !reflect.DeepEqual(tt.result, mapRes) {
- t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
- }
- } else {
- t.Errorf("%d. %q\n\nThe returned result is not type of []byte: %#v\n", i, tt.sql, result)
- }
- }
- }
- func TestProjectPlanError(t *testing.T) {
- var tests = []struct {
- sql string
- data interface{}
- result interface{}
- }{
- {
- sql: "SELECT a FROM test",
- data: errors.New("an error from upstream"),
- result: errors.New("an error from upstream"),
- }, {
- 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)"),
- }, {
- 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})"),
- }, {
- 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: only float64 & int type are supported"),
- }, {
- sql: `SELECT round(a) as r FROM test`,
- data: &xsql.Tuple{
- Emitter: "test",
- Message: xsql.Message{
- "abc": "common string",
- },
- },
- result: errors.New("run Select error: call func round error: only float64 & int type are supported"),
- }, {
- 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{
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.54}},
- {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 4, "a": "dde"}},
- {Emitter: "src2", Message: xsql.Message{"id": 4, "color": "w2"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.54}},
- {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}},
- },
- },
- },
- {
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 2, "a": 89.03}},
- {Emitter: "src2", Message: xsql.Message{"id": 2, "color": "w1"}},
- },
- },
- &xsql.JoinTuple{
- Tuples: []xsql.Tuple{
- {Emitter: "test", Message: xsql.Message{"id": 4, "a": 14.6}},
- {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)"),
- }, {
- sql: "SELECT sum(a) as sum FROM test GROUP BY TumblingWindow(ss, 10)",
- data: xsql.WindowTuplesSet{
- xsql.WindowTuples{
- Emitter: "test",
- Tuples: []xsql.Tuple{
- {
- Emitter: "src1",
- Message: xsql.Message{"a": 53},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": "ddd"},
- }, {
- Emitter: "src1",
- Message: xsql.Message{"a": 123123},
- },
- },
- },
- },
- result: errors.New("run Select error: call func sum error: requires int but found string(ddd)"),
- }, {
- 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 := common.Log.WithField("rule", "TestProjectPlanError")
- ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
- for i, tt := range tests {
- stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- pp := &ProjectPlan{Fields: stmt.Fields, IsAggregate: xsql.IsAggStatement(stmt)}
- pp.isTest = true
- fv, afv := xsql.NewAggregateFunctionValuers()
- result := pp.Apply(ctx, tt.data, fv, afv)
- 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)
- }
- }
- }
|