ekuiper-new/internal/xsql/row_test.go

// Copyright 2022-2023 EMQ Technologies Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package xsql

import (
	"fmt"
	"reflect"
	"strings"
	"sync"
	"testing"

	"github.com/lf-edge/ekuiper/internal/conf"
)

// Row valuer, wildcarder test
// WindowTuples, JoinTuples, GroupTuples are collectionRow
func TestCollectionRow(t *testing.T) {
	tests := []struct {
		rowC     CollectionRow
		value    []string
		wildcard []string
		result   []interface{} // result of valuers and wildcards
	}{
		{
			rowC:     &Tuple{Emitter: "a", Message: map[string]interface{}{"a": 1, "b": "2"}, Timestamp: conf.GetNowInMilli(), Metadata: nil},
			value:    []string{"a", "b"},
			wildcard: []string{""},
			result:   []interface{}{1, "2", map[string]interface{}{"a": 1, "b": "2"}},
		}, {
			rowC:     &Tuple{Emitter: "a", Message: map[string]interface{}{"a": 1, "b": "2"}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"a": 4, "c": 3}, AliasMap: map[string]interface{}{"b": "b1"}}},
			value:    []string{"a", "b", "c"},
			wildcard: []string{""},
			result:   []interface{}{4, "b1", 3, map[string]interface{}{"a": 1, "b": "2"}},
		}, {
			rowC: &JoinTuple{Tuples: []TupleRow{
				&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}},
				&Tuple{Emitter: "src2", Message: Message{"a": 2, "c": "w2"}},
			}},
			value:    []string{"a", "src2.a", "b", "c"},
			wildcard: []string{"", "src1"},
			result:   []interface{}{1, 2, "v1", "w2", map[string]interface{}{"a": 2, "b": "v1", "c": "w2"}, map[string]interface{}{"a": 1, "b": "v1"}},
		}, {
			rowC: &JoinTuple{Tuples: []TupleRow{
				&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}},
				&Tuple{Emitter: "src2", Message: Message{"a": 2, "c": "w2"}},
			}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"a": 4, "d": 3}, AliasMap: map[string]interface{}{"d": 4}}},
			value:    []string{"a", "src2.a", "b", "c", "d"},
			wildcard: []string{"", "src1"},
			result:   []interface{}{4, 2, "v1", "w2", 4, map[string]interface{}{"a": 2, "b": "v1", "c": "w2"}, map[string]interface{}{"a": 1, "b": "v1"}},
		}, {
			rowC:     &GroupedTuples{Content: []TupleRow{&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}}, &Tuple{Emitter: "src1", Message: Message{"a": 2, "b": "v2"}}}},
			value:    []string{"a", "b"},
			wildcard: []string{""},
			result:   []interface{}{1, "v1", map[string]interface{}{"a": 1, "b": "v1"}},
		}, {
			rowC:     &GroupedTuples{Content: []TupleRow{&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}}, &Tuple{Emitter: "src1", Message: Message{"a": 2, "b": "v2"}}}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"a": 4, "d": 3}, AliasMap: map[string]interface{}{"d": 4}}},
			value:    []string{"a", "b", "d"},
			wildcard: []string{""},
			result:   []interface{}{4, "v1", 4, map[string]interface{}{"a": 1, "b": "v1"}},
		}, {
			rowC:     &WindowTuples{Content: []TupleRow{&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}}, &Tuple{Emitter: "src1", Message: Message{"a": 2, "b": "v2"}}}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"a": 4, "d": 3}, AliasMap: map[string]interface{}{"d": 4}}},
			value:    []string{"a", "b", "d"},
			wildcard: []string{""},
			result:   []interface{}{4, "v1", 4, map[string]interface{}{"a": 1, "b": "v1"}},
		}, {
			rowC: &JoinTuples{Content: []*JoinTuple{{Tuples: []TupleRow{
				&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"b": "v2", "$$lag_a": 1}}},
				&Tuple{Emitter: "src2", Message: Message{"a": 2, "c": "w2"}},
			}}}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"a": 4, "d": 3}, AliasMap: map[string]interface{}{"d": 4}}},
			value:    []string{"a", "b", "d"},
			wildcard: []string{""},
			result:   []interface{}{4, "v2", 4, map[string]interface{}{"a": 2, "b": "v1", "c": "w2"}},
		},
	}
	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
	var ok bool
	for i, tt := range tests {
		result := make([]interface{}, len(tt.value)+len(tt.wildcard))
		for j, v := range tt.value {
			var key, table string
			strs := strings.Split(v, ".")
			if len(strs) > 1 {
				key = strs[1]
				table = strs[0]
			} else {
				key = strs[0]
				table = ""
			}
			result[j], ok = tt.rowC.Value(key, table)
			if !ok {
				t.Errorf("%d.%d.%d: %s", i, j, 0, "Value() failed.")
				continue
			}
		}
		for j, v := range tt.wildcard {
			result[len(tt.value)+j], ok = tt.rowC.All(v)
			if !ok {
				t.Errorf("%d.%d.%d: %s", i, j, 1, "Wildcard() failed.")
				continue
			}
		}
		if !reflect.DeepEqual(tt.result, result) {
			t.Errorf("%d result mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.result, result)
		}
	}
}

