Merge branch 'main' into fix-like-expression-panic-on-parse

protocol/kafka_confluent/v2/message.go

@@ -11,14 +11,15 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+
 	"github.com/cloudevents/sdk-go/v2/binding"
 	"github.com/cloudevents/sdk-go/v2/binding/format"
 	"github.com/cloudevents/sdk-go/v2/binding/spec"
-	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
 )
 
 const (
-	prefix         = "ce-"
+	prefix         = "ce_"
 	contentTypeKey = "content-type"
 )
 

protocol/kafka_confluent/v2/message_test.go

@@ -42,14 +42,14 @@ var (
 		TopicPartition: topicPartition,
 		Value:          []byte("hello world!"),
 		Headers: mapToKafkaHeaders(map[string]string{
-			"ce-type":            testEvent.Type(),
-			"ce-source":          testEvent.Source(),
-			"ce-id":              testEvent.ID(),
-			"ce-time":            test.Timestamp.String(),
-			"ce-specversion":     "1.0",
-			"ce-dataschema":      test.Schema.String(),
-			"ce-datacontenttype": "text/json",
-			"ce-subject":         "receiverTopic",
+			"ce_type":            testEvent.Type(),
+			"ce_source":          testEvent.Source(),
+			"ce_id":              testEvent.ID(),
+			"ce_time":            test.Timestamp.String(),
+			"ce_specversion":     "1.0",
+			"ce_dataschema":      test.Schema.String(),
+			"ce_datacontenttype": "text/json",
+			"ce_subject":         "receiverTopic",
 			"exta":               "someext",
 		}),
 	}
@@ -89,14 +89,14 @@ func TestNewMessage(t *testing.T) {
 				TopicPartition: topicPartition,
 				Value:          nil,
 				Headers: mapToKafkaHeaders(map[string]string{
-					"ce-type":            testEvent.Type(),
-					"ce-source":          testEvent.Source(),
-					"ce-id":              testEvent.ID(),
-					"ce-time":            test.Timestamp.String(),
-					"ce-specversion":     "1.0",
-					"ce-dataschema":      test.Schema.String(),
-					"ce-datacontenttype": "text/json",
-					"ce-subject":         "receiverTopic",
+					"ce_type":            testEvent.Type(),
+					"ce_source":          testEvent.Source(),
+					"ce_id":              testEvent.ID(),
+					"ce_time":            test.Timestamp.String(),
+					"ce_specversion":     "1.0",
+					"ce_dataschema":      test.Schema.String(),
+					"ce_datacontenttype": "text/json",
+					"ce_subject":         "receiverTopic",
 				}),
 			},
 			expectedEncoding: binding.EncodingBinary,

sql/v2/README.md

