Revert the null value handling in AggregationResultsBlock when nullHandlingEnabled is false

[jackjlli]

Before this PR, when the result value is null, it will be handled in the following method regardless of the value of nullHandlingEnabled.

  @Override
  public void setColumn(int colId, @Nullable Object value)
      throws IOException {
    _currentRowDataByteBuffer.position(_columnOffsets[colId]);
    _currentRowDataByteBuffer.putInt(_variableSizeDataByteArrayOutputStream.size());
    if (value == null) {
      _currentRowDataByteBuffer.putInt(0);
      _variableSizeDataOutputStream.writeInt(CustomObject.NULL_TYPE_VALUE);
    } else {
      int objectTypeValue = ObjectSerDeUtils.ObjectType.getObjectType(value).getValue();
      byte[] bytes = ObjectSerDeUtils.serialize(value, objectTypeValue);
      _currentRowDataByteBuffer.putInt(bytes.length);
      _variableSizeDataOutputStream.writeInt(objectTypeValue);
      _variableSizeDataByteArrayOutputStream.write(bytes);
    }
  }

While that PR changes the behavior that the result should not be null only when nullHandlingEnabled is set to true, and this breaks some of the use cases running in production. Here is the sample exception:

2025/03/26 19:07:06.365 ERROR [QueryScheduler] [pqr-0] [pinot-server] [] Caught exception while serializing response for requestId: 1549889132000050486, brokerId: broker_1_7001
java.lang.NullPointerException: Cannot invoke "it.unimi.dsi.fastutil.longs.LongSet.size()" because "longSet" is null
        at org.apache.pinot.core.common.ObjectSerDeUtils$24.serialize(ObjectSerDeUtils.java:937) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.common.ObjectSerDeUtils$24.serialize(ObjectSerDeUtils.java:933) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.query.aggregation.function.array.ArrayAggDistinctLongFunction.serializeIntermediateResult(ArrayAggDistinctLongFunction.java:68) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.query.aggregation.function.array.ArrayAggDistinctLongFunction.serializeIntermediateResult(ArrayAggDistinctLongFunction.java:33) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock.getDataTable(AggregationResultsBlock.java:170) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.operator.blocks.InstanceResponseBlock.toDataOnlyDataTable(InstanceResponseBlock.java:114) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.operator.blocks.InstanceResponseBlock.toDataTable(InstanceResponseBlock.java:107) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.query.scheduler.QueryScheduler.serializeResponse(QueryScheduler.java:222) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.query.scheduler.QueryScheduler.processQueryAndSerialize(QueryScheduler.java:156) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.query.scheduler.QueryScheduler.lambda$createQueryFutureTask$0(QueryScheduler.java:123) ~[org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
        at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:75) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82) [com.google.guava.guava-33.4.0-jre.jar:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
        at java.lang.Thread.run(Thread.java:833) [?:?]
2025/03/26 19:07:06.365 INFO [ServerQueryLogger] [pqr-0] [pinot-server] [] Processed requestId=1549889132000050486,table=test_table_OFFLINE,segments(queried/processed/matched/consumingQueried/consumingProcessed/consumingMatched/invalid/limit/value)=60/0/0/-1/0/0/0/0/60,schedulerWaitMs=2,reqDeserMs=2,totalExecMs=5,resSerMs=2,totalTimeMs=11,minConsumingF
reshnessMs=-1,broker=broker_1_7001,numDocsScanned=0,scanInFilter=0,scanPostFilter=0,sched=BinaryWorkloadScheduler,threadCpuTimeNs(total/thread/sysActivity/resSer)=0/0/0/0
2025/03/26 19:07:06.367 ERROR [InstanceRequestHandler] [pqr-0] [pinot-server] [] Query processing error: 
java.lang.Exception: Null query response.
        at org.apache.pinot.core.transport.InstanceRequestHandler$1.onSuccess(InstanceRequestHandler.java:203) [org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at org.apache.pinot.core.transport.InstanceRequestHandler$1.onSuccess(InstanceRequestHandler.java:186) [org.apache.pinot.pinot-core-1.4.0.jar:1.4.0-701bba35005112f9e5b37cadc98fda3f9f8b9976]
        at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1139) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:31) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1307) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:1070) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.AbstractFuture.access$500(AbstractFuture.java:80) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.AbstractFuture$SetFuture.run(AbstractFuture.java:365) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:31) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.ExecutionList.executeListener(ExecutionList.java:145) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.ExecutionList.execute(ExecutionList.java:133) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.ListenableFutureTask.done(ListenableFutureTask.java:113) [com.google.guava.guava-33.4.0-jre.jar:?]
        at java.util.concurrent.FutureTask.finishCompletion(FutureTask.java:381) [?:?]
        at java.util.concurrent.FutureTask.set(FutureTask.java:232) [?:?]
        at java.util.concurrent.FutureTask.run(FutureTask.java:272) [?:?]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
        at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:75) [com.google.guava.guava-33.4.0-jre.jar:?]
        at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82) [com.google.guava.guava-33.4.0-jre.jar:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
        at java.lang.Thread.run(Thread.java:833) [?:?]

And in fact, if the aggregate(int length, AggregationResultHolder aggregationResultHolder, Map<ExpressionContext, BlockValSet> blockValSetMap) method never gets invoked, the Object _value; inside the AggregationResultHolder class will be null, which is a valid case (we can see from the above query log that 0 segment gets processed).

This PR adds back the logic to handle the case when the result could be null.

[Jackie-Jiang]

Good catch!