fix(compute): detect temporal rounding overflow

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/compute/internal/kernels/rounding.go

@@ -873,7 +873,11 @@ func InitRoundTemporalState(_ *exec.KernelCtx, args exec.KernelInitArgs) (exec.K
 	// Pre-calculate constants for this rounding operation
 	rs.unitNanos, rs.isSubDay = unitInNanos(rs.Unit)
 	if rs.isSubDay {
-		rs.roundingInterval = rs.unitNanos * rs.Multiple
+		var err error
+		rs.roundingInterval, err = checkedMulInt64(rs.unitNanos, rs.Multiple)
+		if err != nil {
+			return nil, err
+		}
 		rs.useCalendarOrigin = rs.CalendarBasedOrigin && rs.Unit <= RoundTemporalDay
 	}
 
@@ -912,43 +916,63 @@ func roundTimestamp(ts int64, inputUnit arrow.TimeUnit, tz *time.Location, opts
 
 	// Calendar units with variable duration (year, quarter, month, week) require date arithmetic
 	if !opts.isSubDay {
-		tsNanos := convertToNanos(ts, inputUnit)
+		tsNanos, err := convertToNanos(ts, inputUnit)
+		if err != nil {
+			return 0, err
+		}
 		return roundTimestampCalendar(tsNanos, inputUnit, tz, opts)
 	}
 
 	// Day rounding with timezone requires calendar arithmetic (days vary: 23/24/25 hours due to DST)
 	isUTC := tz == time.UTC || tz.String() == "UTC"
 	if !isUTC && opts.Unit == RoundTemporalDay {
-		tsNanos := convertToNanos(ts, inputUnit)
+		tsNanos, err := convertToNanos(ts, inputUnit)
+		if err != nil {
+			return 0, err
+		}
 		return roundTimestampCalendar(tsNanos, inputUnit, tz, opts)
 	}
 
 	// Sub-day units (hour, minute, second, etc.) use fixed-duration arithmetic
 	// Fast path: round directly in input unit if possible (no origin, compatible units)
 	if canRoundInInputUnit(inputUnit, opts.unitNanos) && !opts.useCalendarOrigin {
 		intervalInInputUnit := opts.roundingInterval / int64(inputUnit.Multiplier())
-		rounded := roundToMultipleInt64(ts, intervalInInputUnit, opts.mode, opts.CeilIsStrictlyGreater)
-		return rounded, nil
+		return roundToMultipleInt64(ts, intervalInInputUnit, opts.mode, opts.CeilIsStrictlyGreater)
 	}
 
 	// Slow path: convert to nanoseconds for calendar origin or incompatible units
-	tsNanos := convertToNanos(ts, inputUnit)
+	tsNanos, err := convertToNanos(ts, inputUnit)
+	if err != nil {
+		return 0, err
+	}
 
 	var origin int64 = 0
 	if opts.useCalendarOrigin {
 		// Calendar origin: round relative to start of day (timezone-aware if tz != nil)
 		if tz != nil {
 			t := time.Unix(0, tsNanos).In(tz)
 			startOfDay := time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, tz)
-			origin = startOfDay.UnixNano()
+			origin, err = timeToNanos(startOfDay)
+			if err != nil {
+				return 0, err
+			}
 		} else {
 			origin = tsNanos
 		}
 	}
 
-	adjusted := tsNanos - origin
-	rounded := roundToMultipleInt64(adjusted, opts.roundingInterval, opts.mode, opts.CeilIsStrictlyGreater)
-	result := origin + rounded
+	adjusted, err := checkedSubInt64(tsNanos, origin)
+	if err != nil {
+		return 0, err
+	}
+	rounded, err := roundToMultipleInt64(adjusted, opts.roundingInterval, opts.mode, opts.CeilIsStrictlyGreater)
+	if err != nil {
+		return 0, err
+	}
+	result, err := checkedAddInt64(origin, rounded)
+	if err != nil {
+		return 0, err
+	}
 
 	return convertFromNanos(result, inputUnit), nil
 }
@@ -959,22 +983,55 @@ func canRoundInInputUnit(inputUnit arrow.TimeUnit, roundingIntervalNanos int64)
 	return roundingIntervalNanos%int64(inputUnit.Multiplier()) == 0
 }
 
