Implement Parquet Support For Egress Pipeline Output

internal/encoding/block/block_test.go

@@ -22,7 +22,7 @@ func TestBlock_Types(t *testing.T) {
 	assert.NoError(t, err)
 
 	schema := b[0].Schema()
-	assert.Equal(t, 8, len(schema))
+	assert.Equal(t, 9, len(schema))
 	assert.Contains(t, schema, "boolean1")
 	assert.Contains(t, schema, "double1")
 	assert.Contains(t, schema, "int1")

internal/encoding/block/from_orc.go

@@ -147,5 +147,6 @@ func convertToString(value interface{}) (string, bool) {
 	if ok {
 		return strconv.FormatInt(valueInt, 10), true
 	}
+
 	return "", false
 }

internal/encoding/block/from_parquet.go

@@ -1,6 +1,7 @@
 package block
 
 import (
+	"fmt"
 	"github.com/kelindar/talaria/internal/column"
 	"github.com/kelindar/talaria/internal/encoding/parquet"
 	"github.com/kelindar/talaria/internal/encoding/typeof"
@@ -65,13 +66,19 @@ func FromParquetBy(payload []byte, partitionBy string, filter *typeof.Schema, ap
 			columnName := cols[i]
 			columnType := schema[columnName]
 
+			fieldHandler, _ := getReverseHandler(columnType.String())
+
 			// Encode to JSON
 			if columnType == typeof.JSON {
 				if encoded, ok := convertToJSON(v); ok {
 					v = encoded
 				}
 			}
 
+			if fieldHandler != nil {
+				v, _ = fieldHandler(v)
+			}
+
 			row.Set(columnName, v)
 		}
 
@@ -92,3 +99,49 @@ func FromParquetBy(payload []byte, partitionBy string, filter *typeof.Schema, ap
 	blocks = append(blocks, last...)
 	return blocks, nil
 }
+
+type fieldHandler func(interface{}) (interface{}, error)
+
+func getReverseHandler(typ string) (fieldHandler func(interface{}) (interface{}, error), err error) {
+
+	switch typ {
+	case "string":
+		fieldHandler = stringHandler
+	}
+
+	return fieldHandler, nil
+}
+
+// Transform runs the computed Values and overwrites/appends them to the set.
+func ApplyCastHandlers(r Row, fieldHandlers []fieldHandler) Row {
+	// Create a new output row and copy the column values from the input
+	schema := make(typeof.Schema, len(r.Schema))
+	out := NewRow(schema, len(r.Values))
+
+	i := 0
+	for k, v := range r.Values {
+		handler := fieldHandlers[i]
+
+		if handler != nil {
+			out.Values[k], _ = handler(v)
+		}
+		out.Schema[k] = r.Schema[k]
+	}
+
+	return out
+}
+
+func stringHandler(s interface{}) (interface{}, error) {
+
+	switch s.(type) {
+	case []byte:
+		buf, ok := s.([]byte)
+		if !ok {
+			return nil, fmt.Errorf("Failed to get bytes from the interface")
+		}
+
+		return string(buf), nil
+	}
+
+	return nil, nil
+}

internal/encoding/merge/merge.go

@@ -18,6 +18,8 @@ func New(mergeFunc string) (Func, error) {
 	switch strings.ToLower(mergeFunc) {
 	case "orc":
 		return ToOrc, nil
+	case "parquet":
+		return ToParquet, nil
 	case "": // Default to "orc" so we don't break existing configs
 		return ToOrc, nil
 	}

internal/encoding/merge/merge_test.go

@@ -12,6 +12,7 @@ import (
 // BenchmarkFlush runs a benchmark for a Merge function for flushing
 // To run it, go in the directory and do 'go test -benchmem -bench=. -benchtime=1s'
 // BenchmarkMerge/orc-8         	       1	7195029600 ns/op	2101578032 B/op	36859501 allocs/op
+// BenchmarkMerge/parquet-12     	       1	18666411036 ns/op	5142058320 B/op	115850463 allocs/op
 func BenchmarkMerge(b *testing.B) {
 
 	// Append some files
@@ -27,6 +28,15 @@ func BenchmarkMerge(b *testing.B) {
 			ToOrc(blocks, blocks[0].Schema())
 		}
 	})
+
+	// Run the actual benchmark
+	b.Run("parquet", func(b *testing.B) {
+		b.ResetTimer()
+		b.ReportAllocs()
+		for n := 0; n < b.N; n++ {
+			ToParquet(blocks, blocks[0].Schema())
+		}
+	})
 }
 
 func TestMergeNew(t *testing.T) {