This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 579f74b8a5ea [SPARK-60573][CONNECT] Preserve the full day-time
interval range in Spark Connect Arrow decoding
579f74b8a5ea is described below
commit 579f74b8a5eab79c2f6a046b1f63ed711290f58a
Author: Linhong Liu <[email protected]>
AuthorDate: Wed Oct 28 10:26:22 2026 -0700
[SPARK-60573][CONNECT] Preserve the full day-time interval range in Spark
Connect Arrow decoding
### What changes were proposed in this pull request?
Read day-time interval values directly as microseconds and convert them
with SparkIntervalUtils.microsToDuration.
### Why are the changes needed?
Arrow's DurationVector.getObject converts microseconds through a signed
64-bit nanosecond count. That intermediate value saturates for valid large
Spark intervals. For example, 10000000000000000 microseconds decodes as
approximately 9223372036854775 microseconds.
### Does this PR introduce _any_ user-facing change?
Yes, a bug fix. The Spark Connect JVM client retains the full day-time
interval range when decoding Arrow results. The wire representation is
unchanged.
### How was this patch tested?
Added six regression cases covering Long.MinValue, Long.MaxValue, -1, 0, 1
and a large positive microsecond count, each with NULL and interval-to-string
decoding. Three large-range cases fail on unchanged Apache code. On JDK 17, all
125 ArrowEncoderSuite tests passed; the six regression cases passed again after
formatting. Connect common/client Scalastyle and Scalafmt checks, plus git diff
--check, passed.
```bash
build/sbt 'connect-client-jvm/testOnly
org.apache.spark.sql.connect.client.arrow.ArrowEncoderSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Closes #59817 from linhongliu-db/fix-connect-interval-range.
Authored-by: Linhong Liu <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 57742c90eee4626c5e869e960f0f27c22d7ee201)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../connect/client/arrow/ArrowEncoderSuite.scala | 31 ++++++++++++++++++++++
.../connect/client/arrow/ArrowVectorReader.scala | 6 ++++-
2 files changed, 36 insertions(+), 1 deletion(-)
diff --git
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
index d4319f2b8e55..6cd9731851f2 100644
---
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
+++
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
@@ -1222,6 +1222,37 @@ class ArrowEncoderSuite extends ConnectFunSuite {
DayTimeIntervalType.DEFAULT.startField,
DayTimeIntervalType.DEFAULT.endField)
})
+
+ for (micros <- Seq(Long.MinValue, Long.MaxValue, -1L, 0L, 1L,
10000000000000000L)) {
+ test(s"day-time intervals retain full microsecond range: $micros") {
+ val duration = Duration.of(micros, ChronoUnit.MICROS)
+ val values = Seq(duration, null, Duration.ZERO)
+ val result = roundTrip(DayTimeIntervalEncoder, values.iterator)
+ try {
+ assert(result.toSeq == values)
+ } finally {
+ result.close()
+ }
+ val strings =
+ roundTripWithDifferentIOEncoders(DayTimeIntervalEncoder,
StringEncoder, values.iterator)
+ try {
+ assert(strings.toSeq == values.map { value =>
+ if (value == null) {
+ null
+ } else {
+ toDayTimeIntervalString(
+ durationToMicros(value),
+ ANSI_STYLE,
+ DayTimeIntervalType.DEFAULT.startField,
+ DayTimeIntervalType.DEFAULT.endField)
+ }
+ })
+ } finally {
+ strings.close()
+ }
+ }
+ }
+
UpCastTestCase(YearMonthIntervalEncoder, i => Period.ofMonths(i))
.test(
StringEncoder,
diff --git
a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowVectorReader.scala
b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowVectorReader.scala
index 1f71ed8f8432..bcaddbc697b8 100644
---
a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowVectorReader.scala
+++
b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowVectorReader.scala
@@ -242,7 +242,11 @@ private[arrow] class ViewVarBinaryVectorReader(v:
ViewVarBinaryVector)
private[arrow] class DurationVectorReader(v: DurationVector)
extends TypedArrowVectorReader[DurationVector](v) {
- override def getDuration(i: Int): Duration = vector.getObject(i)
+ // Spark stores day-time intervals as microseconds. Arrow getObject converts
via a Long
+ // nanosecond count, which saturates for valid large intervals. Decode the
raw microseconds
+ // directly to retain the full Spark interval range.
+ override def getDuration(i: Int): Duration =
+
SparkIntervalUtils.microsToDuration(DurationVector.get(vector.getDataBuffer, i))
override def getString(i: Int): String = {
SparkIntervalUtils.toDayTimeIntervalString(
SparkIntervalUtils.durationToMicros(getDuration(i)),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
