fix: address CodeQL control-flow and API warnings

extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitSequenceNumber.java

@@ -25,6 +25,8 @@
 
 // OrderedSequenceNumber.equals() should be used instead.
 @SuppressWarnings("ComparableImplementedButEqualsNotOverridden")
+// Every Rabbit sequence number is inclusive, so inherited equality and value-only ordering are consistent.
+// codeql[java/inconsistent-compareto-and-equals]
 public class RabbitSequenceNumber extends OrderedSequenceNumber<Long>
 {
   private RabbitSequenceNumber(Long sequenceNumber)

extensions-core/druid-bloom-filter/src/main/java/org/apache/druid/query/expressions/BloomFilterExpressions.java

@@ -228,6 +228,9 @@ public ExprEval eval(final ObjectBinding bindings)
                 matches = filter.testLong(longVal);
               }
               break;
+            case ARRAY:
+            case COMPLEX:
+              break;
           }
 
           return ExprEval.ofLongBoolean(matches);
@@ -293,6 +296,9 @@ public ExprEval eval(final ObjectBinding bindings)
                 matches = filter.testLong(longVal);
               }
               break;
+            case ARRAY:
+            case COMPLEX:
+              break;
           }
 
           return ExprEval.ofLongBoolean(matches);

extensions-core/ec2-extensions/src/test/java/org/apache/druid/indexing/overlord/autoscaling/ec2/EC2AutoScalerTest.java

@@ -140,7 +140,6 @@ public void testIptoIdLookup()
     );
 
     final int n = 150;
-    Assert.assertTrue(n <= 2 * EC2AutoScaler.MAX_AWS_FILTER_VALUES);
 
     List<String> ips = Lists.transform(
         ContiguousSet.create(Range.closedOpen(0, n), DiscreteDomain.integers()).asList(),
@@ -193,7 +192,6 @@ public void testIdToIpLookup()
     );
 
     final int n = 150;
