Repository: snowflakedb/gosnowflake
Pull request #1052 (2024-02-17 00:28:01)

add enableArrowBatchesUtf8Validation to replace invalid utf8 chars in arrow batch
### Description


For historical reasons, snowflake used to allow users to upload invalid utf-8 characters. This caused arrow record to contain invalid utf-8 characters in arrow string columns. 

However by definition arrow.String column should contain utf-8 characters only, see https://arrow.apache.org/docs/cpp/api/datatype.html, 
https://github.com/apache/arrow/blob/a03d957b5b8d0425f9d5b6c98b6ee1efa56a1248/go/arrow/datatype.go#L73-L74

This could cause other upstream services that consume the bad arrow record to break, e.g Rust arrow lib force some basic validation, and this bad arrow record will fail the validation. 

This change introduce `enableArrowBatchesUtf8Validation`, when enabled, will iterate all values of string columns and replace any invalid character. 

### 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
- [x] Extended the README / documentation, if necessary


Yifeng-Sigma: cc: @sfc-gh-wfateem 
sfc-gh-dheyman-1: @Yifeng-Sigma , could you also update `doc.go` and include there this new context parameter?
Yifeng-Sigma: > @Yifeng-Sigma , could you also update `doc.go` and include there this new context parameter?

done

Patch:
diff --git a/converter.go b/converter.go
index 7d7b20e..609de6a 100644
--- a/converter.go
+++ b/converter.go
@@ -14,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"time"
+	"unicode/utf8"
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/array"
@@ -983,6 +984,15 @@ func higherPrecisionEnabled(ctx context.Context) bool {
 	return ok && d
 }
 
+func arrowBatchesUtf8ValidationEnabled(ctx context.Context) bool {
+	v := ctx.Value(enableArrowBatchesUtf8Validation)
+	if v == nil {
+		return false
+	}
+	d, ok := v.(bool)
+	return ok && d
+}
+
 func getArrowBatchesTimestampOption(ctx context.Context) snowflakeArrowBatchesTimestampOption {
 	v := ctx.Value(arrowBatchesTimestampOption)
 	if v == nil {
@@ -1098,6 +1108,26 @@ func arrowToRecord(ctx context.Context, record arrow.Record, pool memory.Allocat
 					}
 				}
 
+				newCol = tb.NewArray()
+				defer newCol.Release()
+			}
+		case textType:
+			if arrowBatchesUtf8ValidationEnabled(ctx) && col.DataType().ID() == arrow.STRING {
+				tb := array.NewStringBuilder(pool)
+				defer tb.Release()
+
+				for i := 0; i < int(numRows); i++ {
+					if col.(*array.String).IsValid(i) {
+						stringValue := col.(*array.String).Value(i)
+						if !utf8.ValidString(stringValue) {
+							logger.Error("Invalid UTF-8 characters detected while reading query response, column: ", srcColumnMeta.Name)
+							stringValue = strings.ToValidUTF8(stringValue, "�")
+						}
+						tb.Append(stringValue)
+					} else {
+						tb.AppendNull()
+					}
+				}
 				newCol = tb.NewArray()
 				defer newCol.Release()
 			}
diff --git a/doc.go b/doc.go
index e1c4042..1f61cab 100644
--- a/doc.go
+++ b/doc.go
@@ -495,6 +495,16 @@ If you want to use timestamps in Arrow batches, you have two options:
  2. You can use native Snowflake values. In that case you will receive complex structs as described above. To transform Snowflake values into the Golang time.Time struct you can use `ArrowSnowflakeTimestampToTime`.
     To enable this feature, you must use `WithArrowBatchesTimestampOption` context with value set to`UseOriginalTimestamp`.
 
