feat: add data schema support (#182)

Author: RobertIndie

.github/workflows/bench.yaml

@@ -12,17 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# 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.
 
 name: Benchmark
 on:

.github/workflows/ci.yaml

@@ -12,17 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# 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.
 
 name: CI
 on:

.github/workflows/license.yaml

@@ -12,17 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# 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.
 
 name: License Check
 on:

.golangci.yml

@@ -0,0 +1,54 @@
+# Copyright 2024 Function Stream Org.
+#
+# 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.
+
+run:
+  deadline: 5m
+  allow-parallel-runners: true
+
+issues:
+  # don't skip warning about doc comments
+  # don't exclude the default set of lint
+  exclude-use-default: false
+  # restore some of the defaults
+  # (fill in the rest as needed)
+  exclude-rules:
+    - path: "api/*"
+      linters:
+        - lll
+    - path: "internal/*"
+      linters:
+        - dupl
+        - lll
+linters:
+  disable-all: true
+  enable:
+    - dupl
+    - errcheck
+    - exportloopref
+    - goconst
+    - gocyclo
+    - gofmt
+    - goimports
+    - gosimple
+    - govet
+    - ineffassign
+    - lll
+    - misspell
+    - nakedret
+    - prealloc
+    - staticcheck
+    - typecheck
+    - unconvert
+    - unparam
+    - unused

benchmark/bench_test.go

@@ -18,15 +18,16 @@ package benchmark
 
 import (
 	"context"
-	"github.com/functionstream/function-stream/fs/api"
-	"github.com/functionstream/function-stream/fs/runtime/wazero"
 	"math/rand"
 	"os"
 	"runtime/pprof"
 	"strconv"
 	"testing"
 	"time"
 
+	"github.com/functionstream/function-stream/fs/api"
+	"github.com/functionstream/function-stream/fs/runtime/wazero"
+
 	"github.com/apache/pulsar-client-go/pulsaradmin"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
 	adminclient "github.com/functionstream/function-stream/admin/client"
@@ -115,12 +116,14 @@ func BenchmarkStressForBasicFuncWithMemoryQueue(b *testing.B) {
 	memoryQueueFactory := contube.NewMemoryQueueFactory(context.Background())
 
 	s, err := server.NewServer(
-		server.WithRuntimeFactoryBuilder(common.WASMRuntime, func(configMap common.ConfigMap) (api.FunctionRuntimeFactory, error) {
-			return wazero.NewWazeroFunctionRuntimeFactory(), nil
-		}),
-		server.WithTubeFactoryBuilder(common.MemoryTubeType, func(configMap common.ConfigMap) (contube.TubeFactory, error) {
-			return memoryQueueFactory, nil
-		}),
+		server.WithRuntimeFactoryBuilder(common.WASMRuntime,
+			func(configMap common.ConfigMap) (api.FunctionRuntimeFactory, error) {
+				return wazero.NewWazeroFunctionRuntimeFactory(), nil
+			}),
+		server.WithTubeFactoryBuilder(common.MemoryTubeType,
+			func(configMap common.ConfigMap) (contube.TubeFactory, error) {
+				return memoryQueueFactory, nil
+			}),
 	)
 	if err != nil {
 		b.Fatal(err)
@@ -134,7 +137,7 @@ func BenchmarkStressForBasicFuncWithMemoryQueue(b *testing.B) {
 	inputTopic := "test-input-" + strconv.Itoa(rand.Int())
 	outputTopic := "test-output-" + strconv.Itoa(rand.Int())
 
-	replicas := int32(1)
+	replicas := int32(5)
 
 	pConfig := &perf.Config{
 		RequestRate: 200000.0,

clients/gofs/gofs.go

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2024 Function Stream Org.
+ *
+ * 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 gofs
+
+import "C"
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+
+	. "github.com/functionstream/function-stream/common/wasm_utils"
+	"github.com/wirelessr/avroschema"
+)
+
+var processFile *os.File
+
+func init() {
+	processFile, _ = os.Open("/process")
+}
+
+var processFunc func([]byte) []byte
+
+//go:wasmimport fs registerSchema
+func registerSchema(inputSchemaPtrSize, outputSchemaPtrSize uint64)
+
+func Register[I any, O any](process func(*I) *O) error {
+	inputSchema, err := avroschema.Reflect(new(I))
+	if err != nil {
+		return err
+	}
+	outputSchema, err := avroschema.Reflect(new(O))
+	if err != nil {
+		return err
+	}
+	processFunc = func(payload []byte) []byte {
+		input := new(I)
+		err = json.Unmarshal(payload, input)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "Failed to parse JSON: %s %s", err, payload)
+		}
+		output := process(input)
+		outputPayload, _ := json.Marshal(output)
+		return outputPayload
+	}
+	registerSchema(PtrSize(StringToPtr(inputSchema)), PtrSize(StringToPtr(outputSchema)))
+	return nil
+}
+
+//export process
+func process() {
+	payload, _ := io.ReadAll(processFile)
+	outputPayload := processFunc(payload)
+	_, _ = processFile.Write(outputPayload)
+}

cmd/client/reload/cmd.go

@@ -19,10 +19,11 @@ package reload
 import (
 	"context"
 	"fmt"
+	"os"
+
 	adminclient "github.com/functionstream/function-stream/admin/client"
 	"github.com/functionstream/function-stream/cmd/client/common"
 	"github.com/spf13/cobra"
-	"os"
 )
 
 var Cmd = &cobra.Command{

common/wasm_utils/wasm_utils.go

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2024 Function Stream Org.
+ *
+ * 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 wasm_utils
+
+import "unsafe"
+
+// StringToPtr returns a pointer and size pair for the given string in a way
+// compatible with WebAssembly numeric types.
+// The returned pointer aliases the string hence the string must be kept alive
+// until ptr is no longer needed.
+func StringToPtr(s string) (uint32, uint32) {
+	ptr := unsafe.Pointer(unsafe.StringData(s))
+	return uint32(uintptr(ptr)), uint32(len(s))
+}
+
+//// StringToLeakedPtr returns a pointer and size pair for the given string in a way
+//// compatible with WebAssembly numeric types.
+//// The pointer is not automatically managed by TinyGo hence it must be freed by the host.
+//func StringToLeakedPtr(s string) (uint32, uint32) {
+//	size := C.ulong(len(s))
+//	ptr := unsafe.Pointer(C.malloc(size))
+//	copy(unsafe.Slice((*byte)(ptr), size), s)
+//	return uint32(uintptr(ptr)), uint32(size)
+//}
+
+func PtrSize(ptr, size uint32) uint64 {
+	return (uint64(ptr) << 32) | uint64(size)
+}
+
+func ExtractPtrSize(ptrSize uint64) (uint32, uint32) {
+	return uint32(ptrSize >> 32), uint32(ptrSize)
+}

examples/basic/main.go

@@ -17,48 +17,26 @@
 package main
 
 import (
-	"encoding/json"
 	"fmt"
-	"io"
+	"github.com/functionstream/function-stream/clients/gofs"
 	"os"
 )
 
+func main() {
+	_, _ = fmt.Fprintln(os.Stderr, "Hello from Go!")
+}
+
 type Person struct {
 	Name     string `json:"name"`
 	Money    int    `json:"money"`
 	Expected int    `json:"expected"`
 }
 
-func main() {
-	_, _ = fmt.Fprintln(os.Stderr, "Hello from Go!")
+func init() {
+	_ = gofs.Register(myProcess)
 }
 
-//export process
-func process() {
-	dataBytes, err := io.ReadAll(os.Stdin)
-	if err != nil {
-		_, _ = fmt.Fprintln(os.Stderr, "Failed to read data:", err)
-		os.Exit(1)
-	}
-
-	var person Person
-	err = json.Unmarshal(dataBytes, &person)
-	if err != nil {
-		_, _ = fmt.Fprintln(os.Stderr, "Failed to parse JSON:", err)
-		os.Exit(1)
-	}
-
-	person.Money++
-
-	jsonBytes, err := json.Marshal(person)
-	if err != nil {
-		_, _ = fmt.Fprintln(os.Stderr, "Failed to encode JSON:", err)
-		os.Exit(1)
-	}
-
-	_, err = fmt.Printf("%s", jsonBytes)
-	if err != nil {
-		_, _ = fmt.Fprintln(os.Stderr, "Failed to print JSON:", err)
-		os.Exit(1)
-	}
+func myProcess(person *Person) *Person {
+	person.Money += 1
+	return person
 }

fs/contube/contube.go

@@ -19,6 +19,7 @@ package contube
 import (
 	"context"
 	"encoding/json"
+
 	"github.com/go-playground/validator/v10"
 	"github.com/pkg/errors"
 )
@@ -31,6 +32,7 @@ var (
 
 type Record interface {
 	GetPayload() []byte
+	GetSchema() string
 	Commit()
 }
 
@@ -110,6 +112,7 @@ type TubeFactory interface {
 
 type RecordImpl struct {
 	payload    []byte
+	schema     string
 	commitFunc func()
 }
 
@@ -120,10 +123,22 @@ func NewRecordImpl(payload []byte, ackFunc func()) *RecordImpl {
 	}
 }
 
+func NewSchemaRecordImpl(payload []byte, schema string, ackFunc func()) *RecordImpl {
+	return &RecordImpl{
+		payload:    payload,
+		schema:     schema,
+		commitFunc: ackFunc,
+	}
+}
+
 func (e *RecordImpl) GetPayload() []byte {
 	return e.payload
 }
 
+func (e *RecordImpl) GetSchema() string {
+	return e.schema
+}
+
 func (e *RecordImpl) Commit() {
 	if e.commitFunc != nil {
 		e.commitFunc()