Native Parquet schema-on-read diverges from Spark for ANSI interval targets

[peterxcli]

Describe the bug

When an explicit Parquet read schema requests an ANSI interval type that does not match the file column's logical type, Comet's native scan differs from Spark in two directions:

• A Parquet DATE column requested as INTERVAL YEAR TO MONTH succeeds in Spark but fails in Comet.
• A Parquet INT column requested as INTERVAL DAY TO SECOND fails in Spark but succeeds in Comet, reinterpreting the integers as microseconds.

The second case is a correctness gap because Comet returns rows for a read that Spark rejects with PARQUET_COLUMN_DATA_TYPE_MISMATCH.

Steps to reproduce

Write a one-column Parquet file, then read it with an explicit schema. Run each read once with Comet disabled for the Spark reference and once with the native Parquet scan enabled.

spark.conf.set("spark.sql.sources.useV1SourceList", "parquet")

spark.sql("SELECT CAST(DATE '2020-01-01' AS DATE) AS c")
  .write.parquet(datePath)
spark.read.schema("c INTERVAL YEAR TO MONTH")
  .parquet(datePath).collect()

spark.sql("SELECT CAST(1 AS INT) AS c")
  .write.parquet(intPath)
spark.read.schema("c INTERVAL DAY TO SECOND")
  .parquet(intPath).collect()

The full observed matrix on Spark 4.1 is:

File column	Requested schema	Spark	Comet
date	interval year to month	succeeds (raw INT32 value interpreted as total months)	fails: expected Interval(YearMonth) but found Date32
int	interval day to second	PARQUET_COLUMN_DATA_TYPE_MISMATCH	succeeds; values are reinterpreted as microseconds
timestamp	interval day to second	error	error
long	interval year to month	error	error

Expected behavior

Comet should match Spark's ParquetVectorUpdaterFactory compatibility rules:

• Accept INT32, including DATE, for YearMonthIntervalType and interpret the raw value as total months.
• Accept INT64 for DayTimeIntervalType.
• Reject INT32/INT for DayTimeIntervalType with Spark-compatible schema-mismatch behavior.
• Continue rejecting the timestamp-to-day-time and long-to-year-month control cases.

Additional context

The compatibility decision belongs in the native Parquet schema adapter. Conversion then reaches parquet_convert_array, whose Arrow can_cast_types path accepts numeric-to-Duration, while its fallback can leave Date32 unconverted for an Interval(YearMonth) target.

