Handle LocalDateTime for min and max aggregator (opensearch-project#20074)

Author: vinaykpud

server/src/main/java/org/opensearch/search/aggregations/metrics/MaxAggregator.java

@@ -62,9 +62,8 @@
 import org.opensearch.search.streaming.StreamingCostMetrics;
 
 import java.io.IOException;
-import java.util.ArrayList;
+import java.time.LocalDateTime;
 import java.util.Arrays;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
@@ -294,6 +293,10 @@ public StreamingCostMetrics getStreamingCostMetrics() {
     @Override
     public InternalAggregation convertRow(Map<String, Object[]> shardResult, int row, SearchContext searchContext) {
         Object[] values = shardResult.get(name);
+        if (values[row].getClass().equals(LocalDateTime.class)) {
+            LocalDateTime value = (LocalDateTime) values[row];
+            return new InternalMax(name, convertLocalDateTimeToEpochMillis(value), formatter, metadata());
+        }
         return new InternalMax(name, ((Number) values[row]).doubleValue(), formatter, metadata());
     }
 }

server/src/main/java/org/opensearch/search/aggregations/metrics/MinAggregator.java

@@ -62,8 +62,7 @@
 import org.opensearch.search.streaming.StreamingCostMetrics;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
+import java.time.LocalDateTime;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
@@ -280,6 +279,10 @@ public StarTreeBucketCollector getStarTreeBucketCollector(
     @Override
     public InternalAggregation convertRow(Map<String, Object[]> shardResult, int row, SearchContext searchContext) {
         Object[] values = shardResult.get(name);
+        if (values[row].getClass().equals(LocalDateTime.class)) {
+            LocalDateTime value = (LocalDateTime) values[row];
+            return new InternalMin(name, convertLocalDateTimeToEpochMillis(value), format, metadata());
+        }
         return new InternalMin(name, ((Number) values[row]).doubleValue(), format, metadata());
     }
 

server/src/main/java/org/opensearch/search/aggregations/metrics/NumericMetricsAggregator.java

@@ -37,6 +37,9 @@
 import org.opensearch.search.sort.SortOrder;
 
 import java.io.IOException;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.Map;
 
 /**
@@ -64,6 +67,19 @@ protected SingleValue(String name, SearchContext context, Aggregator parent, Map
 
         public abstract double metric(long owningBucketOrd);
 
+        /**
+         * Converts a LocalDateTime value to epoch milliseconds for use in aggregation results.
+         * The LocalDateTime is treated as UTC to preserve the exact date-time values
+         * without any timezone conversion.
+         *
+         * @param value the LocalDateTime value to convert
+         * @return the epoch milliseconds representation of the LocalDateTime treated as UTC
+         */
+        protected static double convertLocalDateTimeToEpochMillis(LocalDateTime value) {
+            Instant instant = value.atZone(ZoneOffset.UTC).toInstant();
+            return instant.toEpochMilli();
+        }
+
         @Override
         public BucketComparator bucketComparator(String key, SortOrder order) {
             if (key != null && false == "value".equals(key)) {