Repository: lf-edge/ekuiper
Pull request #3000 (2024-07-10 03:41:09)

feat: support rule trial

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `68.08511%` with `45 lines` in your changes missing coverage. Please review.
> Project coverage is 68.33%. Comparing base [(`f2fdc02`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/f2fdc02215f9b49b9ae428e58b77376b88ef0667?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`58bfd83`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/58bfd837d60966d39d7063d80551f970fb593ffe?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/trial/run.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?src=pr&el=tree&filepath=internal%2Ftrial%2Frun.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdHJpYWwvcnVuLmdv) | 66.13% | [16 Missing and 5 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rest.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?src=pr&el=tree&filepath=internal%2Fserver%2Frest.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3Jlc3QuZ28=) | 65.71% | [9 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/trial/manager.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?src=pr&el=tree&filepath=internal%2Ftrial%2Fmanager.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdHJpYWwvbWFuYWdlci5nbw==) | 73.53% | [5 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/node/sink\_node.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fsink_node.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL3Npbmtfbm9kZS5nbw==) | 25.00% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@              Coverage Diff              @@
##           master-v2    #3000      +/-   ##
=============================================
+ Coverage      68.28%   68.33%   +0.05%     
=============================================
  Files            344      346       +2     
  Lines          33079    33214     +135     
=============================================
+ Hits           22586    22694     +108     
- Misses          8476     8496      +20     
- Partials        2017     2024       +7     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3000?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/server/rest.go b/internal/server/rest.go
index c6e5341a..435dcae4 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -38,6 +38,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/server/middleware"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
+	"github.com/lf-edge/ekuiper/v2/internal/trial"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
@@ -179,9 +180,10 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/connections/status", connectionsStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
-
+	r.HandleFunc("/ruletest", testRuleHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruletest/{name}/start", testRuleStartHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruletest/{name}", testRuleStopHandler).Methods(http.MethodDelete)
 	r.HandleFunc("/v2/data/export", yamlConfigurationExportHandler).Methods(http.MethodGet)
-
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	// Register extended routes
 	for k, v := range components {
@@ -856,3 +858,45 @@ func exportHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Add("Content-Disposition", "Attachment")
 	http.ServeContent(w, r, name, time.Now(), exported)
 }
+
+func testRuleHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		handleError(w, err, "Invalid body", logger)
+		return
+	}
+	id, err := trial.TrialManager.CreateRule(string(body))
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+	result := map[string]any{
+		"id":   id,
+		"port": conf.Config.Source.HttpServerPort,
+	}
+	w.WriteHeader(http.StatusOK)
+	jsonResponse(result, w, logger)
+}
+
+func testRuleStartHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	id := vars["name"]
+	err := trial.TrialManager.StartRule(id)
+	if err != nil {
+		handleError(w, err, "start rule error", logger)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	fmt.Fprintf(w, "Test rule %s was started", id)
+}
+
+func testRuleStopHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	id := vars["name"]
+	trial.TrialManager.StopRule(id)
+	w.WriteHeader(http.StatusOK)
+	fmt.Fprintf(w, "Test rule %s was stopped", id)
+}
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
index c41373e7..5945804e 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -66,9 +66,9 @@ func newSinkNode(ctx api.StreamContext, name string, rOpt def.RuleOption, eoflim
 
 func (s *SinkNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	s.prepareExec(ctx, errCh, "sink")
+	err := s.sink.Connect(ctx)
 	go func() {
 		err := infra.SafeRun(func() error {
-			err := s.sink.Connect(ctx)
 			if err != nil {
 				infra.DrainError(ctx, err, errCh)
 			}
diff --git a/internal/topo/planner/logicalPlan.go b/internal/topo/planner/logicalPlan.go
index 5396b6a4..45f6b593 100644
--- a/internal/topo/planner/logicalPlan.go
+++ b/internal/topo/planner/logicalPlan.go
@@ -58,7 +58,7 @@ type PlanExplainInfo struct {
 	T        PlanType `json:"type"`
 	Info     string   `json:"info"`
 	ID       int64    `json:"id"`
-	Children []int64  `json:"children"`
+	Children []int64  `json:"children,omitempty"`
 }
 
 func (p *baseLogicalPlan) Explain() string {
diff --git a/internal/topo/planner/optimizer.go b/internal/topo/planner/optimizer.go
index ba773aef..8e9ad39f 100644
--- a/internal/topo/planner/optimizer.go
+++ b/internal/topo/planner/optimizer.go
@@ -17,6 +17,7 @@ package planner
 var optRuleList = []logicalOptRule{
 	&columnPruner{},
 	&predicatePushDown{},
+	&pushProjectionPlan{},
 }
 
 func optimize(p LogicalPlan) (LogicalPlan, error) {
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index 6863c8e2..2db84571 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -17,6 +17,7 @@ package planner
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
@@ -147,6 +148,10 @@ func GetExplainInfoFromLogicalPlan(rule *def.Rule) (string, error) {
 	if err != nil {
 		return "", err
 	}
+	return ExplainFromLogicalPlan(lp, rule.Id)
+}
+
+func ExplainFromLogicalPlan(lp LogicalPlan, ruleID string) (string, error) {
 	var setId func(p LogicalPlan, id int64)
 	setId = func(p LogicalPlan, id int64) {
 		p.SetID(id)
@@ -162,15 +167,16 @@ func GetExplainInfoFromLogicalPlan(rule *def.Rule) (string, error) {
 		tmp := ""
 		res := ""
 		for i := 0; i < level; i++ {
-			tmp += "   "
+			tmp += "\t"
 		}
 		p.BuildExplainInfo()
 		if info, ok := p.(RuleRuntimeInfo); ok {
-			info.BuildSchemaInfo(rule.Id)
+			info.BuildSchemaInfo(ruleID)
 		}
 		// Build the explainInfo of the current layer
-		res += tmp + p.Explain() + "\n"
+		res += tmp + strings.TrimSuffix(p.Explain(), "\n")
 		if len(p.Children()) != 0 {
+			res += "\n"
 			for _, v := range p.Children() {
 				res += tmp + getExplainInfo(v, level+1)
 			}
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 92402488..497a7941 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -32,7 +32,7 @@ import (
 
 func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesProp map[string]map[string]any, ruleId string, options *def.RuleOption, index int) (node.DataSourceNode, []node.OperatorNode, int, error) {
 	isSchemaless := t.isSchemaless
-	_, isMock := mockSourcesProp[string(t.name)]
+	mockProps, isMock := mockSourcesProp[string(t.name)]
 	if isMock {
 		t.streamStmt.Options.TYPE = "simulator"
 	}
@@ -60,13 +60,18 @@ func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesPr
 			return nil, nil, 0, err
 		}
 	}
-	return splitSource(ctx, t, si, options, index, ruleId, pp)
+	return splitSource(ctx, t, si, options, mockProps, index, ruleId, pp)
 }
 
-func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, options *def.RuleOption, index int, ruleId string, pp node.UnOperation) (node.DataSourceNode, []node.OperatorNode, int, error) {
+func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, options *def.RuleOption, mockProps map[string]any, index int, ruleId string, pp node.UnOperation) (node.DataSourceNode, []node.OperatorNode, int, error) {
 	// Get all props
 	props := nodeConf.GetSourceConf(t.streamStmt.Options.TYPE, t.streamStmt.Options)
 	sp := &SourcePropsForSplit{}
+	if len(mockProps) > 0 {
+		for k, v := range mockProps {
+			props[k] = v
+		}
+	}
 	_ = cast.MapToStruct(props, sp)
 	// Create the connector node as source node
 	var (
diff --git a/internal/topo/planner/push_proj.go b/internal/topo/planner/push_proj.go
new file mode 100644
index 00000000..8f06617b
--- /dev/null
+++ b/internal/topo/planner/push_proj.go
@@ -0,0 +1,110 @@
+// Copyright 2024 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 planner
+
+import (
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+type pushProjectionPlan struct{}
+
+// pushProjectionPlan inject Projection Plan between the shared Datasource and its father only if the Plan have windowPlan
+// We use Projection to remove the unused column before windowPlan in order to reduce memory consuming
+func (pp *pushProjectionPlan) optimize(plan LogicalPlan) (LogicalPlan, error) {
+	if pp.searchWindowPlan(plan) {
+		ctx := &searchCtx{
+			find: make([]*sharedSource, 0),
+		}
+		pp.searchSharedDataSource(ctx, plan, nil)
+		if len(ctx.find) > 0 {
+			pp.pushProjection(ctx)
+		}
+	}
+	return plan, nil
+}
+
+func (pp *pushProjectionPlan) searchWindowPlan(plan LogicalPlan) bool {
+	switch plan.(type) {
+	case *WindowPlan:
+		return true
+	default:
+	}
+	for _, child := range plan.Children() {
+		search := pp.searchWindowPlan(child)
+		if search {
+			return true
+		}
+	}
+	return false
+}
+
+type searchCtx struct {
+	find []*sharedSource
+}
+
+type sharedSource struct {
+	ds     *DataSourcePlan
+	father LogicalPlan
+}
+
+func (pp *pushProjectionPlan) searchSharedDataSource(ctx *searchCtx, plan, father LogicalPlan) {
+	switch ds := plan.(type) {
+	case *DataSourcePlan:
+		if ds.streamStmt.Options.SHARED {
+			ctx.find = append(ctx.find, &sharedSource{
+				ds:     ds,
+				father: father,
+			})
+		}
+	default:
+	}
+	for _, child := range plan.Children() {
+		pp.searchSharedDataSource(ctx, child, plan)
+	}
+}
+
+func (pp *pushProjectionPlan) pushProjection(ctx *searchCtx) {
+	for _, search := range ctx.find {
+		p := ProjectPlan{
+			fields:      buildFields(search.ds),
+			isAggregate: false,
+			sendMeta:    false,
+			enableLimit: false,
+		}.Init()
+		p.children = []LogicalPlan{search.ds}
+		for index, child := range search.father.Children() {
+			if child.ID() == search.ds.ID() {
+				search.father.Children()[index] = p
+				break
+			}
+		}
+	}
+}
+
+func buildFields(ds *DataSourcePlan) []ast.Field {
+	want := make([]ast.Field, 0)
+	if ds.isWildCard {
+		want = append(want, ast.Field{Expr: &ast.Wildcard{}})
+		return want
+	}
+	for k := range ds.streamFields {
+		want = append(want, ast.Field{Name: k, Expr: &ast.FieldRef{Name: k, StreamName: ds.streamStmt.Name}})
+	}
+	return want
+}
+
+func (pp *pushProjectionPlan) name() string {
+	return "push_projection"
+}
diff --git a/internal/trial/manager.go b/internal/trial/manager.go
new file mode 100644
index 00000000..33d363a3
--- /dev/null
+++ b/internal/trial/manager.go
@@ -0,0 +1,89 @@
+// Copyright 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 trial
+
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/topo"
+)
+
+// TrialManager Manager Initialized in the binder
+var TrialManager = &Manager{
+	runs: make(map[string]Run),
+}
+
+// Manager In memory manager for all trial rules
+type Manager struct {
+	sync.RWMutex
+	// ruleId -> *Topo
+	runs map[string]Run
+}
+
+type Run struct {
+	def  *RunDef
+	topo *topo.Topo
+}
+
+func (m *Manager) CreateRule(ruleDef string) (string, error) {
+	def := &RunDef{}
+	err := json.Unmarshal([]byte(ruleDef), def)
+	if err != nil {
+		return "", fmt.Errorf("fail to parse rule definition %s: %s", ruleDef, err)
+	}
+	m.Lock()
+	defer m.Unlock()
+	// If the rule exists, stop it first
+	if r, ok := m.runs[def.Id]; ok {
+		r.topo.Cancel()
+		conf.Log.Warnf("stop last run of test rule %s", def.Id)
+	}
+	t, err := create(def)
+	if err != nil {
+		return "", err
+	}
+	m.runs[def.Id] = Run{
+		def:  def,
+		topo: t,
+	}
+	return def.Id, nil
+}
+
+func (m *Manager) StopRule(ruleId string) {
+	m.Lock()
+	defer m.Unlock()
+	if r, ok := m.runs[ruleId]; ok {
+		r.topo.Cancel()
+		delete(m.runs, ruleId)
+		httpserver.UnRegisterWebSocketEndpoint(r.def.endpoint)
+	} else {
+		conf.Log.Warnf("try to stop test rule %s but it is not found", ruleId)
+	}
+}
+
+func (m *Manager) StartRule(ruleId string) error {
+	m.RLock()
+	defer m.RUnlock()
+	if r, ok := m.runs[ruleId]; ok {
+		trialRun(r.topo)
+	} else {
+		return fmt.Errorf("try to start test rule %s but it is not found", ruleId)
+	}
+	return nil
+}
diff --git a/internal/trial/run.go b/internal/trial/run.go
new file mode 100644
index 00000000..066e9864
--- /dev/null
+++ b/internal/trial/run.go
@@ -0,0 +1,133 @@
+// Copyright 2023-2024 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 trial
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/topo"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+)
+
+type RunDef struct {
+	Id        string                    `json:"id"`
+	Sql       string                    `json:"sql"`
+	Mock      map[string]map[string]any `json:"mockSource"`
+	SinkProps map[string]any            `json:"sinkProps"`
+
+	endpoint string
+}
+
+func genTrialRuleID(def *RunDef) string {
+	return "$$_" + uuid.New().String() + def.Id
+}
+
+func genTrialRule(rd *RunDef, sinkProps map[string]interface{}) *def.Rule {
+	id := genTrialRuleID(rd)
+	rt := def.GetDefaultRule(id, rd.Sql)
+	rt.Actions = []map[string]interface{}{
+		{
+			"websocket": sinkProps,
+		},
+	}
+	return rt
+}
+
+func create(def *RunDef) (*topo.Topo, error) {
+	endpoint := "/test/" + def.Id
+	def.endpoint = endpoint
+	_, _, err := httpserver.RegisterWebSocketEndpoint(context.Background(), endpoint)
+	if err != nil {
+		return nil, err
+	}
+	sinkProps := map[string]any{
+		"path":      endpoint,
+		"sendError": true,
+	}
+	for k, v := range def.SinkProps {
+		sinkProps[k] = v
+	}
+	trialRule := genTrialRule(def, sinkProps)
+	// Add trial run prefix for rule id to avoid duplicate rule id with real rules in runtime or other trial rule
+	tp, err := planner.PlanSQLWithSourcesAndSinks(trialRule, prepareSourceProps(def.Mock))
+	if err != nil {
+		return nil, fmt.Errorf("fail to run rule %s: %s", def.Id, err)
+	}
+	// Try run
+	// TODO currently some static validations are done in runtime, so start to run to detect them. This adds time penalty for this API.
+	// 	In the future, we should do it in planning.
+	err = infra.SafeRun(func() error {
+		select {
+		case e := <-tp.Open():
+			if e != nil {
+				return e
+			}
+		case <-time.After(10 * time.Millisecond):
+			return nil
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("fail to run rule %s: %s", def.Id, err)
+	}
+	tp.Cancel()
+	return tp, nil
+}
+
+func trialRun(tp *topo.Topo) {
+	go func() {
+		timeout := time.After(5 * time.Minute)
+		err := infra.SafeRun(func() error {
+			select {
+			case err := <-tp.Open():
+				if err != nil {
+					conf.Log.Errorf("closing test run for error: %v", err)
+					tp.Cancel()
+					return err
+				}
+			case <-timeout:
+				tp.GetContext().GetLogger().Debugf("trial run stops after timeout")
+				tp.Cancel()
+			}
+			return nil
+		})
+		if err != nil {
+			conf.Log.Debugf("trial run error: %v", err)
+		}
+	}()
+}
+
+func prepareSourceProps(mockSource map[string]map[string]any) map[string]map[string]any {
+	for source, props := range mockSource {
+		v, ok := props["interval"]
+		if ok {
+			vi, ok := v.(int)
+			if ok {
+				props["interval"] = fmt.Sprintf("%dms", vi)
+				mockSource[source] = props
+				continue
+			}
+		}
+	}
+	return mockSource
+}


Test patch:
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 8e21e258..8867636c 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -96,6 +96,9 @@ func (suite *RestTestSuite) SetupTest() {
 	r.HandleFunc("/data/import/status", configurationStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
+	r.HandleFunc("/ruletest", testRuleHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruletest/{name}/start", testRuleStartHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruletest/{name}", testRuleStopHandler).Methods(http.MethodDelete)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	r.HandleFunc("/metadata/sinks/{name}/confKeys/{confKey}", sinkConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
 	suite.r = r
@@ -340,8 +343,10 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	suite.r.ServeHTTP(w1, req1)
 	returnVal, _ = io.ReadAll(w1.Result().Body) //nolint
 	returnStr := string(returnVal)
-	expect = "{\"type\":\"ProjectPlan\",\"info\":\"Fields:[ * ]\",\"id\":0,\"children\":[1]}\n\n   {\"type\":\"DataSourcePlan\",\"info\":\"StreamName: alert\",\"id\":1,\"children\":null}\n\n"
-	assert.Equal(suite.T(), expect, returnStr)
+	expect = `
+{"type":"ProjectPlan","info":"Fields:[ * ]","id":0,"children":[1]}
+	{"type":"DataSourcePlan","info":"StreamName: alert","id":1}`
+	assert.Equal(suite.T(), strings.Trim(expect, "\n"), returnStr)
 
 	req1, _ = http.NewRequest(http.MethodGet, "http://localhost:8080/rules/rule32211/explain", bytes.NewBufferString("any"))
 	w1 = httptest.NewRecorder()
diff --git a/internal/server/trial_test.go b/internal/server/trial_test.go
new file mode 100644
index 00000000..9451fbc1
--- /dev/null
+++ b/internal/server/trial_test.go
@@ -0,0 +1,119 @@
+// Copyright 2021-2024 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 server
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/processor"
+	"github.com/lf-edge/ekuiper/v2/internal/trial"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func (suite *RestTestSuite) TestRuleTest() {
+	ip := "127.0.0.1"
+	port := 10087
+	httpserver.InitGlobalServerManager(ip, port, nil)
+	connection.InitConnectionManager4Test()
+	conf.IsTesting = true
+	conf.InitConf()
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(suite.T(), err)
+	require.NoError(suite.T(), store.SetupDefault(dataDir))
+
+	rd := &trial.RunDef{
+		Id:  "mock1",
+		Sql: "select * from trialDemo",
+		Mock: map[string]map[string]any{
+			"trialDemo": {
+				"data": []map[string]any{
+					{
+						"a": 1,
+					},
+				},
+				"loop":     true,
+				"interval": 100,
+			},
+		},
+		SinkProps: map[string]any{
+			"sendSingle": true,
+		},
+	}
+	b, err := json.Marshal(rd)
+	require.NoError(suite.T(), err)
+	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/ruletest", bytes.NewBuffer(b))
+	w2 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.NotEqual(suite.T(), http.StatusOK, w2.Code)
+
+	p := processor.NewStreamProcessor()
+	p.ExecStmt("DROP STREAM trialDemo")
+	_, err = p.ExecStmt("CREATE STREAM trialDemo () WITH (DATASOURCE=\"trialDemo\", TYPE=\"simulator\", FORMAT=\"json\", KEY=\"ts\")")
+	require.NoError(suite.T(), err)
+	defer p.ExecStmt("DROP STREAM trialDemo")
+
+	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/ruletest", bytes.NewBuffer(b))
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/mock1"}
+	c1, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	assert.NoError(suite.T(), err)
+	defer c1.Close()
+
+	recvCh := make(chan []byte, 10)
+	closeCh := make(chan struct{}, 10)
+	go func() {
+		_, data, err := c1.ReadMessage()
+		require.NoError(suite.T(), err)
+		recvCh <- data
+	}()
+	go func() {
+		for {
+			select {
+			case <-closeCh:
+				return
+			default:
+				timex.Add(100 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/ruletest/mock1/start", bytes.NewBuffer([]byte{}))
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+	require.Equal(suite.T(), `{"a":1}`, string(<-recvCh))
+	req2, _ = http.NewRequest(http.MethodDelete, "http://localhost:8080/ruletest/mock1", bytes.NewBuffer([]byte{}))
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+	closeCh <- struct{}{}
+}
diff --git a/internal/topo/planner/explainInfo_test.go b/internal/topo/planner/explainInfo_test.go
index 8c29ac5d..2ff61e45 100644
--- a/internal/topo/planner/explainInfo_test.go
+++ b/internal/topo/planner/explainInfo_test.go
@@ -24,7 +24,7 @@ func TestDataSourcePlanExplainInfo(t *testing.T) {
 					"field3": {},
 				},
 			},
-			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test1, Fields:[ field1, field2, field3 ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test1, Fields:[ field1, field2, field3 ]\",\"id\":0}\n",
 			t:   "DataSourcePlan",
 		},
 		{
@@ -53,19 +53,19 @@ func TestDataSourcePlanExplainInfo(t *testing.T) {
 					"s3": {},
 				},
 			},
-			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test3, Fields:[ column1, column2, id ], StreamFields:[ s1, s2, s3 ]\",\"id\":2,\"children\":null}\n",
+			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test3, Fields:[ column1, column2, id ], StreamFields:[ s1, s2, s3 ]\",\"id\":2}\n",
 			t:   "DataSourcePlan",
 		},
 		{
 			p: &DataSourcePlan{
 				name: "test4",
 			},
-			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test4\",\"id\":3,\"children\":null}\n",
+			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test4\",\"id\":3}\n",
 			t:   "DataSourcePlan",
 		},
 		{
 			p:   &DataSourcePlan{},
-			res: "{\"type\":\"DataSourcePlan\",\"info\":\"\",\"id\":4,\"children\":null}\n",
+			res: "{\"type\":\"DataSourcePlan\",\"info\":\"\",\"id\":4}\n",
 			t:   "DataSourcePlan",
 		},
 	}
@@ -108,7 +108,7 @@ func TestAggregatePlanExplainInfo(t *testing.T) {
 					},
 				},
 			},
-			res: "{\"type\":\"AggregatePlan\",\"info\":\"Dimension:{ Call:{ name:lpad, args:[$$default.name, 1] } }\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"AggregatePlan\",\"info\":\"Dimension:{ Call:{ name:lpad, args:[$$default.name, 1] } }\",\"id\":0}\n",
 			t:   "AggregatePlan",
 		},
 		{
@@ -139,7 +139,7 @@ func TestAggregatePlanExplainInfo(t *testing.T) {
 					ast.Dimension{Expr: &ast.FieldRef{Name: "department", StreamName: ast.DefaultStream}},
 				},
 			},
-			res: "{\"type\":\"AggregatePlan\",\"info\":\"Dimension:{ $$default.department }\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"AggregatePlan\",\"info\":\"Dimension:{ $$default.department }\",\"id\":0}\n",
 			t:   "AggregatePlan",
 		},
 	}
@@ -190,7 +190,7 @@ func TestAnalyticFuncsPlanExplainInfo(t *testing.T) {
 					},
 				},
 			},
-			res: "{\"type\":\"AnalyticFuncsPlan\",\"info\":\"Funcs:[ Call:{ name:lag, args:[src1.temp] } ], FieldFuncs:[ Call:{ name:lag, args:[src1.name] }, Call:{ name:latest, args:[Call:{ name:lag, args:[src1.name] }] } ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"AnalyticFuncsPlan\",\"info\":\"Funcs:[ Call:{ name:lag, args:[src1.temp] } ], FieldFuncs:[ Call:{ name:lag, args:[src1.name] }, Call:{ name:latest, args:[Call:{ name:lag, args:[src1.name] }] } ]\",\"id\":0}\n",
 			t:   "AnalyticFuncsPlan",
 		},
 		{
@@ -278,7 +278,7 @@ func TestFilterPlanExplainInfo(t *testing.T) {
 					RHS: &ast.StringLiteral{Val: "v1"},
 				},
 			},
-			res: "{\"type\":\"FilterPlan\",\"info\":\"Condition:{ binaryExpr:{ src1.name = v1 } }, \",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"FilterPlan\",\"info\":\"Condition:{ binaryExpr:{ src1.name = v1 } }, \",\"id\":0}\n",
 			t:   "FilterPlan",
 		},
 		{
@@ -336,7 +336,7 @@ func TestHavingPlanExplainInfo(t *testing.T) {
 					RHS: &ast.IntegerLiteral{Val: 2},
 				},
 			},
-			res: "{\"type\":\"HavingPlan\",\"info\":\"Condition:{ binaryExpr:{ Call:{ name:count, args:[*] } > 2 } }, \",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"HavingPlan\",\"info\":\"Condition:{ binaryExpr:{ Call:{ name:count, args:[*] } > 2 } }, \",\"id\":0}\n",
 			t:   "HavingPlan",
 		},
 		{
@@ -413,7 +413,7 @@ func TestJoinAlignPlanExplainInfo(t *testing.T) {
 			p: &JoinAlignPlan{
 				Emitters: []string{"tableInPlanner"},
 			},
-			res: "{\"type\":\"JoinAlignPlan\",\"info\":\"Emitters:[ tableInPlanner ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"JoinAlignPlan\",\"info\":\"Emitters:[ tableInPlanner ]\",\"id\":0}\n",
 			t:   "JoinAlignPlan",
 		},
 		{
@@ -478,7 +478,7 @@ func TestJoinPlanExplainInfo(t *testing.T) {
 					},
 				}},
 			},
-			res: "{\"type\":\"JoinPlan\",\"info\":\"Joins:[ { joinType:INNER_JOIN, binaryExpr:{ binaryExpr:{ binaryExpr:{ src1.temp > 20 } OR binaryExpr:{ src2.hum > 60 } } AND binaryExpr:{ src1.id1 = src2.id2 } } } ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"JoinPlan\",\"info\":\"Joins:[ { joinType:INNER_JOIN, binaryExpr:{ binaryExpr:{ binaryExpr:{ src1.temp > 20 } OR binaryExpr:{ src2.hum > 60 } } AND binaryExpr:{ src1.id1 = src2.id2 } } } ]\",\"id\":0}\n",
 			t:   "JoinPlan",
 		},
 		{
@@ -557,7 +557,7 @@ func TestLookupPlanExplainInfo(t *testing.T) {
 					},
 				},
 			},
-			res: "{\"type\":\"LookupPlan\",\"info\":\"Join:{ joinType:LEFT_JOIN, expr:binaryExpr:{ left.device_id = good.id } }\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"LookupPlan\",\"info\":\"Join:{ joinType:LEFT_JOIN, expr:binaryExpr:{ left.device_id = good.id } }\",\"id\":0}\n",
 			t:   "LookupPlan",
 		},
 		{
@@ -907,7 +907,7 @@ func TestWatermarkPlanExplainInfo(t *testing.T) {
 				Emitters:      []string{"id", "grade"},
 				SendWatermark: false,
 			},
-			res: "{\"type\":\"WatermarkPlan\",\"info\":\"Emitters:[ id, grade ], SendWatermark:false\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"WatermarkPlan\",\"info\":\"Emitters:[ id, grade ], SendWatermark:false\",\"id\":0}\n",
 			t:   "WatermarkPlan",
 		},
 		{
@@ -969,7 +969,7 @@ func TestWindowFuncPlanExplainInfo(t *testing.T) {
 					},
 				},
 			},
-			res: "{\"type\":\"WindowFuncPlan\",\"info\":\"windowFuncFields:[ {name:AName, expr:}, {name:AName, expr:}, {name:Name}, {name:AName} ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"WindowFuncPlan\",\"info\":\"windowFuncFields:[ {name:AName, expr:}, {name:AName, expr:}, {name:Name}, {name:AName} ]\",\"id\":0}\n",
 			t:   "WindowFuncPlan",
 		},
 		{
@@ -1015,7 +1015,7 @@ func TestOrderPlanExplainInfo(t *testing.T) {
 			p: &OrderPlan{
 				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false, FieldExpr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}},
 			},
