SQL: Implement DATEDIFF function (#47920)

Implement DATEDIFF/TIMESTAMPDIFF function as per the MS-SQL spec:
https://docs.microsoft.com/en-us/sql/t-sql/functions/datediff-transact-sql?view=sql-server-2017
which allows a user to substract two date/datetime fields and return the
difference in the date/time unit specified.

Closes: #47919

docs/reference/sql/functions/date-time.asciidoc

@@ -249,7 +249,7 @@ Currently, using a _precision_ greater than 3 doesn't make any difference to the
 function as the maximum number of second fractional digits returned is 3 (milliseconds).
 
 [[sql-functions-datetime-add]]
-==== `DATE_ADD/DATEADD/TIMESTAMPADD/TIMESTAMP_ADD`
+==== `DATE_ADD/DATEADD/TIMESTAMP_ADD/TIMESTAMPADD`
 
 .Synopsis:
 [source, sql]
@@ -326,6 +326,80 @@ include-tagged::{sql-specs}/docs/docs.csv-spec[dateAddDateQuarters]
 include-tagged::{sql-specs}/docs/docs.csv-spec[dateAddDateMinutes]
 --------------------------------------------------
 
+[[sql-functions-datetime-diff]]
+==== `DATE_DIFF/DATEDIFF/TIMESTAMP_DIFF/TIMESTAMPDIFF`
+
+.Synopsis:
+[source, sql]
+--------------------------------------------------
+DATE_DIFF(
+    string_exp, <1>
+    datetime_exp, <2>
+    datetime_exp) <3>
+--------------------------------------------------
+
+*Input*:
+
+<1> string expression denoting the date/time unit difference between the following two date/datetime expressions
+<2> start date/datetime expression
+<3> end date/datetime expression
+
+*Output*: integer
+
+.Description:
+
+Subtract the second argument from the third argument and return their difference in multiples of the unit
+specified in the first argument. If the second argument (start) is greater than the third argument (end),
+then negative values are returned. If any of the three arguments is `null`, a `null` is returned.
+
+[cols="^,^"]
+|===
+2+h|Datetime difference units
+
+s|unit
+s|abbreviations
+
+| year        | years, yy, yyyy
+| quarter     | quarters, qq, q
+| month       | months, mm, m
+| dayofyear   | dy, y
+| day         | days, dd, d
+| week        | weeks, wk, ww
+| weekday     | weekdays, dw
+| hour        | hours, hh
+| minute      | minutes, mi, n
+| second      | seconds, ss, s
+| millisecond | milliseconds, ms
+| microsecond | microseconds, mcs
+| nanosecond  | nanoseconds, ns
+|===
+
+
+[source, sql]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[dateDiffDateTimeYears]
+--------------------------------------------------
+
+[source, sql]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[dateDiffDateTimeWeeks]
+--------------------------------------------------
+
+[source, sql]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[dateDiffDateTimeSeconds]
+--------------------------------------------------
+
+[source, sql]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[dateDiffDateQuarters]
+--------------------------------------------------
+
+[source, sql]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[dateDiffDateMinutes]
+--------------------------------------------------
+
 [[sql-functions-datetime-part]]
 ==== `DATE_PART/DATEPART`
 

docs/reference/sql/functions/index.asciidoc

@@ -52,6 +52,7 @@
 ** <<sql-functions-current-time>>
 ** <<sql-functions-current-timestamp>>
 ** <<sql-functions-datetime-add>>
+** <<sql-functions-datetime-diff>>
 ** <<sql-functions-datetime-part>>
 ** <<sql-functions-datetime-trunc>>
 ** <<sql-functions-datetime-day>>

x-pack/plugin/sql/qa/src/main/resources/command.csv-spec

@@ -42,9 +42,11 @@ CURRENT_TIME     |SCALAR
 CURRENT_TIMESTAMP|SCALAR
 CURTIME          |SCALAR
 DATEADD          |SCALAR
+DATEDIFF         |SCALAR
 DATEPART         |SCALAR
 DATETRUNC        |SCALAR
 DATE_ADD         |SCALAR
+DATE_DIFF        |SCALAR
 DATE_PART        |SCALAR
 DATE_TRUNC       |SCALAR
 DAY              |SCALAR
@@ -82,7 +84,9 @@ QUARTER          |SCALAR
 SECOND           |SCALAR         
 SECOND_OF_MINUTE |SCALAR
 TIMESTAMPADD     |SCALAR
+TIMESTAMPDIFF    |SCALAR
 TIMESTAMP_ADD    |SCALAR
+TIMESTAMP_DIFF   |SCALAR
 TODAY            |SCALAR
 WEEK             |SCALAR
 WEEK_OF_YEAR     |SCALAR         

x-pack/plugin/sql/qa/src/main/resources/datetime.csv-spec

@@ -296,6 +296,177 @@ null    | 2004-04-30 00:00:00.000Z
 F       | 2002-05-19 00:00:00.000Z
 ;
 
+selectDiffWithDateTime
+schema::diff_year:i|diff_quarter:i|diff_month:i|diff_week:i|diff_day:i|diff_hours:i|diff_min:i|diff_sec:i|diff_millis:i|diff_mcsec:i|diff_nsec:i
+SELECT DATE_DIFF('year', '2019-09-04T11:22:33.123Z'::datetime, '2076-01-12T22:11:55.123Z'::datetime) as diff_year, DATE_DIFF('quarter', '2048-02-14T22:11:55.123Z'::datetime, '2019-09-04T11:22:33.123Z'::datetime) as diff_quarter,
+DATE_DIFF('month', '2019-09-04T11:22:33.123Z'::datetime, '2053-07-02T01:31:13.987Z'::datetime) as diff_month, DATE_DIFF('week', '2019-09-04T11:22:33.123Z'::datetime, '2001-07-14T21:41:43.321Z'::datetime) as diff_week,
+DATE_DIFF('day', '2019-09-04T11:22:33.123Z'::datetime, '2027-05-30T12:44:22.425Z'::datetime) as diff_day,
+DATE_DIFF('hours', '2019-09-04T14:10:12.432Z'::datetime, '2005-08-14T01:12:32.432Z'::datetime) as diff_hours, DATE_DIFF('minutes', '2019-03-05T11:33:22.456Z'::datetime, '2026-04-30T05:29:52.652Z'::datetime) as diff_min,
+DATE_DIFF('second', '2020-01-01T11:22:33.532Z'::datetime, '2019-09-04T15:47:56.987Z'::datetime) as diff_sec, DATE_DIFF('milliseconds', '2019-09-10T11:22:33.123Z'::datetime, '2019-09-12T20:33:22.987Z'::datetime) as diff_millis,
+DATE_DIFF('mcs', '2019-09-04T11:25:21.123456Z'::datetime, '2019-09-04T11:22:33.987654Z'::datetime) as diff_mcsec, DATE_DIFF('nanosecond', '2019-09-04T11:22:43.987654321Z'::datetime, '2019-09-04T11:22:44.123456789Z'::datetime) as diff_nsec;
+
+ diff_year | diff_quarter | diff_month | diff_week | diff_day | diff_hours | diff_min | diff_sec  | diff_millis | diff_mcsec | diff_nsec
+-----------+--------------+------------+-----------+----------+------------+----------+-----------+-------------+------------+----------
+57         | -114         | 406        | -947      | 2825     | -123228    | 3762357  | -10265677 | 205849864   | -167135802 | 135802468
+;
+
+selectDiffWithDate
+schema::diff_year:i|diff_quarter:i|diff_month:i|diff_week:i|diff_day:i|diff_hours:i|diff_min:i|diff_sec:i|diff_millis:i|diff_mcsec:i|diff_nsec:i
+SELECT DATEDIFF('year', '2010-12-10'::date, '2019-09-04'::date) as diff_year, DATEDIFF('quarter', '2042-06-12'::date,'2019-09-04'::date) as diff_quarter,
+DATEDIFF('month', '2019-09-04'::date, '2042-02-24'::date) as diff_month, DATEDIFF('week', '2031-05-17'::date, '2019-09-04'::date) as diff_week, DATEDIFF('day', '2019-09-04'::date, '2051-08-30'::date) as diff_day,
+DATEDIFF('hours', '2027-01-02'::date, '2019-09-04'::date) as diff_hours, DATEDIFF('minutes', '2019-09-04'::date, '2022-11-11'::date) as diff_min, DATEDIFF('second', '2020-02-14'::date, '2019-09-04'::date) as diff_sec,
+DATEDIFF('milliseconds', '2019-09-04'::date, '2019-09-06'::date) as diff_millis, DATEDIFF('mcs', '2019-09-04'::date, '2019-09-04'::date) as diff_mcsec, DATEDIFF('nanosecond', '2019-09-04'::date, '2019-09-04'::date) as diff_nsec;
+
+ diff_year | diff_quarter | diff_month | diff_week | diff_day | diff_hours | diff_min | diff_sec  | diff_millis | diff_mcsec | diff_nsec
+-----------+--------------+------------+-----------+----------+------------+----------+-----------+-------------+------------+----------
+9          | -91          | 269        | -611      | 11683    | -64248     | 1676160  | -14083200 | 172800000   | 0          | 0
+;
+
+selectDateDiffWithField
+schema::emp_no:i|birth_date:ts|hire_date:ts|diff_year:i|diff_quarter:i|diff_month:i|diff_week:i|diff_day:i|diff_min:i|diff_sec:i
+SELECT emp_no, birth_date, hire_date, TIMESTAMP_DIFF('year', birth_date, hire_date) as diff_year, TIMESTAMP_DIFF('quarter', hire_date, birth_date) as diff_quarter,
+TIMESTAMP_DIFF('month', birth_date, hire_date) as diff_month, TIMESTAMP_DIFF('week', hire_date, birth_date) as diff_week, TIMESTAMP_DIFF('day', birth_date, hire_date) as diff_day,
+TIMESTAMP_DIFF('minutes', hire_date, birth_date) as diff_min, TIMESTAMP_DIFF('second', birth_date, hire_date) as diff_sec
+FROM test_emp WHERE emp_no >= 10032 AND emp_no <= 10042 ORDER BY 1;
+
+ emp_no  |      birth_date          |       hire_date          |  diff_year | diff_quarter | diff_month | diff_week | diff_day | diff_min  | diff_sec
+---------+--------------------------+--------------------------+------------+--------------+------------+-----------+----------+-----------+----------
+10032    | 1960-08-09 00:00:00.000Z | 1990-06-20 00:00:00.000Z | 30         | -119         | 358        | -1559     | 10907    | -15706080 | 942364800
+10033    | 1956-11-14 00:00:00.000Z | 1987-03-18 00:00:00.000Z | 31         | -121         | 364        | -1584     | 11081    | -15956640 | 957398400
+10034    | 1962-12-29 00:00:00.000Z | 1988-09-21 00:00:00.000Z | 26         | -103         | 309        | -1343     | 9398     | -13533120 | 811987200
+10035    | 1953-02-08 00:00:00.000Z | 1988-09-05 00:00:00.000Z | 35         | -142         | 427        | -1857     | 12993    | -18709920 | 1122595200
+10036    | 1959-08-10 00:00:00.000Z | 1992-01-03 00:00:00.000Z | 33         | -130         | 389        | -1691     | 11834    | -17040960 | 1022457600
+10037    | 1963-07-22 00:00:00.000Z | 1990-12-05 00:00:00.000Z | 27         | -109         | 329        | -1429     | 9998     | -14397120 | 863827200
+10038    | 1960-07-20 00:00:00.000Z | 1989-09-20 00:00:00.000Z | 29         | -116         | 350        | -1523     | 10654    | -15341760 | 920505600
+10039    | 1959-10-01 00:00:00.000Z | 1988-01-19 00:00:00.000Z | 29         | -113         | 339        | -1477     | 10337    | -14885280 | 893116800
+10040    | null                     | 1993-02-14 00:00:00.000Z | null       | null         | null       | null      | null     | null      | null
+10041    | null                     | 1989-11-12 00:00:00.000Z | null       | null         | null       | null      | null     | null      | null
+10042    | null                     | 1993-03-21 00:00:00.000Z | null       | null         | null       | null      | null     | null      | null
+;
+
+selectDiffWithComplexExpressions1
+schema::gender:s|birth_date:ts|hire_date:ts|diff:i
+SELECT gender, birth_date, hire_date, TIMESTAMPDIFF('months', birth_date + CASE WHEN gender = 'M' THEN INTERVAL 10 years
+WHEN gender = 'F' THEN INTERVAL -10 years ELSE INTERVAL 12 years END, hire_date + INTERVAL 10 month) AS diff
+FROM test_emp WHERE (diff BETWEEN 150 AND 250) OR diff > 500 ORDER BY emp_no LIMIT 10;
+
+  gender  |      birth_date          |       hire_date          | diff
+----------+--------------------------+--------------------------+-----
+M         | 1959-12-03 00:00:00.000Z | 1986-08-28 00:00:00.000Z | 210
+F         | 1953-04-20 00:00:00.000Z | 1989-06-02 00:00:00.000Z | 564
+F         | 1957-05-23 00:00:00.000Z | 1989-02-10 00:00:00.000Z | 511
+F         | 1952-04-19 00:00:00.000Z | 1985-02-18 00:00:00.000Z | 524
+null      | 1963-06-01 00:00:00.000Z | 1989-08-24 00:00:00.000Z | 180
+null      | 1956-02-12 00:00:00.000Z | 1987-03-11 00:00:00.000Z | 239
+null      | 1959-08-19 00:00:00.000Z | 1987-07-02 00:00:00.000Z | 201
+M         | 1960-02-20 00:00:00.000Z | 1988-02-10 00:00:00.000Z | 226
+F         | 1953-09-29 00:00:00.000Z | 1989-12-17 00:00:00.000Z | 565
+F         | 1958-09-05 00:00:00.000Z | 1997-05-19 00:00:00.000Z | 594
+;
+
+selectDiffWithComplexExpressions2
+schema::languages:byte|first_name:s|gender:s|birth_date:ts|hire_date:ts|diff:l
+SELECT languages, first_name, gender, birth_date, hire_date,
+CAST(DATE_DIFF(CASE WHEN gender = 'M' THEN CONCAT(gender, 'onths') WHEN gender = 'F' THEN NULL ELSE 'quarter' END,
+    hire_date + INTERVAL 20 month, birth_date - interval 50 month) AS long) AS diff
+FROM test_emp WHERE languages >= 3 AND first_name LIKE '%y%' ORDER BY diff ASC, languages DESC;
+
+   languages  |   first_name  |  gender  |      birth_date          |       hire_date          |  diff
+--------------+---------------+----------+--------------------------+--------------------------+------
+4             | Mayumi        | M        | 1957-04-04 00:00:00.000Z | 1995-03-13 00:00:00.000Z | -525
+5             | Georgy        | M        | 1956-06-06 00:00:00.000Z | 1992-04-27 00:00:00.000Z | -500
+4             | Jayson        | M        | 1954-09-16 00:00:00.000Z | 1990-01-14 00:00:00.000Z | -494
+5             | Mary          | null     | 1953-11-07 00:00:00.000Z | 1990-01-22 00:00:00.000Z | -168
+5             | Hironoby      | F        | 1952-05-15 00:00:00.000Z | 1988-07-21 00:00:00.000Z | null
+4             | Weiyi         | F        | null                     | 1993-02-14 00:00:00.000Z | null
+3             | Magy          | F        | null                     | 1993-03-21 00:00:00.000Z | null
+;
+
+selectDiffWithComplexExpressions3
+schema::first_name:s|gender:s|birth_date:ts|hire_date:ts|date_diff1:i|date_diff2:i
+SELECT first_name, gender, birth_date, hire_date, DATE_DIFF(CASE WHEN gender = 'M' THEN CONCAT(gender, 'onths') WHEN gender = 'F' THEN 'year' ELSE 'quarter' END,
+birth_date, hire_date + INTERVAL 10 month) AS date_diff1,
+DATE_DIFF(CASE WHEN gender = 'M' THEN CONCAT(gender, 'onths') WHEN gender = 'F' THEN 'year' ELSE 'quarter' END,
+hire_date + INTERVAL 10 month, birth_date - INTERVAL 20 month) AS date_diff2
+FROM test_emp WHERE (date_diff2 IS NULL OR date_diff2 < -455) AND first_name LIKE '%y%' ORDER BY date_diff1 DESC;
+
+  first_name   |     gender |      birth_date          |       hire_date          |   date_diff1  |   date_diff2
+---------------+------------+--------------------------+--------------------------+---------------+-------------
+Mayuko         | M          | 1952-12-24 00:00:00.000Z | 1991-01-26 00:00:00.000Z | 467           | -487
+Mayumi         | M          | 1957-04-04 00:00:00.000Z | 1995-03-13 00:00:00.000Z | 465           | -485
+Saniya         | M          | 1958-02-19 00:00:00.000Z | 1994-09-15 00:00:00.000Z | 449           | -469
+Georgy         | M          | 1956-06-06 00:00:00.000Z | 1992-04-27 00:00:00.000Z | 440           | -460
+Weiyi          | F          | null                     | 1993-02-14 00:00:00.000Z | null          | null
+Magy           | F          | null                     | 1993-03-21 00:00:00.000Z | null          | null
+Yishay         | M          | null                     | 1990-10-20 00:00:00.000Z | null          | null
+;
+
+selectDiffWithComplexExpressions4
+schema::first_name:s|gender:s|birth_date:ts|hire_date:ts|date_diff1:i|date_diff2:i|arithmetics:i
+SELECT first_name, gender, birth_date, hire_date, DATE_DIFF('months', birth_date - INTERVAL 40 months, hire_date + INTERVAL 40 months) AS date_diff1,
+DATE_DIFF('quarter', hire_date + INTERVAL 20 month, birth_date - INTERVAL 20 month) AS date_diff2,
+(DATE_DIFF('months', birth_date - INTERVAL 40 months, hire_date + INTERVAL 40 months) * DATE_DIFF('quarter', hire_date + INTERVAL 20 month, birth_date - INTERVAL 20 month))
+- DATE_DIFF('months', birth_date - INTERVAL 40 months, hire_date + INTERVAL 40 months) + DATE_DIFF('quarter', hire_date + INTERVAL 20 month, birth_date - INTERVAL 20 month) AS arithmetics
+FROM test_emp WHERE ((date_diff1 * date_diff2) - date_diff1 + date_diff2) < -81000 AND first_name LIKE '%y%' ORDER BY date_diff1 DESC;
+
+  first_name | gender | birth_date               | hire_date                | date_diff1 | date_diff2 | arithmetics
+-------------+--------+--------------------------+--------------------------+------------+------------+------------
+Mayuko       | M      | 1952-12-24 00:00:00.000Z | 1991-01-26 00:00:00.000Z | 537        | -165       | -89307
+Mayumi       | M      | 1957-04-04 00:00:00.000Z | 1995-03-13 00:00:00.000Z | 535        | -165       | -88975
+Saniya       | M      | 1958-02-19 00:00:00.000Z | 1994-09-15 00:00:00.000Z | 519        | -160       | -83719
+Mary         | null   | 1953-11-07 00:00:00.000Z | 1990-01-22 00:00:00.000Z | 514        | -158       | -81884
+Hironoby     | F      | 1952-05-15 00:00:00.000Z | 1988-07-21 00:00:00.000Z | 514        | -158       | -81884
+;
+
+dateDiffOrderBy
+schema::emp_no:i | birth_date:ts | hire_date:ts | diff:i
+SELECT emp_no, birth_date, hire_date, DATE_DIFF('hours', hire_date, birth_date) as diff FROM test_emp ORDER BY diff NULLS LAST, emp_no LIMIT 5;
+
+ emp_no   |      birth_date          |       hire_date          |  diff
+----------+--------------------------+--------------------------+--------
+10019     | 1953-01-23 00:00:00.000Z | 1999-04-30 00:00:00.000Z | -405552
+10022     | 1952-07-08 00:00:00.000Z | 1995-08-22 00:00:00.000Z | -378000
+10026     | 1953-04-03 00:00:00.000Z | 1995-03-20 00:00:00.000Z | -367824
+10051     | 1953-07-28 00:00:00.000Z | 1992-10-15 00:00:00.000Z | -343776
+10024     | 1958-09-05 00:00:00.000Z | 1997-05-19 00:00:00.000Z | -339264
+;
+
+dateDiffFilter
+schema::emp_no:i|birth_date:ts|hire_date:ts|diff:i
+SELECT emp_no, birth_date, hire_date, DATE_DIFF('quarter', birth_date, hire_date) as diff FROM test_emp WHERE diff > 100 ORDER BY emp_no LIMIT 5;
+
+ emp_no   |      birth_date          |       hire_date          |  diff
+---------+---------------------------+--------------------------+------
+10001     | 1953-09-02 00:00:00.000Z | 1986-06-26 00:00:00.000Z | 131
+10003     | 1959-12-03 00:00:00.000Z | 1986-08-28 00:00:00.000Z | 107
+10004     | 1954-05-01 00:00:00.000Z | 1986-12-01 00:00:00.000Z | 130
+10005     | 1955-01-21 00:00:00.000Z | 1989-09-12 00:00:00.000Z | 138
+10006     | 1953-04-20 00:00:00.000Z | 1989-06-02 00:00:00.000Z | 144
+;
+
+dateDiffGroupBy
+schema::count:l|diff:i
+SELECT count(*) as count, DATE_DIFF('weeks', birth_date, hire_date) diff FROM test_emp GROUP BY diff ORDER BY 2 LIMIT 5;
+
+  count  |  diff
+---------+------
+10       | null
+1        | 1121
+1        | 1124
+1        | 1168
+1        | 1196
+;
+
+dateDiffHaving
+schema::gender:s|diff:i
+SELECT gender, DATE_DIFF('months', max(birth_date), max(hire_date)) AS diff FROM test_emp GROUP BY gender HAVING DATE_DIFF('months', max(birth_date), max(hire_date)) > 390 ORDER BY 1;
+
+  gender   |  diff
+-----------+------
+null       | 430
+F          | 391
+;
+
 selectDateTruncWithDateTime
 schema::dt_hour:ts|dt_min:ts|dt_sec:ts|dt_millis:s|dt_micro:s|dt_nano:s
 SELECT DATE_TRUNC('hour', '2019-09-04T11:22:33.123Z'::datetime) as dt_hour, DATE_TRUNC('minute', '2019-09-04T11:22:33.123Z'::datetime) as dt_min,

x-pack/plugin/sql/qa/src/main/resources/docs/docs.csv-spec

@@ -238,9 +238,11 @@ CURRENT_TIME     |SCALAR
 CURRENT_TIMESTAMP|SCALAR
 CURTIME          |SCALAR
 DATEADD          |SCALAR
+DATEDIFF         |SCALAR
 DATEPART         |SCALAR
 DATETRUNC        |SCALAR
 DATE_ADD         |SCALAR
+DATE_DIFF        |SCALAR
 DATE_PART        |SCALAR
 DATE_TRUNC       |SCALAR
 DAY              |SCALAR
@@ -278,7 +280,9 @@ QUARTER          |SCALAR
 SECOND           |SCALAR         
 SECOND_OF_MINUTE |SCALAR
 TIMESTAMPADD     |SCALAR
+TIMESTAMPDIFF    |SCALAR
 TIMESTAMP_ADD    |SCALAR
+TIMESTAMP_DIFF   |SCALAR
 TODAY            |SCALAR
 WEEK             |SCALAR
 WEEK_OF_YEAR     |SCALAR         
@@ -2472,6 +2476,57 @@ SELECT DATE_ADD('minutes', 9235, '2019-09-04'::date) AS "+9235 minutes";
 // end::dateAddDateMinutes
 ;
 
+dateDiffDateTimeYears
+// tag::dateDiffDateTimeYears
+SELECT DATE_DIFF('years', '2019-09-04T11:22:33.000Z'::datetime, '2032-09-04T22:33:11.000Z'::datetime) AS "diffInYears";
+
+      diffInYears
+------------------------
+13
+// end::dateDiffDateTimeYears
+;
+
+dateDiffDateTimeWeeks
+// tag::dateDiffDateTimeWeeks
+SELECT DATE_DIFF('week', '2019-09-04T11:22:33.000Z'::datetime, '2016-12-08T22:33:11.000Z'::datetime) AS "diffInWeeks";
+
+      diffInWeeks
+------------------------
+-143
+// end::dateDiffDateTimeWeeks
+;
+
+dateDiffDateTimeSeconds
+// tag::dateDiffDateTimeSeconds
+SELECT DATE_DIFF('seconds', '2019-09-04T11:22:33.123Z'::datetime, '2019-07-12T22:33:11.321Z'::datetime) AS "diffInSeconds";
+
+      diffInSeconds
+------------------------
+-4625362
+// end::dateDiffDateTimeSeconds
+;
+
+dateDiffDateQuarters
+// tag::dateDiffDateQuarters
+SELECT DATE_DIFF('qq', '2019-09-04'::date, '2025-04-25'::date) AS "diffInQuarters";
+
+      diffInQuarters
+------------------------
+23
+// end::dateDiffDateQuarters
+;
+
+dateDiffDateMinutes
+// tag::dateDiffDateMinutes
+SELECT DATE_DIFF('minutes', '2019-09-04'::date, '2015-08-17T22:33:11.567Z'::datetime) AS "diffInMinutes";
+
+      diffInMinutes
+------------------------
+-2128407
+// end::dateDiffDateMinutes
+;
+
+
 datePartDateTimeYears
 // tag::datePartDateTimeYears
 SELECT DATE_PART('year', '2019-09-22T11:22:33.123Z'::datetime) AS "years";

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/FunctionRegistry.java

@@ -32,6 +32,7 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.CurrentDa
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.CurrentDateTime;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.CurrentTime;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateAdd;
+import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateDiff;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DatePart;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTrunc;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DayName;
@@ -197,6 +198,7 @@ public class FunctionRegistry {
                 def(DayOfWeek.class, DayOfWeek::new, "DAY_OF_WEEK", "DAYOFWEEK", "DOW"),
                 def(DayOfYear.class, DayOfYear::new, "DAY_OF_YEAR", "DAYOFYEAR", "DOY"),
                 def(DateAdd.class, DateAdd::new, "DATEADD", "DATE_ADD", "TIMESTAMPADD", "TIMESTAMP_ADD"),
+                def(DateDiff.class, DateDiff::new, "DATEDIFF", "DATE_DIFF", "TIMESTAMPDIFF", "TIMESTAMP_DIFF"),
                 def(DatePart.class, DatePart::new, "DATEPART", "DATE_PART"),
                 def(DateTrunc.class, DateTrunc::new, "DATETRUNC", "DATE_TRUNC"),
                 def(HourOfDay.class, HourOfDay::new, "HOUR_OF_DAY", "HOUR"),

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/Processors.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.sql.expression.function.scalar;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry.Entry;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateAddProcessor;
+import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateDiffProcessor;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DatePartProcessor;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeProcessor;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTruncProcessor;
@@ -92,6 +93,7 @@ public final class Processors {
         entries.add(new Entry(Processor.class, NonIsoDateTimeProcessor.NAME, NonIsoDateTimeProcessor::new));
         entries.add(new Entry(Processor.class, QuarterProcessor.NAME, QuarterProcessor::new));
         entries.add(new Entry(Processor.class, DateAddProcessor.NAME, DateAddProcessor::new));
+        entries.add(new Entry(Processor.class, DateDiffProcessor.NAME, DateDiffProcessor::new));
         entries.add(new Entry(Processor.class, DatePartProcessor.NAME, DatePartProcessor::new));
         entries.add(new Entry(Processor.class, DateTruncProcessor.NAME, DateTruncProcessor::new));
         // math

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateAdd.java

@@ -67,8 +67,8 @@ public class DateAdd extends ThreeArgsDateTimeFunction {
             return DateTimeField.findSimilar(NAME_TO_PART.keySet(), match);
         }
 
-        public static Part resolve(String truncateTo) {
-            return DateTimeField.resolveMatch(NAME_TO_PART, truncateTo);
+        public static Part resolve(String dateTimeUnit) {
+            return DateTimeField.resolveMatch(NAME_TO_PART, dateTimeUnit);
         }
 
         public ZonedDateTime add(ZonedDateTime dateTime, Integer numberOfUnits) {
@@ -134,8 +134,8 @@ public class DateAdd extends ThreeArgsDateTimeFunction {
     }
 
     @Override
-    protected Pipe createPipe(Pipe first, Pipe second, Pipe third, ZoneId zoneId) {
-        return new DateAddPipe(source(), this, first, second, third, zoneId);
+    protected Pipe createPipe(Pipe unit, Pipe numberOfUnits, Pipe timestamp, ZoneId zoneId) {
+        return new DateAddPipe(source(), this, unit, numberOfUnits, timestamp, zoneId);
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateDiff.java

@@ -0,0 +1,225 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
+
+import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
+import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.Expressions;
+import org.elasticsearch.xpack.sql.expression.Nullability;
+import org.elasticsearch.xpack.sql.expression.gen.pipeline.Pipe;
+import org.elasticsearch.xpack.sql.tree.NodeInfo;
+import org.elasticsearch.xpack.sql.tree.Source;
+import org.elasticsearch.xpack.sql.type.DataType;
+
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoField;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+
+import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
+import static org.elasticsearch.xpack.sql.expression.TypeResolutions.isDate;
+import static org.elasticsearch.xpack.sql.expression.TypeResolutions.isString;
+import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.NonIsoDateTimeProcessor.NonIsoDateTimeExtractor;
+
+public class DateDiff extends ThreeArgsDateTimeFunction {
+
+    public enum Part implements DateTimeField {
+
+        YEAR((start, end) ->  end.getYear() - start.getYear(), "years", "yyyy", "yy"),
+        QUARTER((start, end) -> QuarterProcessor.quarter(end) - QuarterProcessor.quarter(start) + (YEAR.diff(start, end) * 4),
+            "quarters", "qq", "q"),
+        MONTH((start, end) -> safeInt(end.getLong(ChronoField.PROLEPTIC_MONTH) - start.getLong(ChronoField.PROLEPTIC_MONTH)),
+            "months", "mm", "m"),
+        DAYOFYEAR((start, end) -> safeInt(diffInDays(start, end)), "dy", "y"),
+        DAY(DAYOFYEAR::diff, "days", "dd", "d"),
+        WEEK((start, end) -> {
+            int extraWeek = NonIsoDateTimeExtractor.WEEK_OF_YEAR.extract(end) -
+                NonIsoDateTimeExtractor.WEEK_OF_YEAR.extract(start) == 0 ? 0 : 1;
+            long diffWeeks = diffInDays(start, end) / 7;
+            if (diffWeeks < 0) {
+                diffWeeks -= extraWeek;
+            } else {
+                diffWeeks += extraWeek;
+            }
+            return safeInt(diffWeeks);
+        }, "weeks", "wk", "ww"),
+        WEEKDAY(DAYOFYEAR::diff,  "weekdays", "dw"),
+        HOUR((start, end) -> safeInt(diffInHours(start, end)),  "hours", "hh"),
+        MINUTE((start, end) -> safeInt(diffInMinutes(start, end)), "minutes", "mi", "n"),
+        SECOND((start, end) -> safeInt(end.toEpochSecond() - start.toEpochSecond()), "seconds", "ss", "s"),
+        MILLISECOND((start, end) -> safeInt(end.toInstant().toEpochMilli() - start.toInstant().toEpochMilli()),
+            "milliseconds", "ms"),
+        MICROSECOND((start, end) -> {
+            long secondsDiff = diffInSeconds(start, end);
+            long microsDiff = end.toInstant().getLong(ChronoField.MICRO_OF_SECOND) -
+                start.toInstant().getLong(ChronoField.MICRO_OF_SECOND);
+            return safeInt(secondsDiff * 1_000_000L + microsDiff);
+        }, "microseconds", "mcs"),
+        NANOSECOND((start, end) -> {
+            long secondsDiff = diffInSeconds(start, end);
+            int nanosDiff = end.getNano() - start.getNano();
+            return safeInt(secondsDiff * 1_000_000_000L + nanosDiff);
+        }, "nanoseconds", "ns");
+
+        private static final Map<String, Part> NAME_TO_PART;
+        private static final List<String> VALID_VALUES;
+
+        static {
+            NAME_TO_PART = DateTimeField.initializeResolutionMap(values());
+            VALID_VALUES = DateTimeField.initializeValidValues(values());
+        }
+
+        private BiFunction<ZonedDateTime, ZonedDateTime, Integer> diffFunction;
+        private Set<String> aliases;
+
+        Part(BiFunction<ZonedDateTime, ZonedDateTime, Integer> diffFunction, String... aliases) {
+            this.diffFunction = diffFunction;
+            this.aliases = Set.of(aliases);
+        }
+
+        @Override
+        public Iterable<String> aliases() {
+            return aliases;
+        }
+
+        public static List<String> findSimilar(String match) {
+            return DateTimeField.findSimilar(NAME_TO_PART.keySet(), match);
+        }
+
+        public static Part resolve(String dateTimeUnit) {
+            return DateTimeField.resolveMatch(NAME_TO_PART, dateTimeUnit);
+        }
+
+        public Integer diff(ZonedDateTime startTimestamp, ZonedDateTime endTimestamp) {
+            return diffFunction.apply(startTimestamp, endTimestamp);
+        }
+
+        private static long diffInSeconds(ZonedDateTime start, ZonedDateTime end) {
+            return end.toEpochSecond() - start.toEpochSecond();
+        }
+
+        private static int safeInt(long diff) {
+            if (diff > Integer.MAX_VALUE || diff < Integer.MIN_VALUE) {
+                throw new SqlIllegalArgumentException("The DATE_DIFF function resulted in an overflow; the number of units " +
+                    "separating two date/datetime instances is too large. Try to use DATE_DIFF with a less precise unit.");
+            } else {
+                return Long.valueOf(diff).intValue();
+            }
+        }
+
+        private static long diffInMinutes(ZonedDateTime start, ZonedDateTime end) {
+            long secondsDiff = diffInSeconds(start, end);
+            if (secondsDiff > 0) {
+                return (long) Math.ceil(secondsDiff / 60.0d);
+            } else {
+                return (long) Math.floor(secondsDiff / 60.0d);
+            }
+        }
+
+        private static long diffInHours(ZonedDateTime start, ZonedDateTime end) {
+            return diffInMinutes(start, end) / 60;
+        }
+
+        private static long diffInDays(ZonedDateTime start, ZonedDateTime end) {
+            return diffInHours(start, end) / 24;
+        }
+    }
+
+    public DateDiff(Source source, Expression unit, Expression startTimestamp, Expression endTimestamp, ZoneId zoneId) {
+        super(source, unit, startTimestamp, endTimestamp, zoneId);
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        TypeResolution resolution = isString(first(), sourceText(), Expressions.ParamOrdinal.FIRST);
+        if (resolution.unresolved()) {
+            return resolution;
+        }
+
+        if (first().foldable()) {
+            String datePartValue = (String) first().fold();
+            if (datePartValue != null && resolveDateTimeField(datePartValue) == false) {
+                List<String> similar = findSimilarDateTimeFields(datePartValue);
+                if (similar.isEmpty()) {
+                    return new TypeResolution(format(null, "first argument of [{}] must be one of {} or their aliases; found value [{}]",
+                        sourceText(),
+                        validDateTimeFieldValues(),
+                        Expressions.name(first())));
+                } else {
+                    return new TypeResolution(format(null, "Unknown value [{}] for first argument of [{}]; did you mean {}?",
+                        Expressions.name(first()),
+                        sourceText(),
+                        similar));
+                }
+            }
+        }
+
+        resolution = isDate(second(), sourceText(), Expressions.ParamOrdinal.SECOND);
+        if (resolution.unresolved()) {
+            return resolution;
+        }
+
+        resolution = isDate(third(), sourceText(), Expressions.ParamOrdinal.THIRD);
+        if (resolution.unresolved()) {
+            return resolution;
+        }
+
+        return TypeResolution.TYPE_RESOLVED;
+    }
+
+    @Override
+    public DataType dataType() {
+        return DataType.INTEGER;
+    }
+
+    @Override
+    protected ThreeArgsDateTimeFunction replaceChildren(Expression newFirst, Expression newSecond, Expression newThird) {
+        return new DateDiff(source(), newFirst, newSecond, newThird, zoneId());
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, DateDiff::new, first(), second(), third(), zoneId());
+    }
+
+    @Override
+    public Nullability nullable() {
+        return Nullability.UNKNOWN;
+    }
+
+    @Override
+    protected Pipe createPipe(Pipe unit, Pipe startTimestamp, Pipe endTimestamp, ZoneId zoneId) {
+        return new DateDiffPipe(source(), this, unit, startTimestamp, endTimestamp, zoneId);
+    }
+
+    @Override
+    protected String scriptMethodName() {
+        return "dateDiff";
+    }
+
+    @Override
+    public Object fold() {
+        return DateDiffProcessor.process(first().fold(), second().fold(), third().fold(), zoneId());
+    }
+
+    @Override
+    protected boolean resolveDateTimeField(String dateTimeField) {
+        return Part.resolve(dateTimeField) != null;
+    }
+
+    @Override
+    protected List<String> findSimilarDateTimeFields(String dateTimeField) {
+        return Part.findSimilar(dateTimeField);
+    }
+
+    @Override
+    protected List<String> validDateTimeFieldValues() {
+        return Part.VALID_VALUES;
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateDiffPipe.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
+
+import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.gen.pipeline.Pipe;
+import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
+import org.elasticsearch.xpack.sql.tree.NodeInfo;
+import org.elasticsearch.xpack.sql.tree.Source;
+
+import java.time.ZoneId;
+
+public class DateDiffPipe extends ThreeArgsDateTimePipe {
+
+    public DateDiffPipe(Source source, Expression expression, Pipe first, Pipe second, Pipe third, ZoneId zoneId) {
+        super(source, expression, first, second, third, zoneId);
+    }
+
+    @Override
+    protected NodeInfo<DateDiffPipe> info() {
+        return NodeInfo.create(this, DateDiffPipe::new, expression(), first(), second(), third(), zoneId());
+    }
+
+    @Override
+    public ThreeArgsDateTimePipe replaceChildren(Pipe newFirst, Pipe newSecond, Pipe newThird) {
+        return new DateDiffPipe(source(), expression(), newFirst, newSecond, newThird, zoneId());
+    }
+
+    @Override
+    protected Processor makeProcessor(Processor first, Processor second, Processor third, ZoneId zoneId) {
+        return new DateDiffProcessor(first, second, third, zoneId);
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateDiffProcessor.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
+import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateDiff.Part;
+import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
+
+import java.io.IOException;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+
+public class DateDiffProcessor extends ThreeArgsDateTimeProcessor {
+
+    public static final String NAME = "dtdiff";
+
+    public DateDiffProcessor(Processor unit, Processor startTimestamp, Processor endTimestamp, ZoneId zoneId) {
+        super(unit, startTimestamp, endTimestamp, zoneId);
+    }
+
+    public DateDiffProcessor(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return NAME;
+    }
+
+    @Override
+    public Object doProcess(Object unit, Object startTimestamp, Object endTimestamp, ZoneId zoneId) {
+        return process(unit, startTimestamp, endTimestamp, zoneId);
+    }
+
+    /**
+     * Used in Painless scripting
+     */
+    public static Object process(Object unit, Object startTimestamp, Object endTimestamp, ZoneId zoneId) {
+        if (unit == null || startTimestamp == null || endTimestamp == null) {
+            return null;
+        }
+        if (unit instanceof String == false) {
+            throw new SqlIllegalArgumentException("A string is required; received [{}]", unit);
+        }
+        Part datePartField = Part.resolve((String) unit);
+        if (datePartField == null) {
+            List<String> similar = Part.findSimilar((String) unit);
+            if (similar.isEmpty()) {
+                throw new SqlIllegalArgumentException("A value of {} or their aliases is required; received [{}]",
+                    Part.values(), unit);
+            } else {
+                throw new SqlIllegalArgumentException("Received value [{}] is not valid date part to add; " +
+                    "did you mean {}?", unit, similar);
+            }
+        }
+
+        if (startTimestamp instanceof ZonedDateTime == false) {
+            throw new SqlIllegalArgumentException("A date/datetime is required; received [{}]", startTimestamp);
+        }
+
+        if (endTimestamp instanceof ZonedDateTime == false) {
+            throw new SqlIllegalArgumentException("A date/datetime is required; received [{}]", endTimestamp);
+        }
+
+        return datePartField.diff(((ZonedDateTime) startTimestamp).withZoneSameInstant(zoneId),
+            ((ZonedDateTime) endTimestamp).withZoneSameInstant(zoneId));
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DatePart.java

@@ -19,7 +19,7 @@ import java.time.temporal.ChronoField;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.function.Function;
+import java.util.function.ToIntFunction;
 
 import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.NonIsoDateTimeProcessor.NonIsoDateTimeExtractor;
 
@@ -49,10 +49,10 @@ public class DatePart extends BinaryDateTimeFunction {
             VALID_VALUES = DateTimeField.initializeValidValues(values());
         }
 
-        private Function<ZonedDateTime, Integer> extractFunction;
+        private ToIntFunction<ZonedDateTime> extractFunction;
         private Set<String> aliases;
 
-        Part(Function<ZonedDateTime, Integer> extractFunction, String... aliases) {
+        Part(ToIntFunction<ZonedDateTime> extractFunction, String... aliases) {
             this.extractFunction = extractFunction;
             this.aliases = Set.of(aliases);
         }
@@ -66,17 +66,17 @@ public class DatePart extends BinaryDateTimeFunction {
             return DateTimeField.findSimilar(NAME_TO_PART.keySet(), match);
         }
 
-        public static Part resolve(String truncateTo) {
-            return DateTimeField.resolveMatch(NAME_TO_PART, truncateTo);
+        public static Part resolve(String dateTimePart) {
+            return DateTimeField.resolveMatch(NAME_TO_PART, dateTimePart);
         }
 
         public Integer extract(ZonedDateTime dateTime) {
-            return extractFunction.apply(dateTime);
+            return extractFunction.applyAsInt(dateTime);
         }
     }
 
-    public DatePart(Source source, Expression truncateTo, Expression timestamp, ZoneId zoneId) {
-        super(source, truncateTo, timestamp, zoneId);
+    public DatePart(Source source, Expression dateTimePart, Expression timestamp, ZoneId zoneId) {
+        super(source, dateTimePart, timestamp, zoneId);
     }
 
     @Override
@@ -85,8 +85,8 @@ public class DatePart extends BinaryDateTimeFunction {
     }
 
     @Override
-    protected BinaryScalarFunction replaceChildren(Expression newTruncateTo, Expression newTimestamp) {
-        return new DatePart(source(), newTruncateTo, newTimestamp, zoneId());
+    protected BinaryScalarFunction replaceChildren(Expression newDateTimePart, Expression newTimestamp) {
+        return new DatePart(source(), newDateTimePart, newTimestamp, zoneId());
     }
 
     @Override
@@ -105,8 +105,8 @@ public class DatePart extends BinaryDateTimeFunction {
     }
 
     @Override
-    protected Pipe createPipe(Pipe left, Pipe right, ZoneId zoneId) {
-        return new DatePartPipe(source(), this, left, right, zoneId);
+    protected Pipe createPipe(Pipe dateTimePart, Pipe timestamp, ZoneId zoneId) {
+        return new DatePartPipe(source(), this, dateTimePart, timestamp, zoneId);
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateTrunc.java

@@ -160,8 +160,8 @@ public class DateTrunc extends BinaryDateTimeFunction {
     }
 
     @Override
-    protected Pipe createPipe(Pipe left, Pipe right, ZoneId zoneId) {
-        return new DateTruncPipe(source(), this, left, right, zoneId);
+    protected Pipe createPipe(Pipe truncateTo, Pipe timestamp, ZoneId zoneId) {
+        return new DateTruncPipe(source(), this, truncateTo, timestamp, zoneId);
     }
 
     @Override

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/whitelist/InternalSqlScriptUtils.java

@@ -10,6 +10,7 @@ import org.elasticsearch.index.fielddata.ScriptDocValues;
 import org.elasticsearch.script.JodaCompatibleZonedDateTime;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateAddProcessor;
+import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateDiffProcessor;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DatePartProcessor;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeFunction;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTruncProcessor;
@@ -377,6 +378,10 @@ public final class InternalSqlScriptUtils {
         return (ZonedDateTime) DateAddProcessor.process(dateField, numberOfUnits, asDateTime(dateTime) , ZoneId.of(tzId));
     }
 
+    public static Integer dateDiff(String dateField, Object dateTime1, Object dateTime2, String tzId) {
+        return (Integer) DateDiffProcessor.process(dateField, asDateTime(dateTime1), asDateTime(dateTime2) , ZoneId.of(tzId));
+    }
+
     public static ZonedDateTime dateTrunc(String truncateTo, Object dateTime, String tzId) {
         return (ZonedDateTime) DateTruncProcessor.process(truncateTo, asDateTime(dateTime) , ZoneId.of(tzId));
     }

x-pack/plugin/sql/src/main/resources/org/elasticsearch/xpack/sql/plugin/sql_whitelist.txt

@@ -116,6 +116,7 @@ class org.elasticsearch.xpack.sql.expression.function.scalar.whitelist.InternalS
   Integer quarter(Object, String)
   Integer weekOfYear(Object, String)
   ZonedDateTime dateAdd(String, Integer, Object, String)
+  Integer dateDiff(String, Object, Object, String)
   ZonedDateTime dateTrunc(String, Object, String)
   Integer datePart(String, Object, String)
   IntervalDayTime intervalDayTime(String, String)

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/analysis/analyzer/VerifierErrorMessagesTests.java

@@ -226,14 +226,69 @@ public class VerifierErrorMessagesTests extends ESTestCase {
             error("SELECT DATE_TRUNC('yyyz', keyword) FROM test"));
     }
 
+    public void testDateAddValidArgs() {
+        accept("SELECT DATE_ADD('weekday', 0, date) FROM test");
+        accept("SELECT DATEADD('dw', 20, date) FROM test");
+        accept("SELECT TIMESTAMP_ADD('years', -10, date) FROM test");
+        accept("SELECT TIMESTAMPADD('dayofyear', 123, date) FROM test");
+        accept("SELECT DATE_ADD('dy', 30, date) FROM test");
+        accept("SELECT DATE_ADD('ms', 1, date::date) FROM test");
+    }
+
+    public void testDateAddInvalidArgs() {
+        assertEquals("1:8: first argument of [DATE_ADD(int, int, date)] must be [string], found value [int] type [integer]",
+            error("SELECT DATE_ADD(int, int, date) FROM test"));
+        assertEquals("1:8: second argument of [DATE_ADD(keyword, 1.2, date)] must be [integer], found value [1.2] " +
+            "type [double]", error("SELECT DATE_ADD(keyword, 1.2, date) FROM test"));
+        assertEquals("1:8: third argument of [DATE_ADD(keyword, int, keyword)] must be [date or datetime], found value [keyword] " +
+            "type [keyword]", error("SELECT DATE_ADD(keyword, int, keyword) FROM test"));
+        assertEquals("1:8: first argument of [DATE_ADD('invalid', int, date)] must be one of [YEAR, QUARTER, MONTH, DAYOFYEAR, " +
+                "DAY, WEEK, WEEKDAY, HOUR, MINUTE, SECOND, MILLISECOND, MICROSECOND, NANOSECOND] " +
+                "or their aliases; found value ['invalid']",
+            error("SELECT DATE_ADD('invalid', int, date) FROM test"));
+        assertEquals("1:8: Unknown value ['sacinds'] for first argument of [DATE_ADD('sacinds', int, date)]; " +
+                "did you mean [seconds, second]?",
+            error("SELECT DATE_ADD('sacinds', int, date) FROM test"));
+        assertEquals("1:8: Unknown value ['dz'] for first argument of [DATE_ADD('dz', int, date)]; " +
+                "did you mean [dd, dw, dy, d]?",
+            error("SELECT DATE_ADD('dz', int, date) FROM test"));
+    }
+
+    public void testDateDiffValidArgs() {
+        accept("SELECT DATE_DIFF('weekday', date, date) FROM test");
+        accept("SELECT DATEDIFF('dw', date::date, date) FROM test");
+        accept("SELECT TIMESTAMP_DIFF('years', date, date) FROM test");
+        accept("SELECT TIMESTAMPDIFF('dayofyear', date, date::date) FROM test");
+        accept("SELECT DATE_DIFF('dy', date, date) FROM test");
+        accept("SELECT DATE_DIFF('ms', date::date, date::date) FROM test");
+    }
+
+    public void testDateDiffInvalidArgs() {
+        assertEquals("1:8: first argument of [DATE_DIFF(int, date, date)] must be [string], found value [int] type [integer]",
+            error("SELECT DATE_DIFF(int, date, date) FROM test"));
+        assertEquals("1:8: second argument of [DATE_DIFF(keyword, keyword, date)] must be [date or datetime], found value [keyword] " +
+            "type [keyword]", error("SELECT DATE_DIFF(keyword, keyword, date) FROM test"));
+        assertEquals("1:8: third argument of [DATE_DIFF(keyword, date, keyword)] must be [date or datetime], found value [keyword] " +
+            "type [keyword]", error("SELECT DATE_DIFF(keyword, date, keyword) FROM test"));
+        assertEquals("1:8: first argument of [DATE_DIFF('invalid', int, date)] must be one of [YEAR, QUARTER, MONTH, DAYOFYEAR, " +
+                "DAY, WEEK, WEEKDAY, HOUR, MINUTE, SECOND, MILLISECOND, MICROSECOND, NANOSECOND] " +
+                "or their aliases; found value ['invalid']",
+            error("SELECT DATE_DIFF('invalid', int, date) FROM test"));
+        assertEquals("1:8: Unknown value ['sacinds'] for first argument of [DATE_DIFF('sacinds', int, date)]; " +
+                "did you mean [seconds, second]?",
+            error("SELECT DATE_DIFF('sacinds', int, date) FROM test"));
+        assertEquals("1:8: Unknown value ['dz'] for first argument of [DATE_DIFF('dz', int, date)]; " +
+                "did you mean [dd, dw, dy, d]?",
+            error("SELECT DATE_DIFF('dz', int, date) FROM test"));
+    }
+
     public void testDateTruncValidArgs() {
         accept("SELECT DATE_TRUNC('decade', date) FROM test");
         accept("SELECT DATE_TRUNC('decades', date) FROM test");
-        accept("SELECT DATE_TRUNC('day', date) FROM test");
-        accept("SELECT DATE_TRUNC('days', date) FROM test");
+        accept("SELECT DATETRUNC('day', date) FROM test");
+        accept("SELECT DATETRUNC('days', date) FROM test");
         accept("SELECT DATE_TRUNC('dd', date) FROM test");
         accept("SELECT DATE_TRUNC('d', date) FROM test");
-
     }
 
     public void testDatePartInvalidArgs() {
@@ -255,41 +310,13 @@ public class VerifierErrorMessagesTests extends ESTestCase {
 
     public void testDatePartValidArgs() {
         accept("SELECT DATE_PART('weekday', date) FROM test");
-        accept("SELECT DATE_PART('dw', date) FROM test");
-        accept("SELECT DATE_PART('tz', date) FROM test");
+        accept("SELECT DATEPART('dw', date) FROM test");
+        accept("SELECT DATEPART('tz', date) FROM test");
         accept("SELECT DATE_PART('dayofyear', date) FROM test");
         accept("SELECT DATE_PART('dy', date) FROM test");
         accept("SELECT DATE_PART('ms', date) FROM test");
     }
 
-    public void testDateAddInvalidArgs() {
-        assertEquals("1:8: first argument of [DATE_ADD(int, int, date)] must be [string], found value [int] type [integer]",
-            error("SELECT DATE_ADD(int, int, date) FROM test"));
-        assertEquals("1:8: second argument of [DATE_ADD(keyword, 1.2, date)] must be [integer], found value [1.2] " +
-            "type [double]", error("SELECT DATE_ADD(keyword, 1.2, date) FROM test"));
-        assertEquals("1:8: third argument of [DATE_ADD(keyword, int, keyword)] must be [date or datetime], found value [keyword] " +
-            "type [keyword]", error("SELECT DATE_ADD(keyword, int, keyword) FROM test"));
-        assertEquals("1:8: first argument of [DATE_ADD('invalid', int, date)] must be one of [YEAR, QUARTER, MONTH, DAYOFYEAR, " +
-                "DAY, WEEK, WEEKDAY, HOUR, MINUTE, SECOND, MILLISECOND, MICROSECOND, NANOSECOND] " +
-                "or their aliases; found value ['invalid']",
-            error("SELECT DATE_ADD('invalid', int, date) FROM test"));
-        assertEquals("1:8: Unknown value ['sacinds'] for first argument of [DATE_ADD('sacinds', int, date)]; " +
-                "did you mean [seconds, second]?",
-            error("SELECT DATE_ADD('sacinds', int, date) FROM test"));
-        assertEquals("1:8: Unknown value ['dz'] for first argument of [DATE_ADD('dz', int, date)]; " +
-                "did you mean [dd, dw, dy, d]?",
-            error("SELECT DATE_ADD('dz', int, date) FROM test"));
-    }
-
-    public void testDateAddValidArgs() {
-        accept("SELECT DATE_ADD('weekday', 0, date) FROM test");
-        accept("SELECT DATEADD('dw', 20, date) FROM test");
-        accept("SELECT TIMESTAMP_ADD('years', -10, date) FROM test");
-        accept("SELECT TIMESTAMPADD('dayofyear', 123, date) FROM test");
-        accept("SELECT DATE_ADD('dy', 30, date) FROM test");
-        accept("SELECT DATE_ADD('ms', 1, date::date) FROM test");
-    }
-
     public void testValidDateTimeFunctionsOnTime() {
         accept("SELECT HOUR_OF_DAY(CAST(date AS TIME)) FROM test");
         accept("SELECT MINUTE_OF_HOUR(CAST(date AS TIME)) FROM test");

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateDiffPipeTests.java

@@ -0,0 +1,153 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
+
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.function.scalar.FunctionTestUtils;
+import org.elasticsearch.xpack.sql.expression.gen.pipeline.Pipe;
+import org.elasticsearch.xpack.sql.tree.AbstractNodeTestCase;
+import org.elasticsearch.xpack.sql.tree.Source;
+import org.elasticsearch.xpack.sql.tree.SourceTests;
+
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Function;
+
+import static org.elasticsearch.xpack.sql.expression.Expressions.pipe;
+import static org.elasticsearch.xpack.sql.expression.function.scalar.FunctionTestUtils.randomDatetimeLiteral;
+import static org.elasticsearch.xpack.sql.expression.function.scalar.FunctionTestUtils.randomStringLiteral;
+import static org.elasticsearch.xpack.sql.tree.SourceTests.randomSource;
+
+public class DateDiffPipeTests extends AbstractNodeTestCase<DateDiffPipe, Pipe> {
+
+    @Override
+    protected DateDiffPipe randomInstance() {
+        return randomDateDiffPipe();
+    }
+
+    private Expression randomDateDiffPipeExpression() {
+        return randomDateDiffPipe().expression();
+    }
+
+    public static DateDiffPipe randomDateDiffPipe() {
+        return (DateDiffPipe) new DateDiff(
+                randomSource(),
+                randomStringLiteral(),
+                randomDatetimeLiteral(),
+                randomDatetimeLiteral(),
+                randomZone())
+                .makePipe();
+    }
+
+    @Override
+    public void testTransform() {
+        // test transforming only the properties (source, expression),
+        // skipping the children (the three parameters of the function) which are tested separately
+        DateDiffPipe b1 = randomInstance();
+
+        Expression newExpression = randomValueOtherThan(b1.expression(), this::randomDateDiffPipeExpression);
+        DateDiffPipe newB = new DateDiffPipe(
+                b1.source(),
+                newExpression,
+                b1.first(),
+                b1.second(),
+                b1.third(),
+                b1.zoneId());
+        assertEquals(newB, b1.transformPropertiesOnly(v -> Objects.equals(v, b1.expression()) ? newExpression : v, Expression.class));
+
+        DateDiffPipe b2 = randomInstance();
+        Source newLoc = randomValueOtherThan(b2.source(), SourceTests::randomSource);
+        newB = new DateDiffPipe(
+                newLoc,
+                b2.expression(),
+                b2.first(),
+                b2.second(),
+                b2.third(),
+                b2.zoneId());
+        assertEquals(newB,
+                b2.transformPropertiesOnly(v -> Objects.equals(v, b2.source()) ? newLoc : v, Source.class));
+    }
+
+    @Override
+    public void testReplaceChildren() {
+        DateDiffPipe b = randomInstance();
+        Pipe newFirst = pipe(((Expression) randomValueOtherThan(b.first(), FunctionTestUtils::randomStringLiteral)));
+        Pipe newSecond = pipe(((Expression) randomValueOtherThan(b.second(), FunctionTestUtils::randomDatetimeLiteral)));
+        Pipe newThird = pipe(((Expression) randomValueOtherThan(b.third(), FunctionTestUtils::randomDatetimeLiteral)));
+        ZoneId newZoneId = randomValueOtherThan(b.zoneId(), ESTestCase::randomZone);
+        DateDiffPipe newB = new DateDiffPipe(b.source(), b.expression(), b.first(), b.second(), b.third(), newZoneId);
+
+        ThreeArgsDateTimePipe transformed = newB.replaceChildren(newFirst, b.second(), b.third());
+        assertEquals(transformed.source(), b.source());
+        assertEquals(transformed.expression(), b.expression());
+        assertEquals(transformed.first(), newFirst);
+        assertEquals(transformed.second(), b.second());
+        assertEquals(transformed.third(), b.third());
+
+        transformed = newB.replaceChildren(b.first(), newSecond, b.third());
+        assertEquals(transformed.source(), b.source());
+        assertEquals(transformed.expression(), b.expression());
+        assertEquals(transformed.first(), b.first());
+        assertEquals(transformed.second(), newSecond);
+        assertEquals(transformed.third(), b.third());
+
+        transformed = newB.replaceChildren(b.first(), b.second(), newThird);
+        assertEquals(transformed.expression(), b.expression());
+        assertEquals(transformed.source(), b.source());
+        assertEquals(transformed.first(), b.first());
+        assertEquals(transformed.second(), b.second());
+        assertEquals(transformed.third(), newThird);
+
+        transformed = newB.replaceChildren(newFirst, newSecond, newThird);
+        assertEquals(transformed.source(), b.source());
+        assertEquals(transformed.expression(), b.expression());
+        assertEquals(transformed.first(), newFirst);
+        assertEquals(transformed.second(), newSecond);
+        assertEquals(transformed.third(), newThird);
+    }
+
+    @Override
+    protected DateDiffPipe mutate(DateDiffPipe instance) {
+        List<Function<DateDiffPipe, DateDiffPipe>> randoms = new ArrayList<>();
+        randoms.add(f -> new DateDiffPipe(f.source(), f.expression(),
+                pipe(((Expression) randomValueOtherThan(f.first(), FunctionTestUtils::randomStringLiteral))),
+                f.second(),
+                f.third(),
+                randomValueOtherThan(f.zoneId(), ESTestCase::randomZone)));
+        randoms.add(f -> new DateDiffPipe(f.source(), f.expression(),
+                f.first(),
+                pipe(((Expression) randomValueOtherThan(f.second(), FunctionTestUtils::randomDatetimeLiteral))),
+                f.third(),
+                randomValueOtherThan(f.zoneId(), ESTestCase::randomZone)));
+        randoms.add(f -> new DateDiffPipe(f.source(), f.expression(),
+                f.first(),
+                f.second(),
+                pipe(((Expression) randomValueOtherThan(f.third(), FunctionTestUtils::randomDatetimeLiteral))),
+            randomValueOtherThan(f.zoneId(), ESTestCase::randomZone)));
+        randoms.add(f -> new DateDiffPipe(f.source(), f.expression(),
+                pipe(((Expression) randomValueOtherThan(f.first(), FunctionTestUtils::randomStringLiteral))),
+                pipe(((Expression) randomValueOtherThan(f.second(), FunctionTestUtils::randomDatetimeLiteral))),
+                pipe(((Expression) randomValueOtherThan(f.third(), FunctionTestUtils::randomDatetimeLiteral))),
+                randomValueOtherThan(f.zoneId(), ESTestCase::randomZone)));
+
+        return randomFrom(randoms).apply(instance);
+    }
+
+    @Override
+    protected DateDiffPipe copy(DateDiffPipe instance) {
+        return new DateDiffPipe(
+            instance.source(),
+            instance.expression(),
+            instance.first(),
+            instance.second(),
+            instance.third(),
+            instance.zoneId());
+    }
+}

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateDiffProcessorTests.java

@@ -0,0 +1,360 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
+
+import org.elasticsearch.common.io.stream.Writeable.Reader;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.sql.AbstractSqlWireSerializingTestCase;
+import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
+import org.elasticsearch.xpack.sql.expression.Literal;
+import org.elasticsearch.xpack.sql.expression.gen.processor.ConstantProcessor;
+import org.elasticsearch.xpack.sql.tree.Source;
+
+import java.time.ZoneId;
+
+import static org.elasticsearch.xpack.sql.expression.Literal.NULL;
+import static org.elasticsearch.xpack.sql.expression.function.scalar.FunctionTestUtils.l;
+import static org.elasticsearch.xpack.sql.expression.function.scalar.FunctionTestUtils.randomDatetimeLiteral;
+import static org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeTestUtils.dateTime;
+import static org.elasticsearch.xpack.sql.util.DateUtils.UTC;
+
+public class DateDiffProcessorTests extends AbstractSqlWireSerializingTestCase<DateDiffProcessor> {
+
+    public static DateDiffProcessor randomDateDiffProcessor() {
+        return new DateDiffProcessor(
+            new ConstantProcessor(randomRealisticUnicodeOfLengthBetween(0, 128)),
+            new ConstantProcessor(DateTimeTestUtils.nowWithMillisResolution()),
+            new ConstantProcessor(DateTimeTestUtils.nowWithMillisResolution()),
+            randomZone());
+    }
+
+    @Override
+    protected DateDiffProcessor createTestInstance() {
+        return randomDateDiffProcessor();
+    }
+
+    @Override
+    protected Reader<DateDiffProcessor> instanceReader() {
+        return DateDiffProcessor::new;
+    }
+
+    @Override
+    protected ZoneId instanceZoneId(DateDiffProcessor instance) {
+        return instance.zoneId();
+    }
+
+    @Override
+    protected DateDiffProcessor mutateInstance(DateDiffProcessor instance) {
+        return new DateDiffProcessor(
+            new ConstantProcessor(ESTestCase.randomRealisticUnicodeOfLength(128)),
+            new ConstantProcessor(DateTimeTestUtils.nowWithMillisResolution()),
+            new ConstantProcessor(DateTimeTestUtils.nowWithMillisResolution()),
+            randomValueOtherThan(instance.zoneId(), ESTestCase::randomZone));
+    }
+
+    public void testInvalidInputs() {
+        SqlIllegalArgumentException siae = expectThrows(SqlIllegalArgumentException.class,
+                () -> new DateDiff(Source.EMPTY, l(5),
+                    randomDatetimeLiteral(), randomDatetimeLiteral(), randomZone()).makePipe().asProcessor().process(null));
+        assertEquals("A string is required; received [5]", siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY,
+                l("days"), l("foo"), randomDatetimeLiteral(), randomZone()).makePipe().asProcessor().process(null));
+        assertEquals("A date/datetime is required; received [foo]", siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY,
+                l("days"), randomDatetimeLiteral(), l("foo"), randomZone()).makePipe().asProcessor().process(null));
+        assertEquals("A date/datetime is required; received [foo]", siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("invalid"),
+                randomDatetimeLiteral(), randomDatetimeLiteral(), randomZone()).makePipe().asProcessor().process(null));
+        assertEquals("A value of [YEAR, QUARTER, MONTH, DAYOFYEAR, DAY, WEEK, WEEKDAY, HOUR, MINUTE, " +
+            "SECOND, MILLISECOND, MICROSECOND, NANOSECOND] or their aliases is required; received [invalid]",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("quertar"),
+                randomDatetimeLiteral(), randomDatetimeLiteral(), randomZone()).makePipe().asProcessor().process(null));
+        assertEquals("Received value [quertar] is not valid date part to add; did you mean [quarter, quarters]?",
+             siae.getMessage());
+    }
+
+    public void testWithNulls() {
+        assertNull(new DateDiff(Source.EMPTY,
+            NULL, randomDatetimeLiteral(), randomDatetimeLiteral(), randomZone()).makePipe().asProcessor().process(null));
+        assertNull(new DateDiff(Source.EMPTY,
+            l("days"), NULL, randomDatetimeLiteral(), randomZone()).makePipe().asProcessor().process(null));
+        assertNull(new DateDiff(Source.EMPTY,
+            l("days"), randomDatetimeLiteral(), NULL, randomZone()).makePipe().asProcessor().process(null));
+        assertNull(new DateDiff(Source.EMPTY,
+            NULL, NULL, NULL, randomZone()).makePipe().asProcessor().process(null));
+    }
+
+    public void testDiff() {
+        ZoneId zoneId = ZoneId.of("Etc/GMT-10");
+
+        Literal dt1 = l(dateTime(2019, 12, 31, 20, 22, 33, 987654321, ZoneId.of("Etc/GMT+5")));
+        Literal dt2 = l(dateTime(2022, 1, 1, 4, 33, 22, 123456789, ZoneId.of("Etc/GMT-5")));
+
+        assertEquals(1, new DateDiff(Source.EMPTY, l("years"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-1, new DateDiff(Source.EMPTY, l("year"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(2, new DateDiff(Source.EMPTY, l("yyyy"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-2, new DateDiff(Source.EMPTY, l("yy"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(7, new DateDiff(Source.EMPTY, l("quarter"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-7, new DateDiff(Source.EMPTY, l("qq"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(8, new DateDiff(Source.EMPTY, l("quarter"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-8, new DateDiff(Source.EMPTY, l("qq"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(23, new DateDiff(Source.EMPTY, l("month"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-23, new DateDiff(Source.EMPTY, l("months"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(24, new DateDiff(Source.EMPTY, l("mm"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-24, new DateDiff(Source.EMPTY, l("m"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(730, new DateDiff(Source.EMPTY, l("dayofyear"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-730, new DateDiff(Source.EMPTY, l("dy"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(730, new DateDiff(Source.EMPTY, l("y"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-730, new DateDiff(Source.EMPTY, l("y"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(730, new DateDiff(Source.EMPTY, l("day"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-730, new DateDiff(Source.EMPTY, l("days"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(730, new DateDiff(Source.EMPTY, l("dd"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-730, new DateDiff(Source.EMPTY, l("dd"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(104, new DateDiff(Source.EMPTY, l("week"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-104, new DateDiff(Source.EMPTY, l("weeks"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(104, new DateDiff(Source.EMPTY, l("wk"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-104, new DateDiff(Source.EMPTY, l("ww"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(730, new DateDiff(Source.EMPTY, l("weekday"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-730, new DateDiff(Source.EMPTY, l("weekdays"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(730, new DateDiff(Source.EMPTY, l("dw"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-730, new DateDiff(Source.EMPTY, l("dw"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(17542, new DateDiff(Source.EMPTY, l("hour"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-17542, new DateDiff(Source.EMPTY, l("hours"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(17542, new DateDiff(Source.EMPTY, l("hh"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-17542, new DateDiff(Source.EMPTY, l("hh"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(1052531, new DateDiff(Source.EMPTY, l("minute"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-1052531, new DateDiff(Source.EMPTY, l("minutes"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(1052531, new DateDiff(Source.EMPTY, l("mi"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-1052531, new DateDiff(Source.EMPTY, l("n"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        dt1 = l(dateTime(2020, 12, 31, 20, 22, 33, 123456789, ZoneId.of("Etc/GMT+5")));
+        dt2 = l(dateTime(2021, 1, 1, 10, 33, 22, 987654321, ZoneId.of("Etc/GMT-5")));
+
+        assertEquals(15049, new DateDiff(Source.EMPTY, l("second"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-15049, new DateDiff(Source.EMPTY, l("seconds"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(15049, new DateDiff(Source.EMPTY, l("ss"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-15049, new DateDiff(Source.EMPTY, l("s"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(15049864, new DateDiff(Source.EMPTY, l("millisecond"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-15049864, new DateDiff(Source.EMPTY, l("milliseconds"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(15049864, new DateDiff(Source.EMPTY, l("ms"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-15049864, new DateDiff(Source.EMPTY, l("ms"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        dt1 = l(dateTime(2020, 12, 31, 20, 22, 33, 123456789, ZoneId.of("Etc/GMT+5")));
+        dt2 = l(dateTime(2021, 1, 1, 6, 33, 22, 987654321, ZoneId.of("Etc/GMT-5")));
+
+        assertEquals(649864198, new DateDiff(Source.EMPTY, l("microsecond"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-649864198, new DateDiff(Source.EMPTY, l("microseconds"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(649864198, new DateDiff(Source.EMPTY, l("mcs"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-649864198, new DateDiff(Source.EMPTY, l("mcs"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        dt1 = l(dateTime(2020, 12, 31, 20, 33, 22, 123456789, ZoneId.of("Etc/GMT+5")));
+        dt2 = l(dateTime(2021, 1, 1, 6, 33, 23, 987654321, ZoneId.of("Etc/GMT-5")));
+
+        assertEquals(1864197532, new DateDiff(Source.EMPTY, l("nanosecond"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-1864197532, new DateDiff(Source.EMPTY, l("nanoseconds"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(1864197532, new DateDiff(Source.EMPTY, l("ns"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-1864197532, new DateDiff(Source.EMPTY, l("ns"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+    }
+
+    public void testDiffEdgeCases() {
+        ZoneId zoneId = ZoneId.of("Etc/GMT-10");
+
+        Literal dt1 = l(dateTime(2010, 12, 31, 18, 0, 0, 0));
+        Literal dt2 = l(dateTime(2019, 1, 1, 18, 0, 0, 0));
+
+        assertEquals(9, new DateDiff(Source.EMPTY, l("years"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-9, new DateDiff(Source.EMPTY, l("year"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(8, new DateDiff(Source.EMPTY, l("yyyy"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-8, new DateDiff(Source.EMPTY, l("yy"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(33, new DateDiff(Source.EMPTY, l("quarter"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-33, new DateDiff(Source.EMPTY, l("qq"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(32, new DateDiff(Source.EMPTY, l("quarter"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-32, new DateDiff(Source.EMPTY, l("qq"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        assertEquals(97, new DateDiff(Source.EMPTY, l("month"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-97, new DateDiff(Source.EMPTY, l("months"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(96, new DateDiff(Source.EMPTY, l("mm"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-96, new DateDiff(Source.EMPTY, l("m"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        dt1 = l(dateTime(1976, 9, 9, 0, 0, 0, 0));
+        dt2 = l(dateTime(1983, 5, 22, 0, 0, 0, 0));
+        assertEquals(350, new DateDiff(Source.EMPTY, l("week"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-350, new DateDiff(Source.EMPTY, l("weeks"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(350, new DateDiff(Source.EMPTY, l("wk"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-350, new DateDiff(Source.EMPTY, l("ww"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+
+        dt1 = l(dateTime(1988, 1, 5, 0, 0, 0, 0));
+        dt2 = l(dateTime(1996, 5, 13, 0, 0, 0, 0));
+        assertEquals(436, new DateDiff(Source.EMPTY, l("week"), dt1, dt2, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-436, new DateDiff(Source.EMPTY, l("weeks"), dt2, dt1, UTC)
+            .makePipe().asProcessor().process(null));
+        assertEquals(436, new DateDiff(Source.EMPTY, l("wk"), dt1, dt2, zoneId)
+            .makePipe().asProcessor().process(null));
+        assertEquals(-436, new DateDiff(Source.EMPTY, l("ww"), dt2, dt1, zoneId)
+            .makePipe().asProcessor().process(null));
+    }
+
+    public void testOverflow() {
+        ZoneId zoneId = ZoneId.of("Etc/GMT-10");
+        Literal dt1 = l(dateTime(-99992022, 12, 31, 20, 22, 33, 123456789, ZoneId.of("Etc/GMT-5")));
+        Literal dt2 = l(dateTime(99992022, 4, 18, 8, 33, 22, 987654321, ZoneId.of("Etc/GMT+5")));
+
+        SqlIllegalArgumentException siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("month"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("dayofyear"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("day"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("week"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("weekday"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("hours"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("minute"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("second"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("milliseconds"), dt2, dt1, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+         siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("mcs"), dt1, dt2, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+
+        siae = expectThrows(SqlIllegalArgumentException.class,
+            () -> new DateDiff(Source.EMPTY, l("nanoseconds"), dt2, dt1, zoneId).makePipe().asProcessor().process(null));
+        assertEquals("The DATE_DIFF function resulted in an overflow; the number of units separating two date/datetime " +
+                "instances is too large. Try to use DATE_DIFF with a less precise unit.",
+            siae.getMessage());
+    }
+}

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/DateTimeTestUtils.java

@@ -11,6 +11,7 @@ import org.elasticsearch.xpack.sql.util.DateUtils;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.OffsetTime;
+import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 
@@ -23,7 +24,12 @@ public class DateTimeTestUtils {
     }
 
     public static ZonedDateTime dateTime(int year, int month, int day, int hour, int minute, int seconds, int nanos) {
-        return ZonedDateTime.of(year, month, day, hour, minute, seconds, nanos, DateUtils.UTC);
+        return dateTime(year, month, day, hour, minute, seconds, nanos, DateUtils.UTC);
+    }
+
+    public static ZonedDateTime dateTime(int year, int month, int day, int hour, int minute, int seconds, int nanos,
+                                         ZoneId zoneId) {
+        return ZonedDateTime.of(year, month, day, hour, minute, seconds, nanos, zoneId);
     }
 
     public static ZonedDateTime dateTime(long millisSinceEpoch) {

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/planner/QueryTranslatorTests.java

@@ -310,6 +310,23 @@ public class QueryTranslatorTests extends ESTestCase {
         assertEquals("[{v=quarter}, {v=int}, {v=date}, {v=Z}, {v=2018-09-04T00:00:00.000Z}]", sc.script().params().toString());
     }
 
+    public void testTranslateDateDiff_WhereClause_Painless() {
+        LogicalPlan p = plan("SELECT int FROM test WHERE DATE_DIFF('week',date, date) > '2018-09-04'::date");
+        assertTrue(p instanceof Project);
+        assertTrue(p.children().get(0) instanceof Filter);
+        Expression condition = ((Filter) p.children().get(0)).condition();
+        assertFalse(condition.foldable());
+        QueryTranslation translation = QueryTranslator.toQuery(condition, false);
+        assertNull(translation.aggFilter);
+        assertTrue(translation.query instanceof ScriptQuery);
+        ScriptQuery sc = (ScriptQuery) translation.query;
+        assertEquals("InternalSqlScriptUtils.nullSafeFilter(InternalSqlScriptUtils.gt(InternalSqlScriptUtils.dateDiff(" +
+                "params.v0,InternalSqlScriptUtils.docValue(doc,params.v1),InternalSqlScriptUtils.docValue(doc,params.v2)," +
+                "params.v3),InternalSqlScriptUtils.asDateTime(params.v4)))",
+            sc.script().toString());
+        assertEquals("[{v=week}, {v=date}, {v=date}, {v=Z}, {v=2018-09-04T00:00:00.000Z}]", sc.script().params().toString());
+    }
+
     public void testTranslateDateTrunc_WhereClause_Painless() {
         LogicalPlan p = plan("SELECT int FROM test WHERE DATE_TRUNC('month', date) > '2018-09-04'::date");
         assertTrue(p instanceof Project);