Repository: segmentio/parquet-go
Pull request #455 (2022-12-12 23:35:56)

Add JSON Support on all fields
Toward resolving #423 

I have proof this works in this PR... but given my lack of familiarity with the lib, @achille-roussel I suspect you can point out where the dragons lay.

parsnips: @achille-roussel I think this is ready for review.
ryandeivert: @parsnips / @achille-roussel any updates on this PR? I'm interested in this functionality, but not entire sure the current work solves for my use-case. 

I have a `map[string]interface{}` that I'd like to apply this to, and given the way #423 is outlined, I think this should be possible.

I currently have a "sanitization" step that converts `map[string]interface{}` to `map[string]string`, where the sanitized string value is the json representation of the interface (for non-string values).

I have to hard-code this sanitization for all map types now, which is very inflexible. Essentially I would like to do away with this step and have the serializer do it for me.

I have my project pointing to @parsnips feature branch (`parsnips:parsnips/full-json`), but can't seem to get a basic example using `structpb.Value` working.

Any thoughts?

Basic example of what I currently do:

```golang
type Example struct {
	FooData   map[string]interface{} `json:"fooData" parquet:"-"`
	PqFooData map[string]string      `parquet:"foodata,optional"`
	BarData   map[string]interface{} `json:"barData" parquet:"-"`
	PqBarData map[string]string      `parquet:"bardata,optional"`
}

func (c *Example) sanitize() {
	jsonCast(c.FooData, &c.PqFooData)
	jsonCast(c.BarData, &c.PqBarData)
}

func jsonCast(source map[string]interface{}, dest *map[string]string) {
	*dest = make(map[string]string)
	for k, v := range source {
		var cast string
		switch i := v.(type) {
		case string:
			cast = i // already string, do not marshal
		default:
			b, err := json.Marshal(i)
			if err != nil {
				log.Printf("could not convert to json: %s (source: %s)", err, source)
				continue
			}
			cast = string(b)
		}
		(*dest)[k] = cast
	}
}

// fileToParquet opens a file for reading and writes Example structs to parquet
func fileToParquet(file string, writer *parquet.GenericWriter[Example]) {
	in, err := os.Open(file)
	if err != nil {
		log.Fatal(err)
	}
	defer in.Close()

        // readFile returns a channel that yields Example structs
	for ex := range readFile(in) {
		ex.sanitize() // prefer to not do this
		_, err = writer.Write([]Example{ex})
		if err != nil {
			log.Printf("failed to write object to parquet: %s", err)
		}
	}
}
```
ryandeivert: update: I was able to solve for the above with a custom type + json marshaler at load time, but would prefer to have a more extensible way to do this if you have any suggestions. thanks for providing a great package 🥇 
parsnips: @ryandeivert I have a repro of this - funnily enough the entire point of writing this was to encode `structpb.Value` as json 😂  I should have a fix today.
parsnips: @ryandeivert This should work for you out of the box now.
parsnips: @bartleyg can you kick off the tests again?
ryandeivert: thanks @parsnips! I may wait for it to merge here to test out, since I've got something _functional_ for the time being. thanks for the great addition!