-// convertToNanos converts a timestamp value to nanoseconds
-func convertToNanos(ts int64, unit arrow.TimeUnit) int64 {
-	return ts * int64(unit.Multiplier())
+func overflowError() error {
+	return fmt.Errorf("%w: temporal rounding overflow", arrow.ErrInvalid)
+}
+
+func checkedAddInt64(left, right int64) (int64, error) {
+	if (right > 0 && left > math.MaxInt64-right) || (right < 0 && left < math.MinInt64-right) {
+		return 0, overflowError()
+	}
+	return left + right, nil
+}
+
+func checkedSubInt64(left, right int64) (int64, error) {
+	if (right > 0 && left < math.MinInt64+right) || (right < 0 && left > math.MaxInt64+right) {
+		return 0, overflowError()
+	}
+	return left - right, nil
+}
+
+func checkedMulInt64(left, right int64) (int64, error) {
+	if left == 0 || right == 0 {
+		return 0, nil
+	}
+	if (left == math.MinInt64 && right == -1) || (right == math.MinInt64 && left == -1) {
+		return 0, overflowError()
+	}
+
+	result := left * right
+	if result/right != left {
+		return 0, overflowError()
+	}
+	return result, nil
+}
+
+// convertToNanos converts a timestamp value to nanoseconds.
+func convertToNanos(ts int64, unit arrow.TimeUnit) (int64, error) {
+	return checkedMulInt64(ts, int64(unit.Multiplier()))
 }
 
 // convertFromNanos converts a nanosecond timestamp to the specified unit
 func convertFromNanos(nanos int64, unit arrow.TimeUnit) int64 {
 	return nanos / int64(unit.Multiplier())
 }
 
-func roundToMultipleInt64(value, multiple int64, mode RoundMode, strictCeil bool) int64 {
+func roundToMultipleInt64(value, multiple int64, mode RoundMode, strictCeil bool) (int64, error) {
 	if multiple == 0 || value%multiple == 0 {
 		if strictCeil && mode == RoundUp {
-			return value + multiple
+			return checkedAddInt64(value, multiple)
 		}
-		return value
+		return value, nil
 	}
 
 	quotient := value / multiple
@@ -983,56 +1040,95 @@ func roundToMultipleInt64(value, multiple int64, mode RoundMode, strictCeil bool
 	switch mode {
 	case RoundDown:
 		if remainder < 0 {
-			return (quotient - 1) * multiple
+			quotient, err := checkedSubInt64(quotient, 1)
+			if err != nil {
+				return 0, err
+			}
+			return checkedMulInt64(quotient, multiple)
 		}
-		return quotient * multiple
+		return checkedMulInt64(quotient, multiple)
 	case RoundUp:
 		if remainder > 0 || (strictCeil && remainder == 0) {
-			return (quotient + 1) * multiple
+			quotient, err := checkedAddInt64(quotient, 1)
+			if err != nil {
+				return 0, err
+			}
+			return checkedMulInt64(quotient, multiple)
 		}
 		if remainder < 0 {
-			return quotient * multiple
+			return checkedMulInt64(quotient, multiple)
+		}
+		quotient, err := checkedAddInt64(quotient, 1)
+		if err != nil {
+			return 0, err
 		}
-		return (quotient + 1) * multiple
+		return checkedMulInt64(quotient, multiple)
 	case HalfUp, HalfDown, HalfToEven:
 		half := multiple / 2
 		absRemainder := remainder
 		if absRemainder < 0 {
 			absRemainder = -absRemainder
 		}
 
-		if absRemainder < half {
-			return quotient * multiple
+		// Odd multiples do not have an exact halfway point. For example,
+		// a remainder of 1 when rounding to multiples of 3 is closer to 0
+		// than to 3, so it must not be treated as a tie.
+		if absRemainder < half || (multiple%2 != 0 && absRemainder == half) {
+			return checkedMulInt64(quotient, multiple)
 		} else if absRemainder > half {
 			if remainder > 0 {
-				return (quotient + 1) * multiple
+				quotient, err := checkedAddInt64(quotient, 1)
+				if err != nil {
+					return 0, err
+				}
+				return checkedMulInt64(quotient, multiple)
 			}
-			return (quotient - 1) * multiple
+			quotient, err := checkedSubInt64(quotient, 1)
+			if err != nil {
+				return 0, err
+			}
+			return checkedMulInt64(quotient, multiple)
 		} else {
 			// Exactly on the halfway point
 			switch mode {
 			case HalfDown:
 				if remainder > 0 {
-					return quotient * multiple
+					return checkedMulInt64(quotient, multiple)
+				}
+				quotient, err := checkedSubInt64(quotient, 1)
+				if err != nil {
+					return 0, err
 				}
-				return (quotient - 1) * multiple
+				return checkedMulInt64(quotient, multiple)
 			case HalfUp:
 				if remainder > 0 {
-					return (quotient + 1) * multiple
+					quotient, err := checkedAddInt64(quotient, 1)
+					if err != nil {
+						return 0, err
+					}
+					return checkedMulInt64(quotient, multiple)
 				}
-				return quotient * multiple
+				return checkedMulInt64(quotient, multiple)
 			case HalfToEven:
 				if quotient%2 == 0 {
-					return quotient * multiple
+					return checkedMulInt64(quotient, multiple)
 				}
 				if remainder > 0 {
-					return (quotient + 1) * multiple
+					quotient, err := checkedAddInt64(quotient, 1)
+					if err != nil {
+						return 0, err
+					}
+					return checkedMulInt64(quotient, multiple)
+				}
+				quotient, err := checkedSubInt64(quotient, 1)
+				if err != nil {
+					return 0, err
 				}
-				return (quotient - 1) * multiple
+				return checkedMulInt64(quotient, multiple)
 			}
 		}
 	}
-	return quotient * multiple
+	return checkedMulInt64(quotient, multiple)
 }
 
 // halfRoundPeriod performs half-rounding by finding the midpoint between period start and end
@@ -1198,10 +1294,21 @@ func roundTimestampCalendar(tsNanos int64, inputUnit arrow.TimeUnit, tz *time.Lo
 	}
 
 	// Convert back to the input unit
-	roundedNanos := rounded.UnixNano()
+	roundedNanos, err := timeToNanos(rounded)
+	if err != nil {
+		return 0, err
+	}
 	return convertFromNanos(roundedNanos, inputUnit), nil
 }
 
+func timeToNanos(value time.Time) (int64, error) {
+	timestamp, err := arrow.TimestampFromTime(value, arrow.Nanosecond)
+	if err != nil {
+		return 0, err
+	}
+	return int64(timestamp), nil
+}
+
 // Kernel execution functions for temporal rounding
 func FloorTemporalKernel(ctx *exec.KernelCtx, batch *exec.ExecSpan, out *exec.ExecResult) error {
 	state := ctx.State.(roundTemporalState)
@@ -1237,7 +1344,12 @@ func roundTemporalExec(ctx *exec.KernelCtx, batch *exec.ExecSpan, out *exec.Exec
 				return 0
 			}
 			// Convert back to days
-			return int32(result / 86400)
+			daysResult := result / 86400
+			if daysResult < math.MinInt32 || daysResult > math.MaxInt32 {
+				*e = overflowError()
+				return 0
+			}
+			return int32(daysResult)
 		}
 		return ScalarUnaryNotNull(fn)(ctx, batch, out)
 

arrow/compute/internal/kernels/temporal_rounding_odd_multiple_test.go

@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build go1.18
+
+package kernels
+
+import "testing"
+
+func TestRoundToMultipleInt64OddMultipleAcrossModes(t *testing.T) {
+	for _, mode := range []RoundMode{HalfDown, HalfUp, HalfToEven} {
+		t.Run(mode.String(), func(t *testing.T) {
+			for _, tc := range []struct {
+				value int64
+				want  int64
+			}{
+				{value: 1, want: 0},
+				{value: 2, want: 3},
+				{value: -1, want: 0},
+				{value: -2, want: -3},
+			} {
+				got, err := roundToMultipleInt64(tc.value, 3, mode, false)
+				if err != nil {
+					t.Fatalf("roundToMultipleInt64(%d, 3, %s) returned an error: %v", tc.value, mode, err)
+				}
+				if got != tc.want {
+					t.Errorf("roundToMultipleInt64(%d, 3, %s) = %d, want %d", tc.value, mode, got, tc.want)
+				}
+			}
+		})
+	}
+}