-			res: "{\"type\":\"OrderPlan\",\"info\":\"SortFields:[ sortField:{ name:name, ascending:false, fieldExpr:{ $$default.name } } ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"OrderPlan\",\"info\":\"SortFields:[ sortField:{ name:name, ascending:false, fieldExpr:{ $$default.name } } ]\",\"id\":0}\n",
 			t:   "OrderPlan",
 		},
 		{
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index 4b60d32d..0a897a6e 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -4863,112 +4863,3 @@ func TestTransformSourceNode(t *testing.T) {
 		})
 	}
 }
-
-func TestGetLogicalPlanForExplain(t *testing.T) {
-	kv, err := store.GetKV("stream")
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	streamSqls := map[string]string{
-		"src1": `CREATE STREAM src1 (
-					id1 BIGINT,
-					temp BIGINT,
-					name string,
-					myarray array(string)
-				) WITH (DATASOURCE="src1", FORMAT="json", KEY="ts");`,
-		"src2": `CREATE STREAM src2 (
-					id2 BIGINT,
-					hum BIGINT
-				) WITH (DATASOURCE="src2", FORMAT="json", KEY="ts", TIMESTAMP_FORMAT="YYYY-MM-dd HH:mm:ss");`,
-		"tableInPlanner": `CREATE TABLE tableInPlanner (
-					id BIGINT,
-					name STRING,
-					value STRING,
-					hum BIGINT
-				) WITH (TYPE="file");`,
-	}
-	types := map[string]ast.StreamType{
-		"src1":           ast.TypeStream,
-		"src2":           ast.TypeStream,
-		"tableInPlanner": ast.TypeTable,
-	}
-	for name, sql := range streamSqls {
-		s, err := json.Marshal(&xsql.StreamInfo{
-			StreamType: types[name],
-			Statement:  sql,
-		})
-		if err != nil {
-			t.Error(err)
-			t.Fail()
-		}
-		err = kv.Set(name, string(s))
-		if err != nil {
-			t.Error(err)
-			t.Fail()
-		}
-	}
-	streams := make(map[string]*ast.StreamStmt)
-	for n := range streamSqls {
-		streamStmt, err := xsql.GetDataSource(kv, n)
-		if err != nil {
-			t.Errorf("fail to get stream %s, please check if stream is created", n)
-			return
-		}
-		streams[n] = streamStmt
-	}
-
-	ref := &ast.AliasRef{
-		Expression: &ast.Call{
-			Name:     "row_number",
-			FuncType: ast.FuncTypeWindow,
-		},
-	}
-	ref.SetRefSource([]string{})
-
-	tests := []struct {
-		rule *def.Rule
-		res  string
-		err  string
-	}{
-		{
-			rule: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "select name, row_number() as index from src1",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
-			},
-			res: "{\"type\":\"ProjectPlan\",\"info\":\"Fields:[ $$alias.index, src1.name ]\",\"id\":0,\"children\":[1]}\n\n   {\"type\":\"WindowFuncPlan\",\"info\":\"windowFuncFields:[ {name:index, expr:$$alias.index} ]\",\"id\":1,\"children\":[2]}\n\n         {\"type\":\"DataSourcePlan\",\"info\":\"StreamName: src1, StreamFields:[ name ]\",\"id\":2,\"children\":null}\n\n",
-		},
-		{
-			rule: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "select name, row_number() from src1",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
-			},
-			res: "{\"type\":\"ProjectPlan\",\"info\":\"Fields:[ src1.name, Call:{ name:row_number } ]\",\"id\":0,\"children\":[1]}\n\n   {\"type\":\"WindowFuncPlan\",\"info\":\"windowFuncFields:[ {name:row_number, expr:Call:{ name:row_number }} ]\",\"id\":1,\"children\":[2]}\n\n         {\"type\":\"DataSourcePlan\",\"info\":\"StreamName: src1, StreamFields:[ name ]\",\"id\":2,\"children\":null}\n\n",
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-
-	for i, tt := range tests {
-		explain, err := GetExplainInfoFromLogicalPlan(tt.rule)
-		if err != nil {
-			t.Errorf(err.Error())
-		}
-		if !reflect.DeepEqual(explain, tt.res) {
-			t.Errorf("case %d: expect validate %v but got %v", i, tt.res, explain)
-		}
-	}
-}
diff --git a/internal/topo/planner/push_proj_test.go b/internal/topo/planner/push_proj_test.go
new file mode 100644
index 00000000..d187c1d1
--- /dev/null
+++ b/internal/topo/planner/push_proj_test.go
@@ -0,0 +1,80 @@
+// Copyright 2024 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 planner
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+func TestPushProjection(t *testing.T) {
+	kv, err := store.GetKV("stream")
+	require.NoError(t, err)
+	require.NoError(t, prepareStream())
+	sql := `select a as c from sharedStream group by countwindow(2)`
+	stmt, err := xsql.NewParser(strings.NewReader(sql)).Parse()
+	assert.NoError(t, err)
+	p, err := createLogicalPlan(stmt, &def.RuleOption{
+		Qos: 0,
+	}, kv)
+	require.NoError(t, err)
+	explain, err := ExplainFromLogicalPlan(p, "")
+	require.NoError(t, err)
+	expect := `
+{"type":"ProjectPlan","info":"Fields:[ $$alias.c ]","id":0,"children":[1]}
+	{"type":"WindowPlan","info":"{ length:2, windowType:COUNT_WINDOW, limit: 0 }","id":1,"children":[2]}
+			{"type":"ProjectPlan","info":"Fields:[ sharedStream.a ]","id":2,"children":[3]}
+					{"type":"DataSourcePlan","info":"StreamName: sharedStream, StreamFields:[ a ]","id":3}`
+	require.Equal(t, strings.TrimPrefix(expect, "\n"), explain)
+}
+
+func prepareStream() error {
+	kv, err := store.GetKV("stream")
+	if err != nil {
+		return err
+	}
+	streamSqls := map[string]string{
+		"sharedStream": `CREATE STREAM sharedStream (
+					a BIGINT,
+					b BIGINT,
+				) WITH (DATASOURCE="src1", SHARED="true");`,
+	}
+	types := map[string]ast.StreamType{
+		"sharedStream": ast.TypeStream,
+	}
+	for name, sql := range streamSqls {
+		s, err := json.Marshal(&xsql.StreamInfo{
+			StreamType: types[name],
+			Statement:  sql,
+		})
+		if err != nil {
+			return err
+		}
+		err = kv.Set(name, string(s))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/internal/trial/manager_test.go b/internal/trial/manager_test.go
new file mode 100644
index 00000000..71a1df12
--- /dev/null
+++ b/internal/trial/manager_test.go
@@ -0,0 +1,172 @@
+// Copyright 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 trial
+
+import (
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/processor"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+// Run two test rules in parallel. Rerun one of the rules
+func TestTrialRule(t *testing.T) {
+	ip := "127.0.0.1"
+	port := 10087
+	httpserver.InitGlobalServerManager(ip, port, nil)
+	connection.InitConnectionManager4Test()
+	conf.IsTesting = true
+	conf.InitConf()
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	p := processor.NewStreamProcessor()
+	p.ExecStmt("DROP STREAM demo")
+	// Test 1 wrong rule
+	mockDef1 := `{"id":"rule1","sql":"select * from demo","mockSource":{"demo":{"data":[{"name":"demo","value":1}],"interval":100,"loop":true}},"sinkProps":{"sendSingle":true}}`
+	_, err = TrialManager.CreateRule(mockDef1)
+	require.Error(t, err)
+	require.Equal(t, "fail to run rule rule1: fail to get stream demo, please check if stream is created", err.Error())
+
+	_, err = p.ExecStmt("CREATE STREAM demo () WITH (DATASOURCE=\"demo\", TYPE=\"simulator\", FORMAT=\"json\", KEY=\"ts\")")
+	require.NoError(t, err)
+	defer p.ExecStmt("DROP STREAM demo")
+
+	// Test 2 valid rule with mock
+	testValidTrial(t, mockDef1)
+
+	// Test 3 Runtime error rule
+	testRuntimeErrorTrial(t)
+
+	// Test 4 Rule without mock
+	testRealSourceTrial(t)
+	require.Equal(t, 0, len(TrialManager.runs))
+}
+
+func testValidTrial(t *testing.T, mockDef1 string) {
+	// Test 2 valid rule with mock
+	id, err := TrialManager.CreateRule(mockDef1)
+	require.NoError(t, err)
+	require.Equal(t, "rule1", id)
+	// Read from ws
+	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/rule1"}
+	c1, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	assert.NoError(t, err)
+	recvCh := make(chan []byte, 10)
+	closeCh := make(chan struct{}, 10)
+	go func() {
+		_, data, err := c1.ReadMessage()
+		require.NoError(t, err)
+		recvCh <- data
+	}()
+	go func() {
+		for {
+			select {
+			case <-closeCh:
+				return
+			default:
+				timex.Add(100 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}()
+	time.Sleep(100 * time.Millisecond)
+	require.NoError(t, TrialManager.StartRule("rule1"))
+	require.Equal(t, []byte(`{"name":"demo","value":1}`), <-recvCh)
+	c1.Close()
+	TrialManager.StopRule("rule1")
+	closeCh <- struct{}{}
+}
+
+func testRuntimeErrorTrial(t *testing.T) {
+	// Test 3 Runtime error rule
+	mockDefErr := `{"id":"ruleErr","sql":"select name + value from demo","mockSource":{"demo":{"data":[{"name":"demo","value":1}],"interval":100,"loop":true}},"sinkProps":{"sendSingle":true}}`
+	id, err := TrialManager.CreateRule(mockDefErr)
+	require.NoError(t, err)
+	require.Equal(t, "ruleErr", id)
+	// Read from ws
+	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/ruleErr"}
+	c2, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	require.NoError(t, err)
+	recvCh := make(chan []byte, 10)
+	closeCh := make(chan struct{}, 10)
+	go func() {
+		_, data, err := c2.ReadMessage()
+		require.NoError(t, err)
+		recvCh <- data
+	}()
+	go func() {
+		for {
+			select {
+			case <-closeCh:
+				return
+			default:
+				timex.Add(100 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}()
+	time.Sleep(10 * time.Millisecond)
+	require.NoError(t, TrialManager.StartRule(id))
+	require.Equal(t, `run Select error: expr: binaryExpr:{ demo.name + demo.value } meet error, err:invalid operation string(demo) + float64(1)`, string(<-recvCh))
+	TrialManager.StopRule(id)
+	closeCh <- struct{}{}
+	c2.Close()
+}
+
+func testRealSourceTrial(t *testing.T) {
+	noMockDef := `{"id":"rule2","sql":"select * from demo","sinkProps":{"sendSingle":true}}`
+	id, err := TrialManager.CreateRule(noMockDef)
+	assert.Equal(t, "rule2", id)
+	assert.NoError(t, err)
+
+	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/rule2"}
+	c3, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	require.NoError(t, err)
+	recvCh := make(chan []byte, 10)
+	closeCh := make(chan struct{}, 10)
+	go func() {
+		_, data, err := c3.ReadMessage()
+		require.NoError(t, err)
+		recvCh <- data
+	}()
+	go func() {
+		for {
+			select {
+			case <-closeCh:
+				return
+			default:
+				timex.Add(time.Second)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}()
+	time.Sleep(10 * time.Millisecond)
+	require.NoError(t, TrialManager.StartRule(id))
+	require.Equal(t, "{\"humidity\":50,\"temperature\":22.5}", string(<-recvCh))
+	TrialManager.StopRule(id)
+	closeCh <- struct{}{}
+	c3.Close()
+}