Marks end criteria reached for the segment if the Index cannot consume more rows #14479

noob-se7en · 2024-11-18T11:04:56Z

Problem: Mutable indexes fails to convert to immutable indexes because implementation of immutable index varies (Like some immutable indexes have size limit or can only handle cardinality upto Integer.MAX, etc).
This can cause Runtime Exception when consuming segment is trying to transition to online.

Example: Num of values for a multi valued column exceeds Integer.MAX_VALUE and overflows resulting in RuntimeException during indexing in forward index. Or Immutable fwd index can only hold 2GB data whereas mutable fwd index is unbounded resulting in RuntimeException when immutable index is built.

Proposed Solution: [Short Term] Stop segment consumption if any problematic index cannot consume more rows.

codecov-commenter · 2024-11-18T11:47:08Z

Codecov Report

Attention: Patch coverage is 88.88889% with 2 lines in your changes missing coverage. Please review.

Project coverage is 64.05%. Comparing base (59551e4) to head (eb3662f).
Report is 1430 commits behind head on master.

Files with missing lines	Patch %	Lines
...e/impl/forward/FixedByteMVMutableForwardIndex.java	66.66%	0 Missing and 1 partial ⚠️
.../pinot/segment/spi/index/mutable/MutableIndex.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #14479      +/-   ##
============================================
+ Coverage     61.75%   64.05%   +2.30%     
- Complexity      207     1574    +1367     
============================================
  Files          2436     2687     +251     
  Lines        133233   147699   +14466     
  Branches      20636    22634    +1998     
============================================
+ Hits          82274    94614   +12340     
- Misses        44911    46146    +1235     
- Partials       6048     6939     +891

