remove size checks from IncrementalIndex to simplify logic (#18219) (#18276)

Co-authored-by: Clint Wylie <cwylie@apache.org>

Author: capistrant

benchmarks/src/test/java/org/apache/druid/benchmark/indexing/OnheapIncrementalIndexBenchmark.java

@@ -51,7 +51,6 @@
 import org.apache.druid.segment.Segment;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexSchema;
-import org.apache.druid.segment.incremental.IndexSizeExceededException;
 import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
 import org.joda.time.Interval;
 import org.junit.Assert;
@@ -253,13 +252,8 @@ public void concurrentAddRead() throws InterruptedException, ExecutionException
           indexExecutor.submit(
               () -> {
                 currentlyRunning.incrementAndGet();
-                try {
-                  for (int i = 0; i < elementsPerAddTask; i++) {
-                    incrementalIndex.add(getLongRow(timestamp + i, 1, DIMENSION_COUNT));
-                  }
-                }
-                catch (IndexSizeExceededException e) {
-                  throw new RuntimeException(e);
+                for (int i = 0; i < elementsPerAddTask; i++) {
+                  incrementalIndex.add(getLongRow(timestamp + i, 1, DIMENSION_COUNT));
                 }
                 currentlyRunning.decrementAndGet();
                 someoneRan.set(true);

indexing-service/src/main/java/org/apache/druid/indexing/overlord/sampler/InputSourceSampler.java

@@ -171,7 +171,7 @@ public SamplerResponse sample(
 
             //keep the index of the row to be added to responseRows for further use
             final int rowIndex = responseRows.size();
-            IncrementalIndexAddResult addResult = index.add(new SamplerInputRow(row, rowIndex), true);
+            IncrementalIndexAddResult addResult = index.add(new SamplerInputRow(row, rowIndex));
             if (addResult.hasParseException()) {
               responseRows.add(new SamplerResponseRow(
                   rawColumns,

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

@@ -27,7 +27,6 @@
 import org.apache.druid.data.input.impl.TimestampSpec;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.segment.incremental.IncrementalIndex;
-import org.apache.druid.segment.incremental.IndexSizeExceededException;
 import org.joda.time.Interval;
 
 import java.util.ArrayList;
@@ -207,12 +206,7 @@ public InputRow get()
   public static void addStreamToIndex(Stream<InputRow> stream, IncrementalIndex index)
   {
     stream.forEachOrdered(row -> {
-      try {
-        index.add(row);
-      }
-      catch (IndexSizeExceededException e) {
-        throw new RuntimeException(e);
-      }
+      index.add(row);
     });
   }
 

processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndex.java

@@ -102,10 +102,10 @@
  * In-memory, row-based data structure used to hold data during ingestion. Realtime tasks query this index using
  * {@link IncrementalIndexCursorFactory}.
  *
- * Concurrency model: {@link #add(InputRow)} and {@link #add(InputRow, boolean)} are not thread-safe, and must be
- * called from a single thread or externally synchronized. However, the methods that support
- * {@link IncrementalIndexCursorFactory} are thread-safe, and may be called concurrently with each other, and with
- * the "add" methods. This concurrency model supports real-time queries of the data in the index.
+ * Concurrency model: {@link #add(InputRow)} is not thread-safe, and must be called from a single thread or externally
+ * synchronized. However, the methods that support {@link IncrementalIndexCursorFactory} are thread-safe, and may be
+ * called concurrently with each other, and with the "add" methods. This concurrency model supports real-time queries
+ * of the data in the index.
  */
 public abstract class IncrementalIndex implements IncrementalIndexRowSelector, ColumnInspector, Iterable<Row>, Closeable
 {
@@ -383,9 +383,8 @@ protected abstract void initAggs(
   // Note: This method does not need to be thread safe.
   protected abstract AddToFactsResult addToFacts(
       IncrementalIndexRow key,
-      InputRowHolder inputRowHolder,
-      boolean skipMaxRowsInMemoryCheck
-  ) throws IndexSizeExceededException;
+      InputRowHolder inputRowHolder
+  );
 
 
   public abstract Iterable<Row> iterableWithPostAggregations(
@@ -453,45 +452,23 @@ public ColumnFormat getColumnFormat(String columnName)
     }
   }
 
-  /**
-   * Adds a new row.  The row might correspond with another row that already exists, in which case this will
-   * update that row instead of inserting a new one.
-   *
-   * Not thread-safe.
-   *
-   * @param row the row of data to add
-   *
-   * @return the number of rows in the data set after adding the InputRow. If any parse failure occurs, a {@link ParseException} is returned in {@link IncrementalIndexAddResult}.
-   *
-   * @throws IndexSizeExceededException this exception is thrown once it reaches max rows limit and skipMaxRowsInMemoryCheck is set to false.
-   */
-  public IncrementalIndexAddResult add(InputRow row) throws IndexSizeExceededException
-  {
-    return add(row, false);
-  }
-
   /**
    * Adds a new row.  The row might correspond with another row that already exists, in which case this will
    * update that row instead of inserting a new one.
    *
    * Not thread-safe.
    *
    * @param row                      the row of data to add
-   * @param skipMaxRowsInMemoryCheck whether or not to skip the check of rows exceeding the max rows or bytes limit
    *
    * @return the number of rows in the data set after adding the InputRow. If any parse failure occurs, a {@link ParseException} is returned in {@link IncrementalIndexAddResult}.
-   *
-   * @throws IndexSizeExceededException this exception is thrown once it reaches max rows limit and skipMaxRowsInMemoryCheck is set to false.
    */
-  public IncrementalIndexAddResult add(InputRow row, boolean skipMaxRowsInMemoryCheck)
-      throws IndexSizeExceededException
+  public IncrementalIndexAddResult add(InputRow row)
   {
     IncrementalIndexRowResult incrementalIndexRowResult = toIncrementalIndexRow(row);
     inputRowHolder.set(row);
     final AddToFactsResult addToFactsResult = addToFacts(
         incrementalIndexRowResult.getIncrementalIndexRow(),
-        inputRowHolder,
-        skipMaxRowsInMemoryCheck
+        inputRowHolder
     );
     updateMaxIngestedTime(row.getTimestamp());
     @Nullable ParseException parseException = getCombinedParseException(

processing/src/main/java/org/apache/druid/segment/incremental/IndexSizeExceededException.java

@@ -224,9 +224,8 @@ protected void initAggs(
   @Override
   protected AddToFactsResult addToFacts(
       IncrementalIndexRow key,
-      InputRowHolder inputRowHolder,
-      boolean skipMaxRowsInMemoryCheck
-  ) throws IndexSizeExceededException
+      InputRowHolder inputRowHolder
+  )
   {
     final List<String> parseExceptionMessages = new ArrayList<>();
     final AtomicLong totalSizeInBytes = getBytesInMemory();
@@ -258,16 +257,6 @@ protected AddToFactsResult addToFacts(
       final int rowIndex = indexIncrement.getAndIncrement();
       aggregators.put(rowIndex, aggs);
 
-      // Last ditch sanity checks
-      if ((numEntries.get() >= maxRowCount || totalSizeInBytes.get() >= maxBytesInMemory)
-          && facts.getPriorIndex(key) == IncrementalIndexRow.EMPTY_ROW_INDEX
-          && !skipMaxRowsInMemoryCheck) {
-        throw new IndexSizeExceededException(
-            "Maximum number of rows [%d] or max size in bytes [%d] reached",
-            maxRowCount,
-            maxBytesInMemory
-        );
-      }
       final int prev = facts.putIfAbsent(key, rowIndex);
       if (IncrementalIndexRow.EMPTY_ROW_INDEX == prev) {
         numEntries.incrementAndGet();

processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java

@@ -30,7 +30,6 @@
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexCursorFactory;
 import org.apache.druid.segment.incremental.IncrementalIndexSchema;
-import org.apache.druid.segment.incremental.IndexSizeExceededException;
 import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
 
 import java.util.List;
@@ -54,13 +53,8 @@ public static CursorFactory toCursorFactory(List<InputRow> inputRows)
         .setMaxRowCount(1000)
         .build();
 
-    try {
-      for (InputRow inputRow : inputRows) {
-        index.add(inputRow);
-      }
-    }
-    catch (IndexSizeExceededException e) {
-      throw new RuntimeException(e);
+    for (InputRow inputRow : inputRows) {
+      index.add(inputRow);
     }
 
     return new IncrementalIndexCursorFactory(index);

processing/src/test/java/org/apache/druid/frame/processor/test/TestFrameProcessorUtils.java

@@ -56,7 +56,6 @@
 import org.apache.druid.segment.column.ValueType;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexSchema;
-import org.apache.druid.segment.incremental.IndexSizeExceededException;
 import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
 import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
 import org.apache.druid.testing.InitializedNullHandlingTest;
@@ -343,12 +342,7 @@ private static QueryableIndex buildIndex(String storeDoubleAsFloat) throws IOExc
 
 
     getStreamOfEvents().forEach(o -> {
-      try {
-        index.add(ROW_PARSER.parseBatch((Map<String, Object>) o).get(0));
-      }
-      catch (IndexSizeExceededException e) {
-        throw new RuntimeException(e);
-      }
+      index.add(ROW_PARSER.parseBatch((Map<String, Object>) o).get(0));
     });
 
     if (oldValue == null) {

processing/src/test/java/org/apache/druid/query/DoubleStorageTest.java

@@ -593,7 +593,7 @@ public static IncrementalIndex createIncrementalIndex(
       Granularity gran,
       int maxRowCount,
       boolean rollup
-  ) throws Exception
+  )
   {
     IncrementalIndex index = new OnheapIncrementalIndex.Builder()
         .setIndexSchema(
@@ -633,7 +633,7 @@ public static IncrementalIndex createIncrementalIndex(
       Granularity gran,
       int maxRowCount,
       boolean rollup
-  ) throws Exception
+  )
   {
     return createIncrementalIndex(
         rows,

processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java

@@ -44,7 +44,6 @@
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexCursorFactory;
 import org.apache.druid.segment.incremental.IncrementalIndexSchema;
-import org.apache.druid.segment.incremental.IndexSizeExceededException;
 import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
 import org.apache.druid.segment.serde.ComplexMetrics;
 import org.apache.druid.testing.InitializedNullHandlingTest;
@@ -68,7 +67,7 @@ public class StringFirstTimeseriesQueryTest extends InitializedNullHandlingTest
   private QueryableIndex queryableIndex;
 
   @Before
-  public void setUp() throws IndexSizeExceededException
+  public void setUp()
   {
     final SerializablePairLongStringComplexMetricSerde serde = new SerializablePairLongStringComplexMetricSerde();
     ComplexMetrics.registerSerde(serde.getTypeName(), serde);