nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: lf-edge/ekuiper
Pull request #844 (2021-06-19 05:56:42)

fix(sql): support short-circuit for logical operation
#835 


Patch:
diff --git a/xsql/ast.go b/xsql/ast.go
index ff32ad98..560bb2f8 100644
--- a/xsql/ast.go
+++ b/xsql/ast.go
@@ -1276,6 +1276,17 @@ func (v *ValuerEval) evalBinaryExpr(expr *BinaryExpr) interface{} {
 	case error:
 		return val
 	}
+	// shortcut for bool
+	switch expr.OP {
+	case AND:
+		if bv, ok := lhs.(bool); ok && !bv {
+			return false
+		}
+	case OR:
+		if bv, ok := lhs.(bool); ok && bv {
+			return true
+		}
+	}
 	if isSliceOrArray(lhs) {
 		return v.evalJsonExpr(lhs, expr.OP, expr.RHS)
 	}


Test patch:
diff --git a/xstream/operators/filter_test.go b/xstream/operators/filter_test.go
index 9dc5159f..a33c1e29 100644
--- a/xstream/operators/filter_test.go
+++ b/xstream/operators/filter_test.go
@@ -322,6 +322,32 @@ func TestFilterPlan_Apply(t *testing.T) {
 			},
 			result: nil,
 		},
+		// Test for short circuit. If not, an error would happen
+		{
+			sql: "SELECT abc FROM tbl WHERE abc*2 > 12 AND abc / 0 < 20",
+			data: &xsql.Tuple{
+				Emitter: "tbl",
+				Message: xsql.Message{
+					"abc": int64(6),
+				},
+			},
+			result: nil,
+		},
+		{
+			sql: "SELECT abc FROM tbl WHERE abc*2+3 > 12 OR abc / 0 < 20",
+			data: &xsql.Tuple{
+				Emitter: "tbl",
+				Message: xsql.Message{
+					"abc": int64(6),
+				},
+			},
+			result: &xsql.Tuple{
+				Emitter: "tbl",
+				Message: xsql.Message{
+					"abc": int64(6),
+				},
+			},
+		},
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
@@ -445,6 +471,16 @@ func TestFilterPlanError(t *testing.T) {
 			},
 			result: errors.New("run Where error: invalid operation int64(50) = string(v1)"),
 		},
+		{
+			sql: "SELECT abc FROM tbl WHERE abc*2+3 > 12 AND abc / 0 < 20",
+			data: &xsql.Tuple{
+				Emitter: "tbl",
+				Message: xsql.Message{
+					"abc": int64(6),
+				},
+			},
+			result: errors.New("run Where error: divided by zero"),
+		},
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
Repository: lf-edge/ekuiper
Pull request #931 (2021-08-11 07:10:49)

fix(operator): join tuple must exclude default stream
In schemaless mode, the analyzer cannot deduce the stream of a field so that the ref source of a fieldRef could be $$default, which means it could be any of the stream. And that is not a real stream to form a join tuple so we should exclude it.

Signed-off-by: Jiyong Huang <huangjy@emqx.io>