Patch:
diff --git a/column_buffer_go18.go b/column_buffer_go18.go
index 539392c..1d4385c 100644
--- a/column_buffer_go18.go
+++ b/column_buffer_go18.go
@@ -3,6 +3,7 @@
 package parquet
 
 import (
+	"encoding/json"
 	"math/bits"
 	"reflect"
 	"time"
@@ -28,6 +29,10 @@ type writeRowsFunc func(columns []ColumnBuffer, rows sparse.Array, levels column
 // parquet schema. The column path indicates the column that the function is
 // being generated for in the parquet schema.
 func writeRowsFuncOf(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
+	if leaf, exists := schema.Lookup(path...); exists && leaf.Node.Type().LogicalType() != nil && leaf.Node.Type().LogicalType().Json != nil {
+		return writeRowsFuncOfJSON(t, schema, path)
+	}
+
 	switch t {
 	case reflect.TypeOf(deprecated.Int96{}):
 		return writeRowsFuncOfRequired(t, schema, path)
@@ -397,6 +402,44 @@ func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRo
 	}
 }
 
+func writeRowsFuncOfJSON(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
+	// If this is a string or a byte array write directly.
+	switch t.Kind() {
+	case reflect.String:
+		return writeRowsFuncOfRequired(t, schema, path)
+	case reflect.Slice:
+		if t.Elem().Kind() == reflect.Uint8 {
+			return writeRowsFuncOfRequired(t, schema, path)
+		}
+	}
+
+	// Otherwise handle with a json.Marshal
+	asStrT := reflect.TypeOf(string(""))
+	writer := writeRowsFuncOfRequired(asStrT, schema, path)
+
+	return func(columns []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
+		if rows.Len() == 0 {
+			return writer(columns, rows, levels)
+		}
+		for i := 0; i < rows.Len(); i++ {
+			val := reflect.NewAt(t, rows.Index(i))
+			asI := val.Interface()
+
+			b, err := json.Marshal(asI)
+			if err != nil {
+				return err
+			}
+
+			asStr := string(b)
+			a := sparse.MakeStringArray([]string{asStr})
+			if err := writer(columns, a.UnsafeArray(), levels); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
 func writeRowsFuncOfTime(_ reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
 	t := reflect.TypeOf(int64(0))
 	elemSize := uintptr(t.Size())
diff --git a/go.mod b/go.mod
index 9804d55..56ebdae 100644
--- a/go.mod
+++ b/go.mod
@@ -11,6 +11,7 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.9
 	github.com/segmentio/encoding v0.3.5
 	golang.org/x/sys v0.0.0-20211110154304-99a53858aa08
+	google.golang.org/protobuf v1.30.0
 )
 
 require github.com/mattn/go-runewidth v0.0.9 // indirect
diff --git a/go.sum b/go.sum
index 5109074..69e2f68 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,8 @@
 github.com/andybalholm/brotli v1.0.3 h1:fpcw+r1N1h0Poc1F/pHbW40cUm/lMEQslZtCkBQ0UnM=
 github.com/andybalholm/brotli v1.0.3/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
@@ -17,3 +20,8 @@ github.com/segmentio/encoding v0.3.5 h1:UZEiaZ55nlXGDL92scoVuw00RmiRCazIEmvPSbSv
 github.com/segmentio/encoding v0.3.5/go.mod h1:n0JeuIqEQrQoPDGsjo8UNd1iA0U8d8+oHAA4E3G3OxM=
 golang.org/x/sys v0.0.0-20211110154304-99a53858aa08 h1:WecRHqgE09JBkh/584XIE6PMz5KKE/vER4izNUi30AQ=
 golang.org/x/sys v0.0.0-20211110154304-99a53858aa08/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
+google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
diff --git a/schema.go b/schema.go
index a62e43f..d6d2845 100644
--- a/schema.go
+++ b/schema.go
@@ -548,14 +548,18 @@ func nodeOf(t reflect.Type, tag []string) Node {
 			}
 		}
 
-		n = Map(
-			makeNodeOf(t.Key(), t.Name(), []string{keyTag}),
-			makeNodeOf(t.Elem(), t.Name(), []string{valueTag}),
-		)
+		if strings.Contains(mapTag, "json") {
+			n = JSON()
+		} else {
+			n = Map(
+				makeNodeOf(t.Key(), t.Name(), []string{keyTag}),
+				makeNodeOf(t.Elem(), t.Name(), []string{valueTag}),
+			)
+		}
 
 		forEachTagOption([]string{mapTag}, func(option, args string) {
 			switch option {
-			case "":
+			case "", "json":
 				return
 			case "optional":
 				n = Optional(n)
diff --git a/type.go b/type.go
index a100c67..cfa8b53 100644
--- a/type.go
+++ b/type.go
@@ -2,6 +2,7 @@ package parquet
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 	"math/bits"
 	"reflect"
@@ -1579,7 +1580,25 @@ func (t *jsonType) EstimateDecodeSize(numValues int, src []byte, enc encoding.En
 }
 
 func (t *jsonType) AssignValue(dst reflect.Value, src Value) error {
-	return byteArrayType{}.AssignValue(dst, src)
+	// Assign value using ByteArrayType for BC...
+	switch dst.Kind() {
+	case reflect.String:
+		return byteArrayType{}.AssignValue(dst, src)
+	case reflect.Slice:
+		if dst.Type().Elem().Kind() == reflect.Uint8 {
+			return byteArrayType{}.AssignValue(dst, src)
+		}
+	}
+
+	// Otherwise handle with json.Unmarshal
+	b := src.byteArray()
+	val := reflect.New(dst.Type()).Elem()
+	err := json.Unmarshal(b, val.Addr().Interface())
+	if err != nil {
+		return err
+	}
+	dst.Set(val)
+	return nil
 }
 
 func (t *jsonType) ConvertValue(val Value, typ Type) (Value, error) {


Test patch:
diff --git a/parquet_go18_test.go b/parquet_go18_test.go
index 840d94d..fbb98eb 100644
--- a/parquet_go18_test.go
+++ b/parquet_go18_test.go
@@ -12,6 +12,7 @@ import (
 	"testing"
 
 	"github.com/segmentio/parquet-go"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 func ExampleReadFile() {
@@ -326,6 +327,85 @@ func TestIssue377(t *testing.T) {
 	assertRowsEqual(t, rows, ods)
 }
 
+func TestIssue423(t *testing.T) {
+	type Inner struct {
+		Value string `parquet:","`
+	}
+	type Outer struct {
+		Label string  `parquet:","`
+		Inner Inner   `parquet:",json"`
+		Slice []Inner `parquet:",json"`
+		// This is the only tricky situation. Because we're delegating to json Marshaler/Unmarshaler
+		// We use the json tags for optionality.
+		Ptr *Inner `json:",omitempty" parquet:",json"`
+
+		// This tests BC behavior that slices of bytes and json strings still get written/read in a BC way.
+		String        string                     `parquet:",json"`
+		Bytes         []byte                     `parquet:",json"`
+		MapOfStructPb map[string]*structpb.Value `parquet:",json"`
+		StructPB      *structpb.Value            `parquet:",json"`
+	}
+
+	writeRows := []Outer{
+		{
+			Label: "welp",
+			Inner: Inner{
+				Value: "this is a string",
+			},
+			Slice: []Inner{
+				{
+					Value: "in a slice",
+				},
+			},
+			Ptr:    nil,
+			String: `{"hello":"world"}`,
+			Bytes:  []byte(`{"goodbye":"world"}`),
+			MapOfStructPb: map[string]*structpb.Value{
+				"answer": structpb.NewNumberValue(42.00),
+			},
+			StructPB: structpb.NewBoolValue(true),
+		},
+		{
+			Label: "foxes",
+			Inner: Inner{
+				Value: "the quick brown fox jumped over the yellow lazy dog.",
+			},
+			Slice: []Inner{
+				{
+					Value: "in a slice",
+				},
+			},
+			Ptr: &Inner{
+				Value: "not nil",
+			},
+			String: `{"hello":"world"}`,
+			Bytes:  []byte(`{"goodbye":"world"}`),
+			MapOfStructPb: map[string]*structpb.Value{
+				"doubleAnswer": structpb.NewNumberValue(84.00),
+			},
+			StructPB: structpb.NewBoolValue(false),
+		},
+	}
+
+	schema := parquet.SchemaOf(new(Outer))
+	fmt.Println(schema.String())
+	buf := new(bytes.Buffer)
+	w := parquet.NewGenericWriter[Outer](buf, schema)
+	_, err := w.Write(writeRows)
+	if err != nil {
+		t.Fatal("write error: ", err)
+	}
+	w.Close()
+
+	file := bytes.NewReader(buf.Bytes())
+	readRows, err := parquet.Read[Outer](file, file.Size())
+	if err != nil {
+		t.Fatal("read error: ", err)
+	}
+
+	assertRowsEqual(t, writeRows, readRows)
+}
+
 func TestReadFileGenericMultipleRowGroupsMultiplePages(t *testing.T) {
 	type MyRow struct {
 		ID    [16]byte `parquet:"id,delta,uuid"`