func TestTupleRow(t *testing.T) {
	// broadcast(clone) -> set -> broadcast -> set -> compare
	tests := []struct {
		rowO TupleRow
		// The multiple values to set or alias; The first value is set in the first broadcast. the next values are set in the second broadcast.
		set    [][]map[string]interface{}
		result [][]map[string]interface{}
	}{
		{
			rowO: &Tuple{Emitter: "a", Message: map[string]interface{}{"a": 1, "b": "2"}, Timestamp: conf.GetNowInMilli(), Metadata: nil},
			set: [][]map[string]interface{}{
				{
					{"a": 2, "c": "3", "@d": 4},
					{"a": 3},
					{"a": 4, "b": "5"},
					{"@d": 5, "e": 5},
				},
				{
					{"c": "4"},
					{"d": "4"},
					{"a": 5, "b": "6"},
					{"a": 6, "@b": "7"},
				},
			},
			result: [][]map[string]interface{}{
				{
					{"a": 3, "b": "2", "c": "3", "d": 4},
					{"a": 4, "b": "5", "c": "3", "d": 4},
					{"a": 2, "b": "2", "c": "3", "d": 5, "e": 5},
				}, {
					{"a": 1, "b": "2", "c": "4", "d": "4"},
					{"a": 5, "b": "6", "c": "4"},
					{"a": 6, "b": "7", "c": "4"},
				},
			},
		}, {
			rowO: &JoinTuple{Tuples: []TupleRow{
				&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}},
				&Tuple{Emitter: "src2", Message: Message{"a": 2, "c": "w2"}},
			}},
			set: [][]map[string]interface{}{
				{
					{"a": 2, "c": "3", "@d": 4},
					{"a": 3},
					{"a": 4, "b": "5"},
					{"@d": 5, "e": 5},
				},
				{
					{"e": "4"},
					{"d": "4"},
					{"a": 5, "b": "6"},
					{"a": 6, "@b": "7"},
				},
			},
			result: [][]map[string]interface{}{
				{
					{"a": 3, "b": "v1", "c": "3", "d": 4},
					{"a": 4, "b": "5", "c": "3", "d": 4},
					{"a": 2, "b": "v1", "c": "3", "d": 5, "e": 5},
				}, {
					{"a": 1, "b": "v1", "c": "w2", "d": "4", "e": "4"},
					{"a": 5, "b": "6", "c": "w2", "e": "4"},
					{"a": 6, "b": "7", "c": "w2", "e": "4"},
				},
			},
		},
	}
	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
	for i, tt := range tests {
		var (
			wg     sync.WaitGroup
			result = make([][]map[string]interface{}, len(tt.set))
		)
		for si, set := range tt.set {
			wg.Add(1)
			go func(si int, set []map[string]interface{}) {
				nr := tt.rowO.Clone()
				for k, v := range set[0] {
					if strings.HasPrefix(k, "@") {
						nr.AppendAlias(k[1:], v)
					} else {
						nr.Set(k, v)
					}
				}
				var wg2 sync.WaitGroup
				result[si] = make([]map[string]interface{}, len(set)-1)
				for j := 1; j < len(set); j++ {
					wg2.Add(1)
					go func(j int) {
						nnr := nr.Clone()
						for k, v := range set[j] {
							if strings.HasPrefix(k, "@") {
								nnr.AppendAlias(k[1:], v)
							} else {
								nnr.Set(k, v)
							}
						}
						result[si][j-1] = nnr.ToMap()
						wg2.Done()
					}(j)
				}
				wg2.Wait()
				wg.Done()
			}(si, set)
		}
		wg.Wait()
		if !reflect.DeepEqual(tt.result, result) {
			t.Errorf("%d result mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.result, result)
		}
	}
}