Flag	Coverage Δ
custom-integration1	`100.00% <ø> (+99.99%)`	⬆️
integration	`100.00% <ø> (+99.99%)`	⬆️
integration1	`100.00% <ø> (+99.99%)`	⬆️
integration2	`0.00% <ø> (ø)`
java-11	`64.02% <88.88%> (+2.31%)`	⬆️
java-21	`63.94% <88.88%> (+2.31%)`	⬆️
skip-bytebuffers-false	`64.05% <88.88%> (+2.31%)`	⬆️
skip-bytebuffers-true	`63.90% <88.88%> (+36.17%)`	⬆️
temurin	`64.05% <88.88%> (+2.30%)`	⬆️
unittests	`64.05% <88.88%> (+2.30%)`	⬆️
unittests1	`56.16% <38.88%> (+9.26%)`	⬆️
unittests2	`34.58% <50.00%> (+6.85%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Jackie-Jiang · 2024-11-18T19:47:56Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

@@ -362,6 +363,13 @@ private boolean endCriteriaReached() {
              _numRowsConsumed, _numRowsIndexed);
          _stopReason = SegmentCompletionProtocol.REASON_FORCE_COMMIT_MESSAGE_RECEIVED;
          return true;
+        } else if (_thresholdForNumOfColValuesEnabled && _realtimeSegment.isNumOfColValuesAboveThreshold()) {


Nice, so it is fairly easy to stop consumption and commit

Jackie-Jiang · 2024-11-18T19:48:48Z

pinot-common/src/main/java/org/apache/pinot/common/protocols/SegmentCompletionProtocol.java

@@ -149,6 +149,7 @@ public enum ControllerResponseStatus {
  public static final String REASON_END_OF_PARTITION_GROUP = "endOfPartitionGroup";
  // Stop reason sent by server as force commit message received
  public static final String REASON_FORCE_COMMIT_MESSAGE_RECEIVED = "forceCommitMessageReceived";
+  public static final String REASON_NUM_OF_COL_VALUES_ABOVE_THRESHOLD = "numColValuesAboveThreshold";


This should be always on. We may introduce a config to turn it off if we are not confident about this new logic, but if it is not very complicated we can remove this config

Considering new impln now where we are keeping conservative size estimation check as well, I guess it makes sense to enable this change only behind a flag?

Jackie-Jiang · 2024-11-18T19:50:52Z

...al/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java

+
+        if (_thresholdForNumOfColValuesEnabled) {
+          int prevCount = indexContainer._valuesInfo.getNumValues();
+          long newCount = prevCount + 1L + values.length;


Total values itself is not enough. We should perform a per-index check (add an api to the MutableIndex and let it return if it can take more values).
E.g. for MV forward index, if we get 1B values, but each value takes more than 2 bytes, we will run into the same exception

Mutable index is unbounded right (No code enforced limit)? From the code I see Realtime Mutable index is always created with dictionary (even for MV VarByte col with noDict enabled in config).

Hence from numOfValues we know that size of mutable index is approx: numOfValues * 4 Bytes.

But we are more interested in the size of immutable index since that's where exception is being thrown. However Immutable index can be larger or even smaller than mutable index as implementation is completely diff. So while building mutable index we need to keep some state to estimate the approx size of the immutable version of the index (Like while building mutable fwd index, we need to keep an estimation of bitmap, numBitsPerValue, header size, etc).

We should perform a per-index check (add an api to the MutableIndex and let it return if it can take more values).

So this might be adding too much complexity to every mutable index since now after every row consumption we need to update estimated size of the corresponding immutable index and it tight couples code with immutable index logic.

Essentially what we want to ensure is that whatever accepted in mutable segment won't cause problem when the mutable segment is sealed.
I'm okay if we only limit total values here, but we need to enhance immutable index so that it can hold 2B values regardless of the bit length per value.

hmm. Okay, let me figure out if we can even do this on ingestion side considering we have 5 different implementations of immutable forward index for MV columns each having diff size limit.

We can keep threshold near 500 million for numOfValues which will be good enough for few immutable indexes but not for varByte fwd index having 4GB Limit (2GB wasted). However this will ensure we never encounter numOfValues overflow and index limit size reached exception.

For now we can figure out a conservative value that will never cause over 2GB buffer. Once we support larger index, we can increate this value

Sure. Will pick the task to keep the indexes unbounded post this.

Jackie-Jiang

Mostly good

Jackie-Jiang · 2024-12-03T20:16:40Z

...al/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java

@@ -1265,6 +1281,10 @@ private boolean isAggregateMetricsEnabled() {
    return _recordIdMap != null;
  }

+  public boolean isIndexCapacityThresholdBreached() {


Suggest name it canAddMore() to be consistent with the method in each index

Jackie-Jiang · 2024-12-03T20:20:01Z

pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java

+  public boolean isIndexCapacityThresholdCheckEnabled() {
+    return _indexCapacityThresholdCheckEnabled;
+  }
+
+  public void setIndexCapacityThresholdCheckEnabled(boolean indexCapacityThresholdCheckEnabled) {
+    _indexCapacityThresholdCheckEnabled = indexCapacityThresholdCheckEnabled;
+  }
+


Given the overhead is very low, I'd suggest making this on by default, or simply remove this flag.
We didn't run into this issue often because it is extremely rare to have more than 450M values in a segment. Before this PR, we usually manually reduce the commit threshold to workaround this, which is not ideal, and the user experience is very bad.

Jackie-Jiang · 2024-12-03T20:21:47Z

...rc/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java

+      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
+      Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached());
+
+      Field realtimeSegmentField = RealtimeSegmentDataManager.class.getDeclaredField("_realtimeSegment");


Using reflection is not reliable. Consider adding a package private method canAddMore() into RealtimeSegmentDataManager and we can override it here

Jackie-Jiang · 2024-12-04T22:58:43Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

@@ -700,6 +707,10 @@ private boolean processStreamEvents(MessageBatch messageBatch, long idlePipeSlee
    return prematureExit;
  }

+  boolean canAddMore() {


(minor) Annotate it with @VisibleForTesting

Jackie-Jiang · 2024-12-04T23:00:01Z

...al/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java

+            // Hence, The below check is a temporary measure to avoid such scenarios until immutable index
+            // implementations are changed.
+            if (!mutableIndex.canAddMore()) {
+              _logger.debug(


Log it as info, and capitalize the first letter

since this log floods logs (logs: numColum x numOfIndexes), I update the condition to if (!_indexCapacityThresholdBreached && !mutableIndex.canAddMore()) condition

Jackie-Jiang · 2024-12-04T23:01:39Z

...st/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplTestUtils.java

+        aggregationConfigs, false);
+  }
+
+  public static MutableSegmentImpl createMutableSegmentImpl(Schema schema, Set<String> noDictionaryColumns,


We can revert this since the flag is removed

missed it, my bad.

…e more rows (apache#14479)

noob-se7en added 7 commits November 16, 2024 00:51

initial refactoring for handling threshold logic

f03043a

Adds logic for endCriteriaReached based upon numValues

09d43d7

add logic inside endCriteriaReached

121d23a

renames var

06523b6

renames constant

f47cb08

nit

da8d691

nit

2cff520

noob-se7en changed the title ~~Exits prematurely if Num Of Values per segment column crosses threshold~~ bugfix : Exits prematurely if Num Of Values per segment column crosses threshold Nov 18, 2024

noob-se7en changed the title ~~bugfix : Exits prematurely if Num Of Values per segment column crosses threshold~~ Exits prematurely if Num Of Values per segment column crosses threshold Nov 18, 2024

noob-se7en changed the title ~~Exits prematurely if Num Of Values per segment column crosses threshold~~ Stops consumption if Num Of Values per segment column crosses threshold Nov 18, 2024

noob-se7en changed the title ~~Stops consumption if Num Of Values per segment column crosses threshold~~ Stops consumption if Num Of Values per segment column crosses a threshold Nov 18, 2024

noob-se7en added 5 commits November 18, 2024 18:09

refactors logic to inside of addNewRow

55bf033

remove comments

4549e5c

removes usage of map

36c5f0d

nit

c2f78cb

fixes lint

0acb3f6

Jackie-Jiang reviewed Nov 18, 2024

View reviewed changes

Jackie-Jiang added enhancement ingestion labels Nov 18, 2024

noob-se7en added 3 commits November 19, 2024 15:13

nit

96e8663

refactoring

ea48ba0

nit

58370ca

noob-se7en changed the title ~~Stops consumption if Num Of Values per segment column crosses a threshold~~ [WIP] Stops consumption if Num Of Values per segment column crosses a threshold Nov 20, 2024

noob-se7en added 6 commits November 21, 2024 21:38

Adds test

fbc9c93

Adds test

8722efc

Adds test for realtimesegmentDataManager

a1467b5

lint

27c4443

refactors config

a2b95b6

resolves commit

6c3df28

noob-se7en changed the title ~~Stops consumption if Index cannot consume more rows~~ Marks end criteria reached for the segment if the Index cannot consume more rows Nov 26, 2024

noob-se7en added 5 commits November 26, 2024 16:27

nit

69b7e32

nit

c60f1bd

refactors method name

5511e1f

nit

ffa984d

nit

783f8c0

noob-se7en requested a review from gortiz November 26, 2024 11:20

noob-se7en added 2 commits November 27, 2024 16:27

Fixes lint

f9e4b9c

fixes lint

3e1dde2

Jackie-Jiang reviewed Dec 3, 2024

View reviewed changes

noob-se7en added 6 commits December 4, 2024 23:44

Addresses PR comment

ed32e3d

fix bug

e57d726

removes reflection in test

24243bd

fix test

541a7d9

fixes lint

9f67ce4

fix log

65fcaab

noob-se7en requested a review from Jackie-Jiang December 4, 2024 19:39

noob-se7en added 2 commits December 5, 2024 01:16

move log to debug

e6c1f43

fixes test

dbaaf5c

Jackie-Jiang reviewed Dec 4, 2024

View reviewed changes

noob-se7en added 3 commits December 5, 2024 11:00

Addresses PR comment

d15a3fd

nit

6960b5f

nit

eb3662f

noob-se7en requested a review from Jackie-Jiang December 5, 2024 06:28

Jackie-Jiang approved these changes Dec 5, 2024

View reviewed changes

Jackie-Jiang merged commit 0e1c12f into apache:master Dec 5, 2024
21 checks passed

davecromberge pushed a commit to davecromberge/pinot that referenced this pull request Dec 10, 2024

Marks end criteria reached for the segment if the Index cannot consum…

da9d2cb

…e more rows (apache#14479)

yashmayya pushed a commit to yashmayya/pinot that referenced this pull request Jan 3, 2025

Marks end criteria reached for the segment if the Index cannot consum…

8ba2efb

…e more rows (apache#14479)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Marks end criteria reached for the segment if the Index cannot consume more rows #14479

Marks end criteria reached for the segment if the Index cannot consume more rows #14479

noob-se7en commented Nov 18, 2024 •

edited

Loading

codecov-commenter commented Nov 18, 2024 •

edited

Loading

Jackie-Jiang Nov 18, 2024

Jackie-Jiang Nov 18, 2024

noob-se7en Nov 20, 2024

Jackie-Jiang Nov 18, 2024

noob-se7en Nov 19, 2024

Jackie-Jiang Nov 19, 2024

noob-se7en Nov 19, 2024

noob-se7en Nov 19, 2024 •

edited

Loading

Jackie-Jiang Nov 20, 2024

noob-se7en Nov 20, 2024

Jackie-Jiang left a comment

Jackie-Jiang Dec 3, 2024

Jackie-Jiang Dec 3, 2024

Jackie-Jiang Dec 3, 2024

Jackie-Jiang Dec 4, 2024

Jackie-Jiang Dec 4, 2024

noob-se7en Dec 5, 2024

Jackie-Jiang Dec 4, 2024

noob-se7en Dec 5, 2024

Marks end criteria reached for the segment if the Index cannot consume more rows #14479

Marks end criteria reached for the segment if the Index cannot consume more rows #14479

Conversation

noob-se7en commented Nov 18, 2024 • edited Loading

codecov-commenter commented Nov 18, 2024 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

noob-se7en Nov 19, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

noob-se7en commented Nov 18, 2024 •

edited

Loading

codecov-commenter commented Nov 18, 2024 •

edited

Loading

noob-se7en Nov 19, 2024 •

edited

Loading