Patch:
diff --git a/internal/topo/operator/join_operator.go b/internal/topo/operator/join_operator.go
index 6301d605..a9e58d12 100644
--- a/internal/topo/operator/join_operator.go
+++ b/internal/topo/operator/join_operator.go
@@ -73,6 +73,10 @@ func (jp *JoinOp) getStreamNames(join *ast.Join) ([]string, error) {
 	ast.WalkFunc(join, func(node ast.Node) bool {
 		if f, ok := node.(*ast.FieldRef); ok {
 			for _, v := range f.RefSources() {
+				// Exclude default stream as it is a virtual stream name.
+				if v == ast.DefaultStream {
+					continue
+				}
 				if _, ok := keys[v]; !ok {
 					srcs = append(srcs, string(v))
 					keys[v] = true


Test patch:
diff --git a/internal/topo/operator/join_test.go b/internal/topo/operator/join_test.go
index f69554e3..eca039e1 100644
--- a/internal/topo/operator/join_test.go
+++ b/internal/topo/operator/join_test.go
@@ -35,7 +35,7 @@ func TestLeftJoinPlan_Apply(t *testing.T) {
 		result interface{}
 	}{
 		{ //0
-			sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
+			sql: "SELECT id1 FROM src1 left join src2 on id1 = id2",
 			data: xsql.WindowTuplesSet{
 				Content: []xsql.WindowTuples{
 					{
Repository: lf-edge/ekuiper
Pull request #1193 (2022-02-28 02:29:16)

fix(parser): validate string literal


Patch:
diff --git a/internal/xsql/lexical.go b/internal/xsql/lexical.go
index 087e3bc5..70b04e36 100644
--- a/internal/xsql/lexical.go
+++ b/internal/xsql/lexical.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 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.
@@ -308,7 +308,10 @@ func (s *Scanner) ScanString() (tok ast.Token, lit string) {
 			buf.WriteRune(ch)
 		}
 	}
-	r, _ := strconv.Unquote(buf.String())
+	r, err := strconv.Unquote(buf.String())
+	if err != nil {
+		return ast.ILLEGAL, "invalid string: " + buf.String()
+	}
 	return ast.STRING, r
 }
 


Test patch:
diff --git a/internal/xsql/parser_test.go b/internal/xsql/parser_test.go
index 2b11eb2e..d9687dc4 100644
--- a/internal/xsql/parser_test.go
+++ b/internal/xsql/parser_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 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.
@@ -1436,6 +1436,10 @@ func TestParser_ParseStatement(t *testing.T) {
 				Sources: []ast.Source{&ast.Table{Name: "tbl"}},
 			},
 		},
+		{
+			s:   `SELECT ".*(/)(?!.*\1)" FROM topic/sensor1 AS t1`,
+			err: `found "invalid string: \".*(/)(?!.*\\1)\"", expected expression.`,
+		},
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
Repository: lf-edge/ekuiper
Pull request #1283 (2022-05-27 10:11:58)

fix(neuron): fix connection count management


Patch:
diff --git a/internal/topo/mock/mock.go b/internal/topo/mock/mock.go
index 1fae885e..c934c9be 100644
--- a/internal/topo/mock/mock.go
+++ b/internal/topo/mock/mock.go
@@ -22,7 +22,7 @@ import (
 )
 
 func newMockContext(ruleId string, opId string) api.StreamContext {
-	contextLogger := conf.Log.WithField("rule", "mockRule0")
+	contextLogger := conf.Log.WithField("rule", ruleId)
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
 	tempStore, _ := state.CreateStore(ruleId, api.AtMostOnce)
 	return ctx.WithMeta(ruleId, opId, tempStore)
diff --git a/internal/topo/neuron/connection.go b/internal/topo/neuron/connection.go
index 1009274a..9bd2b43d 100644
--- a/internal/topo/neuron/connection.go
+++ b/internal/topo/neuron/connection.go
@@ -67,8 +67,8 @@ func createOrGetConnection(sc api.StreamContext, url string) error {
 		sctx := ctx.WithMeta(ruleId, opId, store)
 		memory.CreatePub(NeuronTopic)
 		go run(sctx)
-		connectionCount++
 	}
+	connectionCount++
 	return nil
 }
 


Test patch:
diff --git a/internal/topo/mock/test_sink.go b/internal/topo/mock/test_sink.go
index bb177546..1741637d 100644
--- a/internal/topo/mock/test_sink.go
+++ b/internal/topo/mock/test_sink.go
@@ -21,7 +21,7 @@ import (
 )
 
 func RunSinkCollect(s api.Sink, data []interface{}) error {
-	ctx := newMockContext("rule1", "op1")
+	ctx := newMockContext("ruleSink", "op1")
 	err := s.Open(ctx)
 	if err != nil {
 		return err
diff --git a/internal/topo/neuron/neuron_test.go b/internal/topo/neuron/neuron_test.go
index 8d341f44..17bfec7b 100644
--- a/internal/topo/neuron/neuron_test.go
+++ b/internal/topo/neuron/neuron_test.go
@@ -106,7 +106,6 @@ func TestMultiSourceSink(t *testing.T) {
 	})
 
 	server, ch := mockNeuron(true, true)
-	defer server.Close()
 	wg := sync.WaitGroup{}
 	wg.Add(3)
 	go func() {
@@ -170,4 +169,7 @@ func TestMultiSourceSink(t *testing.T) {
 		t.Errorf("result mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", sexp, actual)
 	}
 	wg.Wait()
+	server.Close()
+	time.Sleep(100 * time.Millisecond)
+	sinkTest(t)
 }
diff --git a/internal/topo/neuron/sink_test.go b/internal/topo/neuron/sink_test.go
index 7a3ea65b..a2a0daa8 100644
--- a/internal/topo/neuron/sink_test.go
+++ b/internal/topo/neuron/sink_test.go
@@ -21,7 +21,7 @@ import (
 	"time"
 )
 
-func TestSink(t *testing.T) {
+func sinkTest(t *testing.T) {
 	server, ch := mockNeuron(false, true)
 	defer server.Close()
 
@@ -79,4 +79,5 @@ func TestSink(t *testing.T) {
 	if !reflect.DeepEqual(actual, exp) {
 		t.Errorf("result mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", exp, actual)
 	}
+	time.Sleep(100 * time.Millisecond)
 }