+### Invalid UTF-8 characters in Arrow batches
+Snowflake previously allowed users to upload data with invalid UTF-8 characters. Consequently, Arrow records containing string columns in Snowflake could include these invalid UTF-8 characters.
+However, according to the Arrow specifications (https://arrow.apache.org/docs/cpp/api/datatype.html
+and https://github.com/apache/arrow/blob/a03d957b5b8d0425f9d5b6c98b6ee1efa56a1248/go/arrow/datatype.go#L73-L74),
+Arrow string columns should only contain UTF-8 characters.
+
+To address this issue and prevent potential downstream disruptions, the context `enableArrowBatchesUtf8Validation`, is introduced.
+When enabled, this feature iterates through all values in string columns, identifying and replacing any invalid characters with `�`.
+This ensures that Arrow records conform to the UTF-8 standards, preventing validation failures in downstream services like the Rust Arrow library that impose strict validation checks.
+
 # Binding Parameters
 
 Binding allows a SQL statement to use a value that is stored in a Golang variable.
diff --git a/util.go b/util.go
index 4e97a9b..e78aeba 100644
--- a/util.go
+++ b/util.go
@@ -20,18 +20,19 @@ import (
 type contextKey string
 
 const (
-	multiStatementCount         contextKey = "MULTI_STATEMENT_COUNT"
-	asyncMode                   contextKey = "ASYNC_MODE_QUERY"
-	queryIDChannel              contextKey = "QUERY_ID_CHANNEL"
-	snowflakeRequestIDKey       contextKey = "SNOWFLAKE_REQUEST_ID"
-	fetchResultByID             contextKey = "SF_FETCH_RESULT_BY_ID"
-	fileStreamFile              contextKey = "STREAMING_PUT_FILE"
-	fileTransferOptions         contextKey = "FILE_TRANSFER_OPTIONS"
-	enableHigherPrecision       contextKey = "ENABLE_HIGHER_PRECISION"
-	arrowBatches                contextKey = "ARROW_BATCHES"
-	arrowAlloc                  contextKey = "ARROW_ALLOC"
-	arrowBatchesTimestampOption contextKey = "ARROW_BATCHES_TIMESTAMP_OPTION"
-	queryTag                    contextKey = "QUERY_TAG"
+	multiStatementCount              contextKey = "MULTI_STATEMENT_COUNT"
+	asyncMode                        contextKey = "ASYNC_MODE_QUERY"
+	queryIDChannel                   contextKey = "QUERY_ID_CHANNEL"
+	snowflakeRequestIDKey            contextKey = "SNOWFLAKE_REQUEST_ID"
+	fetchResultByID                  contextKey = "SF_FETCH_RESULT_BY_ID"
+	fileStreamFile                   contextKey = "STREAMING_PUT_FILE"
+	fileTransferOptions              contextKey = "FILE_TRANSFER_OPTIONS"
+	enableHigherPrecision            contextKey = "ENABLE_HIGHER_PRECISION"
+	enableArrowBatchesUtf8Validation contextKey = "ENABLE_ARROW_BATCHES_UTF8_VALIDATION"
+	arrowBatches                     contextKey = "ARROW_BATCHES"
+	arrowAlloc                       contextKey = "ARROW_ALLOC"
+	arrowBatchesTimestampOption      contextKey = "ARROW_BATCHES_TIMESTAMP_OPTION"
+	queryTag                         contextKey = "QUERY_TAG"
 )
 
 const (
@@ -129,6 +130,15 @@ func WithArrowBatchesTimestampOption(ctx context.Context, option snowflakeArrowB
 	return context.WithValue(ctx, arrowBatchesTimestampOption, option)
 }
 
+// WithArrowBatchesUtf8Validation in combination with WithArrowBatches returns a context that
+// will validate and replace invalid UTF-8 characters in string columns with the replacement character
+// Theoretically, this should not be necessary, because arrow string column is only intended to contain valid UTF-8 characters.
+// However, in practice, it is possible that the data in the string column is not valid UTF-8.
+func WithArrowBatchesUtf8Validation(ctx context.Context) context.Context {
+	return context.WithValue(ctx, enableArrowBatchesUtf8Validation, true)
+
+}
+
 // WithQueryTag returns a context that will set the given tag as the QUERY_TAG
 // parameter on any queries that are run
 func WithQueryTag(ctx context.Context, tag string) context.Context {


Test patch:
diff --git a/converter_test.go b/converter_test.go
index 96d33b7..4c71f75 100644
--- a/converter_test.go
+++ b/converter_test.go
@@ -909,17 +909,19 @@ func TestArrowToRecord(t *testing.T) {
 	}
 
 	for _, tc := range []struct {
-		logical                     string
-		physical                    string
-		sc                          *arrow.Schema
-		rowType                     execResponseRowType
-		values                      interface{}
-		error                       string
-		arrowBatchesTimestampOption snowflakeArrowBatchesTimestampOption
-		nrows                       int
-		builder                     array.Builder
-		append                      func(b array.Builder, vs interface{})
-		compare                     func(src interface{}, rec arrow.Record) int
+		logical                          string
+		physical                         string
+		sc                               *arrow.Schema
+		rowType                          execResponseRowType
+		values                           interface{}
+		expected                         interface{}
+		error                            string
+		arrowBatchesTimestampOption      snowflakeArrowBatchesTimestampOption
+		enableArrowBatchesUtf8Validation bool
+		nrows                            int
+		builder                          array.Builder
+		append                           func(b array.Builder, vs interface{})
+		compare                          func(src interface{}, expected interface{}, rec arrow.Record) int
 	}{
 		{
 			logical:  "fixed",
@@ -946,7 +948,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Decimal128Builder).Append(num)
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]string)
 				for i, dec := range convertedRec.Column(0).(*array.Int64).Int64Values() {
 					num, ok := stringIntToDecimal(srcvs[i])
@@ -978,7 +980,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Decimal128Builder).Append(num)
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]string)
 				for i, dec := range convertedRec.Column(0).(*array.Float64).Float64Values() {
 					num, err := decimal128.FromString(srcvs[i], 38, 37)
@@ -1038,7 +1040,7 @@ func TestArrowToRecord(t *testing.T) {
 			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 arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]int8)
 				for i, f := range convertedRec.Column(0).(*array.Float64).Float64Values() {
 					rawFloat, _ := intToBigFloat(int64(srcvs[i]), 1).Float64()
@@ -1058,7 +1060,7 @@ func TestArrowToRecord(t *testing.T) {
 			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 arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]int16)
 				for i, f := range convertedRec.Column(0).(*array.Float64).Float64Values() {
 					rawFloat, _ := intToBigFloat(int64(srcvs[i]), 1).Float64()
@@ -1078,7 +1080,7 @@ func TestArrowToRecord(t *testing.T) {
 			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 arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]int32)
 				for i, f := range convertedRec.Column(0).(*array.Float64).Float64Values() {
 					rawFloat, _ := intToBigFloat(int64(srcvs[i]), 2).Float64()
@@ -1098,7 +1100,7 @@ func TestArrowToRecord(t *testing.T) {
 			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 arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]int64)
 				for i, f := range convertedRec.Column(0).(*array.Float64).Float64Values() {
 					rawFloat, _ := intToBigFloat(srcvs[i], 5).Float64()
@@ -1135,6 +1137,27 @@ func TestArrowToRecord(t *testing.T) {
 			builder:  array.NewStringBuilder(pool),
 			append:   func(b array.Builder, vs interface{}) { b.(*array.StringBuilder).AppendValues(vs.([]string), valids) },
 		},
+		{
+			logical:                          "text",
+			physical:                         "string with invalid utf8",
+			sc:                               arrow.NewSchema([]arrow.Field{{Type: &arrow.StringType{}}}, nil),
+			rowType:                          execResponseRowType{Type: "TEXT"},
+			values:                           []string{"\xFF", "bar", "baz\xFF\xFF"},
+			expected:                         []string{"�", "bar", "baz��"},
+			enableArrowBatchesUtf8Validation: true,
+			nrows:                            2,
+			builder:                          array.NewStringBuilder(pool),
+			append:                           func(b array.Builder, vs interface{}) { b.(*array.StringBuilder).AppendValues(vs.([]string), valids) },
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
+				arr := convertedRec.Column(0).(*array.String)
+				for i := 0; i < arr.Len(); i++ {
+					if expected.([]string)[i] != string(arr.Value(i)) {
+						return i
+					}
+				}
+				return -1
+			},
+		},
 		{
 			logical: "binary",
 			sc:      arrow.NewSchema([]arrow.Field{{Type: &arrow.BinaryType{}}}, nil),
@@ -1166,7 +1189,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Time64Builder).Append(arrow.Time64(t.UnixNano()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				arr := convertedRec.Column(0).(*array.Time64)
 				for i := 0; i < arr.Len(); i++ {
@@ -1190,7 +1213,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Int64Builder).Append(t.UnixMilli()) // Millisecond for scale = 3
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Nanosecond)) {
@@ -1217,7 +1240,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Nanosecond)) {
@@ -1246,7 +1269,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Microsecond)) {
@@ -1275,7 +1298,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Millisecond)) {
@@ -1304,7 +1327,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Second)) {
@@ -1328,7 +1351,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Int64Builder).Append(t.UnixMilli())
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int { return 0 },
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int { return 0 },
 		},
 		{
 			logical:                     "timestamp_ntz",
@@ -1344,7 +1367,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Int64Builder).Append(t.UnixMilli()) // Millisecond for scale = 3
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i := 0; i < convertedRec.Column(0).Len(); i++ {
 					ts := arrowSnowflakeTimestampToTime(convertedRec.Column(0), timestampNtzType, 3, i, nil)
@@ -1373,7 +1396,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i := 0; i < convertedRec.Column(0).Len(); i++ {
 					ts := arrowSnowflakeTimestampToTime(convertedRec.Column(0), timestampNtzType, 9, i, nil)
@@ -1397,7 +1420,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Int64Builder).Append(t.UnixMilli()) // Millisecond for scale = 3
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Nanosecond)) {
@@ -1424,7 +1447,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Nanosecond)) {
@@ -1453,7 +1476,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Microsecond)) {
@@ -1482,7 +1505,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Millisecond)) {
@@ -1511,7 +1534,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Second)) {
@@ -1535,7 +1558,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Int64Builder).Append(t.UnixMilli()) // Millisecond for scale = 3
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int { return 0 },
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int { return 0 },
 		},
 		{
 			logical:                     "timestamp_ltz",
@@ -1551,7 +1574,7 @@ func TestArrowToRecord(t *testing.T) {
 					b.(*array.Int64Builder).Append(t.UnixMilli()) // Millisecond for scale = 3
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i := 0; i < convertedRec.Column(0).Len(); i++ {
 					ts := arrowSnowflakeTimestampToTime(convertedRec.Column(0), timestampLtzType, 3, i, localTime.Location())
@@ -1580,7 +1603,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(t.Nanosecond()))
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i := 0; i < convertedRec.Column(0).Len(); i++ {
 					ts := arrowSnowflakeTimestampToTime(convertedRec.Column(0), timestampLtzType, 9, i, localTime.Location())
@@ -1608,7 +1631,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(0))      // timezone index - not important in tests
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Nanosecond)) {
@@ -1636,7 +1659,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(2).(*array.Int32Builder).Append(int32(0)) // timezone index - not important in tests
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Nanosecond)) {
@@ -1666,7 +1689,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(2).(*array.Int32Builder).Append(int32(0)) // timezone index - not important in tests
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Microsecond)) {
@@ -1696,7 +1719,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(2).(*array.Int32Builder).Append(int32(0)) // timezone index - not important in tests
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Millisecond)) {
@@ -1726,7 +1749,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(2).(*array.Int32Builder).Append(int32(0)) // timezone index - not important in tests
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i, t := range convertedRec.Column(0).(*array.Timestamp).TimestampValues() {
 					if !srcvs[i].Equal(t.ToTime(arrow.Second)) {
@@ -1754,7 +1777,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(1).(*array.Int32Builder).Append(int32(0))      // timezone index - not important in tests
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i := 0; i < convertedRec.Column(0).Len(); i++ {
 					ts := arrowSnowflakeTimestampToTime(convertedRec.Column(0), timestampTzType, 3, i, nil)
@@ -1784,7 +1807,7 @@ func TestArrowToRecord(t *testing.T) {
 					sb.FieldBuilder(2).(*array.Int32Builder).Append(int32(0)) // timezone index - not important in tests
 				}
 			},
-			compare: func(src interface{}, convertedRec arrow.Record) int {
+			compare: func(src interface{}, expected interface{}, convertedRec arrow.Record) int {
 				srcvs := src.([]time.Time)
 				for i := 0; i < convertedRec.Column(0).Len(); i++ {
 					ts := arrowSnowflakeTimestampToTime(convertedRec.Column(0), timestampTzType, 9, i, nil)
@@ -1853,6 +1876,10 @@ func TestArrowToRecord(t *testing.T) {
 				ctx = WithArrowBatchesTimestampOption(ctx, UseNanosecondTimestamp)
 			}
 
+			if tc.enableArrowBatchesUtf8Validation {
+				ctx = WithArrowBatchesUtf8Validation(ctx)
+			}
+
 			transformedRec, err := arrowToRecord(ctx, rawRec, pool, []execResponseRowType{meta}, localTime.Location())
 			if err != nil {
 				if tc.error == "" || !strings.Contains(err.Error(), tc.error) {
@@ -1865,7 +1892,7 @@ func TestArrowToRecord(t *testing.T) {
 				}
 
 				if tc.compare != nil {
-					idx := tc.compare(tc.values, transformedRec)
+					idx := tc.compare(tc.values, tc.expected, transformedRec)
 					if idx != -1 {
 						t.Fatalf("error: column array value mismatch at index %v", idx)
 					}