feat(sql): support multiple stream join table (#3866)

Signed-off-by: Song Gao <disxiaofei@163.com>

Author: Yisaer

fvt/rule_test.go

@@ -567,3 +567,86 @@ func (s *RuleTestSuite) assertSchemaEquality(expected, actual map[string]any) {
 	// 5. Compare the sorted slices to ensure the nested values are identical.
 	s.Equal(expectedValues, actualValues, "The set of nested schema values should be identical")
 }
+
+func (s *RuleTestSuite) TestJoinWithLookup() {
+	client.DeleteRule("ruleSim11")
+	client.DeleteStream("sim11")
+	client.DeleteStream("sim12")
+	client.DeleteTables("sim13")
+	topic := "test11"
+	subCh := pubsub.CreateSub(topic, nil, topic, 1024)
+	defer pubsub.CloseSourceConsumerChannel(topic, topic)
+	data1 := []map[string]any{
+		{
+			"k": "v1",
+		},
+	}
+	conf1 := map[string]any{
+		"data":     data1,
+		"interval": "1ms",
+		"loop":     false,
+	}
+	resp1, err := client.CreateConf("sources/simulator/confKeys/sim11", conf1)
+	s.Require().NoError(err)
+	s.Require().Equal(http.StatusOK, resp1.StatusCode)
+	streamSql1 := `{"sql": "create stream sim11() WITH (TYPE=\"simulator\", CONF_KEY=\"sim11\")"}`
+	resp1, err = client.CreateStream(streamSql1)
+	s.Require().NoError(err)
+	s.T().Log(GetResponseText(resp1))
+	s.Require().Equal(http.StatusCreated, resp1.StatusCode)
+	data2 := []map[string]any{
+		{
+			"k": "v1",
+		},
+	}
+	conf2 := map[string]any{
+		"data":     data2,
+		"interval": "1ms",
+		"loop":     false,
+	}
+	resp2, err := client.CreateConf("sources/simulator/confKeys/sim12", conf2)
+	s.Require().NoError(err)
+	s.Require().Equal(http.StatusOK, resp2.StatusCode)
+	streamSql2 := `{"sql": "create stream sim12() WITH (TYPE=\"simulator\", CONF_KEY=\"sim12\")"}`
+	resp2, err = client.CreateStream(streamSql2)
+	s.Require().NoError(err)
+	s.T().Log(GetResponseText(resp2))
+	s.Require().Equal(http.StatusCreated, resp2.StatusCode)
+
+	data3 := []map[string]any{
+		{
+			"k": "v1",
+		},
+	}
+	conf3 := map[string]any{
+		"data": data3,
+	}
+	resp3, err := client.CreateConf("sources/simulator/confKeys/sim13", conf3)
+	s.Require().NoError(err)
+	s.Require().Equal(http.StatusOK, resp3.StatusCode)
+	tableSql := `{"sql": "create table sim13() WITH (TYPE=\"simulator\", CONF_KEY=\"sim13\", KIND=\"lookup\")"}`
+	resp3, err = client.CreateStream(tableSql)
+	s.Require().NoError(err)
+	s.T().Log(GetResponseText(resp3))
+	s.Require().Equal(http.StatusCreated, resp3.StatusCode)
+
+	ruleSql := `{
+	   "id": "ruleSim11",
+	   "sql": "SELECT sim11.k as k1, sim12.k as k2, sim13.k as k3 from sim11 inner join sim12 on sim11.k = sim12.k inner join sim13 on sim13.k = sim11.k group by countwindow(2)",
+	   "actions": [
+	       {
+	           "memory": {
+	               "topic": "test11"
+	           }
+	       }
+	   ],
+	   "options": {
+	
+	   }
+	}`
+	resp, err := client.CreateRule(ruleSql)
+	s.Require().NoError(err)
+	s.T().Log(GetResponseText(resp))
+	s.Require().Equal(http.StatusCreated, resp.StatusCode)
+	s.assertRecvMemTuple(subCh, []map[string]any{{"k1": "v1", "k2": "v1", "k3": "v1"}})
+}

fvt/sdk.go

@@ -93,6 +93,10 @@ func (sdk *SDK) CreateStream(streamJson string) (resp *http.Response, err error)
 	return http.Post(sdk.baseUrl.JoinPath("streams").String(), ContentTypeJson, bytes.NewBufferString(streamJson))
 }
 