• Schema adapter:

  datafusion-comet/native/core/src/parquet/schema_adapter.rs

  Lines 773 to 829 in 2963b08

  	// Reject primitive Parquet conversions Spark's vectorized reader rejects
  	// on every supported version (no matching branch in
  	// `ParquetVectorUpdaterFactory.getUpdater`):
  	//
  	// - `INT64 -> Int*` truncates lower bits.
  	// - `INT64 -> Float*` and `INT32 -> Float32` lose precision.
  	// - `Float* -> Int*` and `Float64 -> Float32` truncate / overflow.
  	// - `INT32 -> Timestamp` / `INT64 -> Date32` / `INT64 -> Timestamp`:
  	// date/timestamp-annotated columns surface as Date32 / Timestamp,
  	// so reaching this branch means the column was un-annotated.
  	// - `Date32 -> Timestamp(LTZ)`: Spark only allows Date -> TimestampNTZ.
  	// - `Timestamp -> Date32`: no Timestamp updater branches into Date.
  	//
  	// Deferred to runtime (SPARK-26709). See #4297.
  	let is_spark_rejected_conversion = matches!(
  	(physical_type, target_type),
  	// Long -> narrower int.
  	(
  	DataType::Int64,
  	DataType::Int8 | DataType::Int16 | DataType::Int32,
  	)
  	// Long -> floating point.
  	| (DataType::Int64, DataType::Float32 | DataType::Float64)
  	// Long -> date / timestamp (raw INT64; annotated columns surface as Date32/Timestamp).
  	| (DataType::Int64, DataType::Date32)
  	| (DataType::Int64, DataType::Timestamp(_, _))
  	// Int -> float (DoubleType is allowed via IntegerToDoubleUpdater; FloatType is not).
  	| (
  	DataType::Int8 | DataType::Int16 | DataType::Int32,
  	DataType::Float32,
  	)
  	// Int -> timestamp (raw INT32; DATE-annotated columns surface as Date32).
  	| (
  	DataType::Int8 | DataType::Int16 | DataType::Int32,
  	DataType::Timestamp(_, _),
  	)
  	// Float -> int / Double -> int (no integer branches under FLOAT/DOUBLE).
  	| (
  	DataType::Float32 | DataType::Float64,
  	DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64,
  	)
  	// Double -> float (narrowing).
  	| (DataType::Float64, DataType::Float32)
  	// Date -> Timestamp(LTZ). Spark allows Date -> TimestampNTZ only.
  	| (DataType::Date32, DataType::Timestamp(_, Some(_)))
  	// Timestamp -> Date.
  	| (DataType::Timestamp(_, _), DataType::Date32)
  	);
  	if is_spark_rejected_conversion {
  	let rejection = reject_on_non_empty_expr(
  	child,
  	cast.target_field(),
  	input_field.name(),
  	physical_type,
  	target_type,
  	);
  	return Ok(Transformed::yes(rejection));

• Array conversion:

  datafusion-comet/native/core/src/parquet/parquet_support.rs

  Lines 166 to 260 in 2963b08

  	fn parquet_convert_array(
  	array: ArrayRef,
  	to_type: &DataType,
  	parquet_options: &SparkParquetOptions,
  	) -> DataFusionResult<ArrayRef> {
  	use DataType::*;
  	let from_type = array.data_type().clone();
  	let array = match &from_type {
  	Dictionary(key_type, value_type)
  	if key_type.as_ref() == &Int32
  	&& (value_type.as_ref() == &Utf8 || value_type.as_ref() == &LargeUtf8) =>
  	{
  	let dict_array = array
  	.as_any()
  	.downcast_ref::<DictionaryArray<Int32Type>>()
  	.expect("Expected a dictionary array");
  	let casted_dictionary = DictionaryArray::<Int32Type>::new(
  	dict_array.keys().clone(),
  	parquet_convert_array(Arc::clone(dict_array.values()), to_type, parquet_options)?,
  	);
  	let casted_result = match to_type {
  	Dictionary(_, _) => Arc::new(casted_dictionary.clone()),
  	_ => take(casted_dictionary.values().as_ref(), dict_array.keys(), None)?,
  	};
  	return Ok(casted_result);
  	}
  	_ => array,
  	};
  	let from_type = array.data_type();
  	// Try Comet specific handlers first, then arrow-rs cast if supported,
  	// return uncasted data otherwise
  	match (from_type, to_type) {
  	(Struct(_), Struct(_)) => Ok(parquet_convert_struct_to_struct(
  	array.as_struct(),
  	from_type,
  	to_type,
  	parquet_options,
  	)?),
  	(List(_), List(to_inner_type)) => {
  	let list_arr: &ListArray = array.as_list();
  	let cast_field = parquet_convert_array(
  	Arc::clone(list_arr.values()),
  	to_inner_type.data_type(),
  	parquet_options,
  	)?;
  	Ok(Arc::new(ListArray::new(
  	Arc::clone(to_inner_type),
  	list_arr.offsets().clone(),
  	cast_field,
  	list_arr.nulls().cloned(),
  	)))
  	}
  	(Timestamp(TimeUnit::Microsecond, None), Timestamp(TimeUnit::Microsecond, Some(tz))) => {
  	Ok(Arc::new(
  	array
  	.as_primitive::<TimestampMicrosecondType>()
  	.reinterpret_cast::<TimestampMicrosecondType>()
  	.with_timezone(Arc::clone(tz)),
  	))
  	}
  	(Map(_, ordered_from), Map(_, ordered_to)) if ordered_from == ordered_to =>
  	parquet_convert_map_to_map(array.as_map(), to_type, parquet_options, *ordered_to)
  	,
  	// Iceberg stores UUIDs as 16-byte fixed binary but Spark expects string representation.
  	// Arrow doesn't support casting FixedSizeBinary to Utf8, so we handle it manually.
  	(FixedSizeBinary(16), Utf8) => {
  	let binary_array = array
  	.as_any()
  	.downcast_ref::<FixedSizeBinaryArray>()
  	.expect("Expected a FixedSizeBinaryArray");
  	let string_array: StringArray = binary_array
  	.iter()
  	.map(|opt_bytes| {
  	opt_bytes.map(|bytes| {
  	let uuid = uuid::Uuid::from_bytes(
  	bytes.try_into().expect("Expected 16 bytes")
  	);
  	uuid.to_string()
  	})
  	})
  	.collect();
  	Ok(Arc::new(string_array))
  	}
  	// If Arrow cast supports the cast, delegate the cast to Arrow
  	_ if can_cast_types(from_type, to_type) => {
  	Ok(cast_with_options(&array, to_type, &PARQUET_OPTIONS)?)
  	}
  	_ => Ok(array),

• PR: feat: support ANSI interval types in native Parquet scans #5161
• Review finding: feat: support ANSI interval types in native Parquet scans #5161 (comment)
• Related generic schema-mismatch work: native_datafusion: no error thrown for schema mismatch when reading Parquet with incompatible types #3720 and native_datafusion: silent wrong-answer paths for primitive Parquet type conversions Spark rejects #4297

This issue is limited to schema-on-read compatibility for ANSI interval targets. Matching-schema interval scans are covered by #5060 / #5161. Add focused coverage beside the existing primitive schema-conversion cases in ParquetReadSuite.

[peterxcli]

take