Add support for date_bin() function. (apache#12790)

* Add support for date_bin() function.

Add support for the Postgres SQL date_bin function (ref https://www.postgresql.org/docs/current/functions-datetime.html#FUNCTIONS-DATETIME-BIN).
- The 'stride' or 'binWidth' is specified in the `Period` format. For example:
  `1d`		: for 1 day
  `24h`		: for 24 hours
  `1h30m`	: for 1 hours and 30 minutes
- The granularities supported for `stride` are `seconds`, `minutes`, `hours` and `days`.
- The `source` and `origin` times are in `Timestamp` format which is in local time zone in Pinot.

Example Queries:
`select date_bin(`3h`, '2024-03-10 23:29:55.0', '2024-01-01 00:00:00.0') from myTable`
`select date_bin(`3h`, myTimeColumn, '2024-01-01 00:00:00.0') from myTable`

* Update DateTimeFunctions.java

---------

Co-authored-by: Xiang Fu <xiangfu.1024@gmail.com>

pinot-common/src/main/java/org/apache/pinot/common/function/scalar/DateTimeFunctions.java

@@ -25,6 +25,7 @@
 import org.apache.pinot.common.function.DateTimeUtils;
 import org.apache.pinot.common.function.TimeZoneKey;
 import org.apache.pinot.spi.annotations.ScalarFunction;
+import org.apache.pinot.spi.utils.TimeUtils;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.joda.time.chrono.ISOChronology;
@@ -1161,6 +1162,44 @@ private static long dateTrunc(String unit, long timeValue, String inputTimeUnit,
         TimeUnit.MILLISECONDS);
   }
 
+  /**
+   * Aligns a given timestamp to the nearest bin defined by the specified duration string, starting from an origin
+   * timestamp.
+   *
+   * @param binWidthStr The width of each bin in Period format (e.g., "15m" for 15 minutes, "2d" for 2 days).
+   * @param sourceTimestamp The timestamp to be aligned.
+   * @param originTimestamp The origin timestamp from which binning starts.
+   * @return A java.sql.Timestamp aligned to the nearest bin.
+   */
+  @ScalarFunction(names = {"dateBin", "date_bin"})
+  public static Timestamp dateBin(String binWidthStr, Timestamp sourceTimestamp, Timestamp originTimestamp) {
+    long originMillis = originTimestamp.getTime();
+    long sourceMillis = sourceTimestamp.getTime();
+
+    // Calculate the offset from the origin and adjust to the nearest bin
+    long binnedMillis = dateBin(binWidthStr, sourceMillis, originMillis);
+    return new Timestamp(binnedMillis);
+  }
+
+  /**
+   * Utility method to aligns a given timestamp in epoch Millis to the nearest bin defined by the specified
+   * duration string, starting from an origin timestamp in epoch Millis.
+   *
+   * @param binWidthStr The width of each bin as an ISO-8601 duration string (e.g., "PT15M" for 15 minutes).
+   * @param sourceMillisEpoch The source time in epoch millis to be aligned.
+   * @param originMillisEpoch The origin time in epoch millis from which binning starts.
+   * @return A java.sql.Timestamp aligned to the nearest bin.
+   */
+  public static long dateBin(String binWidthStr, long sourceMillisEpoch, long originMillisEpoch) {
+    long binWidthMillis = TimeUtils.convertPeriodToMillis(binWidthStr);
+    //long binWidthMillis = binWidth.toMillis();
+    long offsetFromOrigin = sourceMillisEpoch - originMillisEpoch;
+    long binCount = offsetFromOrigin / binWidthMillis;
+
+    // Calculate the start of the bin for the given timestamp
+    return originMillisEpoch + binWidthMillis * binCount;
+  }
+
   /**
    * Add a time period to the provided timestamp.
    * e.g. timestampAdd('days', 10, NOW()) will add 10 days to the current timestamp and return the value
@@ -1208,8 +1247,9 @@ public static long[] timestampDiffMV(String unit, long[] timestamp1, long timest
     return results;
   }
 
-  @ScalarFunction(names = {"timestampDiffMVReverse", "timestamp_diff_mv_reverse", "dateDiffMVReverse",
-      "date_diff_mv_reverse"})
+  @ScalarFunction(names = {
+      "timestampDiffMVReverse", "timestamp_diff_mv_reverse", "dateDiffMVReverse", "date_diff_mv_reverse"
+  })
   public static long[] timestampDiffMVReverse(String unit, long timestamp1, long[] timestamp2) {
     long[] results = new long[timestamp2.length];
     for (int i = 0; i < timestamp2.length; i++) {

pinot-core/src/test/java/org/apache/pinot/core/data/function/DateTimeFunctionsTest.java

@@ -19,11 +19,13 @@
 package org.apache.pinot.core.data.function;
 
 import com.google.common.collect.Lists;
+import java.sql.Timestamp;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import org.apache.pinot.common.function.scalar.DateTimeFunctions;
 import org.apache.pinot.segment.local.function.InbuiltFunctionEvaluator;
 import org.apache.pinot.spi.data.readers.GenericRow;
 import org.joda.time.DateTime;
@@ -58,6 +60,7 @@ public void testDateTimeFunctions(String functionExpression, List<String> expect
     testFunction(functionExpression, expectedArguments, row, expectedResult);
   }
 
+
   @DataProvider(name = "dateTimeFunctionsDataProvider")
   public Object[][] dateTimeFunctionsDataProvider() {
     List<Object[]> inputs = new ArrayList<>();
@@ -432,6 +435,48 @@ private static long iso8601ToUtcEpochMillis(String iso8601) {
     return formatter.parseDateTime(iso8601).getMillis();
   }
 
+  @Test
+  public void testDateBin() {
+    assertEquals(DateTimeFunctions.dateBin("2s", Timestamp.valueOf("2024-02-10 23:29:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 23:29:54.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("10s", Timestamp.valueOf("2024-02-10 23:29:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 23:29:50.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("10m", Timestamp.valueOf("2024-02-10 23:29:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 23:20:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("15m", Timestamp.valueOf("2024-02-10 23:29:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 23:15:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("20m", Timestamp.valueOf("2024-02-10 23:29:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 23:20:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("30m", Timestamp.valueOf("2024-02-10 23:00:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 23:00:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("1h", Timestamp.valueOf("2024-02-10 23:00:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 23:00:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("1h15m", Timestamp.valueOf("2024-02-10 23:00:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 22:30:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("2h", Timestamp.valueOf("2024-02-10 23:00:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 22:00:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("24h", Timestamp.valueOf("2024-02-10 23:00:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 00:00:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("1d", Timestamp.valueOf("2024-02-10 23:00:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-10 00:00:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("2d", Timestamp.valueOf("2024-02-09 23:00:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-02-08 00:00:00.0"));
+
+    assertEquals(DateTimeFunctions.dateBin("10d10m", Timestamp.valueOf("2024-02-09 23:00:55.0"),
+        Timestamp.valueOf("2024-01-01 00:00:00.0")), Timestamp.valueOf("2024-01-31 00:30:00.0"));
+  }
+
   @Test
   public void testDateTimeConvert() {
     // EPOCH to EPOCH