123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083 |
- package plans
- import (
- "encoding/json"
- "engine/common"
- "engine/xsql"
- "fmt"
- "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",
- },
- },
- result: []map[string]interface{}{{
- "a": "val_a",
- }},
- },
- {
- 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.000000431Z",
- }},
- },
- {
- 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[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->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,
- }},
- },
- //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),
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- for i, tt := range tests {
- stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- pp := &ProjectPlan{Fields:stmt.Fields}
- result := pp.Apply(nil, tt.data)
- 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_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 * 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 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 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 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 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))
- for i, tt := range tests {
- stmt, _ := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
- pp := &ProjectPlan{Fields:stmt.Fields}
- result := pp.Apply(nil, tt.data)
- 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 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",
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- 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}
- result := pp.Apply(nil, tt.data)
- 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_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,},},
- {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{}{{
- "c": float64(2),
- "r": float64(122),
- },{
- "c": float64(2),
- "r": float64(89),
- }},
- },
- {
- 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": 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) as max 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) as min 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 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": 27 },
- },{
- Emitter: "src1",
- Message: xsql.Message{ "a": 123123 },
- },
- },
- },
- },
- result: []map[string]interface{}{{
- "sum": float64(123203),
- }},
- },
- }
- fmt.Printf("The test bucket size is %d.\n\n", len(tests))
- 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}
- result := pp.Apply(nil, tt.data)
- 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")
- }
- }
- }
|