Repository: snowflakedb/gosnowflake
Pull request #544 (2022-02-08 02:00:17)

Arrow Record Distributed Result Batches
### Description
For performance and ease-of-use purposes, users may need to access data in `arrow.Record` format directly from queries, rather than getting it in `sql.Rows` format and having to covert it manually. This PR covers the implementation and tests to get these records via a new API endpoint called `GetResultBatches`. We use `ResultBatch` objects that can independently fetch chunks of data in `arrow.Record` format. `GetResultBatches` returns such objects in the correct order as to allow the user to compose, in parallel or not, the full data in Arrow format. Below is a sample client request to fetch the data:

```
rbCtx := sf.WithDistributedResultBatches(ctx)
d := sf.SnowflakeDriver{}
db, err := d.Open(dsn)
if err != nil {
	return nil, err
}
snowflakeDb := db.(sf.SnowflakeConnection)

rows, err := snowflakeDb.QueryContext(rbCtx, query, []driver.NamedValue{})
if err != nil {
	return nil, err
}

batches, err := rows.(sf.SnowflakeRows).GetBatches()
if err != nil {
	return nil, err
}
```

For a client using data in arrow.Record format, this can result in more than a 10x speedup in retrieving usable data, as well as provides the client with the ability to manage memory consumption independently.

### Checklist
- [x] Code compiles correctly
- [x] Run ``make fmt`` to fix inconsistent formats
- [x] Run ``make lint`` to get lint errors and fix all of them
- [x] Created tests which fail without the change (if possible)
- [x] All tests passing
- [ ] Extended the README / documentation, if necessary


sfc-gh-jbahk: Local tests pass successfully. Merging.

Patch:
diff --git a/arrow_chunk.go b/arrow_chunk.go
index b9f043b..2ef2170 100644
--- a/arrow_chunk.go
+++ b/arrow_chunk.go
@@ -7,6 +7,7 @@ import (
 	"encoding/base64"
 	"io"
 
+	"github.com/apache/arrow/go/arrow/array"
 	"github.com/apache/arrow/go/arrow/ipc"
 	"github.com/apache/arrow/go/arrow/memory"
 )
@@ -52,6 +53,27 @@ func (arc *arrowResultChunk) decodeArrowChunk(rowType []execResponseRowType, hig
 	}
 }
 
+func (arc *arrowResultChunk) decodeArrowBatch(scd *snowflakeChunkDownloader) (*[]array.Record, error) {
+	var records []array.Record
+
+	for {
+		rawRecord, err := arc.reader.Read()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			return nil, err
+		}
+		record, err := arrowToRecord(rawRecord, scd.RowSet.RowType)
+		rawRecord.Release()
+		if err != nil {
+			return nil, err
+		}
+		record.Retain()
+		records = append(records, record)
+	}
+	return &records, nil
+}
+
 // Build arrow chunk based on RowSet of base64
 func buildFirstArrowChunk(rowsetBase64 string) arrowResultChunk {
 	rowSetBytes, err := base64.StdEncoding.DecodeString(rowsetBase64)
diff --git a/chunk_downloader.go b/chunk_downloader.go
index 642a2b5..56c32e2 100644
--- a/chunk_downloader.go
+++ b/chunk_downloader.go
@@ -18,6 +18,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/apache/arrow/go/arrow/array"
 	"github.com/apache/arrow/go/arrow/ipc"
 	"github.com/apache/arrow/go/arrow/memory"
 )