-    Assert.assertTrue(n <= 2 * EC2AutoScaler.MAX_AWS_FILTER_VALUES);
 
     List<String> ids = Lists.transform(
         ContiguousSet.create(Range.closedOpen(0, n), DiscreteDomain.integers()).asList(),

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

@@ -747,17 +747,15 @@ protected int ruleCombineBins(
 
     // if there are values below the lower limit, fill in array position 1
     // else array position 0
-    while (j != leftBinCount || k != rightBinCount) {
+    if (j != leftBinCount || k != rightBinCount) {
       if (j != leftBinCount && (k == rightBinCount || leftPositions[j] < rightPositions[k])) {
         mergedPositions[pos] = leftPositions[j];
         mergedBins[pos] = leftBins[j];
         ++j;
-        break;
       } else {
         mergedPositions[pos] = rightPositions[k];
         mergedBins[pos] = rightBins[k];
         ++k;
-        break;
       }
     }
 

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

@@ -30,14 +30,17 @@
 import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
 import org.apache.druid.java.util.common.IAE;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.utils.CollectionUtils;
 import org.apache.kafka.common.TopicPartition;
 
 import javax.annotation.Nullable;
 import java.util.Comparator;
 import java.util.Map;
 
-public class KafkaDataSourceMetadata extends SeekableStreamDataSourceMetadata<KafkaTopicPartition, Long> implements Comparable<KafkaDataSourceMetadata>
+// Natural ordering intentionally compares offsets only; inherited equality also includes the non-ordering stream config.
+// codeql[java/inconsistent-compareto-and-equals]
+public class KafkaDataSourceMetadata extends SeekableStreamDataSourceMetadata<KafkaTopicPartition, Long>
+    implements Comparable<KafkaDataSourceMetadata>
 {
   private static final Logger LOGGER = new Logger(KafkaDataSourceMetadata.class);
 

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

@@ -25,6 +25,8 @@
 
 // OrderedSequenceNumber.equals() should be used instead.
 @SuppressWarnings("ComparableImplementedButEqualsNotOverridden")
+// Every Kafka sequence number is inclusive, so inherited equality and value-only ordering are consistent.
+// codeql[java/inconsistent-compareto-and-equals]
 public class KafkaSequenceNumber extends OrderedSequenceNumber<Long>
 {
   private KafkaSequenceNumber(Long sequenceNumber)

extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisSequenceNumber.java

@@ -26,6 +26,8 @@
 
 // OrderedSequenceNumber.equals() should be used instead.
 @SuppressWarnings("ComparableImplementedButEqualsNotOverridden")
+// Ordering intentionally groups equivalent unread/end markers while inherited equality preserves their identities.
+// codeql[java/inconsistent-compareto-and-equals]
 public class KinesisSequenceNumber extends OrderedSequenceNumber<String>
 {
   /**

indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java

@@ -39,6 +39,8 @@ public class Tasks
   public static final int DEFAULT_EMBEDDED_KILL_TASK_PRIORITY = 25;
 
   static {
+    // Keep the independently defined indexing and compaction defaults aligned when either constant changes.
+    // codeql[java/constant-comparison]
     Verify.verify(DEFAULT_MERGE_TASK_PRIORITY == DataSourceCompactionConfig.DEFAULT_COMPACTION_TASK_PRIORITY);
   }
 

indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerTest.java

@@ -1118,6 +1118,8 @@ protected OrderedSequenceNumber createSequenceNumber(Object sequenceNumber)
       if (sequenceNumber == null) {
         return null;
       }
+      // Offset ordering intentionally excludes boundary exclusivity, which value equality includes.
+      // codeql[java/inconsistent-compareto-and-equals]
       return new OrderedSequenceNumber<>(sequenceNumber.toString(), false)
       {
         @Override

indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java

@@ -3404,6 +3404,8 @@ public SeekableStreamDataSourceMetadata<String, String> createDataSourceMetaData
     @Override
     protected OrderedSequenceNumber<String> makeSequenceNumber(String seq, boolean isExclusive)
     {
+      // Offset ordering intentionally excludes boundary exclusivity, which value equality includes.
+      // codeql[java/inconsistent-compareto-and-equals]
       return new OrderedSequenceNumber<>(seq, isExclusive)
       {
         @Override

indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java

@@ -215,6 +215,8 @@ public SeekableStreamDataSourceMetadata<String, String> createDataSourceMetaData
     @Override
     protected OrderedSequenceNumber<String> makeSequenceNumber(String seq, boolean isExclusive)
     {
+      // Offset ordering intentionally excludes boundary exclusivity, which value equality includes.
+      // codeql[java/inconsistent-compareto-and-equals]
       return new OrderedSequenceNumber<>(seq, isExclusive)
       {
         @Override

multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerHolder.java

@@ -348,6 +348,9 @@ private synchronized void updateStateOnQueryComplete(final MSQTaskReportPayload
       case FAILED:
         state = State.FAILED;
         break;
+
+      case RUNNING:
+        break;
     }
   }
 

multi-stage-query/src/main/java/org/apache/druid/msq/indexing/processor/SegmentGeneratorFrameProcessor.java

@@ -218,6 +218,8 @@ private void addFrame(final Frame frame)
     }
   }
 
+  // Input rows are never ordered during indexing, and compareTo always rejects attempts to do so.
+  // codeql[java/inconsistent-compareto-and-equals]
   private class MSQInputRow implements InputRow
   {
     private final Object[] backingArray;

multi-stage-query/src/test/java/org/apache/druid/msq/kernel/controller/BaseControllerQueryKernelTest.java

@@ -210,6 +210,9 @@ public ControllerQueryKernelTester setupStage(
         case FAILED:
           controllerQueryKernel.failStage(stageId);
           break;
+
+        case RETRYING:
+          throw new IAE("Cannot initialize a stage directly in the retrying phase");
       }
       if (!recursiveCall) {
         setupStages.add(stageNumber);

processing/src/main/java/org/apache/druid/extendedset/intset/ConciseSet.java

@@ -657,28 +657,32 @@ private ConciseSet performOperation(ConciseSet other, Operator operator)
         if (!otherItr.isLiteral) {
           int minCount = Math.min(thisItr.count, otherItr.count);
           res.appendFill(minCount, operator.combineLiterals(thisItr.word, otherItr.word));
-          //noinspection NonShortCircuitBooleanExpression
+          // Both iterators must advance before testing whether either is exhausted.
+          // codeql[java/non-short-circuit-evaluation]
           if (!thisItr.prepareNext(minCount) | /* NOT || */ !otherItr.prepareNext(minCount)) {
             break;
           }
         } else {
           res.appendLiteral(operator.combineLiterals(thisItr.toLiteral(), otherItr.word));
           thisItr.word--;
-          //noinspection NonShortCircuitBooleanExpression
+          // Both iterators must advance before testing whether either is exhausted.
+          // codeql[java/non-short-circuit-evaluation]
           if (!thisItr.prepareNext(1) | /* do NOT use "||" */ !otherItr.prepareNext()) {
             break;
           }
         }
       } else if (!otherItr.isLiteral) {
         res.appendLiteral(operator.combineLiterals(thisItr.word, otherItr.toLiteral()));
         otherItr.word--;
-        //noinspection NonShortCircuitBooleanExpression
+        // Both iterators must advance before testing whether either is exhausted.
+        // codeql[java/non-short-circuit-evaluation]
         if (!thisItr.prepareNext() | /* do NOT use  "||" */ !otherItr.prepareNext(1)) {
           break;
         }
       } else {
         res.appendLiteral(operator.combineLiterals(thisItr.word, otherItr.word));
-        //noinspection NonShortCircuitBooleanExpression
+        // Both iterators must advance before testing whether either is exhausted.
+        // codeql[java/non-short-circuit-evaluation]
         if (!thisItr.prepareNext() | /* do NOT use  "||" */ !otherItr.prepareNext()) {
           break;
         }

processing/src/main/java/org/apache/druid/frame/field/FieldWriters.java

@@ -93,6 +93,9 @@ public static FieldWriter create(
             return makeFloatArrayWriter(columnSelectorFactory, columnName, frameType);
           case DOUBLE:
             return makeDoubleArrayWriter(columnSelectorFactory, columnName, frameType);
+          case ARRAY:
+          case COMPLEX:
+            throw new UnsupportedColumnTypeException(columnName, columnType);
         }
       default:
         throw new UnsupportedColumnTypeException(columnName, columnType);

processing/src/main/java/org/apache/druid/java/util/common/guava/FunctionalIterator.java

@@ -59,6 +59,8 @@ public T next()
   @Override
   public void remove()
   {
+    // Iterator.remove is optional; preserve the removal capability of the wrapped iterator.
+    // codeql[java/iterator-remove-failure]
     delegate.remove();
   }
 

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

@@ -103,6 +103,9 @@ public static ExpressionType asArrayType(@Nullable ExpressionType elementType)
           return LONG_ARRAY;
         case DOUBLE:
           return DOUBLE_ARRAY;
+        case ARRAY:
+        case COMPLEX:
+          return elementType;
       }
     }
     return elementType;
@@ -138,6 +141,9 @@ public static ExpressionType fromColumnTypeStrict(@Nullable TypeSignature<ValueT
             return DOUBLE_ARRAY;
           case STRING:
             return STRING_ARRAY;
+          case ARRAY:
+          case COMPLEX:
+            break;
         }
         return ExpressionTypeFactory.getInstance().ofArray(fromColumnTypeStrict(valueType.getElementType()));
       case COMPLEX:
@@ -175,6 +181,9 @@ public static ExpressionType fromColumnType(@Nullable TypeSignature<ValueType> v
             return DOUBLE_ARRAY;
           case STRING:
             return STRING_ARRAY;
+          case ARRAY:
+          case COMPLEX:
+            break;
         }
         return ExpressionTypeFactory.getInstance().ofArray(fromColumnType(valueType.getElementType()));
       case COMPLEX:

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

@@ -79,6 +79,9 @@ public ExpressionType ofArray(ExpressionType elementType)
           return ExpressionType.DOUBLE_ARRAY;
         case LONG:
           return ExpressionType.LONG_ARRAY;
+        case ARRAY:
+        case COMPLEX:
+          break;
       }
     }
     return INTERNER.intern(new ExpressionType(ExprType.ARRAY, null, elementType));

