Repository: snowflakedb/gosnowflake
Pull request #383 (2021-02-22 22:18:51)

SNOW-187551 Support Bulk Array Binding
### Description
Support for bulk array binding for Golang driver using streaming PUTs using the local file system. This can be enabled by setting the parameter `CLIENT_STAGE_ARRAY_BINDING_THRESHOLD` and binding the column arrays via the `Array()` function, e.g. `db.Exec("insert into tbl (?, ?, ?, ?)", Array(intArray), Array(fltArray), Array(boolArray), Array(strArray))`. [Note that the command would work the same even if the parameter is not set] When the binding threshold is reached, we query a PUT command to GS, and process the passed columns and stream them to the local stage. Partial implementation of PUT.

### 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-stakeda: Please add more context to the PR. Did you already implement PUT command? How much did you test it? It is reliable enough to be used for bulk array binding? Which area do you want me to review?
sfc-gh-stakeda: no test case is included.
sfc-gh-jbahk: @sfc-gh-stakeda sorry, I should have added you as a reviewer once I'm finished with this PR. This is still a draft and my next few commits will implement the streaming PUT command.
sfc-gh-jbahk: Streaming PUT writes to the local stage for array binding and does not require the cloud providers. I have another branch working on the full implementation of PUT (and the cloud providers) so please excuse the lint suppression for unused code.
sfc-gh-stakeda: Please note at this moment `CLIENT_STAGE_ARRAY_BINDING_THRESHOLD` cannot be updated by the customers.
sfc-gh-hchaturvedi: @sfc-gh-jbahk have we looked into any GO lang APIs that abstract all the padding and encryption out? As in we just give the buffer and we get the cipherText out?
sfc-gh-jbahk: @sfc-gh-hchaturvedi the `crypto/aes` and `crypto/cipher` packages are the de facto standard packages for Golang.

Patch:
diff --git a/bind_uploader.go b/bind_uploader.go
index 0a47324..82791a6 100644
--- a/bind_uploader.go
+++ b/bind_uploader.go
@@ -3,10 +3,256 @@
 package gosnowflake
 
 import (
+	"bytes"
+	"context"
 	"database/sql/driver"
 	"reflect"
+	"strconv"
+	"strings"
 )
 
