Adding null check to earliest and latest aggs (#15972)

* Adding null check to earliest and latest aggs

* Native tests for null inPairs

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);

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);

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()
   {