[benbellick]

Which issue does this PR close?

Closes #26605.

Rationale for this change

date_bin returns NULL for unrepresentable timestamp bins. Return an error instead, matching PostgreSQL.

I'm not sure whether DataFusion has a specific policy on returning NULL versus an error here, but the PostgreSQL precedent seems clear. Returning NULL because we cannot compute a bin for a non-null timestamp silently hides a computation failure rather than representing missing input.

This prepares date_bin for avoiding aggregate repartitioning in #25345, similar to the date_trunc optimization in #26449. It follows the error behavior established for date_trunc in #25131.

What changes are included in this PR?

Make scalar and array timestamp-binning failures return DATE_BIN timestamp out of range. Preserve NULL inputs and successful wider-arithmetic results.

What is the testing strategy for this PR?

Unit tests and SQLLogicTests cover overflow errors, scalar/array consistency, NULL preservation, and representable results across timestamp precisions.

Are there any user-facing changes?

Previously NULL-producing timestamp-binning failures now fail the query.

---

This PR was developed with AI assistance. All changes have been reviewed, and I take full responsibility for this contribution.

[benbellick]

mirrors the date_trunc nullness-preservation tests

datafusion/datafusion/functions/src/datetime/date_trunc.rs

Lines 1017 to 1063 in c7a7f94

	fn assert_timezone_less_date_trunc_preserves_nullness<T: ArrowTimestampType>() {
	let input = PrimitiveArray::<T>::from_iter([Some(0), None]);
	let timestamp_type = DataType::Timestamp(T::UNIT, None);
	for granularity in [
	"microsecond",
	"millisecond",
	"second",
	"minute",
	"hour",
	"day",
	"week",
	"month",
	"quarter",
	"year",
	] {
	let output = invoke_date_trunc(
	granularity,
	ColumnarValue::Array(Arc::new(input.clone())),
	timestamp_type.clone(),
	input.len(),
	)
	.unwrap();
	let ColumnarValue::Array(output) = output else {
	panic!("expected array output")
	};
	assert_eq!(
	input.nulls(),
	output.nulls(),
	"{granularity} must preserve nullness for {:?}",
	T::UNIT
	);
	}
	}
	/// Verifies the exact null contract required by
	/// [`ScalarUDFImpl::supports_range_key_locality_analysis`] across every
	/// supported granularity and timezone-less timestamp unit. Ordering metadata
	/// and concrete range-split separation are tested separately.
	#[test]
	fn timezone_less_date_trunc_preserves_nullness() {
	assert_timezone_less_date_trunc_preserves_nullness::<TimestampSecondType>();
	assert_timezone_less_date_trunc_preserves_nullness::<TimestampMillisecondType>();
	assert_timezone_less_date_trunc_preserves_nullness::<TimestampMicrosecondType>();
	assert_timezone_less_date_trunc_preserves_nullness::<TimestampNanosecondType>();
	}

[benbellick]

related date_trunc scalar/array consistency test

datafusion/datafusion/functions/src/datetime/date_trunc.rs

Lines 1930 to 1991 in 2717e25

	#[test]
	fn scalar_and_array_accept_timestamps_beyond_nanosecond_range() {
	// 2286-11-20T17:46:40, after the last nanosecond timestamp in 2262
	let seconds = 10_000_000_000;
	let utc: Option<Arc<str>> = Some("UTC".into());
	let cases = [
	// Seconds, truncated to a minute, an hour and a day
	(
	ScalarValue::TimestampSecond(Some(seconds + 59), None),
	"minute",
	),
	(
	ScalarValue::TimestampSecond(Some(seconds + 1), None),
	"hour",
	),
	(ScalarValue::TimestampSecond(Some(seconds + 1), None), "day"),
	// The granularity is the input's own unit
	(ScalarValue::TimestampSecond(Some(seconds), None), "second"),
	// With a time zone
	(
	ScalarValue::TimestampSecond(Some(seconds + 59), utc.clone()),
	"minute",
	),
	// Milliseconds and microseconds, truncated to a coarser unit
	(
	ScalarValue::TimestampMillisecond(Some(seconds * 1_000 + 999), None),
	"second",
	),
	(
	ScalarValue::TimestampMicrosecond(Some(seconds * 1_000_000 + 999), None),
	"millisecond",
	),
	// Before the epoch, with a time zone
	(
	ScalarValue::TimestampMicrosecond(Some(-seconds * 1_000_000 - 1), utc),
	"second",
	),
	];
	for (value, granularity) in cases {
	let (scalar, array) = date_trunc_scalar_and_array(granularity, value.clone());
	let scalar = scalar.unwrap_or_else(|e| {
	panic!("scalar date_trunc('{granularity}', {value:?}) failed: {e}")
	});
	assert_eq!(
	scalar,
	array.unwrap(),
	"date_trunc('{granularity}', {value:?})"
	);
	}
	// `date_trunc('second', to_timestamp_seconds(10000000000))` returns its input
	// rather than an out of range error.
	// See <https://github.com/apache/datafusion/issues/25432>.
	let (scalar, _) = date_trunc_scalar_and_array(
	"second",
	ScalarValue::TimestampSecond(Some(seconds), None),
	);
	assert_eq!(
	scalar.unwrap(),
	ScalarValue::TimestampSecond(Some(seconds), None)
	);
	}

[benbellick]

cc @jayzhan211 since you reviewed the last PR related to a similar fix for date_trunc. Please let me know if anything is unclear, thanks!

[jayzhan211]

Thanks @benbellick ! LGTM