Implement stream-based input record decoder's instead of batch-based

columnifier/columnifier.go

@@ -4,9 +4,9 @@ import "io"
 
 // Columnifier is the interface that converts input file to columnar format file.
 type Columnifier interface {
-	io.WriteCloser
-
+	WriteFromReader(reader io.Reader) (int, error)
 	WriteFromFiles(paths []string) (int, error)
+	Close() error
 }
 
 // NewColumnifier creates a new Columnifier.

columnifier/parquet.go

@@ -1,13 +1,16 @@
 package columnifier
 
 import (
+	"io"
 	"io/ioutil"
+	"os"
 
 	"github.com/reproio/columnify/record"
 
 	"github.com/reproio/columnify/parquet"
 	"github.com/reproio/columnify/schema"
 	"github.com/xitongsys/parquet-go-source/local"
+	"github.com/xitongsys/parquet-go/marshal"
 	parquetSource "github.com/xitongsys/parquet-go/source"
 	"github.com/xitongsys/parquet-go/writer"
 )
@@ -65,17 +68,27 @@ func NewParquetColumnifier(st string, sf string, rt string, output string, confi
 }
 
 // Write reads, converts input binary data and write it to buffer.
-func (c *parquetColumnifier) Write(data []byte) (int, error) {
+func (c *parquetColumnifier) WriteFromReader(reader io.Reader) (int, error) {
 	// Intermediate record type is map[string]interface{}
-	c.w.MarshalFunc = parquet.MarshalMap
-	records, err := record.FormatToMap(data, c.schema, c.rt)
+	c.w.MarshalFunc = marshal.MarshalJSON
+	decoder, err := record.NewJsonDecoder(reader, c.schema, c.rt)
 	if err != nil {
 		return -1, err
 	}
 
 	beforeSize := c.w.Size
-	for _, r := range records {
-		if err := c.w.Write(r); err != nil {
+	for {
+		var v string
+		err = decoder.Decode(&v)
+		if err != nil {
+			if err == io.EOF {
+				break
+			} else {
+				return -1, err
+			}
+		}
+
+		if err := c.w.Write(v); err != nil {
 			return -1, err
 		}
 	}
@@ -103,11 +116,12 @@ func (c *parquetColumnifier) WriteFromFiles(paths []string) (int, error) {
 	var n int
 
 	for _, p := range paths {
-		data, err := ioutil.ReadFile(p)
+		f, err := os.Open(p)
 		if err != nil {
 			return -1, err
 		}
-		if n, err = c.Write(data); err != nil {
+
+		if n, err = c.WriteFromReader(f); err != nil {
 			return -1, err
 		}
 	}

record/avro.go

@@ -3,12 +3,48 @@ package record
 import (
 	"bytes"
 	"fmt"
-
-	"github.com/reproio/columnify/schema"
+	"io"
 
 	"github.com/linkedin/goavro/v2"
+	"github.com/reproio/columnify/schema"
 )
 
+type avroInnerDecoder struct {
+	r *goavro.OCFReader
+}
+
+func newAvroInnerDecoder(r io.Reader) (*avroInnerDecoder, error) {
+	reader, err := goavro.NewOCFReader(r)
+	if err != nil {
+		return nil, err
+	}
+
+	return &avroInnerDecoder{
+		r: reader,
+	}, nil
+}
+
+func (d *avroInnerDecoder) Decode(r *map[string]interface{}) error {
+	if d.r.Scan() {
+		v, err := d.r.Read()
+		if err != nil {
+			return err
+		}
+
+		m, mapOk := v.(map[string]interface{})
+		if !mapOk {
+			return fmt.Errorf("invalid value %v: %w", v, ErrUnconvertibleRecord)
+		}
+
+		flatten := flattenAvroUnion(m)
+		*r = flatten
+	} else if d.r.RemainingBlockItems() == 0 {
+		return io.EOF
+	}
+
+	return d.r.Err()
+}
+
 // flattenAvroUnion flattens nested map type has only 1 element.
 func flattenAvroUnion(in map[string]interface{}) map[string]interface{} {
 	out := make(map[string]interface{})

record/avro_test.go

@@ -2,6 +2,7 @@ package record
 
 import (
 	"bytes"
+	"io"
 	"reflect"
 	"testing"
 
@@ -35,11 +36,10 @@ func TestFlattenAvroUnion(t *testing.T) {
 	}
 }
 
-func TestFormatAvroToMap(t *testing.T) {
+func TestAvroInnerDecoder_Decode(t *testing.T) {
 	cases := []struct {
 		input    []byte
 		expected []map[string]interface{}
-		isErr    bool
 	}{
 		{
 			input: func() []byte {
@@ -113,22 +113,29 @@ func TestFormatAvroToMap(t *testing.T) {
 					"string":  "bar",
 				},
 			},
-			isErr: false,
-		},
-
-		// Not avro
-		{
-			input:    []byte("not-valid-avro"),
-			expected: nil,
-			isErr:    true,
 		},
 	}
 
 	for _, c := range cases {
-		actual, err := FormatAvroToMap(c.input)
+		buf := bytes.NewReader(c.input)
+		d, err := newAvroInnerDecoder(buf)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		actual := make([]map[string]interface{}, 0)
+		for {
+			var v map[string]interface{}
+			err = d.Decode(&v)
+			if err != nil {
+				break
+			}
+			actual = append(actual, v)
+		}
 
-		if err != nil != c.isErr {
-			t.Errorf("expected: %v, but actual: %v\n", c.isErr, err)
+		if err != nil && err != io.EOF {
+			t.Errorf("expected no error or io.EOF, but actual: %v\n", err)
+			continue
 		}
 
 		if !reflect.DeepEqual(actual, c.expected) {

record/csv.go

@@ -17,10 +17,71 @@ const (
 	TsvDelimiter delimiter = '\t'
 )
 
+type csvInnerDecoder struct {
+	r     *csv.Reader
+	names []string
+}
+
+func newCsvInnerDecoder(r io.Reader, s *schema.IntermediateSchema, delimiter delimiter) (*csvInnerDecoder, error) {
+	names, err := getFieldNamesFromSchema(s)
+	if err != nil {
+		return nil, err
+	}
+
+	reader := csv.NewReader(r)
+	reader.Comma = rune(delimiter)
+
+	return &csvInnerDecoder{
+		r:     reader,
+		names: names,
+	}, nil
+}
+
+func (d *csvInnerDecoder) Decode(r *map[string]interface{}) error {
+	values, err := d.r.Read()
+	if err != nil {
+		return err
+	}
+
+	if len(d.names) != len(values) {
+		return fmt.Errorf("incompleted value %v: %w", values, ErrUnconvertibleRecord)
+	}
+
+	*r = make(map[string]interface{})
+	for i, v := range values {
+		n := d.names[i]
+
+		// bool
+		if v != "0" && v != "1" {
+			if vv, err := strconv.ParseBool(v); err == nil {
+				(*r)[n] = vv
+				continue
+			}
+		}
+
+		// int
+		if vv, err := strconv.ParseInt(v, 10, 64); err == nil {
+			(*r)[n] = vv
+			continue
+		}
+
+		// float
+		if vv, err := strconv.ParseFloat(v, 64); err == nil {
+			(*r)[n] = vv
+			continue
+		}
+
+		// others; to string
+		(*r)[n] = v
+	}
+
+	return nil
+}
+
 func getFieldNamesFromSchema(s *schema.IntermediateSchema) ([]string, error) {
 	elems := s.ArrowSchema.Fields()
 
-	if len(elems) < 2 {
+	if len(elems) == 0 {
 		return nil, fmt.Errorf("no element is available: %w", ErrUnconvertibleRecord)
 	}
 

record/csv_test.go

@@ -1,14 +1,16 @@
 package record
 
 import (
+	"bytes"
+	"io"
 	"reflect"
 	"testing"
 
 	"github.com/apache/arrow/go/arrow"
 	"github.com/reproio/columnify/schema"
 )
 
-func TestFormatCsvToMap(t *testing.T) {
+func TestCsvInnerDecoder_Decode(t *testing.T) {
 	cases := []struct {
 		schema    *schema.IntermediateSchema
 		input     []byte
@@ -151,36 +153,28 @@ true	2	2	2.2	2.2	bar	bar`),
 			},
 			isErr: false,
 		},
-
-		// Not csv
-		{
-			schema: schema.NewIntermediateSchema(
-				arrow.NewSchema([]arrow.Field{}, nil),
-				"primitives",
-			),
-			input:    []byte("not-valid-csv"),
-			expected: nil,
-			isErr:    true,
-		},
-
-		// Not tsv
-		{
-			schema: schema.NewIntermediateSchema(
-				arrow.NewSchema([]arrow.Field{}, nil),
-				"primitives",
-			),
-			input:     []byte("not-valid-tsv"),
-			delimiter: TsvDelimiter,
-			expected:  nil,
-			isErr:     true,
-		},
 	}
 
 	for _, c := range cases {
-		actual, err := FormatCsvToMap(c.schema, c.input, c.delimiter)
+		buf := bytes.NewReader(c.input)
+		d, err := newCsvInnerDecoder(buf, c.schema, c.delimiter)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		actual := make([]map[string]interface{}, 0)
+		for {
+			var v map[string]interface{}
+			err = d.Decode(&v)
+			if err != nil {
+				break
+			}
+			actual = append(actual, v)
+		}
 
-		if err != nil != c.isErr {
+		if (err != nil && err != io.EOF) != c.isErr {
 			t.Errorf("expected: %v, but actual: %v\n", c.isErr, err)
+			continue
 		}
 
 		if !reflect.DeepEqual(actual, c.expected) {

record/jsonl.go

@@ -1,12 +1,36 @@
 package record
 
 import (
+	"bufio"
 	"encoding/json"
+	"io"
 	"strings"
 
 	"github.com/reproio/columnify/schema"
 )
 
+type jsonlInnerDecoder struct {
+	s *bufio.Scanner
+}
+
+func newJsonlInnerDecoder(r io.Reader) *jsonlInnerDecoder {
+	return &jsonlInnerDecoder{
+		s: bufio.NewScanner(r),
+	}
+}
+
+func (d *jsonlInnerDecoder) Decode(r *map[string]interface{}) error {
+	if d.s.Scan() {
+		if err := json.Unmarshal(d.s.Bytes(), r); err != nil {
+			return err
+		}
+	} else {
+		return io.EOF
+	}
+
+	return d.s.Err()
+}
+
 func FormatJsonlToMap(data []byte) ([]map[string]interface{}, error) {
 	lines := strings.Split(string(data), "\n")