+func (sdk *SDK) CreateTable(tableJson string) (resp *http.Response, err error) {
+	return http.Post(sdk.baseUrl.JoinPath("tables").String(), ContentTypeJson, bytes.NewBufferString(tableJson))
+}
+
 func (sdk *SDK) DeleteStream(name string) (resp *http.Response, err error) {
 	req, err := http.NewRequest(http.MethodDelete, sdk.baseUrl.JoinPath("streams", name).String(), nil)
 	if err != nil {
@@ -102,6 +106,15 @@ func (sdk *SDK) DeleteStream(name string) (resp *http.Response, err error) {
 	return sdk.httpClient.Do(req)
 }
 
+func (sdk *SDK) DeleteTables(name string) (resp *http.Response, err error) {
+	req, err := http.NewRequest(http.MethodDelete, sdk.baseUrl.JoinPath("tables", name).String(), nil)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	return sdk.httpClient.Do(req)
+}
+
 func (sdk *SDK) GetStreamSchema(name string) (map[string]any, error) {
 	resp, err := http.Get(sdk.baseUrl.JoinPath("streams", name, "schema").String())
 	if err != nil {

internal/binder/io/builtin.go

@@ -56,6 +56,7 @@ func init() {
 
 	modules.RegisterLookupSource("memory", memory.GetLookupSource)
 	modules.RegisterLookupSource("httppull", http.GetLookUpSource)
+	modules.RegisterLookupSource("simulator", func() api.Source { return &simulator.SimulatorLookupSource{} })
 
 	modules.RegisterConnection("mqtt", mqtt.CreateConnection)
 	modules.RegisterConnection("nng", nng.CreateConnection)

internal/io/simulator/lookupSource.go

@@ -0,0 +1,65 @@
+// Copyright 2025 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 simulator
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type SimulatorLookupSource struct {
+	cfg *sLookupConfig
+}
+
+func (s *SimulatorLookupSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	cfg := &sLookupConfig{
+		Data: make([]map[string]any, 0),
+	}
+	if err := cast.MapToStruct(configs, cfg); err != nil {
+		return err
+	}
+	s.cfg = cfg
+	return nil
+}
+
+func (s *SimulatorLookupSource) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (s *SimulatorLookupSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+	return nil
+}
+
+func (s *SimulatorLookupSource) Lookup(ctx api.StreamContext, lookupFields []string, cmpKeys []string, cmpValues []any) ([]map[string]any, error) {
+	res := make([]map[string]any, 0)
+	for _, d := range s.cfg.Data {
+		for index, key := range cmpKeys {
+			value, ok := d[key]
+			if ok && value == cmpValues[index] {
+				selectedRow := make(map[string]any)
+				for _, field := range lookupFields {
+					selectedRow[field] = d[field]
+				}
+				res = append(res, selectedRow)
+			}
+		}
+	}
+	return res, nil
+}
+
+type sLookupConfig struct {
+	Data []map[string]any `json:"data"`
+}

internal/io/simulator/lookupSource_test.go

@@ -0,0 +1,201 @@
+// Copyright 2025 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 simulator
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestSimulatorLookupSource_Lookup(t *testing.T) {
+	tests := []struct {
+		name         string
+		data         []map[string]any
+		lookupFields []string
+		cmpKeys      []string
+		cmpValues    []any
+		expected     []map[string]any
+	}{
+		{
+			name: "single match",
+			data: []map[string]any{
+				{"id": 1, "name": "Alice", "age": 25},
+				{"id": 2, "name": "Bob", "age": 30},
+				{"id": 3, "name": "Charlie", "age": 35},
+			},
+			lookupFields: []string{"id", "name"},
+			cmpKeys:      []string{"id"},
+			cmpValues:    []any{2},
+			expected: []map[string]any{
+				{"id": 2, "name": "Bob"},
+			},
+		},
+		{
+			name: "multiple matches",
+			data: []map[string]any{
+				{"id": 1, "name": "Alice", "age": 25},
+				{"id": 2, "name": "Bob", "age": 30},
+				{"id": 1, "name": "Alice", "age": 28},
+				{"id": 3, "name": "Charlie", "age": 35},
+			},
+			lookupFields: []string{"id", "name"},
+			cmpKeys:      []string{"id"},
+			cmpValues:    []any{1},
+			expected: []map[string]any{
+				{"id": 1, "name": "Alice"},
+				{"id": 1, "name": "Alice"},
+			},
+		},
+		{
+			name: "no match",
+			data: []map[string]any{
+				{"id": 1, "name": "Alice", "age": 25},
+				{"id": 2, "name": "Bob", "age": 30},
+				{"id": 3, "name": "Charlie", "age": 35},
+			},
+			lookupFields: []string{"id", "name"},
+			cmpKeys:      []string{"id"},
+			cmpValues:    []any{99},
+			expected:     []map[string]any{},
+		},
+		{
+			name: "string value match",
+			data: []map[string]any{
+				{"id": 1, "name": "Alice", "age": 25},
+				{"id": 2, "name": "Bob", "age": 30},
+				{"id": 3, "name": "Charlie", "age": 35},
+			},
+			lookupFields: []string{"id", "name"},
+			cmpKeys:      []string{"name"},
+			cmpValues:    []any{"Bob"},
+			expected: []map[string]any{
+				{"id": 2, "name": "Bob"},
+			},
+		},
+		{
+			name: "multiple lookup fields",
+			data: []map[string]any{
+				{"id": 1, "name": "Alice", "age": 25, "city": "New York"},
+				{"id": 2, "name": "Bob", "age": 30, "city": "London"},
+				{"id": 3, "name": "Charlie", "age": 35, "city": "Paris"},
+			},
+			lookupFields: []string{"id", "name", "age"},
+			cmpKeys:      []string{"name"},
+			cmpValues:    []any{"Bob"},
+			expected: []map[string]any{
+				{"id": 2, "name": "Bob", "age": 30},
+			},
+		},
+		{
+			name:         "empty data",
+			data:         []map[string]any{},
+			lookupFields: []string{"id", "name"},
+			cmpKeys:      []string{"id"},
+			cmpValues:    []any{1},
+			expected:     []map[string]any{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create SimulatorLookupSource instance
+			source := &SimulatorLookupSource{}
+
+			// Create mock context
+			ctx := mockContext.NewMockContext("test", "Test")
+
+			// Provision the source with test data
+			configs := map[string]any{
+				"data": tt.data,
+			}
+			err := source.Provision(ctx, configs)
+			require.NoError(t, err)
+
+			// Connect the source
+			err = source.Connect(ctx, func(status string, message string) {
+				// do nothing
+			})
+			require.NoError(t, err)
+
+			// Perform lookup
+			result, err := source.Lookup(ctx, tt.lookupFields, tt.cmpKeys, tt.cmpValues)
+
+			// Verify results
+			assert.NoError(t, err)
+			assert.Equal(t, tt.expected, result)
+
+			// Close the source
+			err = source.Close(ctx)
+			require.NoError(t, err)
+		})
+	}
+}
+
+func TestSimulatorLookupSource_Provision(t *testing.T) {
+	tests := []struct {
+		name    string
+		configs map[string]any
+		wantErr bool
+		errMsg  string
+	}{
+		{
+			name: "valid config",
+			configs: map[string]any{
+				"data": []map[string]any{
+					{"id": 1, "name": "Alice"},
+					{"id": 2, "name": "Bob"},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "empty data config",
+			configs: map[string]any{
+				"data": []map[string]any{},
+			},
+			wantErr: false,
+		},
+		{
+			name: "nil data config",
+			configs: map[string]any{
+				"data": nil,
+			},
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			source := &SimulatorLookupSource{}
+			ctx := mockContext.NewMockContext("test", "Test")
+
+			err := source.Provision(ctx, tt.configs)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				if tt.errMsg != "" {
+					assert.Contains(t, err.Error(), tt.errMsg)
+				}
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, source.cfg)
+			}
+		})
+	}
+}