@@ -18,6 +18,54 @@ expression, err := cesqlparser.Parse("subject = 'Hello world'")
 res, err := expression.Evaluate(event)
 ```
 
+Add a user defined function
+```go
+import (
+    cesql "github.com/cloudevents/sdk-go/sql/v2"
+    cefn "github.com/cloudevents/sdk-go/sql/v2/function"
+    cesqlparser "github.com/cloudevents/sdk-go/sql/v2/parser"
+    ceruntime "github.com/cloudevents/sdk-go/sql/v2/runtime"
+    cloudevents "github.com/cloudevents/sdk-go/v2"
+)
+
+// Create a test event
+event := cloudevents.NewEvent()
+event.SetID("aaaa-bbbb-dddd")
+event.SetSource("https://my-source")
+event.SetType("dev.tekton.event")
+
+// Create and add a new user defined function
+var HasPrefixFunction cesql.Function = cefn.NewFunction(
+    "HASPREFIX",
+    []cesql.Type{cesql.StringType, cesql.StringType},
+    nil,
+    func(event cloudevents.Event, i []interface{}) (interface{}, error) {
+        str := i[0].(string)
+        prefix := i[1].(string)
+
+        return strings.HasPrefix(str, prefix), nil
+    },
+)
+
+err := ceruntime.AddFunction(HasPrefixFunction)
+
+// parse the expression 
+expression, err := cesqlparser.Parse("HASPREFIX(type, 'dev.tekton.event')")
+	if err != nil {
+		fmt.Println("parser err: ", err)
+		os.Exit(1)
+	}
+
+// Evalute the expression with the test event
+res, err := expression.Evaluate(event)
+
+if res.(bool) {
+    fmt.Println("Event type has the prefix")
+} else {
+    fmt.Println("Event type doesn't have the prefix")
+}
+```
+
 ## Development guide
 
 To regenerate the parser, make sure you have [ANTLR4 installed](https://github.com/antlr/antlr4/blob/master/doc/getting-started.md) and then run:

sql/v2/function/function.go

@@ -10,11 +10,13 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 )
 
+type FuncType func(cloudevents.Event, []interface{}) (interface{}, error)
+
 type function struct {
 	name         string
 	fixedArgs    []cesql.Type
 	variadicArgs *cesql.Type
-	fn           func(cloudevents.Event, []interface{}) (interface{}, error)
+	fn           FuncType
 }
 
 func (f function) Name() string {
@@ -39,3 +41,15 @@ func (f function) ArgType(index int) *cesql.Type {
 func (f function) Run(event cloudevents.Event, arguments []interface{}) (interface{}, error) {
 	return f.fn(event, arguments)
 }
+
+func NewFunction(name string,
+	fixedargs []cesql.Type,
+	variadicArgs *cesql.Type,
+	fn FuncType) cesql.Function {
+	return function{
+		name:         name,
+		fixedArgs:    fixedargs,
+		variadicArgs: variadicArgs,
+		fn:           fn,
+	}
+}

sql/v2/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/antlr/antlr4/runtime/Go/antlr v1.4.10
 	github.com/cloudevents/sdk-go/v2 v2.5.0
 	github.com/stretchr/testify v1.8.0
+	gopkg.in/yaml.v2 v2.4.0
 	sigs.k8s.io/yaml v1.3.0
 )
 
@@ -20,7 +21,6 @@ require (
 	go.uber.org/atomic v1.4.0 // indirect
 	go.uber.org/multierr v1.1.0 // indirect
 	go.uber.org/zap v1.10.0 // indirect
-	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 

sql/v2/runtime/functions_resolver.go

@@ -58,6 +58,11 @@ func (table functionTable) AddFunction(function cesql.Function) error {
 	}
 }
 
+// Adds user defined function
+func AddFunction(fn cesql.Function) error {
+	return globalFunctionTable.AddFunction(fn)
+}
+
 func (table functionTable) ResolveFunction(name string, args int) cesql.Function {
 	item := table[strings.ToUpper(name)]
 	if item == nil {

sql/v2/runtime/test/tck/user_defined_functions.yaml

@@ -0,0 +1,27 @@
+name: User defined functions
+tests:
+  - name: HASPREFIX (1)
+    expression: "HASPREFIX('abcdef', 'ab')"
+    result: true
+  - name: HASPREFIX (2)
+    expression: "HASPREFIX('abcdef', 'abcdef')"
+    result: true  
+  - name: HASPREFIX (3)
+    expression: "HASPREFIX('abcdef', '')"
+    result: true  
+  - name: HASPREFIX (4)
+    expression: "HASPREFIX('abcdef', 'gh')"
+    result: false  
+  - name: HASPREFIX (5)
+    expression: "HASPREFIX('abcdef', 'abcdefg')"
+    result: false
+
+  - name: KONKAT (1)
+    expression: "KONKAT('a', 'b', 'c')"
+    result: abc
+  - name: KONKAT (2)
+    expression: "KONKAT()"
+    result: ""
+  - name: KONKAT (3)
+    expression: "KONKAT('a')"
+    result: "a"