fix(arrow/array/arreflect): reject unrepresentable timestamps

arrow/array/arreflect/reflect_go_to_arrow.go

@@ -308,7 +308,11 @@ func appendTemporalValue(b array.Builder, v reflect.Value) error {
 		if err != nil {
 			return err
 		}
-		tb.Append(arrow.Timestamp(t.UnixNano() / int64(unit.Multiplier())))
+		timestamp, err := arrow.TimestampFromTime(t, unit)
+		if err != nil {
+			return err
+		}
+		tb.Append(timestamp)
 	case *array.Date32Builder:
 		t, err := asTime(v)
 		if err != nil {

arrow/array/arreflect/reflect_go_to_arrow_test.go

@@ -132,6 +132,18 @@ func TestBuildTemporalArray(t *testing.T) {
 	})
 }
 
+func TestBuildTemporalArrayRejectsTimestampOverflow(t *testing.T) {
+	mem := checkedMem(t)
+	values := []time.Time{
+		time.Date(1970, time.January, 1, 0, 0, 0, 0, time.UTC),
+		time.Date(3000, time.January, 1, 0, 0, 0, 0, time.UTC),
+	}
+
+	arr, err := FromSlice(values, mem)
+	require.ErrorIs(t, err, arrow.ErrInvalid)
+	require.Nil(t, arr)
+}
+
 func TestBuildDecimalArray(t *testing.T) {
 	mem := checkedMem(t)
 

arrow/datatype_fixedwidth.go

@@ -20,6 +20,7 @@ import (
 	"cmp"
 	"errors"
 	"fmt"
+	"math"
 	"strconv"
 	"sync"
 	"time"
@@ -203,16 +204,38 @@ func TimestampFromTime(val time.Time, unit TimeUnit) (Timestamp, error) {
 	case Second:
 		return Timestamp(val.Unix()), nil
 	case Millisecond:
-		return Timestamp(val.Unix()*1e3 + int64(val.Nanosecond())/1e6), nil
+		return timestampFromTime(val, 1e3, 1e6, unit)
 	case Microsecond:
-		return Timestamp(val.Unix()*1e6 + int64(val.Nanosecond())/1e3), nil
+		return timestampFromTime(val, 1e6, 1e3, unit)
 	case Nanosecond:
-		return Timestamp(val.UnixNano()), nil
+		return timestampFromTime(val, 1e9, 1, unit)
 	default:
 		return 0, fmt.Errorf("%w: unexpected timestamp unit: %s", ErrInvalid, unit)
 	}
 }
 
+func timestampFromTime(val time.Time, multiplier, divisor int64, unit TimeUnit) (Timestamp, error) {
+	seconds := val.Unix()
+	fraction := int64(val.Nanosecond()) / divisor
+	maxSeconds := math.MaxInt64 / multiplier
+	if seconds > maxSeconds || (seconds == maxSeconds && fraction > math.MaxInt64%multiplier) {
+		return 0, fmt.Errorf("%w: timestamp value is out of range for %s", ErrInvalid, unit)
+	}
+
+	minSeconds := math.MinInt64 / multiplier
+	minRemainder := math.MinInt64 % multiplier
+	minFraction := multiplier + minRemainder
+	if seconds < minSeconds {
+		if minSeconds == math.MinInt64 || seconds != minSeconds-1 || fraction < minFraction {
+			return 0, fmt.Errorf("%w: timestamp value is out of range for %s", ErrInvalid, unit)
+		}
+		return Timestamp(math.MinInt64 + fraction - minFraction), nil
+	}
+
+	timestamp := seconds * multiplier
+	return Timestamp(timestamp + fraction), nil
+}
+
 // Time32FromString parses a string to return a Time32 value in the given unit,
 // unit needs to be only seconds or milliseconds and the string should be in the
 // form of HH:MM or HH:MM:SS[.zzz] where the fractions of a second are optional.

arrow/datatype_fixedwidth_test.go

@@ -17,6 +17,7 @@
 package arrow_test
 
 import (
+	"math"
 	"sync"
 	"testing"
 	"time"
@@ -494,6 +495,83 @@ func TestDateFromTime(t *testing.T) {
 	assert.EqualValues(t, wantD32, arrow.Date32FromTime(tm))
 }
 
+func timestampBoundaryTime(value arrow.Timestamp, unit arrow.TimeUnit) time.Time {
+	nanosPerUnit := int64(unit.Multiplier())
+	unitsPerSecond := int64(time.Second) / nanosPerUnit
+	seconds := int64(value) / unitsPerSecond
+	remainder := int64(value) % unitsPerSecond
+	if remainder < 0 {
+		seconds--
+		remainder += unitsPerSecond
+	}
+	return time.Unix(seconds, remainder*nanosPerUnit).UTC()
+}
+
+func TestTimestampFromTimeBoundaries(t *testing.T) {
+	for _, unit := range arrow.TimeUnitValues {
+		t.Run(unit.String(), func(t *testing.T) {
+			maxTime := timestampBoundaryTime(arrow.Timestamp(math.MaxInt64), unit)
+			got, err := arrow.TimestampFromTime(maxTime, unit)
+			require.NoError(t, err)
+			assert.Equal(t, arrow.Timestamp(math.MaxInt64), got)
+
+			minTime := timestampBoundaryTime(arrow.Timestamp(math.MinInt64), unit)
+			got, err = arrow.TimestampFromTime(minTime, unit)
+			require.NoError(t, err)
+			assert.Equal(t, arrow.Timestamp(math.MinInt64), got)
+
+			if unit != arrow.Second {
+				_, err = arrow.TimestampFromTime(maxTime.Add(unit.Multiplier()), unit)
+				assert.ErrorIs(t, err, arrow.ErrInvalid)
+
+				_, err = arrow.TimestampFromTime(minTime.Add(-unit.Multiplier()), unit)
+				assert.ErrorIs(t, err, arrow.ErrInvalid)
+			}
+		})
+	}
+}
+
+func TestTimestampFromTimeFractionalSeconds(t *testing.T) {
+	tests := []struct {
+		unit arrow.TimeUnit
+		want arrow.Timestamp
+	}{
+		{unit: arrow.Second, want: -1},
+		{unit: arrow.Millisecond, want: -500},
+		{unit: arrow.Microsecond, want: -500_000},
+		{unit: arrow.Nanosecond, want: -500_000_000},
+	}
+
+	value := time.Date(1969, time.December, 31, 23, 59, 59, 500_000_000, time.UTC)
+	for _, tc := range tests {
+		t.Run(tc.unit.String(), func(t *testing.T) {
+			got, err := arrow.TimestampFromTime(value, tc.unit)
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+
+	positive := time.Date(1970, time.January, 1, 0, 0, 0, 500_000_000, time.UTC)
+	for _, tc := range []struct {
+		unit arrow.TimeUnit
+		want arrow.Timestamp
+	}{
+		{unit: arrow.Second, want: 0},
+		{unit: arrow.Millisecond, want: 500},
+		{unit: arrow.Microsecond, want: 500_000},
+		{unit: arrow.Nanosecond, want: 500_000_000},
+	} {
+		t.Run("positive_"+tc.unit.String(), func(t *testing.T) {
+			got, err := arrow.TimestampFromTime(positive, tc.unit)
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+
+	_, err := arrow.TimestampFromTime(value, arrow.TimeUnit(99))
+	assert.ErrorIs(t, err, arrow.ErrInvalid)
+}
+
 func TestNarrowestDecimalType(t *testing.T) {
 	tests := []struct {
 		min, max int32