+//lint:file-ignore U1000 Ignore all unused code
+
+const (
+	stageName       = "SYSTEM$BIND"
+	createStageStmt = "CREATE TEMPORARY STAGE " + stageName + " file_format=" +
+		"(type=csv field_optionally_enclosed_by='\"')"
+
+	// size (in bytes) of max input stream (10MB default) as per JDBC specs
+	inputStreamBufferSize = 1024 * 1024 * 10
+)
+
+type bindUploader struct {
+	ctx            context.Context
+	sc             *snowflakeConn
+	stagePath      string
+	closed         bool
+	fileCount      int
+	arrayBindStage string
+	bindings       map[string]execBindParameter
+}
+
+func (bu *bindUploader) upload(bindings []driver.NamedValue) (*execResponseData, error) {
+	if !bu.closed {
+		bindingRows, _ := bu.buildRowsAsBytes(bindings)
+		startIdx := 0
+		numBytes := 0
+		rowNum := 0
+		bu.fileCount = 0
+		var data *execResponseData
+		var err error
+		for rowNum < len(bindingRows) {
+			for numBytes < inputStreamBufferSize && rowNum < len(bindingRows) {
+				numBytes += len(bindingRows[rowNum])
+				rowNum++
+			}
+			// concatenate all byte arrays into 1 and put into input stream
+			var b bytes.Buffer
+			b.Grow(numBytes)
+			for i := startIdx; i < rowNum; i++ {
+				b.Write(bindingRows[i])
+			}
+
+			bu.fileCount++
+			filename := strconv.Itoa(bu.fileCount)
+			data, err = bu.uploadStreamInternal(&b, filename, true)
+			if err != nil {
+				return nil, err
+			}
+			startIdx = rowNum
+			numBytes = 0
+		}
+		return data, nil
+	}
+	return nil, nil
+}
+
+func (bu *bindUploader) uploadStreamInternal(inputStream *bytes.Buffer, dstFilename string, compressData bool) (*execResponseData, error) {
+	err := bu.createStageIfNeeded()
+	if err != nil {
+		return nil, err
+	}
+	stageName := bu.stagePath
+	if stageName == "" {
+		return nil, &SnowflakeError{
+			Number:  ErrBindUpload,
+			Message: "stage name is null",
+		}
+	}
+	if dstFilename == "" {
+		return nil, &SnowflakeError{
+			Number:  ErrBindUpload,
+			Message: "destination file is null",
+		}
+	}
+
+	var putCommand strings.Builder
+	// use a placeholder for source file
+	putCommand.WriteString("put file:///tmp/placeholder ")
+	// add stage name surrounded by quotations in case special chars are used in directory name
+	putCommand.WriteString("'")
+	putCommand.WriteString(stageName)
+	putCommand.WriteString("'")
+	putCommand.WriteString(" overwrite=true")
+	data, err := bu.sc.exec(bu.ctx, putCommand.String(), false, false, false, []driver.NamedValue{})
+	if err != nil {
+		return nil, err
+	}
+
+	sfa := &snowflakeFileTransferAgent{
+		data:                       data.Data,
+		command:                    putCommand.String(),
+		sourceStream:               inputStream,
+		dstFileNameForStreamSource: dstFilename,
+		compressSourceFromStream:   compressData,
+	}
+	sfa.execute()
+	return sfa.result()
+}
+
+func (bu *bindUploader) createStageIfNeeded() error {
+	if bu.arrayBindStage != "" {
+		return nil
+	}
+	data, err := bu.sc.exec(bu.ctx, createStageStmt, false, false, false, []driver.NamedValue{})
+	if !data.Success {
+		code, err := strconv.Atoi(data.Code)
+		if err != nil {
+			return err
+		}
+		return &SnowflakeError{
+			Number:   code,
+			SQLState: data.Data.SQLState,
+			Message:  err.Error(),
+			QueryID:  data.Data.QueryID}
+	}
+	if err != nil {
+		return err
+	}
+	bu.arrayBindStage = stageName
+	if err != nil {
+		newThreshold := "0"
+		bu.sc.cfg.Params[sessionArrayBindStageThreshold] = &newThreshold
+	}
+	return nil
+}
+
+// transpose the columns to rows and write them to a list of bytes
+func (bu *bindUploader) buildRowsAsBytes(columns []driver.NamedValue) ([][]byte, error) {
+	numColumns := len(columns)
+	if columns[0].Value == nil {
+		return nil, &SnowflakeError{
+			Number:  ErrBindSerialization,
+			Message: "no binds found in the first column",
+		}
+	}
+
+	_, column := snowflakeArrayToString(&columns[0])
+	numRows := len(column)
+	csvRows := make([][]byte, 0)
+	rows := make([][]string, 0)
+	for rowIdx := 0; rowIdx < numRows; rowIdx++ {
+		rows = append(rows, make([]string, numColumns))
+	}
+
+	for rowIdx := 0; rowIdx < numRows; rowIdx++ {
+		rows[rowIdx][0] = column[rowIdx]
+	}
+	for colIdx := 1; colIdx < numColumns; colIdx++ {
+		_, column = snowflakeArrayToString(&columns[colIdx])
+		iNumRows := len(column)
+		if iNumRows != numRows {
+			return nil, &SnowflakeError{
+				Number:      ErrBindSerialization,
+				Message:     errMsgBindColumnMismatch,
+				MessageArgs: []interface{}{colIdx, iNumRows, numRows},
+			}
+		}
+		for rowIdx := 0; rowIdx < numRows; rowIdx++ {
+			rows[rowIdx][colIdx] = column[rowIdx] // length of column = number of rows
+		}
+	}
+	for _, row := range rows {
+		csvRows = append(csvRows, bu.createCSVRecord(row))
+	}
+	return csvRows, nil
+}
+
+func (bu *bindUploader) createCSVRecord(data []string) []byte {
+	var b strings.Builder
+	b.Grow(1024)
+	for i := 0; i < len(data); i++ {
+		if i > 0 {
+			b.WriteString(",")
+		}
+		b.WriteString(escapeForCSV(data[i]))
+	}
+	b.WriteString("\n")
+	return []byte(b.String())
+}
+
+func (bu *bindUploader) close() {
+	if !bu.closed {
+		bu.closed = true
+	}
+}
+
+func getBindValues(bindings []driver.NamedValue) (map[string]execBindParameter, error) {
+	tsmode := timestampNtzType
+	idx := 1
+	var err error
+	bindValues := make(map[string]execBindParameter, len(bindings))
+	for _, binding := range bindings {
+		t := goTypeToSnowflake(binding.Value, tsmode)
+		if t == changeType {
+			tsmode, err = dataTypeMode(binding.Value)
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			var val interface{}
+			if t == sliceType {
+				// retrieve array binding data
+				t, val = snowflakeArrayToString(&binding)
+			} else {
+				val, err = valueToString(binding.Value, tsmode)
+				if err != nil {
+					return nil, err
+				}
+			}
+			if t == nullType || t == unSupportedType {
+				t = textType // if null or not supported, pass to GS as text
+			}
+			bindValues[strconv.Itoa(idx)] = execBindParameter{
+				Type:  t.String(),
+				Value: val,
+			}
+			idx++
+		}
+	}
+	return bindValues, nil
+}
+
+func arrayBindValueCount(bindValues []driver.NamedValue) int {
+	if !isArrayBind(bindValues) {
+		return 0
+	}
+	_, arr := snowflakeArrayToString(&bindValues[0])
+	return len(bindValues) * len(arr)
+}
+
+func isArrayBind(bindings []driver.NamedValue) bool {
+	if len(bindings) == 0 {
+		return false
+	}
+	for _, binding := range bindings {
+		if supported := supportedArrayBind(&binding); !supported {
+			return false
+		}
+	}
+	return true
+}
+
 func supportedArrayBind(nv *driver.NamedValue) bool {
 	switch reflect.TypeOf(nv.Value) {
 	case reflect.TypeOf(&intArray{}), reflect.TypeOf(&int32Array{}),
diff --git a/connection.go b/connection.go
index 48a056e..ad9401b 100644
--- a/connection.go
+++ b/connection.go
@@ -31,6 +31,7 @@ const (
 const (
 	sessionClientSessionKeepAlive          = "client_session_keep_alive"
 	sessionClientValidateDefaultParameters = "CLIENT_VALIDATE_DEFAULT_PARAMETERS"
+	sessionArrayBindStageThreshold         = "client_stage_array_binding_threshold"
 	serviceName                            = "service_name"
 )
 
@@ -58,7 +59,7 @@ func (sc *snowflakeConn) isDml(v int64) bool {
 
 // isMultiStmt returns true if the statement type code is of type multistatement
 // Note that the statement type code is also equivalent to type INSERT, so an additional check of the name is required
-func (sc *snowflakeConn) isMultiStmt(data execResponseData) bool {
+func (sc *snowflakeConn) isMultiStmt(data *execResponseData) bool {
 	return data.StatementTypeID == statementTypeIDMulti && data.RowType[0].Name == "multiple statement execution"
 }
 
@@ -86,45 +87,37 @@ func (sc *snowflakeConn) exec(
 	}
 	logger.WithContext(ctx).Infof("parameters: %v", req.Parameters)
 
-	tsmode := timestampNtzType
-	idx := 1
+	requestID := getOrGenerateRequestIDFromContext(ctx)
 	if len(bindings) > 0 {
-		req.Bindings = make(map[string]execBindParameter, len(bindings))
-		for _, binding := range bindings {
-			t := goTypeToSnowflake(binding.Value, tsmode)
-			logger.WithContext(ctx).Debugf("tmode: %v\n", t)
-			if t == changeType {
-				tsmode, err = dataTypeMode(binding.Value)
-				if err != nil {
-					return nil, err
-				}
-			} else {
-				var val interface{}
-				if t == sliceType {
-					// retrieve array binding data
-					t, val = snowflakeArrayToString(&binding)
-				} else {
-					val, err = valueToString(binding.Value, tsmode)
-					if err != nil {
-						return nil, err
-					}
-				}
-				if t == nullType || t == unSupportedType {
-					t = textType // if null or not supported, pass to GS as text
-				}
-				req.Bindings[strconv.Itoa(idx)] = execBindParameter{
-					Type:  t.String(),
-					Value: val,
-				}
-				idx++
+		arrayBindThreshold := sc.getArrayBindStageThreshold()
+		numBinds := arrayBindValueCount(bindings)
+		if 0 < arrayBindThreshold && arrayBindThreshold <= numBinds && !describeOnly && isArrayBind(bindings) {
+			// bulk array insert binding
+			uploader := bindUploader{
+				sc:        sc,
+				ctx:       ctx,
+				stagePath: "@" + stageName + "/" + requestID.String(),
 			}
+			uploader.upload(bindings)
+			req.Bindings = nil
+			req.BindStage = uploader.stagePath
+		} else {
+			// variable or array binding
+			req.Bindings, err = getBindValues(bindings)
+			if err != nil {
+				return nil, err
+			}
+			req.BindStage = ""
 		}
 	}
 	logger.WithContext(ctx).Infof("bindings: %v", req.Bindings)
 
 	headers := make(map[string]string)
 	headers["Content-Type"] = headerContentTypeApplicationJSON
-	headers["accept"] = headerAcceptTypeApplicationSnowflake // TODO v1.1: change to JSON in case of PUT/GET
+	headers["accept"] = headerAcceptTypeApplicationSnowflake
+	if isFileTransfer(query) {
+		headers["accept"] = headerContentTypeApplicationJSON
+	}
 	headers["User-Agent"] = userAgent
 	if serviceName, ok := sc.cfg.Params[serviceName]; ok {
 		headers["X-Snowflake-Service"] = *serviceName
@@ -136,8 +129,6 @@ func (sc *snowflakeConn) exec(
 	}
 
 	var data *execResponse
-
-	requestID := getOrGenerateRequestIDFromContext(ctx)
 	data, err = sc.rest.FuncPostQuery(ctx, sc.rest, &url.Values{}, headers, jsonBody, sc.rest.RequestTimeout, requestID, sc.cfg)
 	if err != nil {
 		return data, err
@@ -297,7 +288,7 @@ func (sc *snowflakeConn) ExecContext(ctx context.Context, query string, args []d
 			insertID:     -1,
 			queryID:      sc.QueryID,
 		}, nil // last insert id is not supported by Snowflake
-	} else if sc.isMultiStmt(data.Data) {
+	} else if sc.isMultiStmt(&data.Data) {
 		return sc.handleMultiExec(ctx, data.Data)
 	}
 	logger.Debug("DDL")
@@ -343,7 +334,7 @@ func (sc *snowflakeConn) QueryContext(ctx context.Context, query string, args []
 	rows.ChunkDownloader = populateChunkDownloader(ctx, sc, data.Data)
 	rows.queryID = sc.QueryID
 
-	if sc.isMultiStmt(data.Data) {
+	if sc.isMultiStmt(&data.Data) {
 		err := sc.handleMultiQuery(ctx, data.Data, rows)
 		if err != nil {
 			return nil, err
@@ -431,6 +422,18 @@ func (sc *snowflakeConn) isClientSessionKeepAliveEnabled() bool {
 	return strings.Compare(*v, "true") == 0
 }
 
+func (sc *snowflakeConn) getArrayBindStageThreshold() int {
+	v, ok := sc.cfg.Params[sessionArrayBindStageThreshold]
+	if !ok {
+		return 0
+	}
+	num, err := strconv.Atoi(*v)
+	if err != nil {
+		return 0
+	}
+	return num
+}
+
 func (sc *snowflakeConn) startHeartBeat() {
 	if !sc.isClientSessionKeepAliveEnabled() {
 		return
@@ -668,7 +671,7 @@ func getAsync(
 			res.insertID = -1
 			if sc.isDml(respd.Data.StatementTypeID) {
 				res.affectedRows, _ = updateRows(respd.Data)
-			} else if sc.isMultiStmt(respd.Data) {
+			} else if sc.isMultiStmt(&respd.Data) {
 				r, err := sc.handleMultiExec(ctx, respd.Data)
 				if err != nil {
 					res.errChannel <- err
@@ -688,7 +691,7 @@ func getAsync(
 			rows.sc = sc
 			rows.ChunkDownloader = populateChunkDownloader(ctx, sc, respd.Data)
 			rows.queryID = respd.Data.QueryID
-			if sc.isMultiStmt(respd.Data) {
+			if sc.isMultiStmt(&respd.Data) {
 				err = sc.handleMultiQuery(ctx, respd.Data, rows)
 				if err != nil {
 					rows.errChannel <- err
diff --git a/driver.go b/driver.go
index 187ca2d..2c369f8 100644
--- a/driver.go
+++ b/driver.go
@@ -7,6 +7,7 @@ import (
 	"database/sql"
 	"database/sql/driver"
 	"net/http"
+	"os"
 	"sync/atomic"
 )
 
@@ -135,3 +136,7 @@ func init() {
 	sql.Register("snowflake", &SnowflakeDriver{})
 	logger.SetLogLevel("error")
 }
+
+func runningOnGithubAction() bool {
+	return os.Getenv("GITHUB_ACTIONS") != ""
+}
diff --git a/encrypt_util.go b/encrypt_util.go
new file mode 100644
index 0000000..0743d78
--- /dev/null
+++ b/encrypt_util.go
@@ -0,0 +1,206 @@
+// Copyright (c) 2021 Snowflake Computing Inc. All right reserved.
+
+package gosnowflake
+
+//lint:file-ignore U1000 TODO SNOW-29352
+
+import (
+	"bytes"
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/json"
+	"io"
+	"io/ioutil"
+	"os"
+)
+
+type snowflakeFileEncryption struct {
+	QueryStageMasterKey string `json:"queryStageMasterKey,omitempty"`
+	QueryID             string `json:"queryId,omitempty"`
+	SMKID               int64  `json:"smkId,omitempty"`
+}
+
+// PUT requests return a single encryptionMaterial object whereas GET requests
+// return a slice (array) of encryptionMaterial objects, both under the field
+// 'encryptionMaterial'
+type encryptionWrapper struct {
+	snowflakeFileEncryption
+	EncryptionMaterials []snowflakeFileEncryption
+}
+
+// override default behavior for wrapper
+func (ew *encryptionWrapper) UnmarshalJSON(data []byte) error {
+	// if GET, unmarshal slice of encryptionMaterial
+	if err := json.Unmarshal(data, &ew.EncryptionMaterials); err == nil {
+		return err
+	}
+	// else (if PUT), unmarshal the encryptionMaterial itself
+	return json.Unmarshal(data, &ew.snowflakeFileEncryption)
+}
+
+type encryptMetadata struct {
+	key     string
+	iv      string
+	matdesc string
+}
+
+// encryptStream encrypts a stream buffer using AES128 block cipher in CBC mode
+// with PKCS5 padding
+func encryptStream(
+	sfe *snowflakeFileEncryption,
+	src io.Reader,
+	out io.Writer,
+	chunkSize int) (*encryptMetadata, error) {
+	if chunkSize == 0 {
+		chunkSize = aes.BlockSize * 4 * 1024
+	}
+	decodedKey, _ := base64.StdEncoding.DecodeString(sfe.QueryStageMasterKey)
+	keySize := len(decodedKey)
+
+	ivData := getSecureRandom(aes.BlockSize)
+	fileKey := getSecureRandom(keySize)
+
+	block, _ := aes.NewCipher(fileKey)
+	mode := cipher.NewCBCEncrypter(block, ivData)
+	cipherText := make([]byte, chunkSize)
+
+	// encrypt file with CBC
+	var err error
+	padded := false
+	for {
+		chunk := make([]byte, chunkSize)
+		n, err := src.Read(chunk)
+		if n == 0 || err != nil {
+			break
+		} else if n%aes.BlockSize != 0 {
+			chunk = padBytesLength(chunk[:n], aes.BlockSize)
+			padded = true
+		}
+		mode.CryptBlocks(cipherText, chunk)
+		out.Write(cipherText[:len(chunk)])
+	}
+	if err != nil {
+		return nil, err
+	}
+	if !padded {
+		blockSizeCipher := bytes.Repeat([]byte{byte(aes.BlockSize)}, aes.BlockSize)
+		chunk := make([]byte, aes.BlockSize)
+		mode.CryptBlocks(chunk, blockSizeCipher)
+		out.Write(chunk)
+	}
+
+	// encrypt key with CFB
+	block, _ = aes.NewCipher(decodedKey)
+	stream := cipher.NewCFBEncrypter(block, ivData)
+	encryptedFileKey := make([]byte, len(fileKey))
+	stream.XORKeyStream(encryptedFileKey, fileKey)
+
+	matDesc := materialDescriptor{
+		sfe.SMKID,
+		sfe.QueryID,
+		keySize * 8,
+	}
+
+	return &encryptMetadata{
+		base64.StdEncoding.EncodeToString(encryptedFileKey),
+		base64.StdEncoding.EncodeToString(ivData),
+		matdescToUnicode(matDesc),
+	}, nil
+}
+
+func encryptFile(sfe *snowflakeFileEncryption, filename string, chunkSize int, tmpDir string) (*encryptMetadata, string, error) {
+	if chunkSize == 0 {
+		chunkSize = aes.BlockSize * 4 * 1024
+	}
+	tmpOutputFile, _ := ioutil.TempFile(tmpDir, baseName(filename)+"#")
+	infile, err := os.OpenFile(filename, os.O_CREATE|os.O_RDONLY, os.ModePerm)
+	if err != nil {
+		return nil, "", err
+	}
+	meta, err := encryptStream(sfe, infile, tmpOutputFile, chunkSize)
+	if err != nil {
+		return nil, "", err
+	}
+	return meta, tmpOutputFile.Name(), nil
+}
+
+func decryptFile(metadata *encryptMetadata, sfe *snowflakeFileEncryption, filename string, chunkSize int, tmpDir string) (string, error) {
+	if chunkSize == 0 {
+		chunkSize = aes.BlockSize * 4 * 1024
+	}
+	decodedKey, _ := base64.StdEncoding.DecodeString(sfe.QueryStageMasterKey)
+	keyBytes, _ := base64.StdEncoding.DecodeString(metadata.key) // encrypted file key
+	ivBytes, _ := base64.StdEncoding.DecodeString(metadata.iv)
+
+	// decrypt file key
+	var err error
+	block, _ := aes.NewCipher(decodedKey)
+	stream := cipher.NewCFBDecrypter(block, ivBytes)
+	fileKey := keyBytes
+	stream.XORKeyStream(fileKey, fileKey)
+
+	// decrypt file
+	block, _ = aes.NewCipher(fileKey)
+	mode := cipher.NewCBCDecrypter(block, ivBytes)
+
+	tmpOutputFile, err := ioutil.TempFile(tmpDir, baseName(filename)+"#")
+	if err != nil {
+		return "", err
+	}
+	infile, err := os.OpenFile(filename, os.O_RDONLY, os.ModePerm)
+	if err != nil {
+		return "", err
+	}
+	var totalFileSize int
+	var prevChunk []byte
+	for {
+		chunk := make([]byte, chunkSize)
+		n, err := infile.Read(chunk)
+		if n == 0 || err != nil {
+			break
+		}
+		totalFileSize += n
+		chunk = chunk[:n]
+		mode.CryptBlocks(chunk, chunk)
+		tmpOutputFile.Write(chunk)
+		prevChunk = chunk
+	}
+	if err != nil {
+		return "", err
+	}
+	if prevChunk != nil {
+		totalFileSize -= paddingOffset(prevChunk)
+	}
+	tmpOutputFile.Truncate(int64(totalFileSize))
+	return tmpOutputFile.Name(), nil
+}
+
+type materialDescriptor struct {
+	SmkID   int64  `json:"smkId"`
+	QueryID string `json:"queryId"`
+	KeySize int    `json:"keySize"`
+}
+
+func matdescToUnicode(matdesc materialDescriptor) string {
+	s, _ := json.Marshal(&matdesc)
+	return string(s)
+}
+
+func getSecureRandom(byteLength int) []byte {
+	token := make([]byte, byteLength)
+	rand.Read(token)
+	return token
+}
+
+func padBytesLength(src []byte, blockSize int) []byte {
+	padLength := blockSize - len(src)%blockSize
+	padText := bytes.Repeat([]byte{byte(padLength)}, padLength)
+	return append(src, padText...)
+}
+
+func paddingOffset(src []byte) int {
+	length := len(src)
+	return int(src[length-1])
+}
diff --git a/errors.go b/errors.go
index 1c08710..625a51c 100644
--- a/errors.go
+++ b/errors.go
@@ -1,4 +1,4 @@
-// Copyright (c) 2017-2019 Snowflake Computing Inc. All right reserved.
+// Copyright (c) 2017-2021 Snowflake Computing Inc. All right reserved.
 
 package gosnowflake
 
@@ -98,6 +98,30 @@ const (
 	// ErrNoDefaultTransactionIsolationLevel is an error code for the case where non default isolation level is specified.
 	ErrNoDefaultTransactionIsolationLevel = 263001
 
+	/* file transfer */
+
+	// ErrInvalidStageFs is an error code denoting an inavlid stage in the file system
+	ErrInvalidStageFs = 264001
+	// ErrFailedToUploadToStage is an error code denoting the failure to upload a file to the stage
+	ErrFailedToUploadToStage = 264003
+	// ErrInvalidStageLocation is an error code denoting an invalid stage location
+	ErrInvalidStageLocation = 264004
+	// ErrLocalPathNotDirectory is an error code denoting a local path that is not a directory
+	ErrLocalPathNotDirectory = 264005
+	// ErrFileNotExists is an error code denoting the file to be transferred does not exist
+	ErrFileNotExists = 264006
+	// ErrCompressionNotSupported is an error code denoting the user specified compression type is not supported
+	ErrCompressionNotSupported = 264007
+	// ErrInternalNotMatchEncryptMaterial is an error code denoting the encryption material specified does not match
+	ErrInternalNotMatchEncryptMaterial = 264008
+
+	/* binding */
+
+	// ErrBindSerialization is an error code for a failed serialization of bind variables
+	ErrBindSerialization = 265001
+	// ErrBindUpload is an error code for the uploading process of bind elements to the stage
+	ErrBindUpload = 265002
+
 	/* converter */
 
 	// ErrInvalidTimestampTz is an error code for the case where a returned TIMESTAMP_TZ internal value is invalid
@@ -156,6 +180,7 @@ const (
 	errMsgOCSPStatusUnknown                  = "OCSP unknown"
 	errMsgOCSPInvalidValidity                = "invalid validity: producedAt: %v, thisUpdate: %v, nextUpdate: %v"
 	errMsgOCSPNoOCSPResponderURL             = "no OCSP server is attached to the certificate. %v"
+	errMsgBindColumnMismatch                 = "column %v has a different number of binds (%v) than column 1 (%v)"
 )
 
 var (
diff --git a/file_compression_type.go b/file_compression_type.go
new file mode 100644
index 0000000..ac95876
--- /dev/null
+++ b/file_compression_type.go
@@ -0,0 +1,131 @@
+// Copyright (c) 2021 Snowflake Computing Inc. All right reserved.
+
+package gosnowflake
+
+import (
+	"fmt"
+	"strings"
+)
+
+type compressionType struct {
+	name          string
+	fileExtension string
+	mimeType      string
+	mimeSubtypes  []string
+	isSupported   bool
+}
+
+var compressionTypes = map[string]compressionType{
+	"GZIP": {
+		"GZIP",
+		".gz",
+		"application",
+		[]string{"gzip", "x-gzip"},
+		true,
+	},
+	"DEFLATE": {
+		"DEFLATE",
+		".deflate",
+		"application",
+		[]string{"zlib", "deflate"},
+		true,
+	},
+	"RAW_DEFLATE": {
+		"RAW_DEFLATE",
+		".raw_deflate",
+		"application",
+		[]string{"raw_deflate"},
+		true,
+	},
+	"BZIP2": {
+		"BZIP2",
+		".bz2",
+		"application",
+		[]string{"bzip2", "x-bzip2", "x-bz2", "x-bzip", "bz2"},
+		true,
+	},
+	"LZIP": {
+		"LZIP",
+		".lz",
+		"application",
+		[]string{"lzip", "x-lzip"},
+		false,
+	},
+	"LZMA": {
+		"LZMA",
+		".lzma",
+		"application",
+		[]string{"lzma", "x-lzma"},
+		false,
+	},
+	"LZO": {
+		"LZO",
+		".lzo",
+		"application",
+		[]string{"lzo", "x-lzo"},
+		false,
+	},
+	"XZ": {
+		"XZ",
+		".xz",
+		"application",
+		[]string{"xz", "x-xz"},
+		false,
+	},
+	"COMPRESS": {
+		"COMPRESS",
+		".Z",
+		"application",
+		[]string{"compress", "x-compress"},
+		false,
+	},
+	"PARQUET": {
+		"PARQUET",
+		".parquet",
+		"snowflake",
+		[]string{"parquet"},
+		true,
+	},
+	"ZSTD": {
+		"ZSTD",
+		".zst",
+		"application",
+		[]string{"zstd", "x-zstd"},
+		true,
+	},
+	"BROTLI": {
+		"BROTLI",
+		".br",
+		"application",
+		[]string{"br", "x-br"},
+		true,
+	},
+	"ORC": {
+		"ORC",
+		".orc",
+		"snowflake",
+		[]string{"orc"},
+		true,
+	},
+}
+
+type fileCompressionType struct {
+	subTypeToMeta map[string]compressionType
+}
+
+func (fct *fileCompressionType) init() {
+	fct.subTypeToMeta = make(map[string]compressionType)
+	for _, meta := range compressionTypes {
+		for _, subType := range meta.mimeSubtypes {
+			fct.subTypeToMeta[subType] = meta
+		}
+	}
+}
+
+func (fct *fileCompressionType) lookupByMimeSubType(mimeSubType string) *compressionType {
+	fmt.Println(mimeSubType)
+	if val, ok := fct.subTypeToMeta[strings.ToLower(mimeSubType)]; ok {
+		return &val
+	}
+	return new(compressionType)
+}
diff --git a/file_transfer_agent.go b/file_transfer_agent.go
new file mode 100644
index 0000000..dbd9584
--- /dev/null
+++ b/file_transfer_agent.go
@@ -0,0 +1,711 @@
+// Copyright (c) 2021 Snowflake Computing Inc. All right reserved.
+
+package gosnowflake
+
+//lint:file-ignore U1000 Ignore all unused code
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+)
+
+type cloudType string
+type commandType string
+
+const (
+	fileProtocol      = "file://"
+	dataSizeThreshold = 64 * 1024 * 1024
+	bigFileThreshold  = 200 * 1024 * 1024
+	injectWaitPut     = 0
+	isWindows         = runtime.GOOS == "windows"
+)
+
+const (
+	uploadCommand   commandType = "UPLOAD"
+	downloadCommand commandType = "DOWNLOAD"
+	unknownCommand  commandType = "UNKNOWN"
+	putRegexp       string      = `(?i)^(?:/\*.*\*/\s*)*put\s+`
+	getRegexp       string      = `(?i)^(?:/\*.*\*/\s*)*get\s+`
+)
+
+const (
+	s3Client    cloudType = "S3"
+	azureClient cloudType = "AZURE"
+	gcsClient   cloudType = "GCS"
+	local       cloudType = "LOCAL_FS"
+)
+
+type resultStatus int
+
+const (
+	errStatus resultStatus = iota
+	uploaded
+	downloaded
+	collision
+	skipped
+	renewToken
+	renewPresignedURL
+	notFoundFile
+	needRetry
+	needRetryWithLowerConcurrency
+)
+
+func (rs resultStatus) String() string {
+	return [...]string{"ERROR", "UPLOADED", "DOWNLOADED", "COLLISION",
+		"SKIPPED", "RENEW_TOKEN", "RENEW_PRESIGNED_URL", "NOT_FOUND_FILE",
+		"NEED_RETRY", "NEED_RETRY_WITH_LOWER_CONCURRENCY"}[rs]
+}
+
+type snowflakeFileTransferAgent struct {
+	data                        execResponseData
+	command                     string
+	commandType                 commandType
+	stageLocationType           cloudType
+	fileMetadata                []*fileMetadata
+	encryptionMaterial          []*snowflakeFileEncryption
+	stageInfo                   *execResponseStageInfo
+	results                     []*fileMetadata
+	srcLocations                []string
+	autoCompress                bool
+	srcCompression              string
+	parallel                    int64
+	overwrite                   bool
+	srcFiles                    []string
+	localLocation               string
+	raisePutGetError            bool
+	srcFileToEncryptionMaterial map[string]*snowflakeFileEncryption
+
+	/* streaming PUT */
+	sourceStream               *bytes.Buffer
+	dstFileNameForStreamSource string
+	compressSourceFromStream   bool
+
+	/* PUT */
+	forcePutOverwrite       bool
+	putCallback             *snowflakeProgressPercentage
+	putAzureCallback        *snowflakeProgressPercentage
+	putCallbackOutputStream *io.Writer
+
+	/* GET */
+	presignedURLs           []string
+	getCallback             *snowflakeProgressPercentage
+	getAzureCallback        *snowflakeProgressPercentage
+	getCallbackOutputStream *io.Writer
+}
+
+func (sfa *snowflakeFileTransferAgent) execute() error {
+	fmt.Println("execute")
+	var err error
+	err = sfa.parseCommand()
+	if err != nil {
+		return err
+	}
+	err = sfa.initFileMetadata()
+	if err != nil {
+		return err
+	}
+
+	if sfa.commandType == uploadCommand {
+		err = sfa.processFileCompressionType()
+		if err != nil {
+			return err
+		}
+	}
+
+	if sfa.stageLocationType == local {
+		if _, err := os.Stat(sfa.stageInfo.Location); os.IsNotExist(err) {
+			err = os.MkdirAll(sfa.stageInfo.Location, os.ModePerm) // TODO what if not enough permissions?
+			if err != nil {
+				panic(err)
+			}
+		}
+	}
+
+	smallFileMetas := make([]*fileMetadata, 0)
+	largeFileMetas := make([]*fileMetadata, 0)
+
+	for _, meta := range sfa.fileMetadata {
+		meta.overwrite = sfa.overwrite
+		meta.sfa = sfa
+		if sfa.stageLocationType != local {
+			meta.putCallback = sfa.putCallback
+			meta.putAzureCallback = sfa.putAzureCallback
+			meta.putCallbackOutputStream = sfa.putCallbackOutputStream
+
+			meta.getCallback = sfa.getCallback
+			meta.getAzureCallback = sfa.getAzureCallback
+			meta.getCallbackOutputStream = sfa.getCallbackOutputStream
+
+			sizeThreshold := dataSizeThreshold
+			if meta.srcFileSize > sizeThreshold {
+				meta.parallel = sfa.parallel
+				largeFileMetas = append(largeFileMetas, meta)
+			} else {
+				meta.parallel = 1
+				smallFileMetas = append(smallFileMetas, meta)
+			}
+		} else {
+			meta.parallel = 1
+			smallFileMetas = append(smallFileMetas, meta)
+		}
+	}
+
+	if sfa.commandType == uploadCommand {
+		sfa.upload(largeFileMetas, smallFileMetas)
+	}
+
+	//if sfa.commandType == downloadCommand {
+	//	// TODO SNOW-206124
+	//}
+	return nil
+}
+
+func (sfa *snowflakeFileTransferAgent) parseCommand() error {
+	fmt.Println("parse command")
+	if sfa.data.Command != "" {
+		sfa.commandType = commandType(sfa.data.Command)
+	} else {
+		sfa.commandType = unknownCommand
+	}
+
+	sfa.initEncryptionMaterial()
+	if len(sfa.data.SrcLocations) == 0 {
+		return &SnowflakeError{
+			Number:  ErrInvalidStageLocation,
+			Message: "failed to parse location",
+		}
+	}
+	sfa.srcLocations = sfa.data.SrcLocations
+
+	if sfa.commandType == uploadCommand {
+		if sfa.sourceStream != nil {
+			sfa.srcFiles = sfa.srcLocations // streaming PUT
+		}
+		sfa.autoCompress = sfa.data.AutoCompress
+		sfa.srcCompression = strings.ToLower(sfa.data.SourceCompression)
+	} else {
+		sfa.srcFiles = sfa.srcLocations
+		sfa.srcFileToEncryptionMaterial = make(map[string]*snowflakeFileEncryption)
+		if len(sfa.data.SrcLocations) == len(sfa.encryptionMaterial) {
+			for i, srcFile := range sfa.srcFiles {
+				sfa.srcFileToEncryptionMaterial[srcFile] = sfa.encryptionMaterial[i]
+			}
+		} else if len(sfa.encryptionMaterial) != 0 {
+			return &SnowflakeError{
+				Number: ErrInternalNotMatchEncryptMaterial,
+				Message: fmt.Sprintf("number of downlodading files doesn't "+
+					"match the encryption materials. files=%v, encmat=%v",
+					len(sfa.data.SrcLocations), len(sfa.encryptionMaterial)),
+			}
+		}
+
+		sfa.localLocation = expandUser(sfa.data.LocalLocation)
+		fi, _ := os.Stat(sfa.localLocation)
+		if !fi.IsDir() {
+			return &SnowflakeError{
+				Number: ErrLocalPathNotDirectory,
+				Message: fmt.Sprintf("the local path is not a directory: %v",
+					sfa.localLocation),
+			}
+		}
+	}
+
+	sfa.parallel = 1
+	if sfa.data.Parallel != 0 {
+		sfa.parallel = sfa.data.Parallel
+	}
+	sfa.overwrite = sfa.overwrite && sfa.forcePutOverwrite
+	sfa.stageLocationType = cloudType(sfa.data.StageInfo.LocationType)
+	sfa.stageInfo = &sfa.data.StageInfo
+	sfa.presignedURLs = make([]string, 0)
+	if len(sfa.data.PresignedURLs) != 0 {
+		sfa.presignedURLs = sfa.data.PresignedURLs
+	}
+
+	if sfa.getStorageClient(sfa.stageLocationType) == nil {
+		return &SnowflakeError{
+			Number: ErrInvalidStageFs,
+			Message: fmt.Sprintf("destination location type is not valid: %v",
+				sfa.stageLocationType),
+		}
+	}
+	return nil
+}
+
+func (sfa *snowflakeFileTransferAgent) initEncryptionMaterial() {
+	sfa.encryptionMaterial = make([]*snowflakeFileEncryption, 0)
+	wrapper := sfa.data.EncryptionMaterial
+
+	if sfa.commandType == uploadCommand {
+		sfa.encryptionMaterial = append(sfa.encryptionMaterial, &wrapper.snowflakeFileEncryption)
+	} else {
+		for _, encmat := range wrapper.EncryptionMaterials {
+			sfa.encryptionMaterial = append(sfa.encryptionMaterial, &encmat)
+		}
+	}
+}
+
+func (sfa *snowflakeFileTransferAgent) initFileMetadata() error {
+	fmt.Println("init file metadata")
+	sfa.fileMetadata = make([]*fileMetadata, 0)
+	if sfa.commandType == uploadCommand {
+		if len(sfa.srcFiles) == 0 {
+			fileName := sfa.data.SrcLocations
+			return &SnowflakeError{
+				Number: ErrFileNotExists,
+				Message: fmt.Sprintf("file does not exist: %v",
+					fileName),
+			}
+		}
+		if sfa.sourceStream != nil {
+			fileName := sfa.srcFiles[0]
+			srcFileSize := sfa.sourceStream.Len()
+			sfa.fileMetadata = append(sfa.fileMetadata, &fileMetadata{
+				name:              baseName(fileName),
+				srcFileName:       fileName,
+				srcStream:         sfa.sourceStream,
+				srcFileSize:       srcFileSize,
+				stageLocationType: sfa.stageLocationType,
+				stageInfo:         sfa.stageInfo,
+			})
+		}
+
+		if len(sfa.encryptionMaterial) > 0 {
+			for _, meta := range sfa.fileMetadata {
+				meta.encryptionMaterial = sfa.encryptionMaterial[0]
+			}
+		}
+	}
+	return nil
+}
+
+func (sfa *snowflakeFileTransferAgent) processFileCompressionType() error {
+	fmt.Println("process file compression type")
+	var userSpecifiedSourceCompression *compressionType
+	var autoDetect bool
+	fct := new(fileCompressionType)
+	fct.init()
+	if sfa.srcCompression == "auto_detect" {
+		autoDetect = true
+	} else if sfa.srcCompression == "none" {
+		autoDetect = false
+	} else {
+		userSpecifiedSourceCompression = fct.lookupByMimeSubType(sfa.srcCompression)
+		if userSpecifiedSourceCompression == nil || !userSpecifiedSourceCompression.isSupported {
+			return &SnowflakeError{
+				Number: ErrCompressionNotSupported,
+				Message: fmt.Sprintf("feature is not supported: %v",
+					userSpecifiedSourceCompression),
+			}
+		}
+		autoDetect = false
+	}
+
+	gzipCompression := compressionTypes["GZIP"]
+	if sfa.sourceStream != nil {
+		fileMeta := sfa.fileMetadata[0]
+		fileMeta.srcCompressionType = userSpecifiedSourceCompression
+		if sfa.compressSourceFromStream {
+			fileMeta.dstCompressionType = &gzipCompression
+			fileMeta.requireCompress = true
+		} else {
+			fileMeta.dstCompressionType = userSpecifiedSourceCompression
+			fileMeta.requireCompress = false
+		}
+
+		// add gz extension if file name doesn't have it
+		if sfa.compressSourceFromStream && strings.HasSuffix(sfa.dstFileNameForStreamSource, gzipCompression.fileExtension) {
+			fileMeta.dstFileName = sfa.dstFileNameForStreamSource + gzipCompression.fileExtension
+		} else {
+			fileMeta.dstFileName = sfa.dstFileNameForStreamSource
+		}
+	} else {
+		for _, meta := range sfa.fileMetadata {
+			fileName := meta.srcFileName
+			var currentFileCompressionType *compressionType
+			if autoDetect {
+				currentFileCompressionType = fct.lookupByMimeSubType(filepath.Ext(fileName))
+				encoding := ""
+				//mimeType := mime.TypeByExtension(filepath.Ext(fileName))
+				//fmt.Println(mimeType)
+				test := make([]byte, 4)
+				if encoding == "" {
+					if sfa.sourceStream != nil {
+						sfa.sourceStream.Read(test)
+					}
+					if fileName[len(fileName)-3:] == ".br" {
+						encoding = "br"
+					} else if len(test) > 0 && bytes.Equal(test[:3], []byte{'O', 'R', 'C'}) {
+						encoding = "orc"
+					} else if len(test) > 0 && bytes.Equal(test, []byte{'P', 'A', 'R', '1'}) {
+						encoding = "parquet"
+					}
+				}
+
+				if encoding != "" {
+					currentFileCompressionType = fct.lookupByMimeSubType(encoding)
+				}
+
+				if currentFileCompressionType != nil && !currentFileCompressionType.isSupported {
+					return &SnowflakeError{
+						Number: ErrCompressionNotSupported,
+						Message: fmt.Sprintf("feature is not supported: %v",
+							currentFileCompressionType),
+					}
+				}
+			} else {
+				currentFileCompressionType = userSpecifiedSourceCompression
+			}
+
+			if currentFileCompressionType != nil {
+				meta.srcCompressionType = currentFileCompressionType
+				if currentFileCompressionType.isSupported {
+					meta.dstCompressionType = currentFileCompressionType
+					meta.requireCompress = false
+					meta.dstFileName = meta.name
+				} else {
+					return &SnowflakeError{
+						Number: ErrCompressionNotSupported,
+						Message: fmt.Sprintf("feature is not supported: %v",
+							currentFileCompressionType),
+					}
+				}
+			} else {
+				meta.requireCompress = sfa.autoCompress
+				meta.srcCompressionType = nil
+				if sfa.autoCompress {
+					dstFileName := meta.name + compressionTypes["GZIP"].fileExtension
+					meta.dstFileName = dstFileName
+					meta.dstCompressionType = &gzipCompression
+				} else {
+					meta.dstFileName = meta.name
+					meta.dstCompressionType = nil
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (sfa *snowflakeFileTransferAgent) getLocalFilePathFromCommand(command string) string {
+	if len(command) == 0 || !strings.Contains(command, fileProtocol) {
+		return ""
+	}
+	if regexp.MustCompile(putRegexp).Match([]byte(command)) {
+		return ""
+	}
+
+	filePathBeginIdx := strings.Index(command, fileProtocol)
+	isFilePathQuoted := command[filePathBeginIdx-1] == '\''
+	filePathBeginIdx += len(fileProtocol)
+
+	filePath := ""
+	var filePathEndIdx = 0
+
+	if isFilePathQuoted {
+		filePathEndIdx = strings.Index(command[filePathBeginIdx:], "'")
+		if filePathEndIdx > filePathBeginIdx {
+			filePath = command[filePathBeginIdx:filePathEndIdx]
+		}
+	} else {
+		indexList := make([]int, 0)
+		delims := []rune{' ', '\n', ';'}
+		for _, delim := range delims {
+			index := strings.Index(command[filePathBeginIdx:], string(delim))
+			if index != -1 {
+				indexList = append(indexList, index)
+			}
+		}
+		filePathEndIdx = -1
+		if getMin(indexList) != -1 {
+			filePathEndIdx = getMin(indexList)
+		}
+
+		if filePathEndIdx > filePathBeginIdx {
+			filePath = command[filePathBeginIdx:filePathEndIdx]
+		} else {
+			filePath = command[filePathBeginIdx:]
+		}
+	}
+	return filePath
+}
+
+func (sfa *snowflakeFileTransferAgent) upload(largeFileMetadata []*fileMetadata, smallFileMetadata []*fileMetadata) {
+	storageClient := sfa.getStorageClient(sfa.stageLocationType)
+
+	for _, meta := range smallFileMetadata {
+		meta.client = storageClient
+	}
+	for _, meta := range largeFileMetadata {
+		meta.client = storageClient
+	}
+
+	if len(smallFileMetadata) > 0 {
+		sfa.uploadFilesParallel(smallFileMetadata)
+	}
+	if len(largeFileMetadata) > 0 {
+		sfa.uploadFilesSequential(largeFileMetadata)
+	}
+}
+
+func (sfa *snowflakeFileTransferAgent) download(largeFileMetadata []fileMetadata, smallFileMetadata []fileMetadata) {
+	// TODO SNOW-206124
+}
+
+func (sfa *snowflakeFileTransferAgent) uploadFilesParallel(fileMetas []*fileMetadata) error {
+	idx := 0
+	fileMetaLen := len(fileMetas)
+	var err error
+	for idx < fileMetaLen {
+		endOfIdx := intMin(fileMetaLen, idx+int(sfa.parallel))
+		targetMeta := fileMetas[idx:endOfIdx]
+		errChan := make(chan error, len(targetMeta))
+		for {
+			var wg sync.WaitGroup
+			results := make([]*fileMetadata, fileMetaLen)
+			for i, meta := range targetMeta {
+				wg.Add(1)
+				go func(k int, m *fileMetadata, errChan chan error) {
+					defer wg.Done()
+					var uploadErr error
+					results[k], uploadErr = sfa.uploadOneFile(m)
+					errChan <- uploadErr
+				}(i, meta, errChan)
+			}
+			wg.Wait()
+			count := 0
+			for err = range errChan {
+				if err != nil {
+					return err
+				}
+				count++
+				if count == len(targetMeta) {
+					break
+				}
+			}
+
+			retryMeta := make([]*fileMetadata, 0)
+			for _, result := range results {
+				if result.resStatus == renewToken || result.resStatus == renewPresignedURL {
+					retryMeta = append(retryMeta, result)
+				} else {
+					sfa.results = append(sfa.results, result)
+				}
+			}
+
+			if len(retryMeta) == 0 {
+				break
+			}
+			for _, result := range retryMeta {
+				if result.resStatus == renewPresignedURL {
+					break
+				}
+			}
+			targetMeta = retryMeta
+		}
+		if endOfIdx == fileMetaLen {
+			break
+		}
+		idx += int(sfa.parallel)
+	}
+	return err
+}
+
+func (sfa *snowflakeFileTransferAgent) uploadFilesSequential(fileMetas []*fileMetadata) error {
+	idx := 0
+	fileMetaLen := len(fileMetas)
+	for idx < fileMetaLen {
+		res, err := sfa.uploadOneFile(fileMetas[idx])
+		if err != nil {
+			return err
+		}
+		sfa.results = append(sfa.results, res)
+		idx++
+		if injectWaitPut > 0 {
+			time.Sleep(injectWaitPut)
+		}
+	}
+	return nil
+}
+
+func (sfa *snowflakeFileTransferAgent) uploadOneFile(meta *fileMetadata) (*fileMetadata, error) {
+	fmt.Println("upload one file")
+	meta.realSrcFileName = meta.srcFileName
+	tmpDir := os.TempDir()
+	meta.tmpDir = tmpDir
+	defer os.RemoveAll(tmpDir) // cleanup
+
+	fileUtil := new(snowflakeFileUtil)
+	if meta.requireCompress {
+		if meta.srcStream != nil {
+			meta.realSrcStream, _ = fileUtil.compressFileWithGzipFromStream(&meta.srcStream)
+		} else {
+			panic("not implemented") // TODO SNOW-29352
+		}
+	}
+
+	if meta.srcStream != nil {
+		if meta.realSrcStream != nil {
+			meta.sha256Digest, meta.uploadSize = fileUtil.getDigestAndSizeForStream(&meta.realSrcStream)
+		} else {
+			meta.sha256Digest, meta.uploadSize = fileUtil.getDigestAndSizeForStream(&meta.srcStream)
+		}
+
+	} else {
+		panic("not implemented") // TODO SNOW-29352
+	}
+
+	client := sfa.getStorageClient(sfa.stageLocationType)
+	err := client.uploadOneFileWithRetry(meta)
+	if err != nil {
+		return nil, err
+	}
+	return meta, nil
+}
+
+func (sfa *snowflakeFileTransferAgent) downloadFilesParallel(fileMetas []fileMetadata) {
+	// TODO SNOW-206124
+}
+
+func (sfa *snowflakeFileTransferAgent) downloadFilesSequential(fileMetas []fileMetadata) {
+	// TODO SNOW-206124
+}
+
+func (sfa *snowflakeFileTransferAgent) downloadOneFile(meta fileMetadata) fileMetadata {
+	// TODO SNOW-206124
+	return fileMetadata{}
+}
+
+func (sfa *snowflakeFileTransferAgent) getStorageClient(stageLocationType cloudType) storageUtil {
+	if stageLocationType == local {
+		return &localUtil{}
+	} else if stageLocationType == s3Client {
+		return nil // TODO SNOW-29352
+	} else if stageLocationType == azureClient {
+		return nil // TODO SNOW-29352
+	} else if stageLocationType == gcsClient {
+		return nil // TODO SNOW-29352
+	}
+	return nil
+}
+
+func (sfa *snowflakeFileTransferAgent) result() (*execResponseData, error) {
+	data := new(execResponseData)
+	rowset := make([]fileTransferResultType, 0)
+	if sfa.commandType == uploadCommand {
+		if len(sfa.results) > 0 {
+			for _, meta := range sfa.results {
+				var srcCompressionType, dstCompressionType *compressionType
+				if meta.srcCompressionType != nil {
+					srcCompressionType = meta.srcCompressionType
+				} else {
+					srcCompressionType = &compressionType{
+						name: "NONE",
+					}
+				}
+				if meta.dstCompressionType != nil {
+					dstCompressionType = meta.dstCompressionType
+				} else {
+					dstCompressionType = &compressionType{
+						name: "NONE",
+					}
+				}
+				errorDetails := meta.errorDetails
+				srcFileSize := meta.srcFileSize
+				dstFileSize := meta.dstFileSize
+				if sfa.raisePutGetError && errorDetails != nil {
+					return nil, &SnowflakeError{
+						Number:  ErrFailedToUploadToStage,
+						Message: errorDetails.Error(),
+					}
+				}
+				rowset = append(rowset, fileTransferResultType{
+					meta.name,
+					meta.srcFileName,
+					meta.dstFileName,
+					srcFileSize,
+					dstFileSize,
+					srcCompressionType,
+					dstCompressionType,
+					meta.resStatus,
+					meta.errorDetails,
+				})
+			}
+			sort.Slice(rowset, func(i, j int) bool {
+				return rowset[i].srcFileName < rowset[j].srcFileName
+			})
+			ccrs := make([][]*string, 0, len(rowset))
+			for _, rs := range rowset {
+				srcFileSize := fmt.Sprintf("%v", rs.srcFileSize)
+				dstFileSize := fmt.Sprintf("%v", rs.dstFileSize)
+				resStatus := rs.resStatus.String()
+				errorStr := ""
+				if rs.errorDetails != nil {
+					errorStr = rs.errorDetails.Error()
+				}
+				ccrs = append(ccrs, []*string{
+					&rs.srcFileName,
+					&rs.dstFileName,
+					&srcFileSize,
+					&dstFileSize,
+					&rs.srcCompressionType.name,
+					&rs.dstCompressionType.name,
+					&resStatus,
+					&errorStr,
+				})
+			}
+			data.RowSet = ccrs
+			cc := make([]chunkRowType, len(ccrs))
+			populateJSONRowSet(cc, ccrs)
+			rt := []execResponseRowType{
+				{Name: "source", ByteLength: 10000, Length: 10000, Type: "TEXT", Scale: 0, Nullable: false},
+				{Name: "target", ByteLength: 10000, Length: 10000, Type: "TEXT", Scale: 0, Nullable: false},
+				{Name: "source_size", ByteLength: 64, Length: 64, Type: "FIXED", Scale: 0, Nullable: false},
+				{Name: "target_size", ByteLength: 64, Length: 64, Type: "FIXED", Scale: 0, Nullable: false},
+				{Name: "source_compression", ByteLength: 10000, Length: 10000, Type: "TEXT", Scale: 0, Nullable: false},
+				{Name: "target_compression", ByteLength: 10000, Length: 10000, Type: "TEXT", Scale: 0, Nullable: false},
+				{Name: "status", ByteLength: 10000, Length: 10000, Type: "TEXT", Scale: 0, Nullable: false},
+				{Name: "message", ByteLength: 10000, Length: 10000, Type: "TEXT", Scale: 0, Nullable: false},
+			}
+			data.RowType = rt
+			//rows.ChunkDownloader = &snowflakeChunkDownloader{
+			//	ctx:                context.Background(),
+			//	CurrentChunk:       cc,
+			//	Total:              int64(len(cc)),
+			//	ChunkMetas:         []execResponseChunk{},
+			//	TotalRowIndex:      -1,
+			//	Qrmk:               "",
+			//	RowSet:             rowSetType{RowType: rt},
+			//	FuncDownload:       nil,
+			//	FuncDownloadHelper: nil,
+			//}
+			return data, nil
+		}
+	}
+	return new(execResponseData), nil
+}
+
+func isFileTransfer(query string) bool {
+	putRe := regexp.MustCompile(putRegexp)
+	getRe := regexp.MustCompile(getRegexp)
+	return putRe.Match([]byte(query)) || getRe.Match([]byte(query))
+}
+
+type snowflakeProgressPercentage struct {
+	filename        string
+	fileSize        float64
+	outputStream    *io.Writer
+	showProgressBar bool
+	seenSoFar       int
+	done            bool
+	startTime       time.Time
+}
diff --git a/file_util.go b/file_util.go
new file mode 100644
index 0000000..d37b54d
--- /dev/null
+++ b/file_util.go
@@ -0,0 +1,152 @@
+// Copyright (c) 2021 Snowflake Computing Inc. All right reserved.
+
+package gosnowflake
+
+//lint:file-ignore U1000 TODO SNOW-29352
+
+import (
+	"bytes"
+	"compress/gzip"
+	"crypto/sha256"
+	"encoding/base64"
+	"io"
+	"io/ioutil"
+	"net/url"
+	usr "os/user"
+	"path/filepath"
+	"strings"
+)
+
+type snowflakeFileUtil struct {
+}
+
+func (util *snowflakeFileUtil) compressFileWithGzipFromStream(srcStream **bytes.Buffer) (*bytes.Buffer, int) {
+	r := getReaderFromBuffer(srcStream)
+	buf, _ := ioutil.ReadAll(r)
+	var c bytes.Buffer
+	w := gzip.NewWriter(&c)
+	w.Write(buf) // write buf to gzip writer
+	w.Close()
+	return &c, c.Len()
+}
+
+func (util *snowflakeFileUtil) getDigestAndSize(src **bytes.Buffer) (string, int64) {
+	chunkSize := 16 * 4 * 1024
+	m := sha256.New()
+	r := getReaderFromBuffer(src)
+	for {
+		chunk := make([]byte, chunkSize)
+		n, err := r.Read(chunk)
+		if n == 0 || err != nil {
+			break
+		}
+		m.Write(chunk[:n])
+	}
+	return base64.StdEncoding.EncodeToString(m.Sum(nil)), int64((*src).Len())
+}
+
+func (util *snowflakeFileUtil) getDigestAndSizeForStream(stream **bytes.Buffer) (string, int64) {
+	return util.getDigestAndSize(stream)
+}
+
+func (util *snowflakeFileUtil) getDigestAndSizeForFile(fileName string) (string, int64) {
+	src, err := ioutil.ReadFile(fileName)
+	if err != nil {
+		panic(err)
+	}
+	buf := bytes.NewBuffer(src)
+	return util.getDigestAndSize(&buf)
+}
+
+// file metadata for PUT/GET
+type fileMetadata struct {
+	name              string
+	srcFileName       string
+	srcFileSize       int
+	stageLocationType cloudType
+	resStatus         resultStatus
+
+	stageInfo          *execResponseStageInfo
+	srcCompressionType *compressionType
+	dstCompressionType *compressionType
+	requireCompress    bool
+	dstFileName        string
+	parallel           int64
+	sha256Digest       string
+	uploadSize         int64
+	encryptionMaterial *snowflakeFileEncryption
+	dstFileSize        int64
+	overwrite          bool
+	sfa                *snowflakeFileTransferAgent
+	client             cloudClient // GetObjectOutput (S3), ContainerURL (Azure), string (GCS)
+	realSrcFileName    string
+	tmpDir             string
+	presignedURL       *url.URL
+	errorDetails       error
+	lastError          error
+	noSleepingTime     bool
+	lastMaxConcurrency int
+	localLocation      string
+	localDigest        string
+
+	/* streaming PUT */
+	srcStream     *bytes.Buffer
+	realSrcStream *bytes.Buffer
+
+	/* PUT */
+	putCallback             *snowflakeProgressPercentage
+	putAzureCallback        *snowflakeProgressPercentage
+	putCallbackOutputStream *io.Writer
+
+	/* GET */
+	getCallback             *snowflakeProgressPercentage
+	getAzureCallback        *snowflakeProgressPercentage
+	getCallbackOutputStream *io.Writer
+}
+
+type fileTransferResultType struct {
+	name               string
+	srcFileName        string
+	dstFileName        string
+	srcFileSize        int
+	dstFileSize        int64
+	srcCompressionType *compressionType
+	dstCompressionType *compressionType
+	resStatus          resultStatus
+	errorDetails       error
+}
+
+type fileHeader struct {
+	digest             string
+	contentLength      int64
+	encryptionMetadata *encryptMetadata
+}
+
+func getReaderFromBuffer(src **bytes.Buffer) io.Reader {
+	var b bytes.Buffer
+	tee := io.TeeReader(*src, &b) // read src to buf
+	*src = &b                     // revert pointer back
+	return tee
+}
+
+func baseName(path string) string {
+	base := filepath.Base(path)
+	if base == "." || base == "/" {
+		return ""
+	}
+	if len(base) > 1 && (path[len(path)-1:] == "." || path[len(path)-1:] == "/") {
+		return ""
+	}
+	return base
+}
+
+func expandUser(path string) string {
+	usr, _ := usr.Current()
+	dir := usr.HomeDir
+	if path == "~" {
+		path = dir
+	} else if strings.HasPrefix(path, "~/") {
+		path = filepath.Join(dir, path[2:])
+	}
+	return path
+}
diff --git a/local_util.go b/local_util.go
new file mode 100644
index 0000000..63bff88
--- /dev/null
+++ b/local_util.go
@@ -0,0 +1,59 @@
+// Copyright (c) 2021 Snowflake Computing Inc. All right reserved.
+
+package gosnowflake
+
+import (
+	"bufio"
+	"io"
+	"os"
+	"path/filepath"
+)
+
+type localUtil struct {
+}
+
+func (util *localUtil) createClient(info *execResponseStageInfo, useAccelerateEndpoint bool) cloudClient {
+	return nil
+}
+
+func (util *localUtil) uploadOneFileWithRetry(meta *fileMetadata) error {
+	var frd *bufio.Reader
+	if meta.srcStream != nil {
+		b := meta.srcStream
+		if meta.realSrcStream != nil {
+			b = meta.realSrcStream
+		}
+		frd = bufio.NewReader(b)
+	} else {
+		f, _ := os.Open(meta.realSrcFileName)
+		defer f.Close()
+		frd = bufio.NewReader(f)
+	}
+
+	output, err := os.OpenFile(filepath.Join(expandUser(meta.stageInfo.Location), meta.dstFileName), os.O_CREATE|os.O_WRONLY, os.ModePerm)
+	if err != nil {
+		return err
+	}
+	defer output.Close()
+	data := make([]byte, meta.uploadSize)
+	for {
+		n, err := frd.Read(data)
+		if err != nil && err != io.EOF {
+			return err
+		}
+		if n == 0 {
+			break
+		}
+
+		if _, err = output.Write(data); err != nil {
+			return err
+		}
+	}
+	meta.dstFileSize = meta.uploadSize
+	meta.resStatus = uploaded
+	return nil
+}
+
+func (util *localUtil) downloadOneFile() {
+	// TODO SNOW-206124
+}
diff --git a/query.go b/query.go
index ac9d6eb..736c583 100644
--- a/query.go
+++ b/query.go
@@ -2,6 +2,8 @@
 
 package gosnowflake
 
+//lint:file-ignore U1000 Ignore all unused code
+
 import (
 	"time"
 )
@@ -26,6 +28,7 @@ type execRequest struct {
 	DescribeOnly bool                         `json:"describeOnly,omitempty"`
 	Parameters   map[string]interface{}       `json:"parameters,omitempty"`
 	Bindings     map[string]execBindParameter `json:"bindings,omitempty"`
+	BindStage    string                       `json:"bindStage,omitempty"`
 }
 
 type execResponseRowType struct {
@@ -45,6 +48,28 @@ type execResponseChunk struct {
 	CompressedSize   int64  `json:"compressedSize"`
 }
 
+type execResponseCredentials struct {
+	AwsKeyID       string `json:"AWS_KEY_ID,omitempty"`
+	AwsSecretKey   string `json:"AWS_SECRET_KEY,omitempty"`
+	AwsToken       string `json:"AWS_TOKEN,omitempty"`
+	AwsID          string `json:"AWS_ID,omitempty"`
+	AwsKey         string `json:"AWS_KEY,omitempty"`
+	AzureSasToken  string `json:"AZURE_SAS_TOKEN,omitempty"`
+	GcsAccessToken string `json:"GCS_ACCESS_TOKEN,omitempty"`
+}
+
+type execResponseStageInfo struct {
+	LocationType          string                  `json:"locationType,omitempty"`
+	Location              string                  `json:"location,omitempty"`
+	Path                  string                  `json:"path,omitempty"`
+	Region                string                  `json:"region,omitempty"`
+	StorageAccount        string                  `json:"storageAccount,omitempty"`
+	IsClientSideEncrypted bool                    `json:"isClientSideEncrypted,omitempty"`
+	Creds                 execResponseCredentials `json:"creds,omitempty"`
+	PresignedURL          string                  `json:"presignedUrl,omitempty"`
+	EndPoint              string                  `json:"endPoint,omitempty"`
+}
+
 // make all data field optional
 type execResponseData struct {
 	// succeed query response data
@@ -79,6 +104,23 @@ type execResponseData struct {
 	// async response placeholders
 	AsyncResult *snowflakeResult `json:"asyncResult,omitempty"`
 	AsyncRows   *snowflakeRows   `json:"asyncRows,omitempty"`
+
+	// file transfer response data
+	UploadInfo              execResponseStageInfo `json:"uploadInfo,omitempty"`
+	LocalLocation           string                `json:"localLocation,omitempty"`
+	SrcLocations            []string              `json:"src_locations,omitempty"`
+	Parallel                int64                 `json:"parallel,omitempty"`
+	Threshold               int64                 `json:"threshold,omitempty"`
+	AutoCompress            bool                  `json:"autoCompress,omitempty"`
+	Overwrite               bool                  `json:"overwrite,omitempty"`
+	SourceCompression       string                `json:"sourceCompression,omitempty"`
+	ShowEncryptionParameter bool                  `json:"clientShowEncryptionParameter,omitempty"`
+	EncryptionMaterial      encryptionWrapper     `json:"encryptionMaterial,omitempty"`
+	PresignedURLs           []string              `json:"presignedUrl,omitempty"`
+	StageInfo               execResponseStageInfo `json:"stageInfo,omitempty"`
+	Command                 string                `json:"command,omitempty"`
+	Kind                    string                `json:"kind,omitempty"`
+	Operation               string                `json:"operation,omitempty"`
 }
 
 type execResponse struct {
diff --git a/storage_client.go b/storage_client.go
new file mode 100644
index 0000000..1c5e3c6
--- /dev/null
+++ b/storage_client.go
@@ -0,0 +1,21 @@
+// Copyright (c) 2021 Snowflake Computing Inc. All right reserved.
+
+package gosnowflake
+
+//lint:file-ignore U1000 TODO SNOW-29352
+
+// implemented by localUtil and cloudUtil
+type storageUtil interface {
+	createClient(*execResponseStageInfo, bool) cloudClient
+	uploadOneFileWithRetry(*fileMetadata) error
+	downloadOneFile()
+}
+
+// implemented by snowflakeS3Util, snowflakeAzureUtil and snowflakeGcsUtil
+type cloudUtil interface {
+	getFileHeader(*fileMetadata, string) *fileHeader
+	uploadFile(string, *fileMetadata, *encryptMetadata, int64)
+	nativeDownloadFile()
+}
+
+type cloudClient interface{}
diff --git a/util.go b/util.go
index 42dfc20..a568d61 100644
--- a/util.go
+++ b/util.go
@@ -1,10 +1,11 @@
-// Copyright (c) 2017-2019 Snowflake Computing Inc. All right reserved.
+// Copyright (c) 2017-2021 Snowflake Computing Inc. All right reserved.
 
 package gosnowflake
 
 import (
 	"context"
 	"database/sql/driver"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -75,6 +76,19 @@ func intMax(a, b int) int {
 	return b
 }
 
+func getMin(arr []int) int {
+	if len(arr) == 0 {
+		return -1
+	}
+	min := arr[0]
+	for _, v := range arr {
+		if v <= min {
+			min = v
+		}
+	}
+	return min
+}
+
 // time.Duration max
 func durationMax(d1, d2 time.Duration) time.Duration {
 	if d1-d2 > 0 {
@@ -125,3 +139,14 @@ func (sta *simpleTokenAccessor) SetTokens(token string, masterToken string, sess
 	sta.masterToken = masterToken
 	sta.sessionID = sessionID
 }
+
+func escapeForCSV(value string) string {
+	if value == "" {
+		return "\"\""
+	}
+	if strings.Contains(value, "\"") || strings.Contains(value, "\n") ||
+		strings.Contains(value, ",") || strings.Contains(value, "\\") {
+		return "\"" + strings.ReplaceAll(value, "\"", "\"\"") + "\""
+	}
+	return value
+}


Test patch:
diff --git a/bindings_test.go b/bindings_test.go
index 6e633e7..6cbd329 100644
--- a/bindings_test.go
+++ b/bindings_test.go
@@ -320,3 +320,67 @@ func TestBindingArray(t *testing.T) {
 		}
 	})
 }
+
+// TestBindingBulkArray tests bulk array binding via the usage of the Array
+// function that converts the passed Golang slice to a Snowflake array type
+func TestBindingBulkArray(t *testing.T) {
+	if runningOnGithubAction() {
+		t.Skip("client_stage_array_binding_threshold value is internal")
+	}
+	runTests(t, dsn, func(dbt *DBTest) {
+		dbt.mustExec(createTableSQL)
+		defer dbt.mustExec(deleteTableSQL)
+		_, err := dbt.db.Exec("ALTER SESSION SET CLIENT_STAGE_ARRAY_BINDING_THRESHOLD = 1")
+		if err != nil {
+			t.Error(err)
+		}
+
+		intArray := []int{1, 2, 3}
+		fltArray := []float64{0.1, 2.34, 5.678}
+		boolArray := []bool{true, false, true}
+		strArray := []string{"test1", "test2", "test3"}
+		dbt.mustExec(insertSQL, Array(&intArray), Array(&fltArray), Array(&boolArray), Array(&strArray))
+		rows := dbt.mustQuery(selectAllSQL)
+		defer rows.Close()
+
+		var v1 int
+		var v2 float64
+		var v3 bool
+		var v4 string
+		if rows.Next() {
+			err := rows.Scan(&v1, &v2, &v3, &v4)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if v1 != 1 && v2 != 0.1 && v3 != true && v4 != "test1" {
+				t.Fatalf("failed to fetch. expected: 1, 0.1, true, test1. got: %v, %v, %v, %v", v1, v2, v3, v4)
+			}
+		} else {
+			t.Error("failed to query")
+		}
+
+		if rows.Next() {
+			err := rows.Scan(&v1, &v2, &v3, &v4)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if v1 != 2 && v2 != 2.34 && v3 != false && v4 != "test2" {
+				t.Fatalf("failed to fetch. expected: 2, 2.34, false, test2. got: %v, %v, %v, %v", v1, v2, v3, v4)
+			}
+		} else {
+			t.Error("failed to query")
+		}
+
+		if rows.Next() {
+			err := rows.Scan(&v1, &v2, &v3, &v4)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if v1 != 3 && v2 != 5.678 && v3 != true && v4 != "test3" {
+				t.Fatalf("failed to fetch. expected: 3, test3. got: %v, %v, %v, %v", v1, v2, v3, v4)
+			}
+		} else {
+			t.Error("failed to query")
+		}
+	})
+}
diff --git a/encrypt_util_test.go b/encrypt_util_test.go
new file mode 100644
index 0000000..cf02f67
--- /dev/null
+++ b/encrypt_util_test.go
@@ -0,0 +1,145 @@
+// Copyright (c) 2021 Snowflake Computing Inc. All right reserved.
+
+package gosnowflake
+
+import (
+	"bufio"
+	"compress/gzip"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"math/rand"
+	"os"
+	"os/exec"
+	"path"
+	"path/filepath"
+	"strconv"
+	"testing"
+	"time"
+)
+
+func TestEncryptDecryptFile(t *testing.T) {
+	encMat := snowflakeFileEncryption{
+		"ztke8tIdVt1zmlQIZm0BMA==",
+		"123873c7-3a66-40c4-ab89-e3722fbccce1",
+		3112,
+	}
+	data := "test data"
+	inputFile := "test_encrypt_decrypt_file"
+
+	fd, err := os.OpenFile(inputFile, os.O_CREATE|os.O_WRONLY, os.ModePerm)
+	if err != nil {
+		t.Error(err)
+	}
+	defer os.Remove(inputFile)
+	if _, err = fd.Write([]byte(data)); err != nil {
+		t.Error(err)
+	}
+
+	metadata, encryptedFile, err := encryptFile(&encMat, inputFile, 0, "")
+	if err != nil {
+		t.Error(err)
+	}
+	defer os.Remove(encryptedFile)
+	decryptedFile, err := decryptFile(metadata, &encMat, encryptedFile, 0, "")
+	if err != nil {
+		t.Error(err)
+	}
+	defer os.Remove(decryptedFile)
+
+	fd, _ = os.OpenFile(decryptedFile, os.O_RDONLY, os.ModePerm)
+	content, _ := ioutil.ReadAll(fd)
+	if string(content) != data {
+		t.Fatalf("data did not match content. expected: %v, got: %v", data, string(content))
+	}
+}
+
+func TestEncryptDecryptLargeFile(t *testing.T) {
+	encMat := snowflakeFileEncryption{
+		"ztke8tIdVt1zmlQIZm0BMA==",
+		"123873c7-3a66-40c4-ab89-e3722fbccce1",
+		3112,
+	}
+	numberOfFiles := 1
+	numberOfLines := 10000
+	tmpDir, _ := ioutil.TempDir("", "data")
+	tmpDir = generateKLinesOfNFiles(numberOfLines, numberOfFiles, false, tmpDir)
+	defer os.RemoveAll(tmpDir)
+	files, err := filepath.Glob(tmpDir + "/file*")
+	if err != nil {
+		t.Error(err)
+	}
+	inputFile := files[0]
+
+	metadata, encryptedFile, err := encryptFile(&encMat, inputFile, 0, "")
+	if err != nil {
+		t.Error(err)
+	}
+	defer os.Remove(encryptedFile)
+	decryptedFile, err := decryptFile(metadata, &encMat, encryptedFile, 0, "")
+	if err != nil {
+		t.Error(err)
+	}
+	defer os.Remove(decryptedFile)
+
+	cnt := 0
+	fd, _ := os.OpenFile(decryptedFile, os.O_RDONLY, os.ModePerm)
+	scanner := bufio.NewScanner(fd)
+	for scanner.Scan() {
+		cnt++
+	}
+	if err = scanner.Err(); err != nil {
+		t.Error(err)
+	}
+	if cnt != 10000 {
+		t.Fatalf("incorrect number of lines. expected: %v, got: %v", 10000, cnt)
+	}
+}
+
+func generateKLinesOfNFiles(k int, n int, compress bool, tmpDir string) string {
+	if tmpDir == "" {
+		tmpDir, _ = ioutil.TempDir(tmpDir, "data")
+	}
+	for i := 0; i < n; i++ {
+		fname := path.Join(tmpDir, "file"+strconv.FormatInt(int64(i), 10))
+		f, _ := os.OpenFile(fname, os.O_CREATE|os.O_WRONLY, os.ModePerm)
+		for j := 0; j < k; j++ {
+			num := rand.Float64() * 10000
+			tm := time.Unix(int64(rand.Float64()*30000-15000), 0)
+			dt := tm.Format("2021-03-01")
+			tm = time.Unix(int64(rand.Float64()*30000-15000), 0)
+			ts := tm.Format("2021-03-01 18:00:00")
+			tm = time.Unix(int64(rand.Float64()*30000-15000), 0)
+			tsltz := tm.Format("2021-03-01 18:00:00")
+			tm = time.Unix(int64(rand.Float64()*30000-15000), 0)
+			tsntz := tm.Format("2021-03-01 18:00:00")
+			tm = time.Unix(int64(rand.Float64()*30000-15000), 0)
+			tstz := tm.Format("2021-03-01 18:00:00")
+			pct := rand.Float64() * 1000
+			ratio := fmt.Sprintf("%.2f", rand.Float64()*1000)
+			rec := fmt.Sprintf("%v,%v,%v,%v,%v,%v,%v,%v\n", num, dt, ts, tsltz, tsntz, tstz, pct, ratio)
+			f.Write([]byte(rec))
+		}
+		if compress {
+			if !isWindows {
+				gzipCmd := exec.Command("gzip", filepath.Join(tmpDir, "file"+strconv.FormatInt(int64(i), 10)))
+				gzipOut, _ := gzipCmd.StdoutPipe()
+				gzipErr, _ := gzipCmd.StderrPipe()
+				gzipCmd.Start()
+				ioutil.ReadAll(gzipOut)
+				ioutil.ReadAll(gzipErr)
+				gzipCmd.Wait()
+			} else {
+				fOut, _ := os.OpenFile(fname+".gz", os.O_CREATE|os.O_WRONLY, os.ModePerm)
+				w := gzip.NewWriter(fOut)
+				defer w.Close()
+				fIn, _ := os.OpenFile(fname, os.O_RDONLY, os.ModePerm)
+				_, err := io.Copy(w, fIn)
+				if err != nil {
+					return ""
+				}
+			}
+		}
+	}
+	return tmpDir
+}
diff --git a/util_test.go b/util_test.go
index 865f9c1..ebc8c3c 100644
--- a/util_test.go
+++ b/util_test.go
@@ -181,3 +181,23 @@ func TestToNamedValues(t *testing.T) {
 		}
 	}
 }
+
+type tcIntArrayMin struct {
+	in  []int
+	out int
+}
+
+func TestGetMin(t *testing.T) {
+	testcases := []tcIntArrayMin{
+		{[]int{1, 2, 3, 4, 5}, 1},
+		{[]int{10, 25, 15, 5, 20}, 5},
+		{[]int{15, 12, 9, 6, 3}, 3},
+		{[]int{123, 123, 123, 123, 123}, 123},
+	}
+	for _, test := range testcases {
+		a := getMin(test.in)
+		if test.out != a {
+			t.Errorf("failed get min. in: %v, expected: %v, got: %v", test.in, test.out, a)
+		}
+	}
+}