fix: address CodeQL nullness and type warnings

extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputSource.java

@@ -61,6 +61,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -160,7 +161,7 @@ public InputSourceReader reader(
         final StructType fullSnapshotSchema = snapshot.getSchema(engine);
         final StructType prunedSchema = pruneSchema(
             fullSnapshotSchema,
-            inputRowSchema.getColumnsFilter()
+            Objects.requireNonNull(inputRowSchema, "inputRowSchema").getColumnsFilter()
         );
 
         final ScanBuilder scanBuilder = snapshot.getScanBuilder(engine);

extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitter.java

@@ -236,7 +236,9 @@ public void close()
         server.close();
       }
     } else {
-      exec.shutdownNow();
+      if (exec != null) {
+        exec.shutdownNow();
+      }
       flush();
 
       try {

extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/ApproximateHistogram.java

@@ -626,6 +626,8 @@ protected ApproximateHistogram foldRule(ApproximateHistogram h, @Nullable float[
     // use preallocated arrays if passed
     if (mergedPositions == null) {
       mergedPositions = new float[this.size];
+    }
+    if (mergedBins == null) {
       mergedBins = new long[this.size];
     }
 

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

@@ -145,7 +145,7 @@
   @Override
   protected int getTaskGroupIdForPartition(KafkaTopicPartition partitionId)
   {
-    Integer taskCount = spec.getIoConfig().getTaskCount();
+    final int taskCount = spec.getIoConfig().getTaskCount();
     if (partitionId.isMultiTopicPartition()) {
       return Math.abs(31 * partitionId.topic().hashCode() + partitionId.partition()) % taskCount;
     } else {

indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HashPartitionMultiPhaseParallelIndexingTest.java

@@ -354,18 +354,17 @@ private void assertHashedPartition(
       List<DataSegment> segmentsInInterval = entry.getValue();
       Assert.assertEquals(expectedIntervalToNumSegments.get(interval).intValue(), segmentsInInterval.size());
       for (DataSegment segment : segmentsInInterval) {
-        HashBasedNumberedShardSpec shardSpec = null;
         if (segment.isTombstone()) {
           Assert.assertSame(TombstoneShardSpec.class, segment.getShardSpec().getClass());
         } else {
           Assert.assertSame(HashBasedNumberedShardSpec.class, segment.getShardSpec().getClass());
-          shardSpec = (HashBasedNumberedShardSpec) segment.getShardSpec();
-          Assert.assertEquals(HashPartitionFunction.MURMUR3_32_ABS, shardSpec.getPartitionFunction());
         }
         List<ScanResultValue> results = querySegment(segment, ImmutableList.of("dim1", "dim2"), tempSegmentDir);
         if (segment.isTombstone()) {
           Assert.assertTrue(results.isEmpty());
         } else {
+          final HashBasedNumberedShardSpec shardSpec = (HashBasedNumberedShardSpec) segment.getShardSpec();
+          Assert.assertEquals(HashPartitionFunction.MURMUR3_32_ABS, shardSpec.getPartitionFunction());
           final int hash = shardSpec.getPartitionFunction().hash(
               HashBasedNumberedShardSpec.serializeGroupKey(
                   getObjectMapper(),

indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/RangePartitionMultiPhaseParallelIndexingTest.java

@@ -455,16 +455,17 @@ private static void assertValuesInRange(List<StringTuple> values, DataSegment se
     Assert.assertTrue(shardSpec.toString(), start != null || end != null);
 
     for (StringTuple value : values) {
+      if (value == null) {
+        Assert.assertNull("null values should be in first partition", start);
+        continue;
+      }
+
       if (start != null) {
         MatcherAssert.assertThat(value.compareTo(start), Matchers.greaterThanOrEqualTo(0));
       }
 
       if (end != null) {
-        if (value == null) {
-          Assert.assertNull("null values should be in first partition", start);
-        } else {
-          MatcherAssert.assertThat(value.compareTo(end), Matchers.lessThan(0));
-        }
+        MatcherAssert.assertThat(value.compareTo(end), Matchers.lessThan(0));
       }
     }
   }

indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java

@@ -271,7 +271,7 @@ public ListenableFuture<TaskStatus> run(Task task)
               try {
                 synchronized (knownTasks) {
                   final TestTaskRunnerWorkItem item2 = knownTasks.get(task.getId());
-                  if (item2.getState() == RunnerTaskState.PENDING) {
+                  if (item2 != null && item2.getState() == RunnerTaskState.PENDING) {
                     knownTasks.put(task.getId(), item2.withState(RunnerTaskState.RUNNING));
                   }
                 }
@@ -282,7 +282,9 @@ public ListenableFuture<TaskStatus> run(Task task)
                         final TestTaskRunnerWorkItem item2;
                         synchronized (knownTasks) {
                           item2 = knownTasks.get(task.getId());
-                          knownTasks.put(task.getId(), item2.withState(RunnerTaskState.NONE));
+                          if (item2 != null) {
+                            knownTasks.put(task.getId(), item2.withState(RunnerTaskState.NONE));
+                          }
                         }
                         if (item2 != null) {
                           item2.setResult(TaskStatus.success(task.getId()));
@@ -494,4 +496,3 @@ public TestTaskRunnerWorkItem withState(final RunnerTaskState newState)
     }
   }
 }
-

processing/src/main/java/org/apache/druid/hll/HyperLogLogCollector.java

@@ -562,32 +562,31 @@ public boolean equals(Object o)
       return false;
     }
 
-    ByteBuffer otherBuffer = ((HyperLogLogCollector) o).storageBuffer;
+    final ByteBuffer otherBuffer = ((HyperLogLogCollector) o).storageBuffer;
 
-    if (storageBuffer != null ? false : otherBuffer != null) {
-      return false;
-    }
-
-    if (storageBuffer == null && otherBuffer == null) {
-      return true;
+    if (storageBuffer == null || otherBuffer == null) {
+      return storageBuffer == otherBuffer;
     }
 
     final ByteBuffer denseStorageBuffer;
     if (storageBuffer.remaining() != getNumBytesForDenseStorage()) {
-      HyperLogLogCollector denseCollector = HyperLogLogCollector.makeCollector(storageBuffer.duplicate());
+      final HyperLogLogCollector denseCollector = HyperLogLogCollector.makeCollector(storageBuffer.duplicate());
       denseCollector.convertToDenseStorage();
       denseStorageBuffer = denseCollector.storageBuffer;
     } else {
       denseStorageBuffer = storageBuffer;
     }
 
+    final ByteBuffer denseOtherBuffer;
     if (otherBuffer.remaining() != getNumBytesForDenseStorage()) {
-      HyperLogLogCollector otherCollector = HyperLogLogCollector.makeCollector(otherBuffer.duplicate());
+      final HyperLogLogCollector otherCollector = HyperLogLogCollector.makeCollector(otherBuffer.duplicate());
       otherCollector.convertToDenseStorage();
-      otherBuffer = otherCollector.storageBuffer;
+      denseOtherBuffer = otherCollector.storageBuffer;
+    } else {
+      denseOtherBuffer = otherBuffer;
     }
 
-    return denseStorageBuffer.equals(otherBuffer);
+    return denseStorageBuffer.equals(denseOtherBuffer);
   }
 
   @Override

processing/src/main/java/org/apache/druid/jackson/DruidDefaultSerializersModule.java

@@ -40,6 +40,7 @@
 
 import java.io.IOException;
 import java.nio.ByteOrder;
+import java.util.Objects;
 
 /**
  *
@@ -152,8 +153,8 @@ public void serialize(Yielder yielder, final JsonGenerator jgen, SerializerProvi
                 } else {
                   final Class<?> clazz = o.getClass();
 
-                  if (serializerClass != clazz) {
-                    serializer = JacksonUtils.getSerializer(provider, clazz);
+                  if (serializer == null || serializerClass != clazz) {
+                    serializer = Objects.requireNonNull(JacksonUtils.getSerializer(provider, clazz));
                     serializerClass = clazz;
                   }
 

processing/src/main/java/org/apache/druid/java/util/common/logger/Logger.java

@@ -295,7 +295,7 @@ private void logException(BiConsumer<String, Throwable> fn, Throwable t, String
 
       fn.accept(message, t);
     } else {
-      if (message.isEmpty()) {
+      if (message == null || message.isEmpty()) {
         fn.accept(t.toString(), null);
       } else {
         fn.accept(StringUtils.nonStrictFormat("%s (%s)", message, t.toString()), null);

processing/src/main/java/org/apache/druid/math/expr/Function.java

@@ -3268,13 +3268,15 @@ public String name()
     @Override
     public ExprEval apply(List<Expr> args, Expr.ObjectBinding bindings)
     {
-      Long left = args.get(0).eval(bindings).asLong();
-      Long right = args.get(1).eval(bindings).asLong();
-      DateTimeZone timeZone = DateTimes.inferTzFromString(args.get(2).eval(bindings).asString());
+      final ExprEval<?> leftEval = args.get(0).eval(bindings);
+      final ExprEval<?> rightEval = args.get(1).eval(bindings);
+      final DateTimeZone timeZone = DateTimes.inferTzFromString(args.get(2).eval(bindings).asString());
 
-      if (left == null || right == null) {
+      if (leftEval.isNumericNull() || rightEval.isNumericNull()) {
         return ExprEval.ofLong(null);
       } else {
+        final long left = leftEval.asLong();
+        final long right = rightEval.asLong();
         return ExprEval.of(DateTimes.subMonths(right, left, timeZone));
       }
 

processing/src/main/java/org/apache/druid/query/ordering/StringComparators.java

@@ -69,6 +69,7 @@ public int compare(String s, String s2)
       // Avoid comparisons for equal references
       // Assuming we mostly compare different strings, checking s.equals(s2) will only make the comparison slower.
       //noinspection StringEquality
+      // codeql[java/reference-equality-on-strings]
       if (s == s2) {
         return 0;
       }
@@ -312,6 +313,7 @@ public int compare(String s, String s2)
     {
       // Optimization
       //noinspection StringEquality
+      // codeql[java/reference-equality-on-strings]
       if (s == s2) {
         return 0;
       }
@@ -374,6 +376,7 @@ public int compare(String o1, String o2)
       // return if o1 and o2 are the same object
       // Assuming we mostly compare different strings, checking o1.equals(o2) will only make the comparison slower.
       //noinspection StringEquality
+      // codeql[java/reference-equality-on-strings]
       if (o1 == o2) {
         return 0;
       }
@@ -450,7 +453,9 @@ public static class VersionComparator extends StringComparator
     @Override
     public int compare(String o1, String o2)
     {
+      // Reference equality is an intentional fast path; value comparison follows below.
       //noinspection StringEquality
+      // codeql[java/reference-equality-on-strings]
       if (o1 == o2) {
         return 0;
       }

processing/src/main/java/org/apache/druid/query/rowsandcols/LazilyDecoratedRowsAndColumns.java

@@ -276,21 +276,15 @@ private Pair<byte[], RowSignature> materializeCursorFactory(CursorFactory cursor
           sortColumns
       );
 
-      final FrameWriter writer = frameWriterFactory.newFrameWriter(columnSelectorFactory);
-      for (; !cursor.isDoneOrInterrupted() && remainingRowsToSkip > 0; remainingRowsToSkip--) {
-        cursor.advance();
-      }
-      for (; !cursor.isDoneOrInterrupted() && remainingRowsToFetch > 0; remainingRowsToFetch--) {
-        writer.addSelection();
-        cursor.advance();
-      }
+      try (final FrameWriter writer = frameWriterFactory.newFrameWriter(columnSelectorFactory)) {
+        for (; !cursor.isDoneOrInterrupted() && remainingRowsToSkip > 0; remainingRowsToSkip--) {
+          cursor.advance();
+        }
+        for (; !cursor.isDoneOrInterrupted() && remainingRowsToFetch > 0; remainingRowsToFetch--) {
+          writer.addSelection();
+          cursor.advance();
+        }
 
-      if (writer == null) {
-        // This means that the accumulate was never called, which can only happen if we didn't have any cursors.
-        // We would only have zero cursors if we essentially didn't match anything, meaning that our RowsAndColumns
-        // should be completely empty.
-        return null;
-      } else {
         final byte[] bytes = writer.toByteArray();
         return Pair.of(bytes, siggy.get());
       }
@@ -383,26 +377,28 @@ private Pair<byte[], RowSignature> naiveMaterialize(RowsAndColumns rac)
     long remainingRowsToSkip = limit.getOffset();
     long remainingRowsToFetch = limit.getLimitOrMax();
 
-    final FrameWriter frameWriter = FrameWriters.makeColumnBasedFrameWriterFactory(
-        memFactory,
-        sigBob.build(),
-        Collections.emptyList()
-    ).newFrameWriter(selectorFactory);
-
-    rowId.set(0);
-    for (; rowId.get() < numRows && remainingRowsToFetch > 0; rowId.incrementAndGet()) {
-      final int theId = rowId.get();
-      if (rowsToSkip != null && rowsToSkip.get(theId)) {
-        continue;
-      }
-      if (remainingRowsToSkip > 0) {
-        remainingRowsToSkip--;
-        continue;
+    try (
+        final FrameWriter frameWriter = FrameWriters.makeColumnBasedFrameWriterFactory(
+            memFactory,
+            sigBob.build(),
+            Collections.emptyList()
+        ).newFrameWriter(selectorFactory)
+    ) {
+      rowId.set(0);
+      for (; rowId.get() < numRows && remainingRowsToFetch > 0; rowId.incrementAndGet()) {
+        final int theId = rowId.get();
+        if (rowsToSkip != null && rowsToSkip.get(theId)) {
+          continue;
+        }
+        if (remainingRowsToSkip > 0) {
+          remainingRowsToSkip--;
+          continue;
+        }
+        remainingRowsToFetch--;
+        frameWriter.addSelection();
       }
-      remainingRowsToFetch--;
-      frameWriter.addSelection();
-    }
 
-    return Pair.of(frameWriter.toByteArray(), sigBob.build());
+      return Pair.of(frameWriter.toByteArray(), sigBob.build());
+    }
   }
 }

processing/src/main/java/org/apache/druid/segment/generator/ColumnValueGenerator.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.segment.generator;
 
+import com.google.common.base.Preconditions;
 import org.apache.commons.math3.distribution.AbstractIntegerDistribution;
 import org.apache.commons.math3.distribution.AbstractRealDistribution;
 import org.apache.commons.math3.distribution.EnumeratedDistribution;
@@ -207,8 +208,18 @@ private void initDistribution()
         distribution = new UniformIntegerDistribution(schema.getStartInt(), schema.getEndInt());
         break;
       case ENUMERATED:
-        for (int i = 0; i < enumeratedValues.size(); i++) {
-          probabilities.add(new Pair<>(enumeratedValues.get(i), enumeratedProbabilities.get(i)));
+        final List<Object> nonNullEnumeratedValues =
+            Preconditions.checkNotNull(enumeratedValues, "enumeratedValues");
+        final List<Double> nonNullEnumeratedProbabilities =
+            Preconditions.checkNotNull(enumeratedProbabilities, "enumeratedProbabilities");
+        Preconditions.checkArgument(
+            nonNullEnumeratedValues.size() == nonNullEnumeratedProbabilities.size(),
+            "enumeratedValues size[%s] must match enumeratedProbabilities size[%s]",
+            nonNullEnumeratedValues.size(),
+            nonNullEnumeratedProbabilities.size()
+        );
+        for (int i = 0; i < nonNullEnumeratedValues.size(); i++) {
+          probabilities.add(new Pair<>(nonNullEnumeratedValues.get(i), nonNullEnumeratedProbabilities.get(i)));
         }
         distribution = new EnumeratedTreeDistribution<>(probabilities);
         break;

processing/src/test/java/org/apache/druid/java/util/common/parsers/JSONFlattenerMakerTest.java

@@ -67,11 +67,11 @@ public void testNumbers() throws JsonProcessingException
   {
     JsonNode node;
     Object result;
-    Integer i1 = 123;
+    final int i1 = 123;
     node = OBJECT_MAPPER.readTree(OBJECT_MAPPER.writeValueAsString(i1));
     Assertions.assertTrue(node.isInt());
     result = FLATTENER_MAKER.finalizeConversionForMap(node);
-    Assertions.assertEquals(i1.longValue(), result);
+    Assertions.assertEquals((long) i1, result);
 
     Long l1 = 1L + Integer.MAX_VALUE;
     node = OBJECT_MAPPER.readTree(OBJECT_MAPPER.writeValueAsString(l1));

processing/src/test/java/org/apache/druid/query/aggregation/firstlast/first/StringFirstBufferAggregatorTest.java

@@ -48,7 +48,7 @@ public void testBufferAggregate()
   {
     final long[] timestamps = {1526724600L, 1526724700L, 1526724800L, 1526725900L, 1526725000L};
     final String[] strings = {"AAAA", "BBBB", "CCCC", "DDDD", "EEEE"};
-    Integer maxStringBytes = 1024;
+    final int maxStringBytes = 1024;
 
     TestLongColumnSelector longColumnSelector = new TestLongColumnSelector(timestamps);
     TestObjectColumnSelector<String> objectColumnSelector = new TestObjectColumnSelector<>(strings);
@@ -85,7 +85,7 @@ public void testBufferAggregateWithFoldCheck()
   {
     final long[] timestamps = {1526724600L, 1526724700L, 1526724800L, 1526725900L, 1526725000L};
     final String[] strings = {"AAAA", "BBBB", "CCCC", "DDDD", "EEEE"};
-    Integer maxStringBytes = 1024;
+    final int maxStringBytes = 1024;
 
     TestLongColumnSelector longColumnSelector = new TestLongColumnSelector(timestamps);
     TestObjectColumnSelector<String> objectColumnSelector = new TestObjectColumnSelector<>(strings);
@@ -123,7 +123,7 @@ public void testNullBufferAggregate()
 
     final long[] timestamps = {2222L, 1111L, 3333L, 4444L, 5555L};
     final String[] strings = {null, "AAAA", "BBBB", "DDDD", "EEEE"};
-    Integer maxStringBytes = 1024;
+    final int maxStringBytes = 1024;
 
     TestLongColumnSelector longColumnSelector = new TestLongColumnSelector(timestamps);
     TestObjectColumnSelector<String> objectColumnSelector = new TestObjectColumnSelector<>(strings);
@@ -162,7 +162,7 @@ public void testNoStringValue()
 
     final long[] timestamps = {1526724000L, 1526724600L};
     final Double[] doubles = {null, 2.00};
-    Integer maxStringBytes = 1024;
+    final int maxStringBytes = 1024;
 
     TestLongColumnSelector longColumnSelector = new TestLongColumnSelector(timestamps);
     TestObjectColumnSelector<Double> objectColumnSelector = new TestObjectColumnSelector<>(doubles);