diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 7f37f43e52..e56af857e6 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -89,12 +89,11 @@ jobs:
           path: fvt_coverage.xml
 
   run_fvt_tests:
-    needs:
-      - run_unit_tests
     uses: ./.github/workflows/run_fvt_tests.yaml
 
   upload_coverage:
     needs:
+      - run_unit_tests
       - run_fvt_tests
     runs-on: ubuntu-latest
     steps:
diff --git a/extensions/impl/sql/source.go b/extensions/impl/sql/source.go
index d506efd830..1e15bf165b 100644
--- a/extensions/impl/sql/source.go
+++ b/extensions/impl/sql/source.go
@@ -208,7 +208,7 @@ func prepareValues(values []interface{}, columnTypes []*sql.ColumnType, columns
 	if len(columnTypes) > 0 {
 		for idx, columnType := range columnTypes {
 			if columnType.ScanType() != nil {
-				values[idx] = reflect.New(reflect.PtrTo(columnType.ScanType())).Interface()
+				values[idx] = reflect.New(reflect.PointerTo(columnType.ScanType())).Interface()
 			} else {
 				values[idx] = new(interface{})
 			}
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index 775c60f51e..055e66533d 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -20,6 +20,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
@@ -120,6 +121,10 @@ func (m *SourceNode) ingestAnyTuple(ctx api.StreamContext, data any, meta map[st
 		for _, mm := range mess {
 			m.ingestTuple(mm, ts)
 		}
+	case []pubsub.MemTuple:
+		for _, mm := range mess {
+			m.ingestTuple(mm.(*xsql.Tuple), ts)
+		}
 	default:
 		// should never happen
 		panic(fmt.Sprintf("receive wrong data %v", data))
diff --git a/test/run_jmeter.sh b/test/run_jmeter.sh
index a8016da0d2..0523245be5 100755
--- a/test/run_jmeter.sh
+++ b/test/run_jmeter.sh
@@ -83,8 +83,8 @@ echo -e "---------------------------------------------\n"
 /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/select_aggr_rule_order.jmx -l jmeter_logs/select_aggr_rule_order.jtl -j jmeter_logs/select_aggr_rule_order.log
 echo -e "---------------------------------------------\n"
 
-# /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/rule_pipeline.jmx -l jmeter_logs/rule_pipeline.jtl -j jmeter_logs/rule_pipeline.log
-# echo -e "---------------------------------------------\n"
+/opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/rule_pipeline.jmx -l jmeter_logs/rule_pipeline.jtl -j jmeter_logs/rule_pipeline.log
+echo -e "---------------------------------------------\n"
 
 if test $with_edgex = true; then
   /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/select_edgex_condition_rule.jmx -Dbase="$base_dir" -Dfvt="$fvt_dir" -l jmeter_logs/select_edgex_condition_rule.jtl -j jmeter_logs/select_edgex_condition_rule.log