ARROW-3626: [Go] implement CSV reader

needs #2871

Author: Sebastien Binet <binet@cern.ch>

Closes #2872 from sbinet/issue-3626 and squashes the following commits:

72d9d9d <Sebastien Binet> ARROW-3626:  implement CSV reader
9dfda70 <Sebastien Binet> ARROW-3625:  add examples for Record and Table
f2bf762 <Sebastien Binet> ARROW-3627:  implement Record builder
8367d64 <Sebastien Binet>  make sure Boolean implements array.Interface
c055186 <Sebastien Binet>  add Release, Reserve and Resize to array.Builder interface
4aaff04 <Sebastien Binet>  test allocator interface

Author: sbinet

go/arrow/array/boolean.go

@@ -72,3 +72,7 @@ func (a *Boolean) setData(data *Data) {
 		a.values = vals.Bytes()
 	}
 }
+
+var (
+	_ Interface = (*Boolean)(nil)
+)

go/arrow/array/builder.go

@@ -31,6 +31,10 @@ const (
 
 // Builder provides an interface to build arrow arrays.
 type Builder interface {
+	// Retain increases the reference count by 1.
+	// Retain may be called simultaneously from multiple goroutines.
+	Retain()
+
 	// Release decreases the reference count by 1.
 	Release()
 
@@ -47,6 +51,14 @@ type Builder interface {
 	// AppendNull adds a new null value to the array being built.
 	AppendNull()
 
+	// Reserve ensures there is enough space for appending n elements
+	// by checking the capacity and calling Resize if necessary.
+	Reserve(n int)
+
+	// Resize adjusts the space allocated by b to n elements. If n is greater than b.Cap(),
+	// additional memory will be allocated. If n is smaller, the allocated memory may reduced.
+	Resize(n int)
+
 	// NewArray creates a new array from the memory buffers used
 	// by the builder and resets the Builder so it can be used to build
 	// a new array.

go/arrow/array/null.go

@@ -86,10 +86,13 @@ func (b *NullBuilder) AppendNull() {
 	b.builder.nulls++
 }
 
+func (*NullBuilder) Reserve(size int) {}
+func (*NullBuilder) Resize(size int)  {}
+
 func (*NullBuilder) init(cap int)                       {}
 func (*NullBuilder) resize(newBits int, init func(int)) {}
 
-// NewArray creates a List array from the memory buffers used by the builder and resets the NullBuilder
+// NewArray creates a Null array from the memory buffers used by the builder and resets the NullBuilder
 // so it can be used to build a new array.
 func (b *NullBuilder) NewArray() Interface {
 	return b.NewNullArray()

go/arrow/array/record.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/apache/arrow/go/arrow"
 	"github.com/apache/arrow/go/arrow/internal/debug"
+	"github.com/apache/arrow/go/arrow/memory"
 )
 
 // RecordReader reads a stream of records.
@@ -115,6 +116,8 @@ type Record interface {
 
 	NumRows() int64
 	NumCols() int64
+
+	Columns() []Interface
 	Column(i int) Interface
 	ColumnName(i int) string
 
@@ -217,6 +220,7 @@ func (rec *simpleRecord) Release() {
 func (rec *simpleRecord) Schema() *arrow.Schema   { return rec.schema }
 func (rec *simpleRecord) NumRows() int64          { return rec.rows }
 func (rec *simpleRecord) NumCols() int64          { return int64(len(rec.arrs)) }
+func (rec *simpleRecord) Columns() []Interface    { return rec.arrs }
 func (rec *simpleRecord) Column(i int) Interface  { return rec.arrs[i] }
 func (rec *simpleRecord) ColumnName(i int) string { return rec.schema.Field(i).Name }
 
@@ -239,6 +243,91 @@ func (rec *simpleRecord) NewSlice(i, j int64) Record {
 	return NewRecord(rec.schema, arrs, j-i)
 }
 
+// RecordBuilder eases the process of building a Record, iteratively, from
+// a known Schema.
+type RecordBuilder struct {
+	refCount int64
+	mem      memory.Allocator
+	schema   *arrow.Schema
+	fields   []Builder
+}
+
+// NewRecordBuilder returns a builder, using the provided memory allocator and a schema.
+func NewRecordBuilder(mem memory.Allocator, schema *arrow.Schema) *RecordBuilder {
+	b := &RecordBuilder{
+		refCount: 1,
+		mem:      mem,
+		schema:   schema,
+		fields:   make([]Builder, len(schema.Fields())),
+	}
+
+	for i, f := range schema.Fields() {
+		b.fields[i] = newBuilder(b.mem, f.Type)
+	}
+
+	return b
+}
+
+// Retain increases the reference count by 1.
+// Retain may be called simultaneously from multiple goroutines.
+func (b *RecordBuilder) Retain() {
+	atomic.AddInt64(&b.refCount, 1)
+}
+
+// Release decreases the reference count by 1.
+func (b *RecordBuilder) Release() {
+	debug.Assert(atomic.LoadInt64(&b.refCount) > 0, "too many releases")
+
+	for _, f := range b.fields {
+		f.Release()
+	}
+
+	if atomic.AddInt64(&b.refCount, -1) == 0 {
+		b.fields = nil
+	}
+}
+
+func (b *RecordBuilder) Schema() *arrow.Schema { return b.schema }
+func (b *RecordBuilder) Fields() []Builder     { return b.fields }
+func (b *RecordBuilder) Field(i int) Builder   { return b.fields[i] }
+
+func (b *RecordBuilder) Reserve(size int) {
+	for _, f := range b.fields {
+		f.Reserve(size)
+	}
+}
+
+// NewRecord creates a new record from the memory buffers and resets the
+// RecordBuilder so it can be used to build a new record.
+//
+// The returned Record must be Release()'d after use.
+//
+// NewRecord panics if the fields' builder do not have the same length.
+func (b *RecordBuilder) NewRecord() Record {
+	cols := make([]Interface, len(b.fields))
+	rows := int64(0)
+
+	defer func(cols []Interface) {
+		for _, col := range cols {
+			if col == nil {
+				continue
+			}
+			col.Release()
+		}
+	}(cols)
+
+	for i, f := range b.fields {
+		cols[i] = f.NewArray()
+		irow := int64(cols[i].Len())
+		if i > 0 && irow != rows {
+			panic(fmt.Errorf("arrow/array: field %d has %d rows. want=%d", i, irow, rows))
+		}
+		rows = irow
+	}
+
+	return NewRecord(b.schema, cols, rows)
+}
+
 var (
 	_ Record       = (*simpleRecord)(nil)
 	_ RecordReader = (*simpleRecords)(nil)

go/arrow/array/record_test.go

@@ -72,6 +72,9 @@ func TestRecord(t *testing.T) {
 	if got, want := rec.NumCols(), int64(2); got != want {
 		t.Fatalf("invalid number of columns: got=%d, want=%d", got, want)
 	}
+	if got, want := rec.Columns()[0], cols[0]; got != want {
+		t.Fatalf("invalid column: got=%q, want=%q", got, want)
+	}
 	if got, want := rec.Column(0), cols[0]; got != want {
 		t.Fatalf("invalid column: got=%q, want=%q", got, want)
 	}
@@ -345,3 +348,43 @@ func TestRecordReader(t *testing.T) {
 		})
 	}
 }
+
+func TestRecordBuilder(t *testing.T) {
+	mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+	defer mem.AssertSize(t, 0)
+
+	schema := arrow.NewSchema(
+		[]arrow.Field{
+			arrow.Field{Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32},
+			arrow.Field{Name: "f2-f64", Type: arrow.PrimitiveTypes.Float64},
+		},
+		nil,
+	)
+
+	b := array.NewRecordBuilder(mem, schema)
+	defer b.Release()
+
+	b.Retain()
+	b.Release()
+
+	b.Field(0).(*array.Int32Builder).AppendValues([]int32{1, 2, 3, 4, 5, 6}, nil)
+	b.Field(0).(*array.Int32Builder).AppendValues([]int32{7, 8, 9, 10}, nil)
+	b.Field(1).(*array.Float64Builder).AppendValues([]float64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)
+
+	rec := b.NewRecord()
+	defer rec.Release()
+
+	if got, want := rec.Schema(), schema; !got.Equal(want) {
+		t.Fatalf("invalid schema: got=%#v, want=%#v", got, want)
+	}
+
+	if got, want := rec.NumRows(), int64(10); got != want {
+		t.Fatalf("invalid number of rows: got=%d, want=%d", got, want)
+	}
+	if got, want := rec.NumCols(), int64(2); got != want {
+		t.Fatalf("invalid number of columns: got=%d, want=%d", got, want)
+	}
+	if got, want := rec.ColumnName(0), schema.Field(0).Name; got != want {
+		t.Fatalf("invalid column name: got=%q, want=%q", got, want)
+	}
+}

go/arrow/array/table.go

@@ -246,6 +246,32 @@ func NewTable(schema *arrow.Schema, cols []Column, rows int64) *simpleTable {
 	return &tbl
 }
 
+// NewTableFromRecords returns a new basic, non-lazy in-memory table.
+//
+// NewTableFromRecords panics if the records and schema are inconsistent.
+func NewTableFromRecords(schema *arrow.Schema, recs []Record) *simpleTable {
+	arrs := make([]Interface, len(recs))
+	cols := make([]Column, len(schema.Fields()))
+
+	defer func(cols []Column) {
+		for i := range cols {
+			cols[i].Release()
+		}
+	}(cols)
+
+	for i := range cols {
+		field := schema.Field(i)
+		for j, rec := range recs {
+			arrs[j] = rec.Column(i)
+		}
+		chunk := NewChunked(field.Type, arrs)
+		cols[i] = *NewColumn(field, chunk)
+		chunk.Release()
+	}
+
+	return NewTable(schema, cols, -1)
+}
+
 func (tbl *simpleTable) Schema() *arrow.Schema { return tbl.schema }
 func (tbl *simpleTable) NumRows() int64        { return tbl.rows }
 func (tbl *simpleTable) NumCols() int64        { return int64(len(tbl.cols)) }

go/arrow/array/table_test.go

@@ -556,6 +556,52 @@ func TestTable(t *testing.T) {
 	}
 }
 
+func TestTableFromRecords(t *testing.T) {
+	mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+	defer mem.AssertSize(t, 0)
+
+	schema := arrow.NewSchema(
+		[]arrow.Field{
+			arrow.Field{Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32},
+			arrow.Field{Name: "f2-f64", Type: arrow.PrimitiveTypes.Float64},
+		},
+		nil,
+	)
+
+	b := array.NewRecordBuilder(mem, schema)
+	defer b.Release()
+
+	b.Field(0).(*array.Int32Builder).AppendValues([]int32{1, 2, 3, 4, 5, 6}, nil)
+	b.Field(0).(*array.Int32Builder).AppendValues([]int32{7, 8, 9, 10}, []bool{true, true, false, true})
+	b.Field(1).(*array.Float64Builder).AppendValues([]float64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)
+
+	rec1 := b.NewRecord()
+	defer rec1.Release()
+
+	b.Field(0).(*array.Int32Builder).AppendValues([]int32{11, 12, 13, 14, 15, 16, 17, 18, 19, 20}, nil)
+	b.Field(1).(*array.Float64Builder).AppendValues([]float64{11, 12, 13, 14, 15, 16, 17, 18, 19, 20}, nil)
+
+	rec2 := b.NewRecord()
+	defer rec2.Release()
+
+	tbl := array.NewTableFromRecords(schema, []array.Record{rec1, rec2})
+	defer tbl.Release()
+
+	if got, want := tbl.Schema(), schema; !got.Equal(want) {
+		t.Fatalf("invalid schema: got=%#v, want=%#v", got, want)
+	}
+
+	if got, want := tbl.NumRows(), int64(20); got != want {
+		t.Fatalf("invalid number of rows: got=%d, want=%d", got, want)
+	}
+	if got, want := tbl.NumCols(), int64(2); got != want {
+		t.Fatalf("invalid number of columns: got=%d, want=%d", got, want)
+	}
+	if got, want := tbl.Column(0).Name(), schema.Field(0).Name; got != want {
+		t.Fatalf("invalid column: got=%q, want=%q", got, want)
+	}
+}
+
 func TestTableReader(t *testing.T) {
 	mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
 	defer mem.AssertSize(t, 0)