[substrait] [sqllogictest] Unsupported cast type: Duration

[gabotechs]

Describe the bug

~10 sqllogictest cases are failing in Substrait round-trip mode because of the following error:

query failed: DataFusion error: This feature is not implemented: Unsupported cast type: Duration(Nanosecond)

To Reproduce

cargo test --test sqllogictests -- --substrait-round-trip timestamps.slt:1607

Expected behavior

The sqllogictest passes in Substrait round-trip mode

Additional context

This is part of a broader initiative to improve the reliability of Substrait-to-DataFusion logical plan conversions.

• [Epic] A collection of Substrait conversion issues #16248

[jkosh44]

take

[jkosh44]

The query that fails looks like

create table foo (val int, ts1 timestamp, ts2 timestamp, i interval) ...
SELECT val, ts1 - ts2 FROM foo ORDER BY ts2 - ts1;

Subtracting two timestamps returns a duration type.

> SELECT arrow_typeof(ts1 - ts2) FROM foo LIMIT 1;
+---------------------------------+
| arrow_typeof(foo.ts1 - foo.ts2) |
+---------------------------------+
| Duration(Nanosecond)            |
+---------------------------------+
1 row(s) fetched. 
Elapsed 0.005 seconds.

There is no duration type in substrait (https://substrait.io/types/type_classes/), so the cast from arrow to substrate fails:

datafusion/datafusion/substrait/src/logical_plan/producer/types.rs

Lines 30 to 289 in 4cf1148

	pub(crate) fn to_substrait_type(
	dt: &DataType,
	nullable: bool,
	) -> datafusion::common::Result<substrait::proto::Type> {
	let nullability = if nullable {
	r#type::Nullability::Nullable as i32
	} else {
	r#type::Nullability::Required as i32
	};
	match dt {
	DataType::Null => internal_err!("Null cast is not valid"),
	DataType::Boolean => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Bool(r#type::Boolean {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Int8 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::I8(r#type::I8 {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::UInt8 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::I8(r#type::I8 {
	type_variation_reference: UNSIGNED_INTEGER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Int16 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::I16(r#type::I16 {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::UInt16 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::I16(r#type::I16 {
	type_variation_reference: UNSIGNED_INTEGER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Int32 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::I32(r#type::I32 {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::UInt32 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::I32(r#type::I32 {
	type_variation_reference: UNSIGNED_INTEGER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Int64 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::I64(r#type::I64 {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::UInt64 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::I64(r#type::I64 {
	type_variation_reference: UNSIGNED_INTEGER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	// Float16 is not supported in Substrait
	DataType::Float32 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Fp32(r#type::Fp32 {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Float64 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Fp64(r#type::Fp64 {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Timestamp(unit, tz) => {
	let precision = match unit {
	TimeUnit::Second => 0,
	TimeUnit::Millisecond => 3,
	TimeUnit::Microsecond => 6,
	TimeUnit::Nanosecond => 9,
	};
	let kind = match tz {
	None => r#type::Kind::PrecisionTimestamp(r#type::PrecisionTimestamp {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	precision,
	}),
	Some(_) => {
	// If timezone is present, no matter what the actual tz value is, it indicates the
	// value of the timestamp is tied to UTC epoch. That's all that Substrait cares about.
	// As the timezone is lost, this conversion may be lossy for downstream use of the value.
	r#type::Kind::PrecisionTimestampTz(r#type::PrecisionTimestampTz {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	precision,
	})
	}
	};
	Ok(substrait::proto::Type { kind: Some(kind) })
	}
	DataType::Date32 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Date(r#type::Date {
	type_variation_reference: DATE_32_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Date64 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Date(r#type::Date {
	type_variation_reference: DATE_64_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Interval(interval_unit) => {
	match interval_unit {
	IntervalUnit::YearMonth => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::IntervalYear(r#type::IntervalYear {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	IntervalUnit::DayTime => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::IntervalDay(r#type::IntervalDay {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	precision: Some(3), // DayTime precision is always milliseconds
	})),
	}),
	IntervalUnit::MonthDayNano => {
	Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::IntervalCompound(
	r#type::IntervalCompound {
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	precision: 9, // nanos
	},
	)),
	})
	}
	}
	}
	DataType::Binary => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Binary(r#type::Binary {
	type_variation_reference: DEFAULT_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::FixedSizeBinary(length) => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::FixedBinary(r#type::FixedBinary {
	length: *length,
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::LargeBinary => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Binary(r#type::Binary {
	type_variation_reference: LARGE_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::BinaryView => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Binary(r#type::Binary {
	type_variation_reference: VIEW_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Utf8 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::String(r#type::String {
	type_variation_reference: DEFAULT_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::LargeUtf8 => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::String(r#type::String {
	type_variation_reference: LARGE_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::Utf8View => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::String(r#type::String {
	type_variation_reference: VIEW_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	})),
	}),
	DataType::List(inner) => {
	let inner_type = to_substrait_type(inner.data_type(), inner.is_nullable())?;
	Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::List(Box::new(r#type::List {
	r#type: Some(Box::new(inner_type)),
	type_variation_reference: DEFAULT_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	}))),
	})
	}
	DataType::LargeList(inner) => {
	let inner_type = to_substrait_type(inner.data_type(), inner.is_nullable())?;
	Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::List(Box::new(r#type::List {
	r#type: Some(Box::new(inner_type)),
	type_variation_reference: LARGE_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	}))),
	})
	}
	DataType::Map(inner, _) => match inner.data_type() {
	DataType::Struct(key_and_value) if key_and_value.len() == 2 => {
	let key_type = to_substrait_type(
	key_and_value[0].data_type(),
	key_and_value[0].is_nullable(),
	)?;
	let value_type = to_substrait_type(
	key_and_value[1].data_type(),
	key_and_value[1].is_nullable(),
	)?;
	Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Map(Box::new(r#type::Map {
	key: Some(Box::new(key_type)),
	value: Some(Box::new(value_type)),
	type_variation_reference: DEFAULT_CONTAINER_TYPE_VARIATION_REF,
	nullability,
	}))),
	})
	}
	_ => plan_err!("Map fields must contain a Struct with exactly 2 fields"),
	},
	DataType::Struct(fields) => {
	let field_types = fields
	.iter()
	.map(|field| to_substrait_type(field.data_type(), field.is_nullable()))
	.collect::<datafusion::common::Result<Vec<_>>>()?;
	Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Struct(r#type::Struct {
	types: field_types,
	type_variation_reference: DEFAULT_TYPE_VARIATION_REF,
	nullability,
	})),
	})
	}
	DataType::Decimal128(p, s) => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Decimal(r#type::Decimal {
	type_variation_reference: DECIMAL_128_TYPE_VARIATION_REF,
	nullability,
	scale: *s as i32,
	precision: *p as i32,
	})),
	}),
	DataType::Decimal256(p, s) => Ok(substrait::proto::Type {
	kind: Some(r#type::Kind::Decimal(r#type::Decimal {
	type_variation_reference: DECIMAL_256_TYPE_VARIATION_REF,
	nullability,
	scale: *s as i32,
	precision: *p as i32,
	})),
	}),
	_ => not_impl_err!("Unsupported cast type: {dt:?}"),
	}
	}

