feat: native support for intervals <-> Neo4j duration

this commit introduces a new way to write from Spark SQL interval types to Neo4j duration type.
commit is additive, i.e. the previous method to write Neo4j duration type via custom struct is
still possible, and should therefore be backwards compatible.

Fixes CONN-341

Author: nord-erik

common/src/main/scala/org/neo4j/spark/converter/DataConverter.scala

@@ -55,11 +55,36 @@ trait DataConverter[T] {
 
 object SparkToNeo4jDataConverter {
   def apply(): SparkToNeo4jDataConverter = new SparkToNeo4jDataConverter()
+
+  def dayTimeIntervalToNeo4j(micros: Long): Value = {
+    val oneSecondInMicros = 1000000L
+    val oneDayInMicros = 24 * 3600 * oneSecondInMicros
+
+    val numberDays = Math.floorDiv(micros, oneDayInMicros)
+    val remainderMicros = Math.floorMod(micros, oneDayInMicros)
+    val numberSeconds = Math.floorDiv(remainderMicros, oneSecondInMicros)
+    val numberNanos = Math.floorMod(remainderMicros, oneSecondInMicros) * 1000
+
+    Values.isoDuration(0L, numberDays, numberSeconds, numberNanos.toInt)
+  }
+
+  // while Neo4j supports years, this driver version's API does not expose it.
+  def yearMonthIntervalToNeo4j(months: Int): Value = {
+    Values.isoDuration(months.toLong, 0L, 0L, 0)
+  }
 }
 
 class SparkToNeo4jDataConverter extends DataConverter[Value] {
 
   override def convert(value: Any, dataType: DataType): Value = {
+    dataType match {
+      case _: DayTimeIntervalType if value != null =>
+        return SparkToNeo4jDataConverter.dayTimeIntervalToNeo4j(value.asInstanceOf[Long])
+      case _: YearMonthIntervalType if value != null =>
+        return SparkToNeo4jDataConverter.yearMonthIntervalToNeo4j(value.asInstanceOf[Int])
+      case _ => // do nothing
+    }
+
     value match {
       case date: java.sql.Date           => convert(date.toLocalDate, dataType)
       case timestamp: java.sql.Timestamp => convert(timestamp.toLocalDateTime, dataType)

common/src/main/scala/org/neo4j/spark/converter/TypeConverter.scala

@@ -18,6 +18,8 @@ package org.neo4j.spark.converter
 
 import org.apache.spark.sql.types.DataType
 import org.apache.spark.sql.types.DataTypes
+import org.apache.spark.sql.types.DayTimeIntervalType
+import org.apache.spark.sql.types.YearMonthIntervalType
 import org.neo4j.driver.types.Entity
 import org.neo4j.spark.converter.CypherToSparkTypeConverter.cleanTerms
 import org.neo4j.spark.converter.CypherToSparkTypeConverter.durationType
@@ -129,6 +131,8 @@ object SparkToCypherTypeConverter {
     DataTypes.DoubleType -> "FLOAT",
     DataTypes.DateType -> "DATE",
     DataTypes.TimestampType -> "LOCAL DATETIME",
+    DayTimeIntervalType() -> "DURATION",
+    YearMonthIntervalType() -> "DURATION",
     durationType -> "DURATION",
     pointType -> "POINT",
     // Cypher graph entities do not allow null values in arrays
@@ -141,6 +145,10 @@ object SparkToCypherTypeConverter {
     DataTypes.createArrayType(DataTypes.DateType, false) -> "LIST<DATE NOT NULL>",
     DataTypes.createArrayType(DataTypes.TimestampType, false) -> "LIST<LOCAL DATETIME NOT NULL>",
     DataTypes.createArrayType(DataTypes.TimestampType, true) -> "LIST<LOCAL DATETIME NOT NULL>",
+    DataTypes.createArrayType(DayTimeIntervalType(), false) -> "LIST<DURATION NOT NULL>",
+    DataTypes.createArrayType(DayTimeIntervalType(), true) -> "LIST<DURATION NOT NULL>",
+    DataTypes.createArrayType(YearMonthIntervalType(), false) -> "LIST<DURATION NOT NULL>",
+    DataTypes.createArrayType(YearMonthIntervalType(), true) -> "LIST<DURATION NOT NULL>",
     DataTypes.createArrayType(durationType, false) -> "LIST<DURATION NOT NULL>",
     DataTypes.createArrayType(pointType, false) -> "LIST<POINT NOT NULL>"
   )

spark-3/src/test/scala/org/neo4j/spark/DataSourceWriterTSE.scala

@@ -21,6 +21,10 @@ import org.apache.spark.SparkException
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.SaveMode
 import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.types.ArrayType
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.DayTimeIntervalType
+import org.apache.spark.sql.types.YearMonthIntervalType
 import org.junit
 import org.junit.Assert._
 import org.junit.Ignore
@@ -30,6 +34,7 @@ import org.neo4j.driver.Transaction
 import org.neo4j.driver.TransactionWork
 import org.neo4j.driver.Value
 import org.neo4j.driver.exceptions.ClientException
+import org.neo4j.driver.exceptions.value.Uncoercible
 import org.neo4j.driver.internal.InternalPoint2D
 import org.neo4j.driver.internal.InternalPoint3D
 import org.neo4j.driver.internal.types.InternalTypeSystem
@@ -74,7 +79,7 @@ case class SimplePerson(name: String, surname: String)
 case class EmptyRow[T](data: T)
 
 class DataSourceWriterTSE extends SparkConnectorScalaBaseTSE {
-  val sparkSession = SparkSession.builder().getOrCreate()
+  val sparkSession: SparkSession = SparkSession.builder().getOrCreate()
 
   import sparkSession.implicits._
 
@@ -414,7 +419,7 @@ class DataSourceWriterTSE extends SparkConnectorScalaBaseTSE {
   }
 
   @Test
-  def `should write nodes with duration values into Neo4j`(): Unit = {
+  def `should write nodes with duration values into Neo4j from struct`(): Unit = {
     val total = 10
     val ds = (1 to total)
       .map(i => i.toLong)
@@ -430,10 +435,10 @@ class DataSourceWriterTSE extends SparkConnectorScalaBaseTSE {
 
     val records = SparkConnectorScalaSuiteIT.session().run(
       """MATCH (p:BeanWithDuration)
-        |RETURN p.data AS data
+        |RETURN p.data AS duration
         |""".stripMargin
     ).list().asScala
-      .map(r => r.get("data").asIsoDuration())
+      .map(r => r.get("duration").asIsoDuration())
       .map(data => (data.months, data.days, data.seconds, data.nanoseconds))
       .toSet
 
@@ -445,7 +450,7 @@ class DataSourceWriterTSE extends SparkConnectorScalaBaseTSE {
   }
 
   @Test
-  def `should write nodes with duration array values into Neo4j`(): Unit = {
+  def `should write nodes with duration array values into Neo4j from struct`(): Unit = {
     val total = 10
     val ds = (1 to total)
       .map(i => i.toLong)
@@ -484,6 +489,175 @@ class DataSourceWriterTSE extends SparkConnectorScalaBaseTSE {
     assertEquals(expected, records)
   }
 
+  private def writeAndGetInterval(expectedDt: Class[_ <: DataType], sql: String): Value = {
+    val id = java.util.UUID.randomUUID().toString
+    val df = sparkSession.sql(s"SELECT '$id' AS id, $sql AS duration")
+
+    df.write
+      .format(classOf[DataSource].getName)
+      .mode(SaveMode.Append)
+      .option("url", SparkConnectorScalaSuiteIT.server.getBoltUrl)
+      .option("labels", "Dur")
+      .save()
+
+    val want = expectedDt.getSimpleName
+    val got = df.schema("duration").dataType
+    assertTrue(s"expected Spark to pick $want but it was $got", expectedDt.isInstance(got))
+
+    SparkConnectorScalaSuiteIT.session().run(
+      s"""MATCH (d:Dur {id: '$id'})
+         |RETURN d.duration AS duration
+         |""".stripMargin
+    ).single().get("duration")
+  }
+
+  @Test
+  def `should write nodes with native Neo4j durations when passed DayTimeIntervalType`(): Unit = {
+    val got = writeAndGetInterval(classOf[DayTimeIntervalType], "INTERVAL '10 05:30:15.123' DAY TO SECOND")
+
+    assertTrue(
+      "can convert SQL day to second interval",
+      try {
+        val _ = got.asIsoDuration()
+        true
+      } catch {
+        case e: Uncoercible => false
+        case e              => throw e // passthrough other exceptions
+      }
+    )
+
+    assertEquals(0L, got.asIsoDuration().months)
+    assertEquals(10L, got.asIsoDuration().days)
+    assertEquals(19815L, got.asIsoDuration().seconds)
+    assertEquals(123000000, got.asIsoDuration().nanoseconds)
+  }
+
+  @Test
+  def `should write nodes with native Neo4j durations when passed YearMonthIntervalType`(): Unit = {
+    val got = writeAndGetInterval(classOf[YearMonthIntervalType], "INTERVAL '4-5' YEAR TO MONTH")
+
+    assertTrue(
+      "can convert SQL year to month interval",
+      try {
+        val _ = got.asIsoDuration()
+        true
+      } catch {
+        case e: Uncoercible => false
+        case e              => throw e // passthrough other exceptions
+      }
+    )
+
+    assertEquals(53L, got.asIsoDuration().months)
+    assertEquals(0L, got.asIsoDuration().days)
+    assertEquals(0L, got.asIsoDuration().seconds)
+    assertEquals(0, got.asIsoDuration().nanoseconds)
+  }
+
+  @Test
+  def `should write nodes with native Neo4j durations when passed timestamp arithmetics`(): Unit = {
+    val intervalQuery = "timestamp('2025-01-02 18:30:00.454') - timestamp('2024-01-01 00:00:00')"
+    val got = writeAndGetInterval(classOf[DayTimeIntervalType], intervalQuery)
+
+    assertTrue(
+      "can convert SQL day to second interval arithmetic",
+      try {
+        val _ = got.asIsoDuration()
+        true
+      } catch {
+        case e: Uncoercible => false
+        case e              => throw e // passthrough other exceptions
+      }
+    )
+
+    assertEquals(0L, got.asIsoDuration().months) // DayTimeIntervalType never returns months
+    assertEquals(367L, got.asIsoDuration().days) // can it capture the leap day?!
+    assertEquals(66600L, got.asIsoDuration().seconds)
+    assertEquals(454000000, got.asIsoDuration().nanoseconds)
+  }
+
+  private def writeAndQueryIfIsDurationArray(
+    expectedInnerDt: Class[_ <: DataType],
+    elemsSql: Seq[String]
+  ): Boolean = {
+    val id = java.util.UUID.randomUUID().toString
+    val sqlArray = elemsSql.mkString("array(", ", ", ")")
+    val df = sparkSession.sql(s"SELECT '$id' AS id, $sqlArray AS durations")
+
+    df.write
+      .format(classOf[DataSource].getName)
+      .mode(SaveMode.Append)
+      .option("url", SparkConnectorScalaSuiteIT.server.getBoltUrl)
+      .option("labels", "DurArr")
+      .save()
+
+    val got = df.schema("durations").dataType
+    val isTypeOk = got match {
+      case ArrayType(et, _) if expectedInnerDt.isInstance(et) => true
+      case _                                                  => false
+    }
+    assertTrue(
+      s"expected Spark to infer ArrayType(${expectedInnerDt.getSimpleName}) but it was $got",
+      isTypeOk
+    )
+
+    val result = SparkConnectorScalaSuiteIT.session().run(
+      s"""MATCH (d:DurArr {id: '$id'})
+         |RETURN d.durations AS durations
+         |""".stripMargin
+    ).single().get("durations")
+
+    try {
+      val _ = result.asList((v: Value) => v.asIsoDuration())
+    } catch {
+      case _: Uncoercible => return false
+      case e              => throw e
+    }
+
+    true
+  }
+
+  @Test
+  def `should write interval day second arrays as native neo4j durations`(): Unit = {
+    assertTrue(
+      "can convert array<DAY TO SECOND> to IsoDuration[]",
+      writeAndQueryIfIsDurationArray(
+        classOf[DayTimeIntervalType],
+        Seq(
+          "INTERVAL '10 05:30:15.123' DAY TO SECOND",
+          "INTERVAL '0 00:00:01.000' DAY TO SECOND"
+        )
+      )
+    )
+  }
+
+  @Test
+  def `should write interval year month arrays as native neo4j durations`(): Unit = {
+    assertTrue(
+      "can convert array<YEAR TO MONTH> to IsoDuration[]",
+      writeAndQueryIfIsDurationArray(
+        classOf[YearMonthIntervalType],
+        Seq(
+          "INTERVAL '1-02' YEAR TO MONTH",
+          "INTERVAL '0-11' YEAR TO MONTH"
+        )
+      )
+    )
+  }
+
+  @Test
+  def `should write interval arithmetic arrays as native neo4j durations`(): Unit = {
+    assertTrue(
+      "can convert array of interval arithmetic to IsoDuration[]",
+      writeAndQueryIfIsDurationArray(
+        classOf[DayTimeIntervalType],
+        Seq(
+          "timestamp('2024-01-02 00:00:00') - timestamp('2024-01-01 00:00:00')",
+          "current_timestamp() - timestamp('2024-01-01 00:00:00')"
+        )
+      )
+    )
+  }
+
   @Test
   def `should write nodes into Neo4j with points`(): Unit = {
     val total = 10