ekuiper-new/xsql/processors/xsql_processor_test.go

package processors

import (
	"encoding/json"
	"fmt"
	"github.com/emqx/kuiper/common"
	"github.com/emqx/kuiper/xsql"
	"github.com/emqx/kuiper/xstream/api"
	"github.com/emqx/kuiper/xstream/nodes"
	"github.com/emqx/kuiper/xstream/test"
	"path"
	"reflect"
	"strings"
	"testing"
	"time"
)

var DbDir = getDbDir()

func getDbDir() string {
	dbDir, err := common.GetAndCreateDataLoc("test")
	if err != nil {
		log.Panic(err)
	}
	log.Infof("db location is %s", dbDir)
	return dbDir
}

func TestStreamCreateProcessor(t *testing.T) {
	var tests = []struct {
		s   string
		r   []string
		err string
	}{
		{
			s: `SHOW STREAMS;`,
			r: []string{"No stream definitions are found."},
		},
		{
			s:   `EXPLAIN STREAM topic1;`,
			err: "Stream topic1 is not found.",
		},
		{
			s: `CREATE STREAM topic1 (
					USERID BIGINT,
					FIRST_NAME STRING,
					LAST_NAME STRING,
					NICKNAMES ARRAY(STRING),
					Gender BOOLEAN,
					ADDRESS STRUCT(STREET_NAME STRING, NUMBER BIGINT),
				) WITH (DATASOURCE="users", FORMAT="AVRO", KEY="USERID");`,
			r: []string{"Stream topic1 is created."},
		},
		{
			s: `CREATE STREAM topic1 (
					USERID BIGINT,
				) WITH (DATASOURCE="users", FORMAT="AVRO", KEY="USERID");`,
			err: "Create stream fails: Item topic1 already exists.",
		},
		{
			s: `EXPLAIN STREAM topic1;`,
			r: []string{"TO BE SUPPORTED"},
		},
		{
			s: `DESCRIBE STREAM topic1;`,
			r: []string{"Fields\n--------------------------------------------------------------------------------\nUSERID\tbigint\nFIRST_NAME\tstring\nLAST_NAME\tstring\nNICKNAMES\t" +
				"array(string)\nGender\tboolean\nADDRESS\tstruct(STREET_NAME string, NUMBER bigint)\n\n" +
				"DATASOURCE: users\nFORMAT: AVRO\nKEY: USERID\n"},
		},
		{
			s: `SHOW STREAMS;`,
			r: []string{"topic1"},
		},
		{
			s: `DROP STREAM topic1;`,
			r: []string{"Stream topic1 is dropped."},
		},
		{
			s:   `DESCRIBE STREAM topic1;`,
			err: "Stream topic1 is not found.",
		},
		{
			s:   `DROP STREAM topic1;`,
			err: "Drop stream fails: topic1 is not found.",
		},
	}

	fmt.Printf("The test bucket size is %d.\n\n", len(tests))

	streamDB := path.Join(getDbDir(), "streamTest")
	for i, tt := range tests {
		results, err := NewStreamProcessor(tt.s, streamDB).Exec()
		if !reflect.DeepEqual(tt.err, errstring(err)) {
			t.Errorf("%d. %q: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.s, tt.err, err)
		} else if tt.err == "" {
			if !reflect.DeepEqual(tt.r, results) {
				t.Errorf("%d. %q\n\nstmt mismatch:\n\ngot=%#v\n\n", i, tt.s, results)
			}
		}
	}
}

func createStreams(t *testing.T) {
	demo := `CREATE STREAM demo (
					color STRING,
					size BIGINT,
					ts BIGINT
				) WITH (DATASOURCE="demo", FORMAT="json", KEY="ts");`
	_, err := NewStreamProcessor(demo, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
	demo1 := `CREATE STREAM demo1 (
					temp FLOAT,
					hum BIGINT,
					ts BIGINT
				) WITH (DATASOURCE="demo1", FORMAT="json", KEY="ts");`
	_, err = NewStreamProcessor(demo1, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
	sessionDemo := `CREATE STREAM sessionDemo (
					temp FLOAT,
					hum BIGINT,
					ts BIGINT
				) WITH (DATASOURCE="sessionDemo", FORMAT="json", KEY="ts");`
	_, err = NewStreamProcessor(sessionDemo, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
}

func dropStreams(t *testing.T) {
	demo := `DROP STREAM demo`
	_, err := NewStreamProcessor(demo, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
	demo1 := `DROP STREAM demo1`
	_, err = NewStreamProcessor(demo1, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
	sessionDemo := `DROP STREAM sessionDemo`
	_, err = NewStreamProcessor(sessionDemo, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
}

func getMockSource(name string, done chan<- struct{}, size int) *nodes.SourceNode {
	var data []*xsql.Tuple
	switch name {
	case "demo":
		data = []*xsql.Tuple{
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "red",
					"size":  3,
					"ts":    1541152486013,
				},
				Timestamp: 1541152486013,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "blue",
					"size":  6,
					"ts":    1541152486822,
				},
				Timestamp: 1541152486822,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "blue",
					"size":  2,
					"ts":    1541152487632,
				},
				Timestamp: 1541152487632,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "yellow",
					"size":  4,
					"ts":    1541152488442,
				},
				Timestamp: 1541152488442,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "red",
					"size":  1,
					"ts":    1541152489252,
				},
				Timestamp: 1541152489252,
			},
		}
	case "demo1":
		data = []*xsql.Tuple{
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  65,
					"ts":   1541152486013,
				},
				Timestamp: 1541152486013,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 27.5,
					"hum":  59,
					"ts":   1541152486823,
				},
				Timestamp: 1541152486823,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 28.1,
					"hum":  75,
					"ts":   1541152487632,
				},
				Timestamp: 1541152487632,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 27.4,
					"hum":  80,
					"ts":   1541152488442,
				},
				Timestamp: 1541152488442,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  62,
					"ts":   1541152489252,
				},
				Timestamp: 1541152489252,
			},
		}
	case "sessionDemo":
		data = []*xsql.Tuple{
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  65,
					"ts":   1541152486013,
				},
				Timestamp: 1541152486013,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 27.5,
					"hum":  59,
					"ts":   1541152486823,
				},
				Timestamp: 1541152486823,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 28.1,
					"hum":  75,
					"ts":   1541152487932,
				},
				Timestamp: 1541152487932,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 27.4,
					"hum":  80,
					"ts":   1541152488442,
				},
				Timestamp: 1541152488442,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  62,
					"ts":   1541152489252,
				},
				Timestamp: 1541152489252,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 26.2,
					"hum":  63,
					"ts":   1541152490062,
				},
				Timestamp: 1541152490062,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 26.8,
					"hum":  71,
					"ts":   1541152490872,
				},
				Timestamp: 1541152490872,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 28.9,
					"hum":  85,
					"ts":   1541152491682,
				},
				Timestamp: 1541152491682,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 29.1,
					"hum":  92,
					"ts":   1541152492492,
				},
				Timestamp: 1541152492492,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 32.2,
					"hum":  99,
					"ts":   1541152493202,
				},
				Timestamp: 1541152493202,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 30.9,
					"hum":  87,
					"ts":   1541152494112,
				},
				Timestamp: 1541152494112,
			},
		}
	}
	return nodes.NewSourceNodeWithSource(name, test.NewMockSource(data[:size], done, false), map[string]string{
		"DATASOURCE": name,
	})
}

