Repository: snowflakedb/gosnowflake
Pull request #1111 (2024-04-19 19:03:37)

SNOW-1264786 Support structured types in arrow batches
### Description

SNOW-1264786 Support for structured types in arrow batches mode

### Checklist
- [ ] Created tests which fail without the change (if possible)
- [ ] Extended the README / documentation, if necessary



Patch:
diff --git a/converter.go b/converter.go
index a8d2c56..65c0343 100644
--- a/converter.go
+++ b/converter.go
@@ -1966,200 +1966,213 @@ func arrowToRecord(ctx context.Context, record arrow.Record, pool memory.Allocat
 	ctxAlloc := compute.WithAllocator(ctx, pool)
 
 	for i, col := range record.Columns() {
-		srcColumnMeta := rowType[i]
+		fieldMetadata := rowType[i].toFieldMetadata()
 
-		// TODO: confirm that it is okay to be using higher precision logic for conversions
-		newCol := col
-		snowflakeType := getSnowflakeType(srcColumnMeta.Type)
-		switch snowflakeType {
-		case fixedType:
+		newCol, err := arrowToRecordSingleColumn(ctxAlloc, s.Field(i), col, fieldMetadata, higherPrecisionEnabled, arrowBatchesTimestampOption, pool, loc, numRows)
+		if err != nil {
+			return nil, err
+		}
+		cols = append(cols, newCol)
+		defer newCol.Release()
+	}
+	newRecord := array.NewRecord(s, cols, numRows)
+	return newRecord, nil
+}
+
+func arrowToRecordSingleColumn(ctx context.Context, field arrow.Field, col arrow.Array, fieldMetadata fieldMetadata, higherPrecisionEnabled bool, timestampOption snowflakeArrowBatchesTimestampOption, pool memory.Allocator, loc *time.Location, numRows int64) (arrow.Array, error) {
+	var err error
+	newCol := col
+	snowflakeType := getSnowflakeType(fieldMetadata.Type)
+	switch snowflakeType {
+	case fixedType:
+		if higherPrecisionEnabled {
+			// do nothing - return decimal as is
+			col.Retain()
+		} else if col.DataType().ID() == arrow.DECIMAL || col.DataType().ID() == arrow.DECIMAL256 {
 			var toType arrow.DataType
-			if higherPrecisionEnabled {
-				// do nothing - return decimal as is
-			} else if col.DataType().ID() == arrow.DECIMAL || col.DataType().ID() == arrow.DECIMAL256 {
-				if srcColumnMeta.Scale == 0 {
-					toType = arrow.PrimitiveTypes.Int64
-				} else {
-					toType = arrow.PrimitiveTypes.Float64
-				}
-				// we're fine truncating so no error for data loss here.
-				// so we use UnsafeCastOptions.
-				newCol, err = compute.CastArray(ctxAlloc, col, compute.UnsafeCastOptions(toType))
-				if err != nil {
-					return nil, err
-				}
-				defer newCol.Release()
-			} else if srcColumnMeta.Scale != 0 && col.DataType().ID() != arrow.INT64 {
-				result, err := compute.Divide(ctxAlloc, compute.ArithmeticOptions{NoCheckOverflow: true},
-					&compute.ArrayDatum{Value: newCol.Data()},
-					compute.NewDatum(math.Pow10(int(srcColumnMeta.Scale))))
-				if err != nil {
-					return nil, err
-				}
-				defer result.Release()
-				newCol = result.(*compute.ArrayDatum).MakeArray()
-				defer newCol.Release()
-			} else if srcColumnMeta.Scale != 0 && col.DataType().ID() == arrow.INT64 {
-				// gosnowflake driver uses compute.Divide() which could bring `integer value not in range: -9007199254740992 to 9007199254740992` error
-				// if we convert int64 to BigDecimal and then use compute.CastArray to convert BigDecimal to float64, we won't have enough precision.
-				// e.g 0.1 as (38,19) will result 0.09999999999999999
-				values := col.(*array.Int64).Int64Values()
-				floatValues := make([]float64, len(values))
-				for i, val := range values {
-					floatValues[i], _ = intToBigFloat(val, srcColumnMeta.Scale).Float64()
-				}
-				builder := array.NewFloat64Builder(memory.NewCheckedAllocator(memory.NewGoAllocator()))
-				builder.AppendValues(floatValues, nil)
-				newCol = builder.NewArray()
-				builder.Release()
-				defer newCol.Release()
+			if fieldMetadata.Scale == 0 {
+				toType = arrow.PrimitiveTypes.Int64
+			} else {
+				toType = arrow.PrimitiveTypes.Float64
 			}
-		case timeType:
-			newCol, err = compute.CastArray(ctxAlloc, col, compute.SafeCastOptions(arrow.FixedWidthTypes.Time64ns))
+			// we're fine truncating so no error for data loss here.
+			// so we use UnsafeCastOptions.
+			newCol, err = compute.CastArray(ctx, col, compute.UnsafeCastOptions(toType))
 			if err != nil {
 				return nil, err
 			}
-			defer newCol.Release()
-		case timestampNtzType, timestampLtzType, timestampTzType:
-			if arrowBatchesTimestampOption == UseOriginalTimestamp {
-				// do nothing - return timestamp as is
+		} else if fieldMetadata.Scale != 0 && col.DataType().ID() != arrow.INT64 {
+			result, err := compute.Divide(ctx, compute.ArithmeticOptions{NoCheckOverflow: true},
+				&compute.ArrayDatum{Value: newCol.Data()},
+				compute.NewDatum(math.Pow10(int(fieldMetadata.Scale))))
+			if err != nil {
+				return nil, err
+			}
+			defer result.Release()
+			newCol = result.(*compute.ArrayDatum).MakeArray()
+		} else if fieldMetadata.Scale != 0 && col.DataType().ID() == arrow.INT64 {
+			// gosnowflake driver uses compute.Divide() which could bring `integer value not in range: -9007199254740992 to 9007199254740992` error
+			// if we convert int64 to BigDecimal and then use compute.CastArray to convert BigDecimal to float64, we won't have enough precision.
+			// e.g 0.1 as (38,19) will result 0.09999999999999999
+			values := col.(*array.Int64).Int64Values()
+			floatValues := make([]float64, len(values))
+			for i, val := range values {
+				floatValues[i], _ = intToBigFloat(val, int64(fieldMetadata.Scale)).Float64()
+			}
+			builder := array.NewFloat64Builder(pool)
+			builder.AppendValues(floatValues, nil)
+			newCol = builder.NewArray()
+			builder.Release()
+		}
+	case timeType:
+		newCol, err = compute.CastArray(ctx, col, compute.SafeCastOptions(arrow.FixedWidthTypes.Time64ns))
+		if err != nil {
+			return nil, err
+		}
+	case timestampNtzType, timestampLtzType, timestampTzType:
+		if timestampOption == UseOriginalTimestamp {
+			// do nothing - return timestamp as is
+			col.Retain()
+		} else {
+			var unit arrow.TimeUnit
+			switch timestampOption {
+			case UseMicrosecondTimestamp:
+				unit = arrow.Microsecond
+			case UseMillisecondTimestamp:
+				unit = arrow.Millisecond
+			case UseSecondTimestamp:
+				unit = arrow.Second
+			case UseNanosecondTimestamp:
+				unit = arrow.Nanosecond
+			}
+			var tb *array.TimestampBuilder
+			if snowflakeType == timestampLtzType {
+				tb = array.NewTimestampBuilder(pool, &arrow.TimestampType{Unit: unit, TimeZone: loc.String()})
 			} else {
-				var unit arrow.TimeUnit
-				switch arrowBatchesTimestampOption {
-				case UseMicrosecondTimestamp:
-					unit = arrow.Microsecond
-				case UseMillisecondTimestamp:
-					unit = arrow.Millisecond
-				case UseSecondTimestamp:
-					unit = arrow.Second
-				case UseNanosecondTimestamp:
-					unit = arrow.Nanosecond
-				}
-				var tb *array.TimestampBuilder
-				if snowflakeType == timestampLtzType {
-					tb = array.NewTimestampBuilder(pool, &arrow.TimestampType{Unit: unit, TimeZone: loc.String()})
-				} else {
-					tb = array.NewTimestampBuilder(pool, &arrow.TimestampType{Unit: unit})
-				}
-				defer tb.Release()
-
-				for i := 0; i < int(numRows); i++ {
-					ts := arrowSnowflakeTimestampToTime(col, snowflakeType, int(srcColumnMeta.Scale), i, loc)
-					if ts != nil {
-						var ar arrow.Timestamp
-						switch arrowBatchesTimestampOption {
-						case UseMicrosecondTimestamp:
-							ar = arrow.Timestamp(ts.UnixMicro())
-						case UseMillisecondTimestamp:
-							ar = arrow.Timestamp(ts.UnixMilli())
-						case UseSecondTimestamp:
-							ar = arrow.Timestamp(ts.Unix())
-						case UseNanosecondTimestamp:
-							ar = arrow.Timestamp(ts.UnixNano())
-							// in case of overflow in arrow timestamp return error
-							// this could only happen for nanosecond case
-							if ts.UTC().Year() != ar.ToTime(arrow.Nanosecond).Year() {
-								return nil, &SnowflakeError{
-									Number:   ErrTooHighTimestampPrecision,
-									SQLState: SQLStateInvalidDataTimeFormat,
-									Message:  fmt.Sprintf("Cannot convert timestamp %v in column %v to Arrow.Timestamp data type due to too high precision. Please use context with WithOriginalTimestamp.", ts.UTC(), srcColumnMeta.Name),
-								}
+				tb = array.NewTimestampBuilder(pool, &arrow.TimestampType{Unit: unit})
+			}
+			defer tb.Release()
+
+			for i := 0; i < int(numRows); i++ {
+				ts := arrowSnowflakeTimestampToTime(col, snowflakeType, int(fieldMetadata.Scale), i, loc)
+				if ts != nil {
+					var ar arrow.Timestamp
+					switch timestampOption {
+					case UseMicrosecondTimestamp:
+						ar = arrow.Timestamp(ts.UnixMicro())
+					case UseMillisecondTimestamp:
+						ar = arrow.Timestamp(ts.UnixMilli())
+					case UseSecondTimestamp:
+						ar = arrow.Timestamp(ts.Unix())
+					case UseNanosecondTimestamp:
+						ar = arrow.Timestamp(ts.UnixNano())
+						// in case of overflow in arrow timestamp return error
+						// this could only happen for nanosecond case
+						if ts.UTC().Year() != ar.ToTime(arrow.Nanosecond).Year() {
+							return nil, &SnowflakeError{
+								Number:   ErrTooHighTimestampPrecision,
+								SQLState: SQLStateInvalidDataTimeFormat,
+								Message:  fmt.Sprintf("Cannot convert timestamp %v in column %v to Arrow.Timestamp data type due to too high precision. Please use context with WithOriginalTimestamp.", ts.UTC(), fieldMetadata.Name),
 							}
 						}
-						tb.Append(ar)
-					} else {
-						tb.AppendNull()
 					}
+					tb.Append(ar)
+				} else {
+					tb.AppendNull()
 				}
-
-				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.WithContext(ctx).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()
+		}
+	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.WithContext(ctx).Error("Invalid UTF-8 characters detected while reading query response, column: ", fieldMetadata.Name)
+						stringValue = strings.ToValidUTF8(stringValue, "�")
 					}
+					tb.Append(stringValue)
+				} else {
+					tb.AppendNull()
 				}
-				newCol = tb.NewArray()
-				defer newCol.Release()
 			}
+			newCol = tb.NewArray()
+		} else {
+			col.Retain()
 		}
-		cols = append(cols, newCol)
-	}
-	return array.NewRecord(s, cols, numRows), nil
-}
-
-func recordToSchema(sc *arrow.Schema, rowType []execResponseRowType, loc *time.Location, timestampOption snowflakeArrowBatchesTimestampOption, withHigherPrecision bool) (*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(srcColumnMeta.Type) {
-		case fixedType:
-			switch f.Type.ID() {
-			case arrow.DECIMAL:
-				if withHigherPrecision {
-					converted = false
-				} else if srcColumnMeta.Scale == 0 {
-					t = &arrow.Int64Type{}
-				} else {
-					t = &arrow.Float64Type{}
-				}
-			default:
-				if withHigherPrecision {
-					converted = false
-				} else if srcColumnMeta.Scale != 0 {
-					t = &arrow.Float64Type{}
-				} else {
-					converted = false
+	case objectType:
+		if structCol, ok := col.(*array.Struct); ok {
+			var internalCols []arrow.Array
+			for i := 0; i < structCol.NumField(); i++ {
+				internalCol := structCol.Field(i)
+				newInternalCol, err := arrowToRecordSingleColumn(ctx, field.Type.(*arrow.StructType).Field(i), internalCol, fieldMetadata.Fields[i], higherPrecisionEnabled, timestampOption, pool, loc, numRows)
+				if err != nil {
+					return nil, err
 				}
+				internalCols = append(internalCols, newInternalCol)
+				defer newInternalCol.Release()
 			}
-		case timeType:
-			t = &arrow.Time64Type{Unit: arrow.Nanosecond}
-		case timestampNtzType, timestampTzType:
-			if timestampOption == UseOriginalTimestamp {
-				// do nothing - return timestamp as is
-				converted = false
-			} else if timestampOption == UseMicrosecondTimestamp {
-				t = &arrow.TimestampType{Unit: arrow.Microsecond}
-			} else if timestampOption == UseMillisecondTimestamp {
-				t = &arrow.TimestampType{Unit: arrow.Millisecond}
-			} else if timestampOption == UseSecondTimestamp {
-				t = &arrow.TimestampType{Unit: arrow.Second}
-			} else {
-				t = &arrow.TimestampType{Unit: arrow.Nanosecond}
+			var fieldNames []string
+			for _, f := range field.Type.(*arrow.StructType).Fields() {
+				fieldNames = append(fieldNames, f.Name)
 			}
-		case timestampLtzType:
-			if timestampOption == UseOriginalTimestamp {
-				// do nothing - return timestamp as is
-				converted = false
-			} else if timestampOption == UseMicrosecondTimestamp {
-				t = &arrow.TimestampType{Unit: arrow.Microsecond, TimeZone: loc.String()}
-			} else if timestampOption == UseMillisecondTimestamp {
-				t = &arrow.TimestampType{Unit: arrow.Millisecond, TimeZone: loc.String()}
-			} else if timestampOption == UseSecondTimestamp {
-				t = &arrow.TimestampType{Unit: arrow.Second, TimeZone: loc.String()}
-			} else {
-				t = &arrow.TimestampType{Unit: arrow.Nanosecond, TimeZone: loc.String()}
+			nullBitmap := memory.NewBufferBytes(structCol.NullBitmapBytes())
+			numberOfNulls := structCol.NullN()
+			return array.NewStructArrayWithNulls(internalCols, fieldNames, nullBitmap, numberOfNulls, 0)
+		}
+	case arrayType:
+		if listCol, ok := col.(*array.List); ok {
+			newCol, err = arrowToRecordSingleColumn(ctx, field.Type.(*arrow.ListType).ElemField(), listCol.ListValues(), fieldMetadata.Fields[0], higherPrecisionEnabled, timestampOption, pool, loc, numRows)
+			if err != nil {
+				return nil, err
 			}
-		default:
-			converted = false
+			defer newCol.Release()
+			newData := array.NewData(arrow.ListOf(newCol.DataType()), listCol.Len(), listCol.Data().Buffers(), []arrow.ArrayData{newCol.Data()}, listCol.NullN(), 0)
+			defer newData.Release()
+			return array.NewListData(newData), nil
+		}
+	case mapType:
+		mapCol := col.(*array.Map)
+		keyCol, err := arrowToRecordSingleColumn(ctx, field.Type.(*arrow.MapType).KeyField(), mapCol.Keys(), fieldMetadata.Fields[0], higherPrecisionEnabled, timestampOption, pool, loc, numRows)
+		if err != nil {
+			return nil, err
+		}
+		defer keyCol.Release()
+		valueCol, err := arrowToRecordSingleColumn(ctx, field.Type.(*arrow.MapType).ItemField(), mapCol.Items(), fieldMetadata.Fields[1], higherPrecisionEnabled, timestampOption, pool, loc, numRows)
+		if err != nil {
+			return nil, err
+		}
+		defer valueCol.Release()
+
+		structArr, err := array.NewStructArray([]arrow.Array{keyCol, valueCol}, []string{"k", "v"})
+		if err != nil {
+			return nil, err
 		}
+		defer structArr.Release()
+		newData := array.NewData(arrow.MapOf(keyCol.DataType(), valueCol.DataType()), mapCol.Len(), mapCol.Data().Buffers(), []arrow.ArrayData{structArr.Data()}, mapCol.NullN(), 0)
+		defer newData.Release()
+		return array.NewMapData(newData), nil
+	default:
+		col.Retain()
+	}
+	return newCol, nil
+}
+
+func recordToSchema(sc *arrow.Schema, rowType []execResponseRowType, loc *time.Location, timestampOption snowflakeArrowBatchesTimestampOption, withHigherPrecision bool) (*arrow.Schema, error) {
+	fields := recordToSchemaRecursive(sc.Fields(), rowType, loc, timestampOption, withHigherPrecision)
+	meta := sc.Metadata()
+	return arrow.NewSchema(fields, &meta), nil
+}
+
+func recordToSchemaRecursive(inFields []arrow.Field, rowType []execResponseRowType, loc *time.Location, timestampOption snowflakeArrowBatchesTimestampOption, withHigherPrecision bool) []arrow.Field {
+	var outFields []arrow.Field
+	for i, f := range inFields {
+		fieldMetadata := rowType[i].toFieldMetadata()
+		converted, t := recordToSchemaSingleField(fieldMetadata, f, withHigherPrecision, timestampOption, loc)
 
 		newField := f
 		if converted {
@@ -2170,10 +2183,104 @@ func recordToSchema(sc *arrow.Schema, rowType []execResponseRowType, loc *time.L
 				Metadata: f.Metadata,
 			}
 		}
-		fields = append(fields, newField)
+		outFields = append(outFields, newField)
 	}
-	meta := sc.Metadata()
-	return arrow.NewSchema(fields, &meta), nil
+	return outFields
+}
+
+func recordToSchemaSingleField(fieldMetadata fieldMetadata, f arrow.Field, withHigherPrecision bool, timestampOption snowflakeArrowBatchesTimestampOption, loc *time.Location) (bool, arrow.DataType) {
+	t := f.Type
+	converted := true
+	switch getSnowflakeType(fieldMetadata.Type) {
+	case fixedType:
+		switch f.Type.ID() {
+		case arrow.DECIMAL:
+			if withHigherPrecision {
+				converted = false
+			} else if fieldMetadata.Scale == 0 {
+				t = &arrow.Int64Type{}
+			} else {
+				t = &arrow.Float64Type{}
+			}
+		default:
+			if withHigherPrecision {
+				converted = false
+			} else if fieldMetadata.Scale != 0 {
+				t = &arrow.Float64Type{}
+			} else {
+				converted = false
+			}
+		}
+	case timeType:
+		t = &arrow.Time64Type{Unit: arrow.Nanosecond}
+	case timestampNtzType, timestampTzType:
+		if timestampOption == UseOriginalTimestamp {
+			// do nothing - return timestamp as is
+			converted = false
+		} else if timestampOption == UseMicrosecondTimestamp {
+			t = &arrow.TimestampType{Unit: arrow.Microsecond}
+		} else if timestampOption == UseMillisecondTimestamp {
+			t = &arrow.TimestampType{Unit: arrow.Millisecond}
+		} else if timestampOption == UseSecondTimestamp {
+			t = &arrow.TimestampType{Unit: arrow.Second}
+		} else {
+			t = &arrow.TimestampType{Unit: arrow.Nanosecond}
+		}
+	case timestampLtzType:
+		if timestampOption == UseOriginalTimestamp {
+			// do nothing - return timestamp as is
+			converted = false
+		} else if timestampOption == UseMicrosecondTimestamp {
+			t = &arrow.TimestampType{Unit: arrow.Microsecond, TimeZone: loc.String()}
+		} else if timestampOption == UseMillisecondTimestamp {
+			t = &arrow.TimestampType{Unit: arrow.Millisecond, TimeZone: loc.String()}
+		} else if timestampOption == UseSecondTimestamp {
+			t = &arrow.TimestampType{Unit: arrow.Second, TimeZone: loc.String()}
+		} else {
+			t = &arrow.TimestampType{Unit: arrow.Nanosecond, TimeZone: loc.String()}
+		}
+	case objectType:
+		converted = false
+		if f.Type.ID() == arrow.STRUCT {
+			var internalFields []arrow.Field
+			for idx, internalField := range f.Type.(*arrow.StructType).Fields() {
+				println(fieldMetadata.String())
+				internalConverted, convertedDataType := recordToSchemaSingleField(fieldMetadata.Fields[idx], internalField, withHigherPrecision, timestampOption, loc)
+				converted = converted || internalConverted
+				if internalConverted {
+					newInternalField := arrow.Field{
+						Name:     internalField.Name,
+						Type:     convertedDataType,
+						Metadata: internalField.Metadata,
+						Nullable: internalField.Nullable,
+					}
+					internalFields = append(internalFields, newInternalField)
+				} else {
+					internalFields = append(internalFields, internalField)
+				}
+			}
+			t = arrow.StructOf(internalFields...)
+		}
+	case arrayType:
+		if _, ok := f.Type.(*arrow.ListType); ok {
+			converted, dataType := recordToSchemaSingleField(fieldMetadata.Fields[0], f.Type.(*arrow.ListType).ElemField(), withHigherPrecision, timestampOption, loc)
+			if converted {
+				t = arrow.ListOf(dataType)
+			}
+		} else {
+			t = f.Type
+		}
+	case mapType:
+		convertedKey, keyDataType := recordToSchemaSingleField(fieldMetadata.Fields[0], f.Type.(*arrow.MapType).KeyField(), withHigherPrecision, timestampOption, loc)
+		convertedValue, valueDataType := recordToSchemaSingleField(fieldMetadata.Fields[1], f.Type.(*arrow.MapType).ItemField(), withHigherPrecision, timestampOption, loc)
+		converted = convertedKey || convertedValue
+		if converted {
+			t = arrow.MapOf(keyDataType, valueDataType)
+		}
+	default:
+		converted = false
+	}
+	return converted, t
 }
 
 // TypedNullTime is required to properly bind the null value with the snowflakeType as the Snowflake functions
diff --git a/query.go b/query.go
index a861709..f1415c5 100644
--- a/query.go
+++ b/query.go
@@ -4,6 +4,7 @@ package gosnowflake
 
 import (
 	"encoding/json"
+	"fmt"
 	"time"
 )
 
@@ -73,6 +74,10 @@ type fieldMetadata struct {
 	Fields []fieldMetadata `json:"fields"`
 }
 
+func (fm *fieldMetadata) String() string {
+	return fmt.Sprintf("fieldMetadata: %v, %v, %v, %v", fm.Name, fm.Type, fm.Scale, fm.Fields)
+}
+
 type execResponseChunk struct {
 	URL              string `json:"url"`
 	RowCount         int    `json:"rowCount"`


Test patch:
diff --git a/arrow_test.go b/arrow_test.go
index 1cab04a..0b64988 100644
--- a/arrow_test.go
+++ b/arrow_test.go
@@ -15,27 +15,6 @@ import (
 	"database/sql/driver"
 )
 
-// A test just to show Snowflake version
-func TestCheckVersion(t *testing.T) {
-	conn := openConn(t)
-	defer conn.Close()
-
-	rows, err := conn.QueryContext(context.Background(), "SELECT current_version()")
-	if err != nil {
-		t.Error(err)
-	}
-	defer rows.Close()
-
-	if !rows.Next() {
-		t.Fatalf("failed to find any row")
-	}
-	var s string
-	if err = rows.Scan(&s); err != nil {
-		t.Fatal(err)
-	}
-	println(s)
-}
-
 func TestArrowBatchHighPrecision(t *testing.T) {
 	runDBTest(t, func(dbt *DBTest) {
 		ctx := WithArrowBatches(context.Background())
diff --git a/structured_type_test.go b/structured_type_test.go
index 4ac7891..4963502 100644
--- a/structured_type_test.go
+++ b/structured_type_test.go
@@ -3,6 +3,10 @@ package gosnowflake
 import (
 	"context"
 	"database/sql"
+	"database/sql/driver"
+	"github.com/apache/arrow/go/v15/arrow"
+	"github.com/apache/arrow/go/v15/arrow/array"
+	"github.com/apache/arrow/go/v15/arrow/memory"
 	"math/big"
 	"reflect"
 	"strings"
@@ -1345,6 +1349,247 @@ func TestWithHigherPrecision(t *testing.T) {
 	})
 }
 
+func TestStructuredTypeInArrowBatchesSimple(t *testing.T) {
+	skipStructuredTypesTestsOnGHActions(t)
+	runDBTest(t, func(dbt *DBTest) {
+		dbt.enableStructuredTypes()
+		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+		defer pool.AssertSize(t, 0)
+		ctx := WithArrowBatches(WithArrowAllocator(context.Background(), pool))
+
+		dbt.forceNativeArrow()
+		var err error
+		var rows driver.Rows
+		err = dbt.conn.Raw(func(sc any) error {
+			rows, err = sc.(driver.QueryerContext).QueryContext(ctx, "SELECT 1, {'s': 'some string'}::OBJECT(s VARCHAR)", nil)
+			return err
+		})
+		assertNilF(t, err)
+		defer rows.Close()
+		batches, err := rows.(SnowflakeRows).GetArrowBatches()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(batches), 0)
+		batch, err := batches[0].Fetch()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(*batch), 0)
+		for _, record := range *batch {
+			assertEqualE(t, record.Column(0).(*array.Int8).Value(0), int8(1))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(0).(*array.String).Value(0), "some string")
+			record.Release()
+		}
+	})
+}
+
+func TestStructuredTypeInArrowBatches(t *testing.T) {
+	skipStructuredTypesTestsOnGHActions(t)
+	runDBTest(t, func(dbt *DBTest) {
+		dbt.enableStructuredTypes()
+		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+		defer pool.AssertSize(t, 0)
+		ctx := WithArrowBatches(WithArrowAllocator(context.Background(), pool))
+
+		dbt.forceNativeArrow()
+		var err error
+		var rows driver.Rows
+		err = dbt.conn.Raw(func(sc any) error {
+			rows, err = sc.(driver.QueryerContext).QueryContext(ctx, "SELECT 1, {'s': 'some string', 'i': 1, 'time': '11:22:33'::TIME, 'date': '2024-04-16'::DATE, 'ltz': '2024-04-16 11:22:33'::TIMESTAMPLTZ, 'tz': '2025-04-16 22:33:11 +0100'::TIMESTAMPTZ, 'ntz': '2026-04-16 15:22:31'::TIMESTAMPNTZ}::OBJECT(s VARCHAR, i INTEGER, time TIME, date DATE, ltz TIMESTAMPLTZ, tz TIMESTAMPTZ, ntz TIMESTAMPNTZ)", nil)
+			return err
+		})
+		assertNilF(t, err)
+		defer rows.Close()
+		batches, err := rows.(SnowflakeRows).GetArrowBatches()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(batches), 0)
+		batch, err := batches[0].Fetch()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(*batch), 0)
+		for _, record := range *batch {
+			assertEqualE(t, record.Column(0).(*array.Int8).Value(0), int8(1))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(0).(*array.String).Value(0), "some string")
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(1).(*array.Int64).Value(0), int64(1))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(2).(*array.Time64).Value(0).ToTime(arrow.Nanosecond), time.Date(1970, 1, 1, 11, 22, 33, 0, time.UTC))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(3).(*array.Date32).Value(0).ToTime(), time.Date(2024, 4, 16, 0, 0, 0, 0, time.UTC))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(4).(*array.Timestamp).Value(0).ToTime(arrow.Nanosecond), time.Date(2024, 4, 16, 11, 22, 33, 0, time.UTC))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(5).(*array.Timestamp).Value(0).ToTime(arrow.Nanosecond), time.Date(2025, 4, 16, 21, 33, 11, 0, time.UTC))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(6).(*array.Timestamp).Value(0).ToTime(arrow.Nanosecond), time.Date(2026, 4, 16, 15, 22, 31, 0, time.UTC))
+			record.Release()
+		}
+	})
+}
+
+func TestStructuredTypeInArrowBatchesWithTimestampOptionAndHigherPrecisionAndUtf8Validation(t *testing.T) {
+	skipStructuredTypesTestsOnGHActions(t)
+	runDBTest(t, func(dbt *DBTest) {
+		dbt.enableStructuredTypes()
+		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+		defer pool.AssertSize(t, 0)
+		dbt.forceNativeArrow()
+		var err error
+		var rows driver.Rows
+		err = dbt.conn.Raw(func(sc any) error {
+			ctx := WithArrowBatchesUtf8Validation(WithHigherPrecision(WithArrowBatchesTimestampOption(WithArrowBatches(WithArrowAllocator(context.Background(), pool)), UseOriginalTimestamp)))
+			rows, err = sc.(driver.QueryerContext).QueryContext(ctx, "SELECT 1, {'i': 123, 'f': 12.34, 'n0': 321, 'n19': 1.5, 's': 'some string', 'bi': TO_BINARY('616263', 'HEX'), 'bool': true, 'time': '11:22:33', 'date': '2024-04-18', 'ntz': '2024-04-01 11:22:33', 'tz': '2024-04-02 11:22:33 +0100', 'ltz': '2024-04-03 11:22:33'}::OBJECT(i INTEGER, f DOUBLE, n0 NUMBER(38, 0), n19 NUMBER(38, 19), s VARCHAR, bi BINARY, bool BOOLEAN, time TIME, date DATE, ntz TIMESTAMP_NTZ, tz TIMESTAMP_TZ, ltz TIMESTAMP_LTZ)", nil)
+			return err
+		})
+		assertNilF(t, err)
+		defer rows.Close()
+		batches, err := rows.(SnowflakeRows).GetArrowBatches()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(batches), 0)
+		batch, err := batches[0].Fetch()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(*batch), 0)
+		for _, record := range *batch {
+			assertEqualE(t, record.Column(0).(*array.Int8).Value(0), int8(1))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(0).(*array.Decimal128).Value(0).LowBits(), uint64(123))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(1).(*array.Float64).Value(0), 12.34)
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(2).(*array.Decimal128).Value(0).LowBits(), uint64(321))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(3).(*array.Decimal128).Value(0).LowBits(), uint64(15000000000000000000))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(4).(*array.String).Value(0), "some string")
+			assertDeepEqualE(t, record.Column(1).(*array.Struct).Field(5).(*array.Binary).Value(0), []byte{'a', 'b', 'c'})
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(6).(*array.Boolean).Value(0), true)
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(7).(*array.Time64).Value(0).ToTime(arrow.Nanosecond), time.Date(1970, 1, 1, 11, 22, 33, 0, time.UTC))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(8).(*array.Date32).Value(0).ToTime(), time.Date(2024, 4, 18, 0, 0, 0, 0, time.UTC))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(9).(*array.Struct).Field(0).(*array.Int64).Value(0), int64(1711970553))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(9).(*array.Struct).Field(1).(*array.Int32).Value(0), int32(0))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(10).(*array.Struct).Field(0).(*array.Int64).Value(0), int64(1712053353))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(10).(*array.Struct).Field(1).(*array.Int32).Value(0), int32(0))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(11).(*array.Struct).Field(0).(*array.Int64).Value(0), int64(1712143353))
+			assertEqualE(t, record.Column(1).(*array.Struct).Field(11).(*array.Struct).Field(1).(*array.Int32).Value(0), int32(0))
+			record.Release()
+		}
+	})
+}
+
+func TestStructuredTypeInArrowBatchesWithEmbeddedObject(t *testing.T) {
+	skipStructuredTypesTestsOnGHActions(t)
+	runDBTest(t, func(dbt *DBTest) {
+		dbt.enableStructuredTypes()
+		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+		defer pool.AssertSize(t, 0)
+		dbt.forceNativeArrow()
+		var err error
+		var rows driver.Rows
+		err = dbt.conn.Raw(func(sc any) error {
+			ctx := WithArrowBatches(WithArrowAllocator(context.Background(), pool))
+			rows, err = sc.(driver.QueryerContext).QueryContext(ctx, "SELECT {'o': {'s': 'some string'}}::OBJECT(o OBJECT(s VARCHAR))", nil)
+			return err
+		})
+		assertNilF(t, err)
+		defer rows.Close()
+		batches, err := rows.(SnowflakeRows).GetArrowBatches()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(batches), 0)
+		batch, err := batches[0].Fetch()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(*batch), 0)
+		for _, record := range *batch {
+			assertEqualE(t, record.Column(0).(*array.Struct).Field(0).(*array.Struct).Field(0).(*array.String).Value(0), "some string")
+			record.Release()
+		}
+	})
+}
+
+func TestStructuredTypeInArrowBatchesAsNull(t *testing.T) {
+	skipStructuredTypesTestsOnGHActions(t)
+	runDBTest(t, func(dbt *DBTest) {
+		dbt.enableStructuredTypes()
+		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+		defer pool.AssertSize(t, 0)
+		dbt.forceNativeArrow()
+		var err error
+		var rows driver.Rows
+		err = dbt.conn.Raw(func(sc any) error {
+			ctx := WithArrowBatches(WithArrowAllocator(context.Background(), pool))
+			rows, err = sc.(driver.QueryerContext).QueryContext(ctx, "SELECT {'s': 'some string'}::OBJECT(s VARCHAR) UNION SELECT null", nil)
+			return err
+		})
+		assertNilF(t, err)
+		defer rows.Close()
+		batches, err := rows.(SnowflakeRows).GetArrowBatches()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(batches), 0)
+		batch, err := batches[0].Fetch()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(*batch), 0)
+		for _, record := range *batch {
+			assertFalseF(t, record.Column(0).IsNull(0))
+			assertTrueE(t, record.Column(0).IsNull(1))
+			record.Release()
+		}
+	})
+}
+
+func TestStructuredArrayInArrowBatches(t *testing.T) {
+	skipStructuredTypesTestsOnGHActions(t)
+	runDBTest(t, func(dbt *DBTest) {
+		dbt.enableStructuredTypes()
+		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+		defer pool.AssertSize(t, 0)
+		ctx := WithArrowBatches(WithArrowAllocator(context.Background(), pool))
+
+		dbt.forceNativeArrow()
+		var err error
+		var rows driver.Rows
+		err = dbt.conn.Raw(func(sc any) error {
+			rows, err = sc.(driver.QueryerContext).QueryContext(ctx, "SELECT [1, 2, 3]::ARRAY(INTEGER) UNION SELECT [4, 5, 6]::ARRAY(INTEGER)", nil)
+			return err
+		})
+		assertNilF(t, err)
+		defer rows.Close()
+		batches, err := rows.(SnowflakeRows).GetArrowBatches()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(batches), 0)
+		batch, err := batches[0].Fetch()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(*batch), 0)
+		record := (*batch)[0]
+		defer record.Release()
+		assertEqualE(t, record.Column(0).(*array.List).ListValues().(*array.Int64).Value(0), int64(1))
+		assertEqualE(t, record.Column(0).(*array.List).ListValues().(*array.Int64).Value(1), int64(2))
+		assertEqualE(t, record.Column(0).(*array.List).ListValues().(*array.Int64).Value(2), int64(3))
+		assertEqualE(t, record.Column(0).(*array.List).ListValues().(*array.Int64).Value(3), int64(4))
+		assertEqualE(t, record.Column(0).(*array.List).ListValues().(*array.Int64).Value(4), int64(5))
+		assertEqualE(t, record.Column(0).(*array.List).ListValues().(*array.Int64).Value(5), int64(6))
+		assertEqualE(t, record.Column(0).(*array.List).Offsets()[0], int32(0))
+		assertEqualE(t, record.Column(0).(*array.List).Offsets()[1], int32(3))
+		assertEqualE(t, record.Column(0).(*array.List).Offsets()[2], int32(6))
+	})
+}
+
+func TestStructuredMapInArrowBatches(t *testing.T) {
+	skipStructuredTypesTestsOnGHActions(t)
+	runDBTest(t, func(dbt *DBTest) {
+		dbt.enableStructuredTypes()
+		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+		defer pool.AssertSize(t, 0)
+		ctx := WithArrowBatches(WithArrowAllocator(context.Background(), pool))
+
+		dbt.forceNativeArrow()
+		var err error
+		var rows driver.Rows
+		err = dbt.conn.Raw(func(sc any) error {
+			rows, err = sc.(driver.QueryerContext).QueryContext(ctx, "SELECT {'a': 'b', 'c': 'd'}::MAP(VARCHAR, VARCHAR)", nil)
+			return err
+		})
+		assertNilF(t, err)
+		defer rows.Close()
+		batches, err := rows.(SnowflakeRows).GetArrowBatches()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(batches), 0)
+		batch, err := batches[0].Fetch()
+		assertNilF(t, err)
+		assertNotEqualF(t, len(*batch), 0)
+		for _, record := range *batch {
+			assertEqualE(t, record.Column(0).(*array.Map).Keys().(*array.String).Value(0), "a")
+			assertEqualE(t, record.Column(0).(*array.Map).Keys().(*array.String).Value(1), "c")
+			assertEqualE(t, record.Column(0).(*array.Map).Items().(*array.String).Value(0), "b")
+			assertEqualE(t, record.Column(0).(*array.Map).Items().(*array.String).Value(1), "d")
+			record.Release()
+		}
+	})
+}
+
 func forAllStructureTypeFormats(dbt *DBTest, f func(t *testing.T, format string)) {
 	for _, tc := range []struct {
 		name        string