[SPARK-37277][PYTHON][SQL] Support DayTimeIntervalType in pandas UDF and Arrow optimization

[HyukjinKwon]

What changes were proposed in this pull request?

This PR proposes to support DayTimeIntervalType in pandas UDF and Arrow optimization.

• Change the mapping of Arrow's IntervalType to DurationType for DayTimeIntervalType (migration guide updated for Arrow developer APIs).
• Add a type mapping for other code paths: numpy.timedelta64 <> pyarrow.duration("us") <> DayTimeIntervalType

Why are the changes needed?

For changing the mapping of Arrow's Interval type to Duration type for DayTimeIntervalType, please refer to #32340 (comment).

DayTimeIntervalType is already mapped to the concept of duration instead of calendar instance: it's is matched to pyarrow.duration("us"), datetime.timedelta, and java.util.Duration.

Spark SQL example

scala> sql("SELECT timestamp '2029-01-01 00:00:00' - timestamp '2019-01-01 00:00:00'").show()

+-------------------------------------------------------------------+
|(TIMESTAMP '2029-01-01 00:00:00' - TIMESTAMP '2019-01-01 00:00:00')|
+-------------------------------------------------------------------+
|                                               INTERVAL '3653 00...|
+-------------------------------------------------------------------+

scala> sql("SELECT timestamp '2029-01-01 00:00:00' - timestamp '2019-01-01 00:00:00'").printSchema()

root
 |-- (TIMESTAMP '2029-01-01 00:00:00' - TIMESTAMP '2019-01-01 00:00:00'): interval day to second (nullable = false)

Python example:

>>> import datetime
>>> datetime.datetime.now() - datetime.datetime.now()
datetime.timedelta(days=-1, seconds=86399, microseconds=999996)

pandas / PyArrow example:

>>> import pyarrow as pa
>>> import pandas as pd
>>> pdf = pd.DataFrame({'a': [datetime.datetime.now() - datetime.datetime.now()]})
>>> tbl = pa.Table.from_pandas(pdf)
>>> tbl.schema
a: duration[ns]
-- schema metadata --
pandas: '{"index_columns": [{"kind": "range", "name": null, "start": 0, "' + 368

Does this PR introduce any user-facing change?

Yes, after this change, users can use DayTimeIntervalType in SparkSession.createDataFrame(pandas_df), DataFrame.to_pandas, and pandas UDFs:

>>> import datetime
>>> import pandas as pd
>>> from pyspark.sql.functions import pandas_udf
>>>
>>> @pandas_udf("interval day to second")
... def noop(s: pd.Series) -> pd.Series:
...     assert s.iloc[0] == datetime.timedelta(microseconds=123)
...     return s
...
>>> df = spark.createDataFrame(pd.DataFrame({"a": [pd.Timedelta(microseconds=123)]}))
>>> df.toPandas()
                       a
0 0 days 00:00:00.000123

How was this patch tested?

Manually tested, and unittests were added.

[HyukjinKwon]

cc @cloud-fan, @BryanCutler @ueshin @viirya @Peng-Lei @Yikun @MaxGekk FYI

[cloud-fan]

do we need to care about precision here? (microsecond)

[cloud-fan]

oh, so it's just an int64 physically, and the type annotation (or logical type) decides its semantic

[Peng-Lei]

LGTM for *.scala and *.java

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/49844/

[viirya]

Are you actually meaning len(pdf) != 0? Or I miss-read the code/comment?

[HyukjinKwon]

Ohh comments are wrong. Let me rewrite.

[HyukjinKwon]

This is, BTW, to work around a bug from Arrow <> pandas.

For some reasons, pd.Series(pd.Timedelta(...), dtype="object") created from Arrow becomes float64 when you cast with series.astype("timedelta64[us]") when the data is non-empty - this cannot be reproduced with plain pandas Series.

So, here I avoided it by just skipping the casting because the type becomes correct when it is not empty. When data is empty, the type becomes object, and it has to be casted.

[viirya]

Thanks for updating it. Looks good now.

[viirya]

Don't want to keep map type ....?

[HyukjinKwon]

Map type was added from #30393 but the doc fix was missing 😂

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/49844/

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/49850/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/49850/

[HyukjinKwon]

Thanks guys.

Merged to master.

[SparkQA]

Test build #145371 has finished for PR 34631 at commit d90d8bf.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #145377 has finished for PR 34631 at commit eb2a55e.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.