@@ -34,6 +35,7 @@ type chunkDownloader interface {
 	getRowType() []execResponseRowType
 	setNextChunkDownloader(downloader chunkDownloader)
 	getNextChunkDownloader() chunkDownloader
+	getArrowBatches() []*ArrowBatch
 }
 
 type snowflakeChunkDownloader struct {
@@ -55,9 +57,11 @@ type snowflakeChunkDownloader struct {
 	ChunksFinalErrors  []*chunkError
 	ChunksMutex        *sync.Mutex
 	DoneDownloadCond   *sync.Cond
+	FirstBatch         *ArrowBatch
 	NextDownloader     chunkDownloader
 	Qrmk               string
 	QueryResultFormat  string
+	ArrowBatches       []*ArrowBatch
 	RowSet             rowSetType
 	FuncDownload       func(context.Context, *snowflakeChunkDownloader, int)
 	FuncDownloadHelper func(context.Context, *snowflakeChunkDownloader, int) error
@@ -88,6 +92,9 @@ func (scd *snowflakeChunkDownloader) nextResultSet() error {
 }
 
 func (scd *snowflakeChunkDownloader) start() error {
+	if usesArrowBatches(scd.ctx) {
+		return scd.startArrowBatches()
+	}
 	scd.CurrentChunkSize = len(scd.RowSet.JSON) // cache the size
 	scd.CurrentIndex = -1                       // initial chunks idx
 	scd.CurrentChunkIndex = -1                  // initial chunk
@@ -231,6 +238,13 @@ func (scd *snowflakeChunkDownloader) getRowType() []execResponseRowType {
 	return scd.RowSet.RowType
 }
 
+func (scd *snowflakeChunkDownloader) getArrowBatches() []*ArrowBatch {
+	if scd.FirstBatch.rec == nil {
+		return scd.ArrowBatches
+	}
+	return append([]*ArrowBatch{scd.FirstBatch}, scd.ArrowBatches...)
+}
+
 func getChunk(
 	ctx context.Context,
 	scd *snowflakeChunkDownloader,
@@ -245,6 +259,33 @@ func getChunk(
 	return newRetryHTTP(ctx, scd.sc.rest.Client, http.NewRequest, u, headers, timeout).execute()
 }
 
+func (scd *snowflakeChunkDownloader) startArrowBatches() error {
+	var err error
+	chunkMetaLen := len(scd.ChunkMetas)
+	firstArrowChunk := buildFirstArrowChunk(scd.RowSet.RowSetBase64)
+	scd.FirstBatch = &ArrowBatch{
+		idx:                0,
+		scd:                scd,
+		funcDownloadHelper: scd.FuncDownloadHelper,
+	}
+	// decode first chunk if possible
+	if firstArrowChunk.allocator != nil {
+		scd.FirstBatch.rec, err = firstArrowChunk.decodeArrowBatch(scd)
+		if err != nil {
+			return err
+		}
+	}
+	scd.ArrowBatches = make([]*ArrowBatch, chunkMetaLen)
+	for i := 0; i < chunkMetaLen; i++ {
+		scd.ArrowBatches[i] = &ArrowBatch{
+			idx:                i,
+			scd:                scd,
+			funcDownloadHelper: scd.FuncDownloadHelper,
+		}
+	}
+	return nil
+}
+
 /* largeResultSetReader is a reader that wraps the large result set with leading and tailing brackets. */
 type largeResultSetReader struct {
 	status int
@@ -380,6 +421,14 @@ func decodeChunk(scd *snowflakeChunkDownloader, idx int, bufStream *bufio.Reader
 			int(scd.totalUncompressedSize()),
 			memory.NewGoAllocator(),
 		}
+		if usesArrowBatches(scd.ctx) {
+			if scd.ArrowBatches[idx].rec, err = arc.decodeArrowBatch(scd); err != nil {
+				return err
+			}
+			// updating metadata
+			scd.ArrowBatches[idx].rowCount = countArrowBatchRows(scd.ArrowBatches[idx].rec)
+			return nil
+		}
 		highPrec := higherPrecisionEnabled(scd.ctx)
 		respd, err = arc.decodeArrowChunk(scd.RowSet.RowType, highPrec)
 		if err != nil {
@@ -512,6 +561,10 @@ func (scd *streamChunkDownloader) getRowType() []execResponseRowType {
 	return scd.RowSet.RowType
 }
 
+func (scd *streamChunkDownloader) getArrowBatches() []*ArrowBatch {
+	return nil
+}
+
 func useStreamDownloader(ctx context.Context) bool {
 	val := ctx.Value(streamChunkDownload)
 	if val == nil {
@@ -630,3 +683,43 @@ func copyChunkStream(body io.Reader, rows chan<- []*string) error {
 	}
 	return nil
 }
+
+// ArrowBatch object represents a chunk of data, or subset of rows, retrievable in array.Record format
+type ArrowBatch struct {
+	rec                *[]array.Record
+	idx                int
+	rowCount           int
+	scd                *snowflakeChunkDownloader
+	funcDownloadHelper func(context.Context, *snowflakeChunkDownloader, int) error
+}
+
+// Fetch returns an array of records representing a chunk in the query
+func (rb *ArrowBatch) Fetch() (*[]array.Record, error) {
+	// chunk has already been downloaded
+	if rb.rec != nil {
+		// updating metadata
+		rb.rowCount = countArrowBatchRows(rb.rec)
+		return rb.rec, nil
+	}
+	if err := rb.funcDownloadHelper(context.Background(), rb.scd, rb.idx); err != nil {
+		return nil, err
+	}
+	return rb.rec, nil
+}
+
+func usesArrowBatches(ctx context.Context) bool {
+	val := ctx.Value(arrowBatches)
+	if val == nil {
+		return false
+	}
+	a, ok := val.(bool)
+	return a && ok
+}
+
+func countArrowBatchRows(recs *[]array.Record) int {
+	var cnt int
+	for _, r := range *recs {
+		cnt += int(r.NumRows())
+	}
+	return cnt
+}
diff --git a/converter.go b/converter.go
index 660096c..debd3e8 100644
--- a/converter.go
+++ b/converter.go
@@ -17,6 +17,7 @@ import (
 	"github.com/apache/arrow/go/arrow"
 	"github.com/apache/arrow/go/arrow/array"
 	"github.com/apache/arrow/go/arrow/decimal128"
+	"github.com/apache/arrow/go/arrow/memory"
 )
 
 const format = "2006-01-02 15:04:05.999999999"
@@ -777,3 +778,264 @@ func higherPrecisionEnabled(ctx context.Context) bool {
 	d, ok := v.(bool)
 	return ok && d
 }
+
+func arrowToRecord(record array.Record, rowType []execResponseRowType) (array.Record, error) {
+	s, err := recordToSchema(record.Schema(), rowType)
+	if err != nil {
+		return nil, err
+	}
+
+	var cols []array.Interface
+	numRows := record.NumRows()
+	pool := memory.NewGoAllocator()
+
+	for i, col := range record.Columns() {
+		srcColumnMeta := rowType[i]
+		data := col.Data()
+
+		//TODO: confirm that it is okay to be using higher precision logic for conversions
+		var newCol = col
+		switch getSnowflakeType(strings.ToUpper(srcColumnMeta.Type)) {
+		case fixedType:
+			switch col.DataType().ID() {
+			case arrow.DECIMAL:
+				if srcColumnMeta.Scale == 0 {
+					ib := array.NewInt64Builder(pool)
+					for i, num := range array.NewDecimal128Data(data).Values() {
+						if !col.IsNull(i) {
+							bi := decimalToBigInt(num)
+							val := bi.Int64()
+							ib.Append(val)
+						} else {
+							ib.AppendNull()
+						}
+					}
+					newCol = ib.NewArray()
+				} else {
+					fb := array.NewFloat64Builder(pool)
+					for i, num := range array.NewDecimal128Data(data).Values() {
+						if !col.IsNull(i) {
+							f := decimalToBigFloat(num, srcColumnMeta.Scale)
+							val, _ := f.Float64()
+							fb.Append(val)
+						} else {
+							fb.AppendNull()
+						}
+					}
+					newCol = fb.NewArray()
+				}
+			case arrow.INT8:
+				if srcColumnMeta.Scale != 0 {
+					fb := array.NewFloat64Builder(pool)
+					for i, val := range array.NewInt8Data(data).Int8Values() {
+						if !col.IsNull(i) {
+							f := intToBigFloat(int64(val), srcColumnMeta.Scale)
+							val, _ := f.Float64()
+							fb.Append(val)
+						} else {
+							fb.AppendNull()
+						}
+					}
+					newCol = fb.NewArray()
+				}
+			case arrow.INT16:
+				if srcColumnMeta.Scale != 0 {
+					fb := array.NewFloat64Builder(pool)
+					for i, val := range array.NewInt16Data(data).Int16Values() {
+						if !col.IsNull(i) {
+							f := intToBigFloat(int64(val), srcColumnMeta.Scale)
+							val, _ := f.Float64()
+							fb.Append(val)
+						} else {
+							fb.AppendNull()
+						}
+					}
+					newCol = fb.NewArray()
+				}
+			case arrow.INT32:
+				if srcColumnMeta.Scale != 0 {
+					fb := array.NewFloat64Builder(pool)
+					for i, val := range array.NewInt32Data(data).Int32Values() {
+						if !col.IsNull(i) {
+							f := intToBigFloat(int64(val), srcColumnMeta.Scale)
+							val, _ := f.Float64()
+							fb.Append(val)
+						} else {
+							fb.AppendNull()
+						}
+					}
+					newCol = fb.NewArray()
+				}
+			case arrow.INT64:
+				if srcColumnMeta.Scale != 0 {
+					fb := array.NewFloat64Builder(pool)
+					for i, val := range array.NewInt64Data(data).Int64Values() {
+						if !col.IsNull(i) {
+							f := intToBigFloat(val, srcColumnMeta.Scale)
+							val, _ := f.Float64()
+							fb.Append(val)
+						} else {
+							fb.AppendNull()
+						}
+					}
+					newCol = fb.NewArray()
+				}
+			}
+		case timeType:
+			tb := array.NewTime64Builder(pool, &arrow.Time64Type{})
+			if col.DataType().ID() == arrow.INT64 || col.DataType().ID() == arrow.TIME64 {
+				for i, i64 := range array.NewInt64Data(data).Int64Values() {
+					if !col.IsNull(i) {
+						tb.Append(arrow.Time64(i64))
+					} else {
+						tb.AppendNull()
+					}
+				}
+			} else {
+				for i, i32 := range array.NewInt32Data(data).Int32Values() {
+					if !col.IsNull(i) {
+						tb.Append(arrow.Time64(int64(i32)))
+					} else {
+						tb.AppendNull()
+					}
+				}
+			}
+			newCol = tb.NewArray()
+		case timestampNtzType:
+			tb := array.NewTimestampBuilder(pool, &arrow.TimestampType{})
+			if col.DataType().ID() == arrow.STRUCT {
+				structData := array.NewStructData(data)
+				epoch := array.NewInt64Data(structData.Field(0).Data()).Int64Values()
+				fraction := array.NewInt32Data(structData.Field(1).Data()).Int32Values()
+				for i := 0; i < int(numRows); i++ {
+					if !col.IsNull(i) {
+						val := time.Unix(epoch[i], int64(fraction[i]))
+						tb.Append(arrow.Timestamp(val.UnixNano()))
+					} else {
+						tb.AppendNull()
+					}
+				}
+			} else {
+				for i, t := range array.NewInt64Data(data).Int64Values() {
+					if !col.IsNull(i) {
+						val := time.Unix(0, t*int64(math.Pow10(9-int(srcColumnMeta.Scale)))).UTC()
+						tb.Append(arrow.Timestamp(val.UnixNano()))
+					} else {
+						tb.AppendNull()
+					}
+				}
+			}
+			newCol = tb.NewArray()
+		case timestampLtzType:
+			tb := array.NewTimestampBuilder(pool, &arrow.TimestampType{TimeZone: "UTC"})
+			if col.DataType().ID() == arrow.STRUCT {
+				structData := array.NewStructData(data)
+				epoch := array.NewInt64Data(structData.Field(0).Data()).Int64Values()
+				fraction := array.NewInt32Data(structData.Field(1).Data()).Int32Values()
+				for i := 0; i < int(numRows); i++ {
+					if !col.IsNull(i) {
+						val := time.Unix(epoch[i], int64(fraction[i]))
+						tb.Append(arrow.Timestamp(val.UnixNano()))
+					} else {
+						tb.AppendNull()
+					}
+				}
+			} else {
+				for i, t := range array.NewInt64Data(data).Int64Values() {
+					if !col.IsNull(i) {
+						q := t / int64(math.Pow10(int(srcColumnMeta.Scale)))
+						r := t % int64(math.Pow10(int(srcColumnMeta.Scale)))
+						val := time.Unix(q, r)
+						tb.Append(arrow.Timestamp(val.UnixNano()))
+					} else {
+						tb.AppendNull()
+					}
+				}
+			}
+			newCol = tb.NewArray()
+		case timestampTzType:
+			tb := array.NewTimestampBuilder(pool, &arrow.TimestampType{})
+			structData := array.NewStructData(data)
+			if structData.NumField() == 2 {
+				epoch := array.NewInt64Data(structData.Field(0).Data()).Int64Values()
+				timezone := array.NewInt32Data(structData.Field(1).Data()).Int32Values()
+				for i := 0; i < int(numRows); i++ {
+					if !col.IsNull(i) {
+						loc := Location(int(timezone[i]) - 1440)
+						tt := time.Unix(epoch[i], 0)
+						val := tt.In(loc)
+						tb.Append(arrow.Timestamp(val.UnixNano()))
+					} else {
+						tb.AppendNull()
+					}
+				}
+			} else {
+				epoch := array.NewInt64Data(structData.Field(0).Data()).Int64Values()
+				fraction := array.NewInt32Data(structData.Field(1).Data()).Int32Values()
+				timezone := array.NewInt32Data(structData.Field(2).Data()).Int32Values()
+				for i := 0; i < int(numRows); i++ {
+					if !col.IsNull(i) {
+						loc := Location(int(timezone[i]) - 1440)
+						tt := time.Unix(epoch[i], int64(fraction[i]))
+						val := tt.In(loc)
+						tb.Append(arrow.Timestamp(val.UnixNano()))
+					} else {
+						tb.AppendNull()
+					}
+				}
+			}
+			newCol = tb.NewArray()
+		}
+		cols = append(cols, newCol)
+	}
+	return array.NewRecord(s, cols, numRows), nil
+}
+
+func recordToSchema(sc *arrow.Schema, rowType []execResponseRowType) (*arrow.Schema, error) {
+	var fields []arrow.Field
+	for i := 0; i < len(sc.Fields()); i++ {
+		f := sc.Field(i)
+		srcColumnMeta := rowType[i]
+		converted := true
+
+		var t arrow.DataType
+		switch getSnowflakeType(strings.ToUpper(srcColumnMeta.Type)) {
+		case fixedType:
+			switch f.Type.ID() {
+			case arrow.DECIMAL:
+				if srcColumnMeta.Scale == 0 {
+					t = &arrow.Int64Type{}
+				} else {
+					t = &arrow.Float64Type{}
+				}
+			default:
+				if srcColumnMeta.Scale != 0 {
+					t = &arrow.Float64Type{}
+				} else {
+					converted = false
+				}
+			}
+		case timeType:
+			t = &arrow.Time64Type{}
+		case timestampNtzType, timestampTzType:
+			t = &arrow.TimestampType{}
+		case timestampLtzType:
+			t = &arrow.TimestampType{TimeZone: "UTC"}
+		default:
+			converted = false
+		}
+
+		var newField = f
+		if converted {
+			newField = arrow.Field{
+				Name:     f.Name,
+				Type:     t,
+				Nullable: f.Nullable,
+				Metadata: f.Metadata,
+			}
+		}
+		fields = append(fields, newField)
+	}
+	meta := sc.Metadata()
+	return arrow.NewSchema(fields, &meta), nil
+}
diff --git a/result.go b/result.go
index 7fbe3cf..e08f419 100644
--- a/result.go
+++ b/result.go
@@ -13,10 +13,11 @@ const (
 	QueryFailed queryStatus = "queryFailed"
 )
 
-// SnowflakeResult provides the associated query ID
+// SnowflakeResult provides an API for methods exposed to the clients
 type SnowflakeResult interface {
 	GetQueryID() string
 	GetStatus() queryStatus
+	GetArrowBatches() ([]*ArrowBatch, error)
 }
 
 type snowflakeResult struct {
@@ -50,6 +51,13 @@ func (res *snowflakeResult) GetStatus() queryStatus {
 	return res.status
 }
 
+func (res *snowflakeResult) GetArrowBatches() ([]*ArrowBatch, error) {
+	return nil, &SnowflakeError{
+		Number:  ErrNotImplemented,
+		Message: errMsgNotImplemented,
+	}
+}
+
 func (res *snowflakeResult) waitForAsyncExecStatus() error {
 	// if async exec, block until execution is finished
 	if res.status == QueryStatusInProgress {
diff --git a/rows.go b/rows.go
index bc15f5f..0663c97 100644
--- a/rows.go
+++ b/rows.go
@@ -144,6 +144,11 @@ func (rows *snowflakeRows) GetStatus() queryStatus {
 	return rows.status
 }
 
+// GetArrowBatches returns an array of ArrowBatch objects to retrieve data in array.Record format
+func (rows *snowflakeRows) GetArrowBatches() ([]*ArrowBatch, error) {
+	return rows.ChunkDownloader.getArrowBatches(), nil
+}
+
 func (rows *snowflakeRows) Next(dest []driver.Value) (err error) {
 	if err = rows.waitForAsyncQueryStatus(); err != nil {
 		return err
diff --git a/util.go b/util.go
index 33733bf..cdcf79b 100644
--- a/util.go
+++ b/util.go
@@ -23,6 +23,7 @@ const (
 	fileStreamFile        contextKey = "STREAMING_PUT_FILE"
 	fileTransferOptions   contextKey = "FILE_TRANSFER_OPTIONS"
 	enableHigherPrecision contextKey = "ENABLE_HIGHER_PRECISION"
+	arrowBatches          contextKey = "ARROW_BATCHES"
 )
 
 const (
@@ -83,6 +84,12 @@ func WithHigherPrecision(ctx context.Context) context.Context {
 	return context.WithValue(ctx, enableHigherPrecision, true)
 }
 
+// WithArrowBatches returns a context that allows users to retrieve
+// array.Record download workers upon querying
+func WithArrowBatches(ctx context.Context) context.Context {
+	return context.WithValue(ctx, arrowBatches, true)
+}
+
 // Get the request ID from the context if specified, otherwise generate one
 func getOrGenerateRequestIDFromContext(ctx context.Context) uuid {
 	requestID, ok := ctx.Value(snowflakeRequestIDKey).(uuid)


Test patch:
diff --git a/chunk_test.go b/chunk_test.go
index deab65b..bbe16f3 100644
--- a/chunk_test.go
+++ b/chunk_test.go
@@ -5,11 +5,13 @@ package gosnowflake
 import (
 	"bytes"
 	"context"
+	"database/sql/driver"
 	"encoding/json"
 	"fmt"
 	"io"
 	"math/rand"
 	"strings"
+	"sync"
 	"testing"
 )
 
@@ -387,3 +389,78 @@ func TestWithStreamDownloader(t *testing.T) {
 		}
 	})
 }
+
+func TestWithArrowBatches(t *testing.T) {
+	ctx := WithArrowBatches(context.Background())
+	numrows := 3000 // approximately 6 ArrowBatch objects
+	config, err := ParseDSN(dsn)
+	if err != nil {
+		t.Error(err)
+	}
+	sc, err := buildSnowflakeConn(ctx, *config)
+	if err != nil {
+		t.Error(err)
+	}
+	if err = authenticateWithConfig(sc); err != nil {
+		t.Error(err)
+	}
+
+	query := fmt.Sprintf(selectRandomGenerator, numrows)
+	rows, err := sc.QueryContext(ctx, query, []driver.NamedValue{})
+	if err != nil {
+		t.Error(err)
+	}
+	defer rows.Close()
+
+	// getting result batches
+	batches, err := rows.(*snowflakeRows).GetArrowBatches()
+	if err != nil {
+		t.Error(err)
+	}
+	numBatches := len(batches)
+	maxWorkers := 10 // enough for 3000 rows
+	type count struct {
+		m       sync.Mutex
+		recVal  int
+		metaVal int
+	}
+	cnt := count{recVal: 0}
+	var wg sync.WaitGroup
+	chunks := make(chan int, numBatches)
+
+	// kicking off download workers - each of which will call fetch on a different result batch
+	for w := 1; w <= maxWorkers; w++ {
+		wg.Add(1)
+		go func(wg *sync.WaitGroup, chunks <-chan int) {
+			defer wg.Done()
+
+			for i := range chunks {
+				rec, err := batches[i].Fetch()
+				if err != nil {
+					t.Error(err)
+				}
+				for _, r := range *rec {
+					cnt.m.Lock()
+					cnt.recVal += int(r.NumRows())
+					cnt.m.Unlock()
+				}
+				cnt.m.Lock()
+				cnt.metaVal += batches[i].rowCount
+				cnt.m.Unlock()
+			}
+		}(&wg, chunks)
+	}
+	for j := 0; j < numBatches; j++ {
+		chunks <- j
+	}
+	close(chunks)
+
+	// wait for workers to finish fetching and check row counts
+	wg.Wait()
+	if cnt.recVal != numrows {
+		t.Errorf("number of rows from records didn't match. expected: %v, got: %v", numrows, cnt.recVal)
+	}
+	if cnt.metaVal != numrows {
+		t.Errorf("number of rows from arrow batch metadata didn't match. expected: %v, got: %v", numrows, cnt.metaVal)
+	}
+}
diff --git a/converter_test.go b/converter_test.go
index fe57373..89981db 100644
--- a/converter_test.go
+++ b/converter_test.go
@@ -556,3 +556,414 @@ func TestArrowToValue(t *testing.T) {
 
 	}
 }
+
+func TestArrowToRecord(t *testing.T) {
+	pool := memory.NewCheckedAllocator(memory.NewGoAllocator())
+	var valids []bool // AppendValues() with an empty valid array adds every value by default
+
+	localTime := time.Date(2019, 2, 6, 14, 17, 31, 123456789, time.FixedZone("-08:00", -8*3600))
+
+	field1 := arrow.Field{Name: "epoch", Type: &arrow.Int64Type{}}
+	field2 := arrow.Field{Name: "timezone", Type: &arrow.Int32Type{}}
+	tzStruct := arrow.StructOf(field1, field2)
+
+	type testObj struct {
+		field1 int
+		field2 string
+	}
+
+	for _, tc := range []struct {
+		logical  string
+		physical string
+		sc       *arrow.Schema
+		rowType  execResponseRowType
+		values   interface{}
+		nrows    int
+		builder  array.Builder
+		append   func(b array.Builder, vs interface{})
+		compare  func(src interface{}, rec array.Record) int
+	}{
+		{
+			logical:  "fixed",
+			physical: "number", // default: number(38, 0)
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int64Type{}}}, nil),
+			values:   []int64{1, 2},
+			nrows:    2,
+			builder:  array.NewInt64Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int64Builder).AppendValues(vs.([]int64), valids) },
+		},
+		{
+			logical:  "fixed",
+			physical: "number(38,0)",
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Decimal128Type{Precision: 30, Scale: 2}}}, nil),
+			values:   []string{"10000000000000000000000000000000000000", "-12345678901234567890123456789012345678"},
+			nrows:    2,
+			builder:  array.NewDecimal128Builder(pool, &arrow.Decimal128Type{Precision: 30, Scale: 2}),
+			append: func(b array.Builder, vs interface{}) {
+				for _, s := range vs.([]string) {
+					num, ok := stringIntToDecimal(s)
+					if !ok {
+						t.Fatalf("failed to convert to Int64")
+					}
+					b.(*array.Decimal128Builder).Append(num)
+				}
+			},
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]string)
+				for i, dec := range array.NewInt64Data(convertedRec.Column(0).Data()).Int64Values() {
+					num, ok := stringIntToDecimal(srcvs[i])
+					if !ok {
+						return i
+					}
+					srcDec := decimalToBigInt(num).Int64()
+					if srcDec != dec {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical:  "fixed",
+			physical: "number(38,37)",
+			rowType:  execResponseRowType{Scale: 37},
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Decimal128Type{Precision: 38, Scale: 37}}}, nil),
+			values:   []string{"1.2345678901234567890123456789012345678", "-9.9999999999999999999999999999999999999"},
+			nrows:    2,
+			builder:  array.NewDecimal128Builder(pool, &arrow.Decimal128Type{Precision: 38, Scale: 37}),
+			append: func(b array.Builder, vs interface{}) {
+				for _, s := range vs.([]string) {
+					num, ok := stringFloatToDecimal(s, 37)
+					if !ok {
+						t.Fatalf("failed to convert to decimal")
+					}
+					b.(*array.Decimal128Builder).Append(num)
+				}
+			},
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]string)
+				for i, dec := range array.NewFloat64Data(convertedRec.Column(0).Data()).Float64Values() {
+					num, ok := stringFloatToDecimal(srcvs[i], 37)
+					if !ok {
+						return i
+					}
+					srcDec, _ := decimalToBigFloat(num, 37).Float64()
+					if srcDec != dec {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical:  "fixed",
+			physical: "int8",
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int8Type{}}}, nil),
+			values:   []int8{1, 2},
+			nrows:    2,
+			builder:  array.NewInt8Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int8Builder).AppendValues(vs.([]int8), valids) },
+		},
+		{
+			logical:  "fixed",
+			physical: "int16",
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int16Type{}}}, nil),
+			values:   []int16{1, 2},
+			nrows:    2,
+			builder:  array.NewInt16Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int16Builder).AppendValues(vs.([]int16), valids) },
+		},
+		{
+			logical:  "fixed",
+			physical: "int32",
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int32Type{}}}, nil),
+			values:   []int32{1, 2},
+			nrows:    2,
+			builder:  array.NewInt32Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int32Builder).AppendValues(vs.([]int32), valids) },
+		},
+		{
+			logical:  "fixed",
+			physical: "int64",
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int64Type{}}}, nil),
+			values:   []int64{1, 2},
+			nrows:    2,
+			builder:  array.NewInt64Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int64Builder).AppendValues(vs.([]int64), valids) },
+		},
+		{
+			logical:  "fixed",
+			physical: "float8",
+			rowType:  execResponseRowType{Scale: 1},
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int8Type{}}}, nil),
+			values:   []int8{10, 16},
+			nrows:    2,
+			builder:  array.NewInt8Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int8Builder).AppendValues(vs.([]int8), valids) },
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]int8)
+				for i, f := range array.NewFloat64Data(convertedRec.Column(0).Data()).Float64Values() {
+					rawFloat, _ := intToBigFloat(int64(srcvs[i]), 1).Float64()
+					if rawFloat != f {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical:  "fixed",
+			physical: "float16",
+			rowType:  execResponseRowType{Scale: 1},
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int16Type{}}}, nil),
+			values:   []int16{20, 26},
+			nrows:    2,
+			builder:  array.NewInt16Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int16Builder).AppendValues(vs.([]int16), valids) },
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]int16)
+				for i, f := range array.NewFloat64Data(convertedRec.Column(0).Data()).Float64Values() {
+					rawFloat, _ := intToBigFloat(int64(srcvs[i]), 1).Float64()
+					if rawFloat != f {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical:  "fixed",
+			physical: "float32",
+			rowType:  execResponseRowType{Scale: 2},
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int32Type{}}}, nil),
+			values:   []int32{200, 265},
+			nrows:    2,
+			builder:  array.NewInt32Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int32Builder).AppendValues(vs.([]int32), valids) },
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]int32)
+				for i, f := range array.NewFloat64Data(convertedRec.Column(0).Data()).Float64Values() {
+					rawFloat, _ := intToBigFloat(int64(srcvs[i]), 2).Float64()
+					if rawFloat != f {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical:  "fixed",
+			physical: "float64",
+			rowType:  execResponseRowType{Scale: 5},
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Int64Type{}}}, nil),
+			values:   []int64{12345, 234567},
+			nrows:    2,
+			builder:  array.NewInt64Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Int64Builder).AppendValues(vs.([]int64), valids) },
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]int64)
+				for i, f := range array.NewFloat64Data(convertedRec.Column(0).Data()).Float64Values() {
+					rawFloat, _ := intToBigFloat(srcvs[i], 5).Float64()
+					if rawFloat != f {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical: "boolean",
+			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.BooleanType{}}}, nil),
+			values:  []bool{true, false},
+			nrows:   2,
+			builder: array.NewBooleanBuilder(pool),
+			append:  func(b array.Builder, vs interface{}) { b.(*array.BooleanBuilder).AppendValues(vs.([]bool), valids) },
+		},
+		{
+			logical:  "real",
+			physical: "float",
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.Float64Type{}}}, nil),
+			values:   []float64{1, 2},
+			nrows:    2,
+			builder:  array.NewFloat64Builder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.Float64Builder).AppendValues(vs.([]float64), valids) },
+		},
+		{
+			logical:  "text",
+			physical: "string",
+			sc:       arrow.NewSchema([]arrow.Field{{Type: &arrow.StringType{}}}, nil),
+			values:   []string{"foo", "bar"},
+			nrows:    2,
+			builder:  array.NewStringBuilder(pool),
+			append:   func(b array.Builder, vs interface{}) { b.(*array.StringBuilder).AppendValues(vs.([]string), valids) },
+		},
+		{
+			logical: "binary",
+			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.BinaryType{}}}, nil),
+			values:  [][]byte{[]byte("foo"), []byte("bar")},
+			nrows:   2,
+			builder: array.NewBinaryBuilder(pool, arrow.BinaryTypes.Binary),
+			append:  func(b array.Builder, vs interface{}) { b.(*array.BinaryBuilder).AppendValues(vs.([][]byte), valids) },
+		},
+		{
+			logical: "date",
+			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.Date32Type{}}}, nil),
+			values:  []time.Time{time.Now(), localTime},
+			nrows:   2,
+			builder: array.NewDate32Builder(pool),
+			append: func(b array.Builder, vs interface{}) {
+				for _, d := range vs.([]time.Time) {
+					b.(*array.Date32Builder).Append(arrow.Date32(d.Unix()))
+				}
+			},
+		},
+		{
+			logical: "time",
+			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.Time64Type{}}}, nil),
+			values:  []time.Time{time.Now(), time.Now()},
+			nrows:   2,
+			builder: array.NewTime64Builder(pool, &arrow.Time64Type{}),
+			append: func(b array.Builder, vs interface{}) {
+				for _, t := range vs.([]time.Time) {
+					b.(*array.Time64Builder).Append(arrow.Time64(t.UnixNano()))
+				}
+			},
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]time.Time)
+				for i, t := range array.NewTime64Data(convertedRec.Column(0).Data()).Time64Values() {
+					if srcvs[i].UnixNano() != int64(t) {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical: "timestamp_ntz",
+			values:  []time.Time{time.Now(), localTime},
+			nrows:   2,
+			rowType: execResponseRowType{Scale: 9},
+			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.TimestampType{}}}, nil),
+			builder: array.NewTimestampBuilder(pool, &arrow.TimestampType{}),
+			append: func(b array.Builder, vs interface{}) {
+				for _, t := range vs.([]time.Time) {
+					b.(*array.TimestampBuilder).Append(arrow.Timestamp(t.UnixNano()))
+				}
+			},
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]time.Time)
+				for i, t := range array.NewTimestampData(convertedRec.Column(0).Data()).TimestampValues() {
+					if srcvs[i].UnixNano() != int64(t) {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical: "timestamp_ltz",
+			values:  []time.Time{time.Now(), localTime},
+			nrows:   2,
+			rowType: execResponseRowType{Scale: 9},
+			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.TimestampType{}}}, nil),
+			builder: array.NewTimestampBuilder(pool, &arrow.TimestampType{}),
+			append: func(b array.Builder, vs interface{}) {
+				for _, t := range vs.([]time.Time) {
+					b.(*array.TimestampBuilder).Append(arrow.Timestamp(t.UnixNano()))
+				}
+			},
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]time.Time)
+				for i, t := range array.NewTimestampData(convertedRec.Column(0).Data()).TimestampValues() {
+					if srcvs[i].UnixNano() != int64(t) {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical: "timestamp_tz",
+			values:  []time.Time{time.Now(), localTime},
+			nrows:   2,
+			sc:      arrow.NewSchema([]arrow.Field{{Type: arrow.StructOf(field1, field2)}}, nil),
+			builder: array.NewStructBuilder(pool, tzStruct),
+			append: func(b array.Builder, vs interface{}) {
+				sb := b.(*array.StructBuilder)
+				valids = []bool{true, true}
+				sb.AppendValues(valids)
+				for _, t := range vs.([]time.Time) {
+					sb.FieldBuilder(0).(*array.Int64Builder).Append(t.Unix())
+					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.UnixNano()))
+				}
+			},
+			compare: func(src interface{}, convertedRec array.Record) int {
+				srcvs := src.([]time.Time)
+				for i, t := range array.NewTimestampData(convertedRec.Column(0).Data()).TimestampValues() {
+					if srcvs[i].Unix() != time.Unix(0, int64(t)).Unix() {
+						return i
+					}
+				}
+				return -1
+			},
+		},
+		{
+			logical: "array",
+			values:  [][]string{{"foo", "bar"}, {"baz", "quz", "quux"}},
+			nrows:   2,
+			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.StringType{}}}, nil),
+			builder: array.NewStringBuilder(pool),
+			append: func(b array.Builder, vs interface{}) {
+				for _, a := range vs.([][]string) {
+					b.(*array.StringBuilder).Append(fmt.Sprint(a))
+				}
+			},
+		},
+		{
+			logical: "object",
+			values:  []testObj{{0, "foo"}, {1, "bar"}},
+			nrows:   2,
+			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.StringType{}}}, nil),
+			builder: array.NewStringBuilder(pool),
+			append: func(b array.Builder, vs interface{}) {
+				for _, o := range vs.([]testObj) {
+					b.(*array.StringBuilder).Append(fmt.Sprint(o))
+				}
+			},
+		},
+	} {
+		testName := tc.logical
+		if tc.physical != "" {
+			testName += " " + tc.physical
+		}
+		t.Run(testName, func(t *testing.T) {
+			b := tc.builder
+			tc.append(b, tc.values)
+			arr := b.NewArray()
+			defer arr.Release()
+			rawRec := array.NewRecord(tc.sc, []array.Interface{arr}, int64(tc.nrows))
+			meta := tc.rowType
+			meta.Type = tc.logical
+
+			transformedRec, err := arrowToRecord(rawRec, []execResponseRowType{meta})
+			if err != nil {
+				t.Fatalf("error: %s", err)
+			}
+
+			if tc.compare != nil {
+				idx := tc.compare(tc.values, transformedRec)
+				if idx != -1 {
+					t.Fatalf("error: column array value mistmatch at index %v", idx)
+				}
+			} else {
+				for i, c := range transformedRec.Columns() {
+					rawCol := rawRec.Column(i)
+					if rawCol != c {
+						t.Fatalf("error: expected column %s, got column %s", rawCol, c)
+					}
+				}
+			}
+		})
+
+	}
+}