processing/src/main/java/org/apache/druid/query/PrioritizedExecutorService.java

@@ -213,6 +213,8 @@ public int getActiveTasks()
   }
 }
 
+// Tasks with equal scheduling keys remain distinct futures, so identity equality is intentional.
+// codeql[java/inconsistent-compareto-and-equals]
 class PrioritizedListenableFutureTask<V> implements RunnableFuture<V>,
     ListenableFuture<V>,
     PrioritizedRunnable,

processing/src/main/java/org/apache/druid/query/filter/InDimFilter.java

@@ -141,7 +141,7 @@ public InDimFilter(String dimension, Set<String> values)
   {
     this(
         dimension,
-        values instanceof ValuesSet ? (ValuesSet) values : new ValuesSet(values),
+        values instanceof ValuesSet valuesSet ? valuesSet : new ValuesSet(values),
         null,
         null,
         null
@@ -161,7 +161,7 @@ public InDimFilter(String dimension, Collection<String> values, @Nullable Extrac
   {
     this(
         dimension,
-        values instanceof ValuesSet ? (ValuesSet) values : new ValuesSet(values),
+        values instanceof ValuesSet valuesSet ? valuesSet : new ValuesSet(values),
         extractionFn,
         null,
         null

processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferMinMaxOffsetHeap.java

@@ -308,10 +308,8 @@ private void siftDown(Comparator comparator, int pos)
         int minGcOffset = buf.getInt(minGrandchild * Integer.BYTES);
         int cmp = comparator.compare(minChildOffset, minGcOffset);
         minIndex = (cmp > 0) ? minGrandchild : minChild;
-      } else if (minChild > -1) {
-        minIndex = minChild;
       } else {
-        break;
+        minIndex = minChild;
       }
       if (minIndex == minGrandchild) {
         int offset = buf.getInt(pos * Integer.BYTES);
@@ -337,6 +335,8 @@ private void siftDown(Comparator comparator, int pos)
             }
           }
           minChild = findMinChild(comparator, minIndex);
+        } else {
+          break;
         }
         pos = minIndex;
       } else {

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

@@ -68,7 +68,8 @@ 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
+      // Identity is only a fast path; ORDERING performs the content comparison for distinct strings.
+      // codeql[java/reference-equality-on-strings]
       if (s == s2) {
         return 0;
       }
@@ -311,7 +312,8 @@ public int compare(String s, String s2)
     public int compare(String s, String s2)
     {
       // Optimization
-      //noinspection StringEquality
+      // Identity is only a fast path; ORDERING performs the content comparison for distinct strings.
+      // codeql[java/reference-equality-on-strings]
       if (s == s2) {
         return 0;
       }
@@ -373,7 +375,8 @@ 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
+      // Identity is only a fast path; numeric and lexical comparison handles distinct strings.
+      // codeql[java/reference-equality-on-strings]
       if (o1 == o2) {
         return 0;
       }
@@ -450,7 +453,8 @@ public static class VersionComparator extends StringComparator
     @Override
     public int compare(String o1, String o2)
     {
-      //noinspection StringEquality
+      // Identity is only a fast path; version comparison handles distinct strings.
+      // codeql[java/reference-equality-on-strings]
       if (o1 == o2) {
         return 0;
       }

processing/src/main/java/org/apache/druid/query/topn/types/TopNColumnAggregatesProcessorFactory.java

@@ -74,6 +74,10 @@ public TopNColumnAggregatesProcessor<?> makeColumnSelectorStrategy(
           return new FloatTopNColumnAggregatesProcessor(converter);
         case DOUBLE:
           return new DoubleTopNColumnAggregatesProcessor(converter);
+        case STRING:
+        case ARRAY:
+        case COMPLEX:
+          break;
       }
     }
 

processing/src/main/java/org/apache/druid/segment/data/CompressedBlockReader.java

@@ -68,7 +68,6 @@ public static Supplier<CompressedBlockReader> fromByteBuffer(
     if (versionFromBuffer == VERSION) {
       final CompressionStrategy compression = CompressionStrategy.forId(buffer.get());
       final int blockSize = buffer.getInt();
-      assert CompressedPools.BUFFER_SIZE == blockSize;
       Preconditions.checkState(
           blockSize <= CompressedPools.BUFFER_SIZE,
           "Maximum block size must be less than " + CompressedPools.BUFFER_SIZE

processing/src/main/java/org/apache/druid/segment/data/VSizeColumnarInts.java

@@ -150,6 +150,25 @@ public int compareTo(VSizeColumnarInts o)
     return retVal;
   }
 
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof VSizeColumnarInts)) {
+      return false;
+    }
+    final VSizeColumnarInts that = (VSizeColumnarInts) o;
+    return numBytes == that.numBytes && buffer.equals(that.buffer);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return 31 * numBytes + buffer.hashCode();
+  }
+
   public int getNumBytes()
   {
     return numBytes;

processing/src/main/java/org/apache/druid/segment/filter/ExpressionFilter.java

@@ -175,6 +175,9 @@ public boolean matches(boolean includeUnknown)
               }
 
               return Arrays.stream(result).filter(Objects::nonNull).anyMatch(o -> Evals.asBoolean((double) o));
+            case ARRAY:
+            case COMPLEX:
+              break;
           }
         }
         return eval.asBoolean();