.

Interestingly, Duration doesn't really seem to be a fully supported data type by Datafusion. It's not listed as a supported SQL datatype: https://datafusion.apache.org/user-guide/sql/data_types.html#date-time-types. You can't directly create a table with a duration column type:

CREATE TABLE t (a duration);
This feature is not implemented: Unsupported SQL type Custom(ObjectName([Identifier(Ident { value: "duration", quote_style: None, span: Span(Location(1,19)..Location(1,27)) })]), [])

but you can indirectly create a table with a duration column type

> CREATE TABLE t AS SELECT '2023-03-15T15:00:20.000000123'::timestamp - '2023-01-20T23:00:00.000000099'::timestamp;
0 row(s) fetched. 
Elapsed 0.009 seconds.

> SHOW COLUMNS FROM t;
+---------------+--------------+------------+-------------------------------------------------------------------------------+----------------------+-------------+
| table_catalog | table_schema | table_name | column_name                                                                   | data_type            | is_nullable |
+---------------+--------------+------------+-------------------------------------------------------------------------------+----------------------+-------------+
| datafusion    | public       | t          | Utf8("2023-03-15T15:00:20.000000123") - Utf8("2023-01-20T23:00:00.000000099") | Duration(Nanosecond) | NO          |
+---------------+--------------+------------+-------------------------------------------------------------------------------+----------------------+-------------+
1 row(s) fetched. 
Elapsed 0.010 seconds.

and as shown above, some expressions return a duration.

One solution would be for Datafusion to stop using Duration and stick to Interval. There has been recent work to add more support for Durations, so I don't think this is desirable:

• Support AVG for durations #15458
• Support Duration in min/max agg functions #15310
• Implement GroupsAccumulator for min/max Duration #15322

The other solution would be to add suport for casting Arrow Duration types to Substrait. There's probably multiple approaches to this:

• Use INTERVAL_COMPOUND<P> (or maybe INTERVAL_DAY<P> is enough). This would be a lossy conversion, because the round trip would look like Arrow Duration -> Substrait Interval -> Arrow Interval.
• Use some type of user defined type, STRUCT, or NSTRUCT.

[jkosh44]

Of course, yet another solution would be to add the Duration type to substrait, but they'd need to be interested in doing that.