Support reading ANSI interval columns (YearMonthIntervalType / DayTimeIntervalType) in the native Parquet scan

[andygrove]

What is the problem the feature request solves?

Comet's native Parquet scan does not support the ANSI interval types, so any table with an INTERVAL YEAR TO MONTH or INTERVAL DAY TO SECOND column falls the entire scan back to Spark, including all the other columns in that table.

CREATE TABLE t(ym INTERVAL YEAR TO MONTH, dt INTERVAL DAY TO SECOND) USING parquet;
INSERT INTO t VALUES (make_ym_interval(1, 2), make_dt_interval(1, 2, 3, 4.5));
SELECT ym, dt FROM t;

Scan parquet spark_catalog.default.t [COMET: Unsupported schema
  StructType(StructField(ym,YearMonthIntervalType(0,1),true),StructField(dt,DayTimeIntervalType(0,3),true)):
  Unsupported ym of type YearMonthIntervalType(0,1)]

The gate is DataTypeSupport.isTypeSupported, which CometScanTypeChecker delegates to:

datafusion-comet/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala

Lines 51 to 56 in 1d3044f

	dt match {
	case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType |
	BinaryType | StringType | _: DecimalType | DateType | TimestampType | TimestampNTZType |
	CalendarIntervalType =>
	true
	case StructType(fields) =>

Note that the list includes CalendarIntervalType but not the two ANSI interval types, which is backwards for a scan gate. Spark's ParquetTable.supportsDataType accepts only AtomicType and friends; CalendarIntervalType extends DataType directly, so it can never appear in a Parquet file, while YearMonthIntervalType / DayTimeIntervalType extend AnsiIntervalType extends AtomicType and are written as annotated INT32 / INT64. So the one type listed is unreachable for scans, and the two that are reachable are rejected.

This is now the main thing blocking interval work from reaching real queries. After #4898 and #4976 the types round-trip through codegen dispatch, CometSink, CometLocalTableScanExec and native shuffle, and I verified that interval columns already work natively for joins, multi-key sorts, first/last, coalesce/if and comparisons. But none of that is reachable from a Parquet-backed table today, so the only way to get an interval column into a Comet operator is to construct it inline with make_dt_interval / make_ym_interval, which is exactly what all the existing tests do.

Describe the potential solution

• Add YearMonthIntervalType and DayTimeIntervalType to DataTypeSupport.isTypeSupported, and confirm the native Parquet reader decodes Spark's physical layout: YearMonthIntervalType is INT32 total months and DayTimeIntervalType is INT64 microseconds, matching the Arrow mappings already chosen in native/core/src/execution/serde.rs (Interval(YearMonth) and Duration(Microsecond)).
• Verify both scan implementations (CometScanExec / V1 and CometBatchScanExec / V2) and both spark.comet.scan.impl modes.
• Decide what to do with the CalendarIntervalType entry. It appears to be unreachable for scans; if it is only needed for CometLocalTableScanExec (which extends DataTypeSupport), it would be clearer to override there rather than widen the shared scan-facing gate.
• Add scan tests that write ANSI interval columns to Parquet and read them back, including nulls, negative values, the full startField/endField range (INTERVAL YEAR, INTERVAL MONTH, INTERVAL DAY TO HOUR, ...), and dictionary-encoded and multi-row-group files.

Additional context

• Part of [EPIC] Support Spark interval types (CalendarInterval / YearMonthInterval / DayTimeInterval) and interval expressions #4540. Follows feat: add CalendarIntervalType support #4898 (CalendarIntervalType support) and feat: support interval codegen dispatch for nested values and native shuffle #4976 (nested values and native shuffle).
• Reproduced on 1d3044f35 with the Spark 4.1 profile.
• Related gates that still exclude the ANSI interval types and will surface as fallbacks once scans work, worth tracking separately:
  • QueryPlanSerde.supportedScalarSortElementType already carries a TODO: Include SparkSQL's YearMonthIntervalType and DayTimeIntervalType. Single-column sort on an interval falls back, while multi-column sort is not checked at all and produces correct results, so the restriction looks removable.
  • AggSerde.minMaxDataTypeSupported / avgDataTypeSupported / sumDataTypeSupported reject intervals, so min / max / sum / avg fall back even though Spark supports all four over ANSI intervals.
  • CometShuffleExchangeExec.supportedHashPartitioningDataType and the columnar-shuffle supportedSerializableDataType reject all three interval types, so GROUP BY <interval> falls back unless shuffle mode is native.
  • arithmetic.scala's supportedDataType rejects intervals, so UnaryMinus and Abs fall back, even though native/spark-expr/src/math_funcs/negative.rs already has overflow-checked interval negation that is currently unreachable. That one is Math expressions: revisit interval / input-type gating and add regression tests once interval support lands #4756.

[peterxcli]

take