fix: [GoSDK] support Timestamptz row insert (#53775)

getColumnCreators has no cases for FieldTypeTimestamptz or
FieldTypeGeometry, so a row-based insert into a collection containing
either field silently produced a nil column and panicked later in
AnyToColumns' AppendValue call (nil interface method dispatch).

Add column creators for both field types:
- Timestamptz maps to NewColumnTimestamptz; ColumnTimestamptz now
overrides AppendValue to accept time.Time, *time.Time and ISO 8601
strings (RFC3339Nano), matching the wire format the server expects.
- Geometry maps to NewColumnGeometryWKT.

Also add an exhaustive regression test enumerating all supported scalar
field types, asserting getColumnCreators yields a non-nil column, so a
future field type missing a creator is caught by CI instead of panicking
at runtime.

Related to #53764

---------

Signed-off-by: Congqi Xia <[email protected]>

client/column/nullable_test.go

@@ -298,10 +298,10 @@ func (s *NullableScalarSuite) TestBasic() {
 		s.NoError(err)
 		s.Equal(entity.FieldTypeTimestamptz, column.Type())
 		s.Equal(name, column.Name())
-		// verify time values are converted to RFC3339Nano strings
+		// verify time values are converted to normalized RFC3339Nano strings
 		expectedStrings := []string{
-			compactData[0].Format(time.RFC3339Nano),
-			compactData[1].Format(time.RFC3339Nano),
+			formatTimestamptz(compactData[0]),
+			formatTimestamptz(compactData[1]),
 		}
 		s.Equal(expectedStrings, column.Data())
 		for i := 0; i < len(validData); i++ {
@@ -313,9 +313,9 @@ func (s *NullableScalarSuite) TestBasic() {
 		// sparse mode
 		sparseData := []time.Time{now, {}, now.Add(2 * time.Hour)}
 		sparseExpected := []string{
-			sparseData[0].Format(time.RFC3339Nano),
-			sparseData[1].Format(time.RFC3339Nano),
-			sparseData[2].Format(time.RFC3339Nano),
+			formatTimestamptz(sparseData[0]),
+			formatTimestamptz(sparseData[1]),
+			formatTimestamptz(sparseData[2]),
 		}
 		column, err = NewNullableColumnTimestamptz(name, sparseData, validData, WithSparseNullableMode[string](true))
 		s.NoError(err)

client/column/scalar.go

@@ -225,18 +225,46 @@ func NewColumnTimestamptz(name string, values []time.Time) *ColumnTimestamptz {
 			name:      name,
 			fieldType: entity.FieldTypeTimestamptz,
 			values: lo.Map(values, func(t time.Time, _ int) string {
-				return t.Format(time.RFC3339Nano)
+				return formatTimestamptz(t)
 			}),
 		},
 	}
 }
 
+// formatTimestamptz normalizes a time.Time to the timestamptz wire format:
+// the UTC instant truncated to microsecond precision, matching how the server
+// stores timestamptz (UTC microseconds). Formatting the original location
+// directly is unsafe: historical offsets may carry seconds (e.g. Shanghai LMT
+// +08:05:43) which RFC3339 (minute precision) cannot represent, shifting the
+// instant on re-parse.
+func formatTimestamptz(t time.Time) string {
+	return t.UTC().Truncate(time.Microsecond).Format(time.RFC3339Nano)
+}
+
 func (c *ColumnTimestamptz) Slice(start, end int) Column {
 	return &ColumnTimestamptz{
 		genericColumnBase: c.genericColumnBase.slice(start, end),
 	}
 }
 
+// AppendValue appends a value into the column. It accepts time.Time, *time.Time
+// and ISO 8601 string (RFC3339Nano), since the column stores timestamptz as
+// RFC3339Nano strings internally. time.Time values are normalized to UTC with
+// microsecond precision before formatting.
+func (c *ColumnTimestamptz) AppendValue(a any) error {
+	switch v := a.(type) {
+	case time.Time:
+		return c.genericColumnBase.AppendValue(formatTimestamptz(v))
+	case *time.Time:
+		if v == nil {
+			return c.AppendNull()
+		}
+		return c.genericColumnBase.AppendValue(formatTimestamptz(*v))
+	default:
+		return c.genericColumnBase.AppendValue(a)
+	}
+}
+
 var _ Column = (*ColumnTimestampTzIsoString)(nil)
 
 type ColumnTimestampTzIsoString struct {

client/column/scalar_test.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"strings"
 	"testing"
 	"time"
 
@@ -245,11 +246,11 @@ func (s *ScalarSuite) TestBasic() {
 		column := NewColumnTimestamptz(name, data)
 		s.Equal(entity.FieldTypeTimestamptz, column.Type())
 		s.Equal(name, column.Name())
-		// verify data is converted to RFC3339Nano format
+		// verify data is converted to normalized RFC3339Nano format
 		expectedStrings := []string{
-			data[0].Format(time.RFC3339Nano),
-			data[1].Format(time.RFC3339Nano),
-			data[2].Format(time.RFC3339Nano),
+			formatTimestamptz(data[0]),
+			formatTimestamptz(data[1]),
+			formatTimestamptz(data[2]),
 		}
 		s.Equal(expectedStrings, column.Data())
 
@@ -294,6 +295,48 @@ func (s *ScalarSuite) TestBasic() {
 	})
 }
 
+func (s *ScalarSuite) TestTimestamptzAppendValue() {
+	name := fmt.Sprintf("field_%d", rand.Intn(1000))
+	column := NewColumnTimestamptz(name, nil)
+
+	now := time.Now().UTC().Truncate(time.Nanosecond)
+	s.NoError(column.AppendValue(now))
+	s.NoError(column.AppendValue(now.Add(time.Hour)))
+	s.NoError(column.AppendValue(now.Format(time.RFC3339Nano)))
+
+	expected := []string{
+		formatTimestamptz(now),
+		formatTimestamptz(now.Add(time.Hour)),
+		// raw ISO string inputs are stored as-is, without normalization
+		now.Format(time.RFC3339Nano),
+	}
+	s.Equal(expected, column.Data())
+	s.Equal(expected, column.FieldData().GetScalars().GetStringData().GetData())
+
+	s.Error(column.AppendValue(now.UnixMilli()))
+}
+
+func (s *ScalarSuite) TestTimestamptzNormalization() {
+	shanghai, err := time.LoadLocation("Asia/Shanghai")
+	s.Require().NoError(err)
+	// Pre-1901 Shanghai LMT offset carries seconds (+08:05:43) that RFC3339
+	// (minute precision) cannot represent; formatting the raw location emits
+	// +08:05 and shifts the instant by 43s on re-parse.
+	lmt := time.Date(1900, 1, 1, 0, 0, 0, 0, shanghai)
+	orig := lmt.UTC()
+
+	column := NewColumnTimestamptz("ts", []time.Time{lmt})
+	s.Require().NoError(column.AppendValue(lmt))
+
+	s.Require().Len(column.Data(), 2)
+	for i, got := range column.Data() {
+		parsed, err := time.Parse(time.RFC3339Nano, got)
+		s.Require().NoError(err, "value %d %q must be RFC3339Nano parseable", i, got)
+		s.Equal(orig, parsed.UTC(), "value %d %q must preserve the instant of %v", i, got, lmt)
+		s.True(strings.HasSuffix(got, "Z"), "value %d %q must be normalized to UTC", i, got)
+	}
+}
+
 func (s *ScalarSuite) TestSlice() {
 	n := 100
 	s.Run("column_bool", func() {

client/milvusclient/results.go

@@ -254,15 +254,23 @@ func (ds DataSet) fillData(data reflect.Value, dataType reflect.Type, idx int) e
 			if err != nil {
 				return err
 			}
+			converted, err := row.CoerceValue(fieldType.Elem(), val)
+			if err != nil {
+				return err
+			}
 			ptr := reflect.New(fieldType.Elem())
-			ptr.Elem().Set(reflect.ValueOf(val))
+			ptr.Elem().Set(reflect.ValueOf(converted))
 			field.Set(ptr)
 		} else {
 			val, err := ds[i].Get(idx)
 			if err != nil {
 				return err
 			}
-			field.Set(reflect.ValueOf(val))
+			converted, err := row.CoerceValue(fieldType, val)
+			if err != nil {
+				return err
+			}
+			field.Set(reflect.ValueOf(converted))
 		}
 	}
 	return nil

client/milvusclient/results_test.go

@@ -18,6 +18,7 @@ package milvusclient
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/suite"
 
@@ -81,6 +82,64 @@ func (s *ResultSetSuite) TestResultsetUnmarshal() {
 	s.Error(err)
 }
 
+func (s *ResultSetSuite) TestResultsetUnmarshalTimestamptz() {
+	type TsData struct {
+		Ts time.Time `milvus:"name:ts"`
+	}
+	type NullableTsData struct {
+		Ts *time.Time `milvus:"name:ts"`
+	}
+	type StringTsData struct {
+		Ts string `milvus:"name:ts"`
+	}
+
+	// the server returns timestamptz as ISO strings in the collection timezone
+	isoData := []string{
+		"2024-01-01T00:00:00+08:00",
+		"2024-06-15T12:30:45.123456Z",
+	}
+	ds := DataSet([]column.Column{
+		column.NewColumnTimestamptzIsoString("ts", isoData),
+	})
+
+	// time.Time target
+	var timeReceiver []*TsData
+	err := ds.Unmarshal(&timeReceiver)
+	s.NoError(err)
+	s.Require().Len(timeReceiver, 2)
+	for i, row := range timeReceiver {
+		parsed, err := time.Parse(time.RFC3339Nano, isoData[i])
+		s.NoError(err)
+		s.True(parsed.Equal(row.Ts), "row %d ts = %v, want %v", i, row.Ts, parsed)
+	}
+
+	// string target unchanged
+	var strReceiver []*StringTsData
+	err = ds.Unmarshal(&strReceiver)
+	s.NoError(err)
+	s.Require().Len(strReceiver, 2)
+	for i, row := range strReceiver {
+		s.Equal(isoData[i], row.Ts)
+	}
+
+	// nullable *time.Time target, including a null row (compact mode)
+	nullableCol, err := column.NewNullableColumnTimestamptzIsoString("ts", []string{"2024-01-01T00:00:00Z"}, []bool{true, false})
+	s.NoError(err)
+	dsN := DataSet([]column.Column{nullableCol})
+	var nullReceiver []*NullableTsData
+	err = dsN.Unmarshal(&nullReceiver)
+	s.NoError(err)
+	s.Require().Len(nullReceiver, 2)
+	s.Require().NotNil(nullReceiver[0].Ts)
+	s.True(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC).Equal(*nullReceiver[0].Ts))
+	s.Nil(nullReceiver[1].Ts)
+
+	// unparseable string into time.Time target returns error
+	badCol := column.NewColumnTimestamptzIsoString("ts", []string{"not-a-timestamp"})
+	err = DataSet([]column.Column{badCol}).Unmarshal(&[]*TsData{})
+	s.Error(err)
+}
+
 func (s *ResultSetSuite) TestSearchResultUnmarshal() {
 	type MyData struct {
 		A     int64     `milvus:"name:id"`

client/row/data.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"reflect"
 	"strconv"
+	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/samber/lo"
@@ -227,9 +228,14 @@ func getColumnCreators(sch *entity.Schema) map[string]columnCreator {
 			case entity.FieldTypeText:
 				data := make([]string, 0, rowsLen)
 				col = column.NewColumnText(field.Name, data)
+			case entity.FieldTypeTimestamptz:
+				col = column.NewColumnTimestamptz(field.Name, nil)
 			case entity.FieldTypeJSON:
 				data := make([][]byte, 0, rowsLen)
 				col = column.NewColumnJSONBytes(field.Name, data)
+			case entity.FieldTypeGeometry:
+				data := make([]string, 0, rowsLen)
+				col = column.NewColumnGeometryWKT(field.Name, data)
 			case entity.FieldTypeArray:
 				if field.ElementType == entity.FieldTypeStruct {
 					structColumn, err := column.NewColumnStructArrayFromSchema(field.Name, field.StructSchema)
@@ -327,6 +333,30 @@ func NewArrayColumn(f *entity.Field) column.Column {
 	}
 }
 
+var timeType = reflect.TypeOf(time.Time{})
+
+// CoerceValue converts value into a value assignable to targetType when
+// unmarshalling read-back data into typed struct fields.
+//
+// TIMESTAMPTZ columns are transported as RFC3339Nano ISO strings, so string
+// values are parsed into time.Time (or *time.Time) targets; string targets and
+// any other type pair keep the value unchanged.
+func CoerceValue(targetType reflect.Type, value any) (any, error) {
+	if value == nil {
+		return nil, nil
+	}
+	if targetType == timeType {
+		if s, ok := value.(string); ok {
+			t, err := time.Parse(time.RFC3339Nano, s)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse timestamptz string %q: %w", s, err)
+			}
+			return t, nil
+		}
+	}
+	return value, nil
+}
+
 func SetField(receiver any, fieldName string, value any) error {
 	candidates, err := rowutil.ParseFields(reflect.ValueOf(receiver))
 	if err != nil {
@@ -344,12 +374,20 @@ func SetField(receiver any, fieldName string, value any) error {
 			if value == nil {
 				candidate.Value.Set(reflect.Zero(candidate.Value.Type()))
 			} else {
+				converted, err := CoerceValue(candidate.Value.Type().Elem(), value)
+				if err != nil {
+					return err
+				}
 				ptr := reflect.New(candidate.Value.Type().Elem())
-				ptr.Elem().Set(reflect.ValueOf(value))
+				ptr.Elem().Set(reflect.ValueOf(converted))
 				candidate.Value.Set(ptr)
 			}
 		} else {
-			candidate.Value.Set(reflect.ValueOf(value))
+			converted, err := CoerceValue(candidate.Value.Type(), value)
+			if err != nil {
+				return err
+			}
+			candidate.Value.Set(reflect.ValueOf(converted))
 		}
 	}
 

client/row/data_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/suite"
 
@@ -374,6 +375,186 @@ func (s *RowsSuite) TestNullablePointerColumns() {
 	})
 }
 
+func (s *RowsSuite) TestRowsToTimestamptzColumn() {
+	s.Run("plain_time", func() {
+		type TimestamptzRow struct {
+			ID        int64     `milvus:"primary_key"`
+			CreatedAt time.Time `milvus:"name:created_at"`
+		}
+
+		schema := entity.NewSchema().
+			WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+			WithField(entity.NewField().WithName("created_at").WithDataType(entity.FieldTypeTimestamptz))
+
+		now := time.Now().UTC().Truncate(time.Microsecond)
+		rows := []any{
+			&TimestamptzRow{ID: 1, CreatedAt: now},
+			&TimestamptzRow{ID: 2, CreatedAt: now.Add(time.Hour)},
+		}
+
+		columns, err := AnyToColumns(rows, false, schema)
+		s.Require().NoError(err)
+		s.Require().Len(columns, 2)
+
+		for _, col := range columns {
+			if col.Name() != "created_at" {
+				continue
+			}
+			s.Equal(entity.FieldTypeTimestamptz, col.Type())
+			s.Equal([]string{
+				now.Format(time.RFC3339Nano),
+				now.Add(time.Hour).Format(time.RFC3339Nano),
+			}, col.FieldData().GetScalars().GetStringData().GetData())
+		}
+	})
+
+	s.Run("nullable_pointer", func() {
+		type TimestamptzRow struct {
+			ID        int64      `milvus:"primary_key"`
+			CreatedAt *time.Time `milvus:"name:created_at"`
+		}
+
+		schema := entity.NewSchema().
+			WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+			WithField(entity.NewField().WithName("created_at").WithDataType(entity.FieldTypeTimestamptz).WithNullable(true))
+
+		now := time.Now().UTC().Truncate(time.Microsecond)
+		rows := []any{
+			&TimestamptzRow{ID: 1, CreatedAt: &now},
+			&TimestamptzRow{ID: 2, CreatedAt: nil},
+		}
+
+		columns, err := AnyToColumns(rows, false, schema)
+		s.Require().NoError(err)
+		s.Require().Len(columns, 2)
+
+		for _, col := range columns {
+			if col.Name() != "created_at" {
+				continue
+			}
+			s.Equal(entity.FieldTypeTimestamptz, col.Type())
+			s.True(col.Nullable())
+			isNull0, err := col.IsNull(0)
+			s.NoError(err)
+			s.False(isNull0)
+			isNull1, err := col.IsNull(1)
+			s.NoError(err)
+			s.True(isNull1)
+			s.Equal([]string{now.Format(time.RFC3339Nano)},
+				col.FieldData().GetScalars().GetStringData().GetData())
+		}
+	})
+}
+
+func (s *RowsSuite) TestColumnCreatorsAllScalarTypes() {
+	// Every scalar field type must yield a non-nil column from getColumnCreators.
+	// A missing case silently leaves the column nil and panics later in
+	// AnyToColumns' AppendValue, so enumerate all supported scalar types here.
+	scalarTypes := []entity.FieldType{
+		entity.FieldTypeBool,
+		entity.FieldTypeInt8,
+		entity.FieldTypeInt16,
+		entity.FieldTypeInt32,
+		entity.FieldTypeInt64,
+		entity.FieldTypeFloat,
+		entity.FieldTypeDouble,
+		entity.FieldTypeString,
+		entity.FieldTypeVarChar,
+		entity.FieldTypeText,
+		entity.FieldTypeJSON,
+		entity.FieldTypeTimestamptz,
+		entity.FieldTypeGeometry,
+	}
+	for _, dataType := range scalarTypes {
+		s.Run(dataType.Name(), func() {
+			field := entity.NewField().WithName("f").WithDataType(dataType)
+			creator := getColumnCreators(entity.NewSchema().WithField(field))["f"]
+			s.Require().NotNil(creator)
+			col, err := creator(1)
+			s.Require().NoError(err)
+			s.Require().NotNil(col, "field type %s must have a column creator", dataType.Name())
+			// FieldTypeString is normalized to a VarChar column by design.
+			if dataType == entity.FieldTypeString {
+				s.Equal(entity.FieldTypeVarChar, col.Type())
+			} else {
+				s.Equal(dataType, col.Type())
+			}
+			s.Zero(col.Len())
+		})
+	}
+}
+
+func (s *RowsSuite) TestRowsToGeometryColumn() {
+	type GeometryRow struct {
+		ID       int64  `milvus:"primary_key"`
+		Location string `milvus:"name:location"`
+	}
+
+	schema := entity.NewSchema().
+		WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+		WithField(entity.NewField().WithName("location").WithDataType(entity.FieldTypeGeometry))
+
+	rows := []any{
+		&GeometryRow{ID: 1, Location: "POINT (1 1)"},
+		&GeometryRow{ID: 2, Location: "POINT (2 2)"},
+	}
+
+	columns, err := AnyToColumns(rows, false, schema)
+	s.Require().NoError(err)
+	s.Require().Len(columns, 2)
+
+	for _, col := range columns {
+		if col.Name() != "location" {
+			continue
+		}
+		s.Equal(entity.FieldTypeGeometry, col.Type())
+		s.Equal([]string{"POINT (1 1)", "POINT (2 2)"},
+			col.FieldData().GetScalars().GetGeometryWktData().GetData())
+	}
+}
+
+func (s *RowsSuite) TestSetFieldTimestamptz() {
+	s.Run("time_value", func() {
+		type Row struct {
+			Ts time.Time
+		}
+		row := &Row{}
+		err := SetField(row, "Ts", "2024-01-01T00:00:00Z")
+		s.NoError(err)
+		s.Equal(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), row.Ts)
+	})
+
+	s.Run("pointer_value", func() {
+		type Row struct {
+			Ts *time.Time
+		}
+		row := &Row{}
+		err := SetField(row, "Ts", "2024-01-01T00:00:00Z")
+		s.NoError(err)
+		s.Require().NotNil(row.Ts)
+		s.Equal(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), *row.Ts)
+	})
+
+	s.Run("string_target_kept", func() {
+		type Row struct {
+			Ts string
+		}
+		row := &Row{}
+		err := SetField(row, "Ts", "2024-01-01T00:00:00Z")
+		s.NoError(err)
+		s.Equal("2024-01-01T00:00:00Z", row.Ts)
+	})
+
+	s.Run("parse_failure", func() {
+		type Row struct {
+			Ts time.Time
+		}
+		row := &Row{}
+		err := SetField(row, "Ts", "not-a-timestamp")
+		s.Error(err)
+	})
+}
+
 func (s *RowsSuite) TestRowsToTextColumnWithSchema() {
 	type TextRow struct {
 		ID      int64