func TestSingleSQL(t *testing.T) {
	var tests = []struct {
		name string
		sql  string
		r    [][]map[string]interface{}
		m    map[string]interface{}
	}{
		{
			name: `rule1`,
			sql:  `SELECT * FROM demo`,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
					"size":  float64(3),
					"ts":    float64(1541152486013),
				}},
				{{
					"color": "blue",
					"size":  float64(6),
					"ts":    float64(1541152486822),
				}},
				{{
					"color": "blue",
					"size":  float64(2),
					"ts":    float64(1541152487632),
				}},
				{{
					"color": "yellow",
					"size":  float64(4),
					"ts":    float64(1541152488442),
				}},
				{{
					"color": "red",
					"size":  float64(1),
					"ts":    float64(1541152489252),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo_0_records_out_total":  int64(5),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(5),
				"kuiper_op_project_0_records_out_total":  int64(5),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(5),
				"kuiper_sink_MockSink_0_records_out_total": int64(5),

				"kuiper_source_demo_0_exceptions_total":  int64(0),
				"kuiper_source_demo_0_records_in_total":  int64(5),
				"kuiper_source_demo_0_records_out_total": int64(5),
			},
		}, {
			name: `rule2`,
			sql:  `SELECT color, ts FROM demo where size > 3`,
			r: [][]map[string]interface{}{
				{{
					"color": "blue",
					"ts":    float64(1541152486822),
				}},
				{{
					"color": "yellow",
					"ts":    float64(1541152488442),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo_0_records_out_total":  int64(5),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(2),
				"kuiper_op_project_0_records_out_total":  int64(2),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(2),
				"kuiper_sink_MockSink_0_records_out_total": int64(2),

				"kuiper_source_demo_0_exceptions_total":  int64(0),
				"kuiper_source_demo_0_records_in_total":  int64(5),
				"kuiper_source_demo_0_records_out_total": int64(5),

				"kuiper_op_filter_0_exceptions_total":   int64(0),
				"kuiper_op_filter_0_process_latency_ms": int64(0),
				"kuiper_op_filter_0_records_in_total":   int64(5),
				"kuiper_op_filter_0_records_out_total":  int64(2),
			},
		}, {
			name: `rule3`,
			sql:  `SELECT size as Int8, ts FROM demo where size > 3`,
			r: [][]map[string]interface{}{
				{{
					"Int8": float64(6),
					"ts":   float64(1541152486822),
				}},
				{{
					"Int8": float64(4),
					"ts":   float64(1541152488442),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo_0_records_out_total":  int64(5),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(2),
				"kuiper_op_project_0_records_out_total":  int64(2),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(2),
				"kuiper_sink_MockSink_0_records_out_total": int64(2),

				"kuiper_source_demo_0_exceptions_total":  int64(0),
				"kuiper_source_demo_0_records_in_total":  int64(5),
				"kuiper_source_demo_0_records_out_total": int64(5),

				"kuiper_op_filter_0_exceptions_total":   int64(0),
				"kuiper_op_filter_0_process_latency_ms": int64(0),
				"kuiper_op_filter_0_records_in_total":   int64(5),
				"kuiper_op_filter_0_records_out_total":  int64(2),
			},
		},
	}
	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
	createStreams(t)
	defer dropStreams(t)
	done := make(chan struct{})
	//defer close(done)
	for i, tt := range tests {
		p := NewRuleProcessor(DbDir)
		parser := xsql.NewParser(strings.NewReader(tt.sql))
		var sources []*nodes.SourceNode
		if stmt, err := xsql.Language.Parse(parser); err != nil {
			t.Errorf("parse sql %s error: %s", tt.sql, err)
		} else {
			if selectStmt, ok := stmt.(*xsql.SelectStatement); !ok {
				t.Errorf("sql %s is not a select statement", tt.sql)
			} else {
				streams := xsql.GetStreams(selectStmt)
				for _, stream := range streams {
					source := getMockSource(stream, done, 5)
					sources = append(sources, source)
				}
			}
		}
		tp, inputs, err := p.createTopoWithSources(&api.Rule{Id: tt.name, Sql: tt.sql, Options:map[string]interface{}{
			"bufferLength": float64(100),
		}}, sources)
		if err != nil {
			t.Error(err)
		}
		mockSink := test.NewMockSink()
		sink := nodes.NewSinkNodeWithSink("MockSink", mockSink)
		tp.AddSink(inputs, sink)
		count := len(sources)
		errCh := tp.Open()
		func() {
			for {
				select {
				case err = <-errCh:
					t.Log(err)
					tp.Cancel()
					return
				case <-done:
					count--
					log.Infof("%d sources remaining", count)
					if count <= 0 {
						log.Info("stream stopping")
						time.Sleep(1 * time.Second)
						tp.Cancel()
						return
					}
				default:
				}
			}
		}()
		results := mockSink.GetResults()
		var maps [][]map[string]interface{}
		for _, v := range results {
			var mapRes []map[string]interface{}
			err := json.Unmarshal(v, &mapRes)
			if err != nil {
				t.Errorf("Failed to parse the input into map")
				continue
			}
			maps = append(maps, mapRes)
		}
		//if !reflect.DeepEqual(tt.r, maps) {
		//	t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.r, maps)
		//	continue
		//}
		metrics := tp.GetMetrics()
		log.Infof("metrics: %v", metrics)
		for k, v := range tt.m {
			if v != metrics[k] {
				t.Errorf("%d. %q\n\nmetrics mismatch for %s:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, k, v, metrics[k])
				break
			}
		}
	}
}

func TestWindow(t *testing.T) {
	common.IsTesting = true
	var tests = []struct {
		name string
		sql  string
		size int
		r    [][]map[string]interface{}
		m    map[string]interface{}
	}{
		{
			name: `rule1`,
			sql:  `SELECT * FROM demo GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
			size: 5,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
					"size":  float64(3),
					"ts":    float64(1541152486013),
				}, {
					"color": "blue",
					"size":  float64(6),
					"ts":    float64(1541152486822),
				}},
				{{
					"color": "red",
					"size":  float64(3),
					"ts":    float64(1541152486013),
				}, {
					"color": "blue",
					"size":  float64(6),
					"ts":    float64(1541152486822),
				}, {
					"color": "blue",
					"size":  float64(2),
					"ts":    float64(1541152487632),
				}},
				{{
					"color": "blue",
					"size":  float64(2),
					"ts":    float64(1541152487632),
				}, {
					"color": "yellow",
					"size":  float64(4),
					"ts":    float64(1541152488442),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo_0_records_out_total":  int64(5),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(3),
				"kuiper_op_project_0_records_out_total":  int64(3),

				"kuiper_sink_mockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_mockSink_0_records_in_total":  int64(3),
				"kuiper_sink_mockSink_0_records_out_total": int64(3),

				"kuiper_source_demo_0_exceptions_total":  int64(0),
				"kuiper_source_demo_0_records_in_total":  int64(5),
				"kuiper_source_demo_0_records_out_total": int64(5),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(5),
				"kuiper_op_window_0_records_out_total":  int64(3),
			},
		}, {
			name: `rule2`,
			sql:  `SELECT color, ts FROM demo where size > 2 GROUP BY tumblingwindow(ss, 1)`,
			size: 5,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
					"ts":    float64(1541152486013),
				}, {
					"color": "blue",
					"ts":    float64(1541152486822),
				}},
				{{
					"color": "yellow",
					"ts":    float64(1541152488442),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo_0_records_out_total":  int64(5),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(2),
				"kuiper_op_project_0_records_out_total":  int64(2),

				"kuiper_sink_mockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_mockSink_0_records_in_total":  int64(2),
				"kuiper_sink_mockSink_0_records_out_total": int64(2),

				"kuiper_source_demo_0_exceptions_total":  int64(0),
				"kuiper_source_demo_0_records_in_total":  int64(5),
				"kuiper_source_demo_0_records_out_total": int64(5),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(5),
				"kuiper_op_window_0_records_out_total":  int64(3),

				"kuiper_op_filter_0_exceptions_total":   int64(0),
				"kuiper_op_filter_0_process_latency_ms": int64(0),
				"kuiper_op_filter_0_records_in_total":   int64(3),
				"kuiper_op_filter_0_records_out_total":  int64(2),
			},
		}, {
			name: `rule3`,
			sql:  `SELECT color, temp, ts FROM demo INNER JOIN demo1 ON demo.ts = demo1.ts GROUP BY SlidingWindow(ss, 1)`,
			size: 5,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
					"temp":  25.5,
					"ts":    float64(1541152486013),
				}}, {{
					"color": "red",
					"temp":  25.5,
					"ts":    float64(1541152486013),
				}}, {{
					"color": "red",
					"temp":  25.5,
					"ts":    float64(1541152486013),
				}}, {{
					"color": "blue",
					"temp":  28.1,
					"ts":    float64(1541152487632),
				}}, {{
					"color": "blue",
					"temp":  28.1,
					"ts":    float64(1541152487632),
				}}, {{
					"color": "blue",
					"temp":  28.1,
					"ts":    float64(1541152487632),
				}, {
					"color": "yellow",
					"temp":  27.4,
					"ts":    float64(1541152488442),
				}}, {{
					"color": "yellow",
					"temp":  27.4,
					"ts":    float64(1541152488442),
				}}, {{
					"color": "yellow",
					"temp":  27.4,
					"ts":    float64(1541152488442),
				}, {
					"color": "red",
					"temp":  25.5,
					"ts":    float64(1541152489252),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo_0_records_out_total":  int64(5),

				"kuiper_op_preprocessor_demo1_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo1_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo1_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo1_0_records_out_total":  int64(5),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(8),
				"kuiper_op_project_0_records_out_total":  int64(8),

				"kuiper_sink_mockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_mockSink_0_records_in_total":  int64(8),
				"kuiper_sink_mockSink_0_records_out_total": int64(8),

				"kuiper_source_demo_0_exceptions_total":  int64(0),
				"kuiper_source_demo_0_records_in_total":  int64(5),
				"kuiper_source_demo_0_records_out_total": int64(5),

				"kuiper_source_demo1_0_exceptions_total":  int64(0),
				"kuiper_source_demo1_0_records_in_total":  int64(5),
				"kuiper_source_demo1_0_records_out_total": int64(5),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(10),
				"kuiper_op_window_0_records_out_total":  int64(10),

				"kuiper_op_join_0_exceptions_total":   int64(0),
				"kuiper_op_join_0_process_latency_ms": int64(0),
				"kuiper_op_join_0_records_in_total":   int64(10),
				"kuiper_op_join_0_records_out_total":  int64(8),
			},
		}, {
			name: `rule4`,
			sql:  `SELECT color FROM demo GROUP BY SlidingWindow(ss, 2), color ORDER BY color`,
			size: 5,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
				}}, {{
					"color": "blue",
				}, {
					"color": "red",
				}}, {{
					"color": "blue",
				}, {
					"color": "red",
				}}, {{
					"color": "blue",
				}, {
					"color": "yellow",
				}}, {{
					"color": "blue",
				}, {
					"color": "red",
				}, {
					"color": "yellow",
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo_0_records_out_total":  int64(5),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(5),
				"kuiper_op_project_0_records_out_total":  int64(5),

				"kuiper_sink_mockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_mockSink_0_records_in_total":  int64(5),
				"kuiper_sink_mockSink_0_records_out_total": int64(5),

				"kuiper_source_demo_0_exceptions_total":  int64(0),
				"kuiper_source_demo_0_records_in_total":  int64(5),
				"kuiper_source_demo_0_records_out_total": int64(5),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(5),
				"kuiper_op_window_0_records_out_total":  int64(5),

				"kuiper_op_aggregate_0_exceptions_total":   int64(0),
				"kuiper_op_aggregate_0_process_latency_ms": int64(0),
				"kuiper_op_aggregate_0_records_in_total":   int64(5),
				"kuiper_op_aggregate_0_records_out_total":  int64(5),

				"kuiper_op_order_0_exceptions_total":   int64(0),
				"kuiper_op_order_0_process_latency_ms": int64(0),
				"kuiper_op_order_0_records_in_total":   int64(5),
				"kuiper_op_order_0_records_out_total":  int64(5),
			},
		}, {
			name: `rule5`,
			sql:  `SELECT temp FROM sessionDemo GROUP BY SessionWindow(ss, 2, 1) `,
			size: 11,
			r: [][]map[string]interface{}{
				{{
					"temp": 25.5,
				}, {
					"temp": 27.5,
				}}, {{
					"temp": 28.1,
				}, {
					"temp": 27.4,
				}, {
					"temp": 25.5,
				}}, {{
					"temp": 26.2,
				}, {
					"temp": 26.8,
				}, {
					"temp": 28.9,
				}, {
					"temp": 29.1,
				}, {
					"temp": 32.2,
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_sessionDemo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_sessionDemo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_sessionDemo_0_records_in_total":   int64(11),
				"kuiper_op_preprocessor_sessionDemo_0_records_out_total":  int64(11),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(3),
				"kuiper_op_project_0_records_out_total":  int64(3),

				"kuiper_sink_mockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_mockSink_0_records_in_total":  int64(3),
				"kuiper_sink_mockSink_0_records_out_total": int64(3),

				"kuiper_source_sessionDemo_0_exceptions_total":  int64(0),
				"kuiper_source_sessionDemo_0_records_in_total":  int64(11),
				"kuiper_source_sessionDemo_0_records_out_total": int64(11),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(11),
				"kuiper_op_window_0_records_out_total":  int64(3),
			},
		}, {
			name: `rule6`,
			sql:  `SELECT max(temp) as m, count(color) as c FROM demo INNER JOIN demo1 ON demo.ts = demo1.ts GROUP BY SlidingWindow(ss, 1)`,
			size: 5,
			r: [][]map[string]interface{}{
				{{
					"m": 25.5,
					"c": float64(1),
				}}, {{
					"m": 25.5,
					"c": float64(1),
				}}, {{
					"m": 25.5,
					"c": float64(1),
				}}, {{
					"m": 28.1,
					"c": float64(1),
				}}, {{
					"m": 28.1,
					"c": float64(1),
				}}, {{
					"m": 28.1,
					"c": float64(2),
				}}, {{
					"m": 27.4,
					"c": float64(1),
				}}, {{
					"m": 27.4,
					"c": float64(2),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demo_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo_0_records_out_total":  int64(5),

				"kuiper_op_preprocessor_demo1_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo1_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo1_0_records_in_total":   int64(5),
				"kuiper_op_preprocessor_demo1_0_records_out_total":  int64(5),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(8),
				"kuiper_op_project_0_records_out_total":  int64(8),

				"kuiper_sink_mockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_mockSink_0_records_in_total":  int64(8),
				"kuiper_sink_mockSink_0_records_out_total": int64(8),

				"kuiper_source_demo_0_exceptions_total":  int64(0),
				"kuiper_source_demo_0_records_in_total":  int64(5),
				"kuiper_source_demo_0_records_out_total": int64(5),

				"kuiper_source_demo1_0_exceptions_total":  int64(0),
				"kuiper_source_demo1_0_records_in_total":  int64(5),
				"kuiper_source_demo1_0_records_out_total": int64(5),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(10),
				"kuiper_op_window_0_records_out_total":  int64(10),

				"kuiper_op_join_0_exceptions_total":   int64(0),
				"kuiper_op_join_0_process_latency_ms": int64(0),
				"kuiper_op_join_0_records_in_total":   int64(10),
				"kuiper_op_join_0_records_out_total":  int64(8),
			},
		},
	}
	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
	createStreams(t)
	defer dropStreams(t)
	done := make(chan struct{})
	defer close(done)
	common.ResetMockTicker()
	for i, tt := range tests {
		p := NewRuleProcessor(DbDir)
		parser := xsql.NewParser(strings.NewReader(tt.sql))
		var sources []*nodes.SourceNode
		if stmt, err := xsql.Language.Parse(parser); err != nil {
			t.Errorf("parse sql %s error: %s", tt.sql, err)
		} else {
			if selectStmt, ok := stmt.(*xsql.SelectStatement); !ok {
				t.Errorf("sql %s is not a select statement", tt.sql)
			} else {
				streams := xsql.GetStreams(selectStmt)
				for _, stream := range streams {
					source := getMockSource(stream, done, tt.size)
					sources = append(sources, source)
				}
			}
		}
		tp, inputs, err := p.createTopoWithSources(&api.Rule{Id: tt.name, Sql: tt.sql}, sources)
		if err != nil {
			t.Error(err)
		}
		mockSink := test.NewMockSink()
		sink := nodes.NewSinkNodeWithSink("mockSink", mockSink)
		tp.AddSink(inputs, sink)
		count := len(sources)
		errCh := tp.Open()
		func() {
			for {
				select {
				case err = <-errCh:
					t.Log(err)
					tp.Cancel()
					return
				case <-done:
					count--
					log.Infof("%d sources remaining", count)
					if count <= 0 {
						log.Info("stream stopping")
						time.Sleep(1 * time.Second)
						tp.Cancel()
						return
					}
				default:
				}
			}
		}()
		results := mockSink.GetResults()
		var maps [][]map[string]interface{}
		for _, v := range results {
			var mapRes []map[string]interface{}
			err := json.Unmarshal(v, &mapRes)
			if err != nil {
				t.Errorf("Failed to parse the input into map")
				continue
			}
			maps = append(maps, mapRes)
		}
		if !reflect.DeepEqual(tt.r, maps) {
			t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.r, maps)
		}
		metrics := tp.GetMetrics()
		for k, v := range tt.m {
			if v != metrics[k] {
				t.Errorf("%d. %q\n\nmetrics mismatch for %s:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, k, v, metrics[k])
				break
			}
		}
	}
}

func createEventStreams(t *testing.T) {
	demo := `CREATE STREAM demoE (
					color STRING,
					size BIGINT,
					ts BIGINT
				) WITH (DATASOURCE="demoE", FORMAT="json", KEY="ts", TIMESTAMP="ts");`
	_, err := NewStreamProcessor(demo, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
	demo1 := `CREATE STREAM demo1E (
					temp FLOAT,
					hum BIGINT,
					ts BIGINT
				) WITH (DATASOURCE="demo1E", FORMAT="json", KEY="ts", TIMESTAMP="ts");`
	_, err = NewStreamProcessor(demo1, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
	sessionDemo := `CREATE STREAM sessionDemoE (
					temp FLOAT,
					hum BIGINT,
					ts BIGINT
				) WITH (DATASOURCE="sessionDemoE", FORMAT="json", KEY="ts", TIMESTAMP="ts");`
	_, err = NewStreamProcessor(sessionDemo, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
}

func dropEventStreams(t *testing.T) {
	demo := `DROP STREAM demoE`
	_, err := NewStreamProcessor(demo, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
	demo1 := `DROP STREAM demo1E`
	_, err = NewStreamProcessor(demo1, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
	sessionDemo := `DROP STREAM sessionDemoE`
	_, err = NewStreamProcessor(sessionDemo, path.Join(DbDir, "stream")).Exec()
	if err != nil {
		t.Log(err)
	}
}

func getEventMockSource(name string, done chan<- struct{}, size int) *nodes.SourceNode {
	var data []*xsql.Tuple
	switch name {
	case "demoE":
		data = []*xsql.Tuple{
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "red",
					"size":  3,
					"ts":    1541152486013,
				},
				Timestamp: 1541152486013,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "blue",
					"size":  2,
					"ts":    1541152487632,
				},
				Timestamp: 1541152487632,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "red",
					"size":  1,
					"ts":    1541152489252,
				},
				Timestamp: 1541152489252,
			},
			{ //dropped item
				Emitter: name,
				Message: map[string]interface{}{
					"color": "blue",
					"size":  6,
					"ts":    1541152486822,
				},
				Timestamp: 1541152486822,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"color": "yellow",
					"size":  4,
					"ts":    1541152488442,
				},
				Timestamp: 1541152488442,
			},
			{ //To lift the watermark and issue all windows
				Emitter: name,
				Message: map[string]interface{}{
					"color": "yellow",
					"size":  4,
					"ts":    1541152492342,
				},
				Timestamp: 1541152488442,
			},
		}
	case "demo1E":
		data = []*xsql.Tuple{
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 27.5,
					"hum":  59,
					"ts":   1541152486823,
				},
				Timestamp: 1541152486823,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  65,
					"ts":   1541152486013,
				},
				Timestamp: 1541152486013,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 27.4,
					"hum":  80,
					"ts":   1541152488442,
				},
				Timestamp: 1541152488442,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 28.1,
					"hum":  75,
					"ts":   1541152487632,
				},
				Timestamp: 1541152487632,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  62,
					"ts":   1541152489252,
				},
				Timestamp: 1541152489252,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  62,
					"ts":   1541152499252,
				},
				Timestamp: 1541152499252,
			},
		}
	case "sessionDemoE":
		data = []*xsql.Tuple{
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  65,
					"ts":   1541152486013,
				},
				Timestamp: 1541152486013,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 28.1,
					"hum":  75,
					"ts":   1541152487932,
				},
				Timestamp: 1541152487932,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 27.5,
					"hum":  59,
					"ts":   1541152486823,
				},
				Timestamp: 1541152486823,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 25.5,
					"hum":  62,
					"ts":   1541152489252,
				},
				Timestamp: 1541152489252,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 27.4,
					"hum":  80,
					"ts":   1541152488442,
				},
				Timestamp: 1541152488442,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 26.2,
					"hum":  63,
					"ts":   1541152490062,
				},
				Timestamp: 1541152490062,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 28.9,
					"hum":  85,
					"ts":   1541152491682,
				},
				Timestamp: 1541152491682,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 26.8,
					"hum":  71,
					"ts":   1541152490872,
				},
				Timestamp: 1541152490872,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 29.1,
					"hum":  92,
					"ts":   1541152492492,
				},
				Timestamp: 1541152492492,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 30.9,
					"hum":  87,
					"ts":   1541152494112,
				},
				Timestamp: 1541152494112,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 32.2,
					"hum":  99,
					"ts":   1541152493202,
				},
				Timestamp: 1541152493202,
			},
			{
				Emitter: name,
				Message: map[string]interface{}{
					"temp": 32.2,
					"hum":  99,
					"ts":   1541152499202,
				},
				Timestamp: 1541152499202,
			},
		}
	}
	return nodes.NewSourceNodeWithSource(name, test.NewMockSource(data[:size], done, true), map[string]string{
		"DATASOURCE": name,
	})
}

func TestEventWindow(t *testing.T) {
	common.IsTesting = true
	var tests = []struct {
		name string
		sql  string
		size int
		r    [][]map[string]interface{}
		m    map[string]interface{}
	}{
		{
			name: `rule1`,
			sql:  `SELECT * FROM demoE GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
			size: 6,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
					"size":  float64(3),
					"ts":    float64(1541152486013),
				}},
				{{
					"color": "red",
					"size":  float64(3),
					"ts":    float64(1541152486013),
				}, {
					"color": "blue",
					"size":  float64(2),
					"ts":    float64(1541152487632),
				}},
				{{
					"color": "blue",
					"size":  float64(2),
					"ts":    float64(1541152487632),
				}, {
					"color": "yellow",
					"size":  float64(4),
					"ts":    float64(1541152488442),
				}}, {{
					"color": "yellow",
					"size":  float64(4),
					"ts":    float64(1541152488442),
				}, {
					"color": "red",
					"size":  float64(1),
					"ts":    float64(1541152489252),
				}}, {{
					"color": "red",
					"size":  float64(1),
					"ts":    float64(1541152489252),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demoE_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demoE_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demoE_0_records_in_total":   int64(6),
				"kuiper_op_preprocessor_demoE_0_records_out_total":  int64(6),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(5),
				"kuiper_op_project_0_records_out_total":  int64(5),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(5),
				"kuiper_sink_MockSink_0_records_out_total": int64(5),

				"kuiper_source_demoE_0_exceptions_total":  int64(0),
				"kuiper_source_demoE_0_records_in_total":  int64(6),
				"kuiper_source_demoE_0_records_out_total": int64(6),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(6),
				"kuiper_op_window_0_records_out_total":  int64(5),
			},
		}, {
			name: `rule2`,
			sql:  `SELECT color, ts FROM demoE where size > 2 GROUP BY tumblingwindow(ss, 1)`,
			size: 6,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
					"ts":    float64(1541152486013),
				}},
				{{
					"color": "yellow",
					"ts":    float64(1541152488442),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demoE_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demoE_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demoE_0_records_in_total":   int64(6),
				"kuiper_op_preprocessor_demoE_0_records_out_total":  int64(6),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(2),
				"kuiper_op_project_0_records_out_total":  int64(2),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(2),
				"kuiper_sink_MockSink_0_records_out_total": int64(2),

				"kuiper_source_demoE_0_exceptions_total":  int64(0),
				"kuiper_source_demoE_0_records_in_total":  int64(6),
				"kuiper_source_demoE_0_records_out_total": int64(6),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(6),
				"kuiper_op_window_0_records_out_total":  int64(4),

				"kuiper_op_filter_0_exceptions_total":   int64(0),
				"kuiper_op_filter_0_process_latency_ms": int64(0),
				"kuiper_op_filter_0_records_in_total":   int64(4),
				"kuiper_op_filter_0_records_out_total":  int64(2),
			},
		}, {
			name: `rule3`,
			sql:  `SELECT color, temp, ts FROM demoE INNER JOIN demo1E ON demoE.ts = demo1E.ts GROUP BY SlidingWindow(ss, 1)`,
			size: 6,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
					"temp":  25.5,
					"ts":    float64(1541152486013),
				}}, {{
					"color": "red",
					"temp":  25.5,
					"ts":    float64(1541152486013),
				}}, {{
					"color": "blue",
					"temp":  28.1,
					"ts":    float64(1541152487632),
				}}, {{
					"color": "blue",
					"temp":  28.1,
					"ts":    float64(1541152487632),
				}, {
					"color": "yellow",
					"temp":  27.4,
					"ts":    float64(1541152488442),
				}}, {{
					"color": "yellow",
					"temp":  27.4,
					"ts":    float64(1541152488442),
				}, {
					"color": "red",
					"temp":  25.5,
					"ts":    float64(1541152489252),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demoE_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demoE_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demoE_0_records_in_total":   int64(6),
				"kuiper_op_preprocessor_demoE_0_records_out_total":  int64(6),

				"kuiper_op_preprocessor_demo1E_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo1E_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo1E_0_records_in_total":   int64(6),
				"kuiper_op_preprocessor_demo1E_0_records_out_total":  int64(6),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(5),
				"kuiper_op_project_0_records_out_total":  int64(5),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(5),
				"kuiper_sink_MockSink_0_records_out_total": int64(5),

				"kuiper_source_demoE_0_exceptions_total":  int64(0),
				"kuiper_source_demoE_0_records_in_total":  int64(6),
				"kuiper_source_demoE_0_records_out_total": int64(6),

				"kuiper_source_demo1E_0_exceptions_total":  int64(0),
				"kuiper_source_demo1E_0_records_in_total":  int64(6),
				"kuiper_source_demo1E_0_records_out_total": int64(6),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(12),
				"kuiper_op_window_0_records_out_total":  int64(5),

				"kuiper_op_join_0_exceptions_total":   int64(0),
				"kuiper_op_join_0_process_latency_ms": int64(0),
				"kuiper_op_join_0_records_in_total":   int64(5),
				"kuiper_op_join_0_records_out_total":  int64(5),
			},
		}, {
			name: `rule4`,
			sql:  `SELECT color FROM demoE GROUP BY SlidingWindow(ss, 2), color ORDER BY color`,
			size: 6,
			r: [][]map[string]interface{}{
				{{
					"color": "red",
				}}, {{
					"color": "blue",
				}, {
					"color": "red",
				}}, {{
					"color": "blue",
				}, {
					"color": "yellow",
				}}, {{
					"color": "blue",
				}, {
					"color": "red",
				}, {
					"color": "yellow",
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demoE_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demoE_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demoE_0_records_in_total":   int64(6),
				"kuiper_op_preprocessor_demoE_0_records_out_total":  int64(6),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(4),
				"kuiper_op_project_0_records_out_total":  int64(4),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(4),
				"kuiper_sink_MockSink_0_records_out_total": int64(4),

				"kuiper_source_demoE_0_exceptions_total":  int64(0),
				"kuiper_source_demoE_0_records_in_total":  int64(6),
				"kuiper_source_demoE_0_records_out_total": int64(6),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(6),
				"kuiper_op_window_0_records_out_total":  int64(4),

				"kuiper_op_aggregate_0_exceptions_total":   int64(0),
				"kuiper_op_aggregate_0_process_latency_ms": int64(0),
				"kuiper_op_aggregate_0_records_in_total":   int64(4),
				"kuiper_op_aggregate_0_records_out_total":  int64(4),

				"kuiper_op_order_0_exceptions_total":   int64(0),
				"kuiper_op_order_0_process_latency_ms": int64(0),
				"kuiper_op_order_0_records_in_total":   int64(4),
				"kuiper_op_order_0_records_out_total":  int64(4),
			},
		}, {
			name: `rule5`,
			sql:  `SELECT temp FROM sessionDemoE GROUP BY SessionWindow(ss, 2, 1) `,
			size: 12,
			r: [][]map[string]interface{}{
				{{
					"temp": 25.5,
				}}, {{
					"temp": 28.1,
				}, {
					"temp": 27.4,
				}, {
					"temp": 25.5,
				}}, {{
					"temp": 26.2,
				}, {
					"temp": 26.8,
				}, {
					"temp": 28.9,
				}, {
					"temp": 29.1,
				}, {
					"temp": 32.2,
				}}, {{
					"temp": 30.9,
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_sessionDemoE_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_sessionDemoE_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_sessionDemoE_0_records_in_total":   int64(12),
				"kuiper_op_preprocessor_sessionDemoE_0_records_out_total":  int64(12),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(4),
				"kuiper_op_project_0_records_out_total":  int64(4),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(4),
				"kuiper_sink_MockSink_0_records_out_total": int64(4),

				"kuiper_source_sessionDemoE_0_exceptions_total":  int64(0),
				"kuiper_source_sessionDemoE_0_records_in_total":  int64(12),
				"kuiper_source_sessionDemoE_0_records_out_total": int64(12),

				"kuiper_op_window_0_exceptions_total":   int64(0),
				"kuiper_op_window_0_process_latency_ms": int64(0),
				"kuiper_op_window_0_records_in_total":   int64(12),
				"kuiper_op_window_0_records_out_total":  int64(4),
			},
		}, {
			name: `rule6`,
			sql:  `SELECT max(temp) as m, count(color) as c FROM demoE INNER JOIN demo1E ON demoE.ts = demo1E.ts GROUP BY SlidingWindow(ss, 1)`,
			size: 6,
			r: [][]map[string]interface{}{
				{{
					"m": 25.5,
					"c": float64(1),
				}}, {{
					"m": 25.5,
					"c": float64(1),
				}}, {{
					"m": 28.1,
					"c": float64(1),
				}}, {{
					"m": 28.1,
					"c": float64(2),
				}}, {{
					"m": 27.4,
					"c": float64(2),
				}},
			},
			m: map[string]interface{}{
				"kuiper_op_preprocessor_demoE_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demoE_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demoE_0_records_in_total":   int64(6),
				"kuiper_op_preprocessor_demoE_0_records_out_total":  int64(6),

				"kuiper_op_preprocessor_demo1E_0_exceptions_total":   int64(0),
				"kuiper_op_preprocessor_demo1E_0_process_latency_ms": int64(0),
				"kuiper_op_preprocessor_demo1E_0_records_in_total":   int64(6),
				"kuiper_op_preprocessor_demo1E_0_records_out_total":  int64(6),

				"kuiper_op_project_0_exceptions_total":   int64(0),
				"kuiper_op_project_0_process_latency_ms": int64(0),
				"kuiper_op_project_0_records_in_total":   int64(5),
				"kuiper_op_project_0_records_out_total":  int64(5),

				"kuiper_sink_MockSink_0_exceptions_total":  int64(0),
				"kuiper_sink_MockSink_0_records_in_total":  int64(5),
				"kuiper_sink_MockSink_0_records_out_total": int64(5),

				"kuiper_source_demoE_0_exceptions_total":  int64(0),
				"kuiper_source_demoE_0_records_in_total":  int64(6),
				"kuiper_source_demoE_0_records_out_total": int64(6),

				"kuiper_source_demo1E_0_exceptions_total":  int64(0),
				"kuiper_source_demo1E_0_records_in_total":  int64(6),
				"kuiper_source_demo1E_0_records_out_total": int64(6),

				"kuiper_op_window_0_exceptions_total":  int64(0),
				"kuiper_op_window_0_records_in_total":  int64(12),
				"kuiper_op_window_0_records_out_total": int64(5),

				"kuiper_op_join_0_exceptions_total":   int64(0),
				"kuiper_op_join_0_process_latency_ms": int64(0),
				"kuiper_op_join_0_records_in_total":   int64(5),
				"kuiper_op_join_0_records_out_total":  int64(5),
			},
		},
	}
	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
	createEventStreams(t)
	defer dropEventStreams(t)
	done := make(chan struct{})
	defer close(done)
	common.ResetMockTicker()
	//mock ticker
	realTicker := time.NewTicker(500 * time.Millisecond)
	tickerDone := make(chan bool)
	go func() {
		ticker := common.GetTicker(1000).(*common.MockTicker)
		timer := common.GetTimer(1000).(*common.MockTimer)
		for {
			select {
			case <-tickerDone:
				log.Infof("real ticker exiting...")
				return
			case t := <-realTicker.C:
				ts := common.TimeToUnixMilli(t)
				if ticker != nil {
					go ticker.DoTick(ts)
				}
				if timer != nil {
					go timer.DoTick(ts)
				}
			}
		}

	}()
	for i, tt := range tests {
		p := NewRuleProcessor(DbDir)
		parser := xsql.NewParser(strings.NewReader(tt.sql))
		var sources []*nodes.SourceNode
		if stmt, err := xsql.Language.Parse(parser); err != nil {
			t.Errorf("parse sql %s error: %s", tt.sql, err)
		} else {
			if selectStmt, ok := stmt.(*xsql.SelectStatement); !ok {
				t.Errorf("sql %s is not a select statement", tt.sql)
			} else {
				streams := xsql.GetStreams(selectStmt)
				for _, stream := range streams {
					source := getEventMockSource(stream, done, tt.size)
					sources = append(sources, source)
				}
			}
		}
		tp, inputs, err := p.createTopoWithSources(&api.Rule{
			Id: tt.name, Sql: tt.sql,
			Options: map[string]interface{}{
				"isEventTime":   true,
				"lateTolerance": float64(1000),
			},
		}, sources)
		if err != nil {
			t.Error(err)
		}
		mockSink := test.NewMockSink()
		sink := nodes.NewSinkNodeWithSink("MockSink", mockSink)
		tp.AddSink(inputs, sink)
		count := len(sources)
		errCh := tp.Open()
		func() {
			for {
				select {
				case err = <-errCh:
					t.Log(err)
					tp.Cancel()
					return
				case <-done:
					count--
					log.Infof("%d sources remaining", count)
					if count <= 0 {
						log.Info("stream stopping")
						time.Sleep(1 * time.Second)
						tp.Cancel()
						return
					}
				default:
				}
			}
		}()
		results := mockSink.GetResults()
		var maps [][]map[string]interface{}
		for _, v := range results {
			var mapRes []map[string]interface{}
			err := json.Unmarshal(v, &mapRes)
			if err != nil {
				t.Errorf("Failed to parse the input into map")
				continue
			}
			maps = append(maps, mapRes)
		}
		if !reflect.DeepEqual(tt.r, maps) {
			t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.r, maps)
		}
		metrics := tp.GetMetrics()
		for k, v := range tt.m {
			if v != metrics[k] {
				t.Errorf("%d. %q\n\nmetrics mismatch for %s:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, k, v, metrics[k])
				break
			}
		}
	}
	realTicker.Stop()
	tickerDone <- true
	close(tickerDone)
}

func errstring(err error) string {
	if err != nil {
		return err.Error()
	}
	return ""
}