From c7823bca9860ea14624597ff669bfaa5c3734116 Mon Sep 17 00:00:00 2001
From: Soumyava <93540295+somu-imply@users.noreply.github.com>
Date: Mon, 18 Mar 2024 21:14:14 -0700
Subject: [PATCH] Adding null check to earliest and latest aggs (#15972)

* Adding null check to earliest and latest aggs

* Native tests for null inPairs
---
 .../first/NumericFirstVectorAggregator.java   |  2 +-
 .../last/NumericLastVectorAggregator.java     |  2 +-
 .../FloatFirstVectorAggregationTest.java      | 38 +++++++++++++++++++
 3 files changed, 40 insertions(+), 2 deletions(-)

diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstVectorAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstVectorAggregator.java
index b487ac817d5..6f6bfec014c 100644
--- a/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstVectorAggregator.java
+++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstVectorAggregator.java
@@ -95,7 +95,7 @@ public abstract class NumericFirstVectorAggregator implements VectorAggregator
 
       if (objectsWhichMightBeNumeric != null) {
         final SerializablePair<Long, Number> inPair = (SerializablePair<Long, Number>) objectsWhichMightBeNumeric[index];
-        if (inPair.lhs != null && inPair.lhs < firstTime) {
+        if (inPair != null && inPair.lhs != null && inPair.lhs < firstTime) {
           firstTime = inPair.lhs;
           if (useDefault || inPair.rhs != null) {
             updateTimeWithValue(buf, position, firstTime, index);
diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastVectorAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastVectorAggregator.java
index ff5fa3af9e9..a556c3fea18 100644
--- a/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastVectorAggregator.java
+++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastVectorAggregator.java
@@ -96,7 +96,7 @@ public abstract class NumericLastVectorAggregator implements VectorAggregator
 
     if (objectsWhichMightBeNumeric != null) {
       final SerializablePair<Long, Number> inPair = (SerializablePair<Long, Number>) objectsWhichMightBeNumeric[index];
-      if (inPair.lhs != null && inPair.lhs >= lastTime) {
+      if (inPair != null && inPair.lhs != null && inPair.lhs >= lastTime) {
         lastTime = inPair.lhs;
         if (useDefault || inPair.rhs != null) {
           updateTimeWithValue(buf, position, lastTime, index);
diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/first/FloatFirstVectorAggregationTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/first/FloatFirstVectorAggregationTest.java
index c7e8210d6c6..28044de5e4e 100644
--- a/processing/src/test/java/org/apache/druid/query/aggregation/first/FloatFirstVectorAggregationTest.java
+++ b/processing/src/test/java/org/apache/druid/query/aggregation/first/FloatFirstVectorAggregationTest.java
@@ -64,12 +64,18 @@ public class FloatFirstVectorAggregationTest extends InitializedNullHandlingTest
       new SerializablePairLongFloat(2345300L, 4.2F)
   };
 
+  private final SerializablePairLongFloat[] null_pairs = {
+      null, null, null, null
+  };
+
 
 
   private VectorObjectSelector selector;
+  private VectorObjectSelector selector1;
   private BaseLongVectorValueSelector timeSelector;
   private ByteBuffer buf;
   private FloatFirstVectorAggregator target;
+  private FloatFirstVectorAggregator target1;
 
   private FloatFirstAggregatorFactory floatFirstAggregatorFactory;
   private VectorColumnSelectorFactory selectorFactory;
@@ -119,6 +125,27 @@ public class FloatFirstVectorAggregationTest extends InitializedNullHandlingTest
       }
     };
 
+    selector1 = new VectorObjectSelector()
+    {
+      @Override
+      public Object[] getObjectVector()
+      {
+        return null_pairs;
+      }
+
+      @Override
+      public int getMaxVectorSize()
+      {
+        return 4;
+      }
+
+      @Override
+      public int getCurrentVectorSize()
+      {
+        return 0;
+      }
+    };
+
     nonFloatValueSelector = new BaseLongVectorValueSelector(new NoFilterVectorOffset(
         LONG_VALUES.length,
         0,
@@ -219,6 +246,7 @@ public class FloatFirstVectorAggregationTest extends InitializedNullHandlingTest
     };
 
     target = new FloatFirstVectorAggregator(timeSelector, selector);
+    target1 = new FloatFirstVectorAggregator(timeSelector, selector1);
     clearBufferForPositions(0, 0);
 
     floatFirstAggregatorFactory = new FloatFirstAggregatorFactory(NAME, FIELD_NAME, TIME_COL);
@@ -252,6 +280,16 @@ public class FloatFirstVectorAggregationTest extends InitializedNullHandlingTest
     Assert.assertEquals(pairs[0].rhs, result.rhs, EPSILON);
   }
 
+  @Test
+  public void aggregateNulls1()
+  {
+    target1.init(buf, 0);
+    target1.aggregate(buf, 0, 0, VALUES.length);
+    Pair<Long, Float> result = (Pair<Long, Float>) target1.get(buf, 0);
+    Assert.assertEquals(Long.MAX_VALUE, result.lhs.longValue());
+    Assert.assertNull(result.rhs);
+  }
+
   @Test
   public void aggregateWithNulls()
   {