diff --git a/client/column/nullable_test.go b/client/column/nullable_test.go index 631ad234fa..fd703b95f0 100644 --- a/client/column/nullable_test.go +++ b/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) diff --git a/client/column/scalar.go b/client/column/scalar.go index 12c718e9bb..3ac8dcb8e6 100644 --- a/client/column/scalar.go +++ b/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 { diff --git a/client/column/scalar_test.go b/client/column/scalar_test.go index 5c5530d451..3472fe6737 100644 --- a/client/column/scalar_test.go +++ b/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() { diff --git a/client/milvusclient/results.go b/client/milvusclient/results.go index da8ca9140e..30c4782e5c 100644 --- a/client/milvusclient/results.go +++ b/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 diff --git a/client/milvusclient/results_test.go b/client/milvusclient/results_test.go index 9098ff0939..496342e25f 100644 --- a/client/milvusclient/results_test.go +++ b/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"` diff --git a/client/row/data.go b/client/row/data.go index 70eb5327e2..f5ea5e7955 100644 --- a/client/row/data.go +++ b/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)) } } diff --git a/client/row/data_test.go b/client/row/data_test.go index ac24adfc7f..9a45dd7267 100644 --- a/client/row/data_test.go +++ b/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