Repository: redpanda-data/connect (language: go)

Issue:
Raw output for jq processor
**Given:**

```json
{"id":"1054fe28-a59e-4878-84f0-43b6fa41c2fb","msg":"sample \"log\" message","ts":1641393111}
```

```jq
[.ts, .id, .msg] | @csv
```

See this pipeline for reproduction: https://studio.benthos.dev/app/view/s/session-2733fe6c-d2b1-43eb-a9e1-b4aabc8f6689/gh-1045.yaml/graph

```yaml
input:
  label: sample_message
  generate:
    mapping: root = {"id":"1054fe28-a59e-4878-84f0-43b6fa41c2fc","msg":"sample \"log\" message","ts":1641393111}
    interval: "0"
    count: 1
pipeline:
  processors:
    - label: to_csv
      jq:
        query: '[.ts,.id,.msg] | @csv'
output:
  stdout:
    codec: lines

```


**Actual output:**

```
"1641393111,\"1054fe28-a59e-4878-84f0-43b6fa41c2fb\",\"sample \"\"log\"\" message\""
```

**Desired output:**

```
1641393111,"1054fe28-a59e-4878-84f0-43b6fa41c2fb","sample ""log"" message"
```

With pure `jq` this is possible with using a `--raw-output` output flag ([playground](https://jqplay.org/s/YBFbo0z3pi))


Discussion / hints:
Okay this is going to look odd and needs some explanation but adding a bloblang mapping `root = this` afterwards will escape the raw string:

```yaml
input:
  label: sample_message
  generate:
    mapping: root = {"id":"1054fe28-a59e-4878-84f0-43b6fa41c2fc","msg":"sample \"log\" message","ts":1641393111}
    interval: "0"
    count: 1
pipeline:
  processors:
    - label: to_csv
      jq:
        query: '[.ts,.id,.msg] | @csv'
    - bloblang: root = this
output:
  stdout:
    codec: lines
```

What happens there is that the result of the `jq` processor is a valid JSON string, which is itself a valid JSON document so bloblang is happy to gobble it up and reference `this` as a string value. When you assign `root` to a string value (or []byte type) the resulting message assumes the raw value (you can think of bloblang as `--raw-output` by default, so you'd need to `this.quote()` to get the same output as jq is giving).

Fix patch:
diff --git a/lib/processor/jq.go b/lib/processor/jq.go
index 1dd41b9855..7178571fde 100644
--- a/lib/processor/jq.go
+++ b/lib/processor/jq.go
@@ -1,6 +1,8 @@
 package processor
 
 import (
+	"bytes"
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -90,6 +92,7 @@ pipeline:
 		FieldSpecs: docs.FieldSpecs{
 			docs.FieldCommon("query", "The jq query to filter and transform messages with."),
 			docs.FieldAdvanced("raw", "Whether to process the input as a raw string instead of as JSON."),
+			docs.FieldAdvanced("output_raw", "Whether to output raw strings instead of JSON texts."),
 		},
 	}
 }
@@ -98,8 +101,9 @@ pipeline:
 
 // JQConfig contains configuration fields for the JQ processor.
 type JQConfig struct {
-	Query string `json:"query" yaml:"query"`
-	Raw   bool   `json:"raw" yaml:"raw"`
+	Query     string `json:"query" yaml:"query"`
+	Raw       bool   `json:"raw" yaml:"raw"`
+	OutputRaw bool   `json:"output_raw" yaml:"output_raw"`
 }
 
 // NewJQConfig returns a JQConfig with default values.
@@ -228,7 +232,22 @@ func (j *JQ) ProcessMessage(msg types.Message) ([]types.Message, types.Response)
 			emitted = append(emitted, out)
 		}
 
-		if len(emitted) > 1 {
+		if j.conf.OutputRaw {
+			raw, err := j.marshalRaw(emitted)
+			if err != nil {
+				j.log.Debugf("%s", err)
+				j.mErr.Incr(1)
+				return false, err
+			}
+
+			if len(raw) == 0 {
+				j.mDroppedParts.Incr(1)
+				return false, nil
+			}
+
+			part.Set(raw)
+			return true, nil
+		} else if len(emitted) > 1 {
 			if err = part.SetJSON(emitted); err != nil {
 				j.log.Debugf("Failed to set part JSON: %v\n", err)
 				j.mErr.Incr(1)
@@ -269,3 +288,30 @@ func (*JQ) CloseAsync() {
 func (*JQ) WaitForClose(timeout time.Duration) error {
 	return nil
 }
+
+func (j *JQ) marshalRaw(values []interface{}) ([]byte, error) {
+	buf := bytes.NewBufferString("")
+
+	for index, el := range values {
+		var rawResult []byte
+
+		val, isString := el.(string)
+		if isString {
+			rawResult = []byte(val)
+		} else {
+			marshalled, err := json.Marshal(el)
+			if err != nil {
+				return nil, fmt.Errorf("Failed marshal JQ result at index %d: %w", index, err)
+			}
+
+			rawResult = marshalled
+		}
+
+		if _, err := buf.Write(rawResult); err != nil {
+			return nil, fmt.Errorf("Failed to write JQ result at index %d: %w", index, err)
+		}
+	}
+
+	bs := buf.Bytes()
+	return bs, nil
+}
diff --git a/website/docs/components/processors/jq.md b/website/docs/components/processors/jq.md
index 0b65fd8e06..3bc7347a0c 100644
--- a/website/docs/components/processors/jq.md
+++ b/website/docs/components/processors/jq.md
@@ -42,6 +42,7 @@ label: ""
 jq:
   query: .
   raw: false
+  output_raw: false
 ```
 
 </TabItem>
@@ -90,6 +91,14 @@ Default: `"."`
 Whether to process the input as a raw string instead of as JSON.
 
 
+Type: `bool`  
+Default: `false`  
+
+### `output_raw`
+
+Whether to output raw strings instead of JSON texts.
+
+
 Type: `bool`  
 Default: `false`  
 


Test patch:
diff --git a/lib/processor/jq_test.go b/lib/processor/jq_test.go
index 7979885070..585572cef5 100644
--- a/lib/processor/jq_test.go
+++ b/lib/processor/jq_test.go
@@ -121,12 +121,98 @@ func TestJQ(t *testing.T) {
 			input:  `{"foo":{"bar":true}}`,
 			output: `true`,
 		},
+		{
+			name:   "convert to csv",
+			path:   "[.ts,.id,.msg] | @csv",
+			input:  `{"id":"1054fe28","msg":"sample \"log\"","ts":1641393111}`,
+			output: `"1641393111,\"1054fe28\",\"sample \"\"log\"\"\""`,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			conf := NewConfig()
+			conf.JQ.Query = test.path
+
+			jSet, err := NewJQ(conf, nil, log.Noop(), metrics.Noop())
+			require.NoError(t, err)
+
+			inMsg := message.New(
+				[][]byte{
+					[]byte(test.input),
+				},
+			)
+			msgs, _ := jSet.ProcessMessage(inMsg)
+			require.Len(t, msgs, 1)
+			assert.Equal(t, test.output, string(message.GetAllBytes(msgs[0])[0]))
+		})
+	}
+}
+
+func TestJQ_OutputRaw(t *testing.T) {
+	type jTest struct {
+		name   string
+		path   string
+		input  string
+		output string
+	}
+
+	tests := []jTest{
+		{
+			name:   "select obj",
+			path:   ".foo.bar",
+			input:  `{"foo":{"bar":{"baz":1}}}`,
+			output: `{"baz":1}`,
+		},
+		{
+			name:   "select array",
+			path:   ".foo.bar",
+			input:  `{"foo":{"bar":["baz","qux"]}}`,
+			output: `["baz","qux"]`,
+		},
+		{
+			name:   "select obj as str",
+			path:   ".foo.bar",
+			input:  `{"foo":{"bar":"{\"baz\":1}"}}`,
+			output: `{"baz":1}`,
+		},
+		{
+			name:   "select str",
+			path:   ".foo.bar",
+			input:  `{"foo":{"bar":"hello world"}}`,
+			output: `hello world`,
+		},
+		{
+			name:   "select float",
+			path:   ".foo.bar",
+			input:  `{"foo":{"bar":0.123}}`,
+			output: `0.123`,
+		},
+		{
+			name:   "select int",
+			path:   ".foo.bar",
+			input:  `{"foo":{"bar":123}}`,
+			output: `123`,
+		},
+		{
+			name:   "select bool",
+			path:   ".foo.bar",
+			input:  `{"foo":{"bar":true}}`,
+			output: `true`,
+		},
+		{
+			name:   "convert to csv",
+			path:   "[.ts,.id,.msg] | @csv",
+			input:  `{"id":"1054fe28","msg":"sample \"log\"","ts":1641393111}`,
+			output: `1641393111,"1054fe28","sample ""log"""`,
+		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			conf := NewConfig()
 			conf.JQ.Query = test.path
+			conf.JQ.OutputRaw = true
 
 			jSet, err := NewJQ(conf, nil, log.Noop(), metrics.Noop())
 			require.NoError(t, err)