package operator import ( "encoding/json" "errors" "fmt" "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" "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": cast.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": cast.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{}{{ "": "value", }}, }, { sql: `SELECT 3.4 FROM test`, data: &xsql.Tuple{ Emitter: "test", Message: xsql.Message{}, }, result: []map[string]interface{}{{ "": 3.4, }}, }, { sql: `SELECT 5 FROM test`, data: &xsql.Tuple{ Emitter: "test", Message: xsql.Message{}, }, result: []map[string]interface{}{{ "": 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[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": []interface{}{ map[string]interface{}{"b": "hello3"}, map[string]interface{}{"b": "hello4"}, map[string]interface{}{"b": "hello5"}, }, }}, }, { 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, }, }}, }, { 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, }, }}, }, { 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, }, }}, }, { 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": []interface{}{ 3.14, 3.141, 3.1415, 3.14159, }, }}, }, { 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": []interface{}{ 3.14, }, }}, }, { 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", }}, }, { 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", }}, }, } 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{Fields: stmt.Fields, SendMeta: true} fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters) 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["kuiper_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 %#v is not type of []byte\n", result, 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{ Content: []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{ Content: []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{ Content: []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{ Content: []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{ Content: []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{ Content: []xsql.JoinTuple{ { Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}}, {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}}, {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}}, }, }, { 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{ Content: []xsql.JoinTuple{ { Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}}, {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}}, {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}}, }, }, { 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{ { Content: []xsql.DataValuer{ &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{ { Content: []xsql.DataValuer{ &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{ { Content: []xsql.DataValuer{ &xsql.Tuple{ Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}, }, &xsql.Tuple{ Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}, }, }, }, { Content: []xsql.DataValuer{ &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{ { Content: []xsql.DataValuer{ &xsql.Tuple{ Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}, }, &xsql.Tuple{ Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}, }, }, }, { Content: []xsql.DataValuer{ &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{ { Content: []xsql.DataValuer{ &xsql.JoinTuple{ Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}}, {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}}, }, }, }, }, { Content: []xsql.DataValuer{ &xsql.JoinTuple{ Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}}, {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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{ Content: []xsql.JoinTuple{ { Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}}, {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}}, {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}}, }, }, { 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{ Content: []xsql.JoinTuple{ { Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id": 1, "f1": "v1"}}, {Emitter: "src2", Message: xsql.Message{"id": 2, "f2": "w2"}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id": 2, "f1": "v2"}}, {Emitter: "src2", Message: xsql.Message{"id": 4, "f2": "w3"}}, }, }, { 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{ { Content: []xsql.DataValuer{ &xsql.Tuple{ Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}, }, &xsql.Tuple{ Emitter: "src1", Message: xsql.Message{"id1": 3, "f1": "v1"}, }, }, }, { Content: []xsql.DataValuer{ &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{ { Content: []xsql.DataValuer{ &xsql.JoinTuple{ Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}}, {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}}, }, }, }, }, { Content: []xsql.DataValuer{ &xsql.JoinTuple{ Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}}, {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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{ { Content: []xsql.DataValuer{ &xsql.JoinTuple{ Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 1, "f1": "v1"}}, {Emitter: "src2", Message: xsql.Message{"id2": 2, "f2": "w2"}}, }, }, }, }, { Content: []xsql.DataValuer{ &xsql.JoinTuple{ Tuples: []xsql.Tuple{ {Emitter: "src1", Message: xsql.Message{"id1": 2, "f1": "v2"}}, {Emitter: "src2", Message: xsql.Message{"id2": 4, "f2": "w3"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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 := 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{Fields: stmt.Fields} fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters) 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["kuiper_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{ Content: []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{ Content: []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{ Content: []xsql.JoinTuple{ { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}}, {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}}, {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}}, }, }, { 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{ Content: []xsql.JoinTuple{ { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 1, "a": 65.55}}, {Emitter: "test1", Message: xsql.Message{"id": 1, "b": 12}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 2, "a": 73.499}}, {Emitter: "test1", Message: xsql.Message{"id": 2, "b": 34}}, }, }, { 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{ Content: []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}}, }, }, { 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}}, }, }, { 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 := 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{Fields: stmt.Fields, IsAggregate: ast.IsAggStatement(stmt)} fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters) 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["kuiper_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, 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{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, WindowRange: &xsql.WindowRange{ WindowStart: 1541152486013, WindowEnd: 1541152487013, }, }, { Content: []xsql.DataValuer{ &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"}}, }, }, }, WindowRange: &xsql.WindowRange{ WindowStart: 1541152486013, WindowEnd: 1541152487013, }, }, }, result: []map[string]interface{}{{ "c": float64(2), "r": float64(122), "ws": float64(1541152486013), "we": float64(1541152487013), }, { "c": float64(2), "r": float64(89), "ws": float64(1541152486013), "we": float64(1541152487013), }}, }, { 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{ { Content: []xsql.DataValuer{ &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}}, {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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(1), "a": 122.33, "s": 122.33, "min": 122.33, "max": 122.33, }, { "c": float64(2), "s": 103.63, "a": 51.815, "min": 14.6, "max": 89.03, }}, }, { sql: "SELECT avg(a) FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color", data: xsql.GroupedTuplesSet{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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), window_start(), window_end() FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10)", data: &xsql.JoinTupleSets{ Content: []xsql.JoinTuple{ { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 1, "a": 122.33}}, {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}}, {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 5, "a": 177.51}}, {Emitter: "src2", Message: xsql.Message{"id": 5, "color": "w2"}}, }, }, }, WindowRange: &xsql.WindowRange{ WindowStart: 1541152486013, WindowEnd: 1541152487013, }, }, result: []map[string]interface{}{{ "min": 68.55, "window_start": float64(1541152486013), "window_end": float64(1541152487013), }}, }, { 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.JoinTupleSets{ Content: []xsql.JoinTuple{ { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 1}}, {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}}, }, }, { Tuples: []xsql.Tuple{ {Emitter: "test", Message: xsql.Message{"id": 1, "a": 68.55}}, {Emitter: "src2", Message: xsql.Message{"id": 1, "color": "w2"}}, }, }, { 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{}{{ "all": float64(3), "c": float64(2), "a": 123.03, "s": 246.06, "min": 68.55, "max": 177.51, }}, }, { sql: "SELECT sum(a), window_start() as ws, window_end() FROM test GROUP BY TumblingWindow(ss, 10)", data: xsql.WindowTuplesSet{ Content: []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}, }, }, }, }, WindowRange: &xsql.WindowRange{ WindowStart: 1541152486013, WindowEnd: 1541152487013, }, }, result: []map[string]interface{}{{ "sum": float64(123203), "ws": float64(1541152486013), "window_end": float64(1541152487013), }}, }, { sql: "SELECT sum(a) as s FROM test GROUP BY TumblingWindow(ss, 10)", data: xsql.WindowTuplesSet{ Content: []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{ Content: []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(*) 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.WindowTuplesSet{ Content: []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{"s": 123123}, }, }, }, }, }, result: []map[string]interface{}{{ "all": float64(3), "c": float64(2), "a": float64(40), "s": float64(80), "min": float64(27), "max": float64(53), }}, }, { 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{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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", }}, }, { sql: "SELECT * FROM test Inner Join test1 on test.id = test1.id GROUP BY TumblingWindow(ss, 10), test1.color", data: xsql.GroupedTuplesSet{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, WindowRange: &xsql.WindowRange{ WindowStart: 1541152486013, WindowEnd: 1541152487013, }, }, { Content: []xsql.DataValuer{ &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"}}, }, }, }, WindowRange: &xsql.WindowRange{ WindowStart: 1541152486013, WindowEnd: 1541152487013, }, }, }, result: []map[string]interface{}{{ "a": 122.33, "c": float64(2), "color": "w2", "id": float64(1), "r": float64(122), }, { "a": 89.03, "c": float64(2), "color": "w1", "id": float64(2), "r": float64(89), }}, }, { 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{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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{}{{ "r1": []interface{}{122.33, 177.51}, }, {"r1": []interface{}{89.03, 14.6}}}, }, { sql: "SELECT collect(*)[1] as c1 FROM test GROUP BY TumblingWindow(ss, 10)", data: xsql.WindowTuplesSet{ Content: []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}, }, }, }, }, WindowRange: &xsql.WindowRange{ WindowStart: 1541152486013, WindowEnd: 1541152487013, }, }, result: []map[string]interface{}{{ "c1": map[string]interface{}{ "a": float64(27), }, }}, }, { sql: "SELECT collect(*)[1]->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)", data: xsql.WindowTuplesSet{ Content: []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{}{{ "c1": float64(27), }}, }, { sql: "SELECT collect(*)[1]->sl[0] as c1 FROM test GROUP BY TumblingWindow(ss, 10)", data: xsql.WindowTuplesSet{ Content: []xsql.WindowTuples{{ Emitter: "test", Tuples: []xsql.Tuple{ { Emitter: "src1", Message: xsql.Message{"a": 53, "sl": []string{"hello", "world"}}, }, { Emitter: "src1", Message: xsql.Message{"a": 27, "sl": []string{"new", "horizon"}}, }, { Emitter: "src1", Message: xsql.Message{"a": 123123, "sl": []string{"south", "africa"}}, }, }, }, }, }, result: []map[string]interface{}{{ "c1": "new", }}, }, { 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{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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{}{ { "r1": []interface{}{ map[string]interface{}{"a": 122.33, "c": float64(2), "color": "w2", "id": float64(1), "r": float64(122)}, map[string]interface{}{"a": 177.51, "color": "w2", "id": float64(5)}}, }, { "r1": []interface{}{ map[string]interface{}{"a": 89.03, "c": float64(2), "color": "w1", "id": float64(2), "r": float64(89)}, map[string]interface{}{"a": 14.6, "color": "w1", "id": float64(4)}}, }, }, }, { sql: "SELECT deduplicate(a, false)->a as c1 FROM test GROUP BY TumblingWindow(ss, 10)", data: xsql.WindowTuplesSet{ Content: []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{}{{ "c1": float64(123123), }}, }, { sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)", data: xsql.WindowTuplesSet{ Content: []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": 53}, }, }, }, }, }, result: []map[string]interface{}{{}}, }, { sql: "SELECT deduplicate(a, false) as c1 FROM test GROUP BY TumblingWindow(ss, 10)", data: xsql.WindowTuplesSet{ Content: []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": 53}, }, }, }, }, }, result: []map[string]interface{}{{}}, }, } 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{Fields: stmt.Fields, IsAggregate: true} fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters) 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["kuiper_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{ { Content: []xsql.DataValuer{ &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"}}, }, }, }, }, { Content: []xsql.DataValuer{ &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{ Content: []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 := 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{Fields: stmt.Fields, IsAggregate: ast.IsAggStatement(stmt)} fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters) 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) } } }