This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new fe892b228a8 Clean up multi-value limit enforcement in mutable segment
(#19431)
fe892b228a8 is described below
commit fe892b228a83437bb144a6c5abdd925c1e3faeda
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Tue Sep 1 23:01:20 2026 -0700
Clean up multi-value limit enforcement in mutable segment (#19431)
---
.../inputformat/clplog/CLPLogRecordExtractor.java | 4 +-
.../indexsegment/mutable/MutableSegmentImpl.java | 101 ++++-----
.../impl/forward/CLPMutableForwardIndex.java | 6 +-
.../forward/FixedByteMVMutableForwardIndex.java | 97 ++++-----
.../segment/index/forward/ForwardIndexType.java | 9 +-
...leSegmentImplNumMultiValuesValidationTest.java} | 238 ++++++++++-----------
.../FixedByteMVMutableForwardIndexTest.java | 66 +++---
.../segment/spi/index/ForwardIndexConfig.java | 5 +
.../mutable/provider/MutableIndexContext.java | 41 ++--
9 files changed, 258 insertions(+), 309 deletions(-)
diff --git
a/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java
b/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java
index d59073a3255..bfbbd08a300 100644
---
a/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java
+++
b/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java
@@ -34,7 +34,7 @@ import java.util.Set;
import javax.annotation.Nullable;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.segment.local.segment.index.forward.ForwardIndexType;
+import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
import org.apache.pinot.spi.data.readers.BaseRecordExtractor;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordExtractorConfig;
@@ -51,7 +51,7 @@ import org.slf4j.LoggerFactory;
/// `_dictionaryVars`, `_encodedVars`. All other fields are extracted as plain
JSON values.
public class CLPLogRecordExtractor extends BaseRecordExtractor<Map<String,
Object>> {
// The maximum number of variables that can be stored in a cell (row of a
single column).
- private static final int MAX_VARIABLES_PER_CELL =
ForwardIndexType.MAX_MULTI_VALUES_PER_ROW;
+ private static final int MAX_VARIABLES_PER_CELL =
ForwardIndexConfig.DEFAULT_MAX_NUM_MULTI_VALUES;
private static final Logger LOGGER =
LoggerFactory.getLogger(CLPLogRecordExtractor.class);
private final ServerMetrics _serverMetrics = ServerMetrics.get();
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
index 6f93fad11e2..ea54e6f9bd1 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
@@ -59,7 +59,6 @@ import
org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentConfig;
import
org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentStatsHistory;
import
org.apache.pinot.segment.local.realtime.impl.dictionary.BaseOffHeapMutableDictionary;
import
org.apache.pinot.segment.local.realtime.impl.dictionary.SameValueMutableDictionary;
-import
org.apache.pinot.segment.local.realtime.impl.forward.FixedByteMVMutableForwardIndex;
import
org.apache.pinot.segment.local.realtime.impl.forward.SameValueMutableForwardIndex;
import
org.apache.pinot.segment.local.realtime.impl.invertedindex.MultiColumnRealtimeLuceneTextIndex;
import
org.apache.pinot.segment.local.realtime.impl.nullvalue.MutableNullValueVector;
@@ -166,7 +165,7 @@ public class MutableSegmentImpl implements MutableSegment {
private final File _consumerDir;
private final Map<String, IndexContainer> _indexContainerMap = new
HashMap<>();
- private final MultiValueRowLimit[] _multiValueRowLimits;
+ private final MultiValueLimit[] _multiValueLimits;
private final IdMap<FixedIntArray> _recordIdMap;
private final int _numKeyColumns;
// Cache the physical (non-virtual) field specs
@@ -311,7 +310,7 @@ public class MutableSegmentImpl implements MutableSegment {
// Initialize for each column
boolean hasColumnWithReuseMutableTextIndex = false;
- List<MultiValueRowLimit> multiValueRowLimits = new ArrayList<>();
+ List<MultiValueLimit> multiValueLimits = new ArrayList<>();
for (FieldSpec fieldSpec : _physicalFieldSpecs) {
String column = fieldSpec.getName();
@@ -330,21 +329,27 @@ public class MutableSegmentImpl implements MutableSegment
{
Optional.ofNullable(config.getIndexConfigByCol().get(column)).orElse(FieldIndexConfigs.EMPTY);
VectorIndexConfig vectorIndexConfig =
indexConfigs.getConfig(StandardIndexes.vector());
boolean isDictionary = !isNoDictionaryColumn(indexConfigs, fieldSpec,
column);
- MutableIndexContext context =
- MutableIndexContext.builder()
- .withFieldSpec(fieldSpec)
- .withMemoryManager(_memoryManager)
- .withDictionary(isDictionary)
- .withCapacity(_capacity)
- .offHeap(_offHeap)
- .withSegmentName(_segmentName)
-
.withEstimatedCardinality(_statsHistory.getEstimatedCardinality(column))
-
.withEstimatedColSize(_statsHistory.getEstimatedAvgColSize(column))
-
.withAvgNumMultiValues(_statsHistory.getEstimatedAvgColSize(column))
- .withMaxNumMultiValuesPerRowOverride(
- vectorIndexConfig.isEnabled() ?
vectorIndexConfig.getVectorDimension() : 0)
- .withConsumerDir(_consumerDir)
- .withFixedLengthBytes(fixedByteSize).build();
+ MutableIndexContext.Builder contextBuilder =
MutableIndexContext.builder()
+ .withFieldSpec(fieldSpec)
+ .withMemoryManager(_memoryManager)
+ .withDictionary(isDictionary)
+ .withCapacity(_capacity)
+ .offHeap(_offHeap)
+ .withSegmentName(_segmentName)
+
.withEstimatedCardinality(_statsHistory.getEstimatedCardinality(column))
+ .withEstimatedColSize(_statsHistory.getEstimatedAvgColSize(column))
+ .withAvgNumMultiValues(config.getAvgNumMultiValues())
+ .withConsumerDir(_consumerDir)
+ .withFixedLengthBytes(fixedByteSize);
+ if (vectorIndexConfig.isEnabled()) {
+ // A vector column holds one value per dimension, which may exceed the
default cap
+
contextBuilder.withMaxNumMultiValues(vectorIndexConfig.getVectorDimension());
+ }
+ MutableIndexContext context = contextBuilder.build();
+
+ if (!fieldSpec.isSingleValueField()) {
+ multiValueLimits.add(new MultiValueLimit(column,
context.getMaxNumMultiValues()));
+ }
// Partition info
PartitionFunction partitionFunction = null;
@@ -411,14 +416,6 @@ public class MutableSegmentImpl implements MutableSegment {
String sourceColumn = columnAggregatorPair.getLeft();
ValueAggregator valueAggregator = columnAggregatorPair.getRight();
- // Capture the row cap from the concrete writer before a SameValue
wrapper hides the type.
- MutableIndex unwrappedForwardIndex =
mutableIndexes.get(StandardIndexes.forward());
- if (!fieldSpec.isSingleValueField()
- && unwrappedForwardIndex instanceof FixedByteMVMutableForwardIndex
fixedByteMVIndex) {
- multiValueRowLimits.add(
- new MultiValueRowLimit(column,
fixedByteMVIndex.getMaxNumberOfMultiValuesPerRow()));
- }
-
// TODO this can be removed after forward index contents no longer
depends on text index configs
// If the raw value is provided, use it for the forward/dictionary index
of this column by wrapping the
// already created MutableIndex with a SameValue implementation. This
optimization can only be done when
@@ -453,7 +450,7 @@ public class MutableSegmentImpl implements MutableSegment {
nullValueVector, sourceColumn, valueAggregator));
}
_hasColumnWithReuseMutableTextIndex = hasColumnWithReuseMutableTextIndex;
- _multiValueRowLimits =
multiValueRowLimits.toArray(MultiValueRowLimit[]::new);
+ _multiValueLimits = multiValueLimits.toArray(new MultiValueLimit[0]);
_partitionDedupMetadataManager = config.getPartitionDedupMetadataManager();
_dedupTimeColumn =
@@ -657,12 +654,12 @@ public class MutableSegmentImpl implements MutableSegment
{
int numDocsIndexed = _numDocsIndexed;
if (isUpsertEnabled()) {
// Validate the incoming row before partial-upsert strategies can copy
or expand oversized MV values.
- validateLengthOfMVColumns(row);
+ validateNumMultiValues(row);
RecordInfo recordInfo = getRecordInfo(row, numDocsIndexed);
GenericRow updatedRow =
_partitionUpsertMetadataManager.updateRecord(row, recordInfo);
if (_isPartialUpsert) {
// Strategies such as APPEND and UNION can produce a merged row that
is larger than the incoming row.
- validateLengthOfMVColumns(updatedRow);
+ validateNumMultiValues(updatedRow);
}
trackMismatchedPartition(mismatchedPartitionIndexContainer,
mismatchedPartition, mismatchedPartitionValue);
@@ -708,7 +705,7 @@ public class MutableSegmentImpl implements MutableSegment {
}
// Validate before dedup or partition tracking so a rejected row cannot
leave metadata state behind.
- validateLengthOfMVColumns(row);
+ validateNumMultiValues(row);
trackMismatchedPartition(mismatchedPartitionIndexContainer,
mismatchedPartition, mismatchedPartitionValue);
if (isDedupEnabled()) {
@@ -824,21 +821,21 @@ public class MutableSegmentImpl implements MutableSegment
{
return new ComparisonColumns(comparisonValues, comparableIndex);
}
- /// @param row
- /// @throws UnsupportedOperationException if the length of an MV column
exceeds the maximum number of values allowed
- /// in a single row of the forward index
- private void validateLengthOfMVColumns(GenericRow row)
- throws UnsupportedOperationException {
- for (int i = 0; i < _multiValueRowLimits.length; i++) {
- MultiValueRowLimit rowLimit = _multiValueRowLimits[i];
- Object value = row.getValue(rowLimit._column);
- if (value == null) {
- continue;
- }
- Object[] values = (Object[]) value;
- if (values.length > rowLimit._maxNumberOfMultiValuesPerRow) {
- throw new UnsupportedOperationException("MV column '" +
rowLimit._column + "' has " + values.length
- + " values, exceeding the maximum of " +
rowLimit._maxNumberOfMultiValuesPerRow + " values per row.");
+ /// Validates that no multi-value column in the row holds more values than
its forward index can store in a single
+ /// multi-value entry. Must run before any column of the row is indexed so
that a rejected row leaves no partial
+ /// state behind.
+ ///
+ /// @throws IllegalStateException if a multi-value column exceeds its
maximum number of values
+ private void validateNumMultiValues(GenericRow row) {
+ for (MultiValueLimit limit : _multiValueLimits) {
+ Object value = row.getValue(limit._column);
+ if (value != null) {
+ int numValues = ((Object[]) value).length;
+ if (numValues > limit._maxNumMultiValues) {
+ throw new IllegalStateException(
+ String.format("Number of values: %d in MV column: %s exceeds the
maximum allowed: %d", numValues,
+ limit._column, limit._maxNumMultiValues));
+ }
}
}
}
@@ -998,9 +995,6 @@ public class MutableSegmentImpl implements MutableSegment {
MutableIndex mutableIndex = indexEntry.getValue();
mutableIndex.add(values, dictIds, docId);
updateIndexCapacityThresholdBreached(mutableIndex,
indexEntry.getKey(), column);
- } catch (IllegalArgumentException e) {
- // Row-limit violations must fail the document. Other index errors
stay fail-soft (#16316).
- throw e;
} catch (Exception e) {
recordIndexingError(indexEntry.getKey(), e);
}
@@ -1693,15 +1687,8 @@ public class MutableSegmentImpl implements
MutableSegment {
}
}
- /// Immutable fixed-byte MV column descriptor used by the per-record length
validation hot path.
- private static final class MultiValueRowLimit {
- private final String _column;
- private final int _maxNumberOfMultiValuesPerRow;
-
- private MultiValueRowLimit(String column, int
maxNumberOfMultiValuesPerRow) {
- _column = column;
- _maxNumberOfMultiValuesPerRow = maxNumberOfMultiValuesPerRow;
- }
+ /// Per-column cap on the number of values in a multi-value entry, as
configured on the mutable index context.
+ private record MultiValueLimit(String _column, int _maxNumMultiValues) {
}
private class IndexContainer implements Closeable {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndex.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndex.java
index 15f25d1220c..2341a6e6471 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndex.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndex.java
@@ -27,7 +27,7 @@ import java.io.IOException;
import
org.apache.pinot.segment.local.realtime.impl.dictionary.StringOffHeapMutableDictionary;
import
org.apache.pinot.segment.local.segment.creator.impl.stats.CLPStatsProvider;
import
org.apache.pinot.segment.local.segment.creator.impl.stats.StringColumnPreIndexStatsCollector;
-import org.apache.pinot.segment.local.segment.index.forward.ForwardIndexType;
+import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
import org.apache.pinot.segment.spi.index.mutable.MutableDictionary;
import org.apache.pinot.segment.spi.index.mutable.MutableForwardIndex;
import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
@@ -71,10 +71,10 @@ public class CLPMutableForwardIndex implements
MutableForwardIndex {
_logTypeFwdIndex =
new FixedByteSVMutableForwardIndex(true, DataType.INT, capacity,
memoryManager, columnName + "_logType.fwd");
_dictVarsFwdIndex =
- new
FixedByteMVMutableForwardIndex(ForwardIndexType.MAX_MULTI_VALUES_PER_ROW, 20,
capacity, Integer.BYTES,
+ new
FixedByteMVMutableForwardIndex(ForwardIndexConfig.DEFAULT_MAX_NUM_MULTI_VALUES,
20, capacity, Integer.BYTES,
memoryManager, columnName + "_dictVars.fwd", true, DataType.INT);
_encodedVarsFwdIndex =
- new
FixedByteMVMutableForwardIndex(ForwardIndexType.MAX_MULTI_VALUES_PER_ROW, 20,
capacity, Long.BYTES,
+ new
FixedByteMVMutableForwardIndex(ForwardIndexConfig.DEFAULT_MAX_NUM_MULTI_VALUES,
20, capacity, Long.BYTES,
memoryManager, columnName + "_encodedVars.fwd", true,
DataType.LONG);
_clpMessageDecoder = new
MessageDecoder(BuiltInVariableHandlingRuleVersions.VariablesSchemaV2,
BuiltInVariableHandlingRuleVersions.VariableEncodingMethodsV1);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/FixedByteMVMutableForwardIndex.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/FixedByteMVMutableForwardIndex.java
index bf91fb2b06a..2c69ab7c657 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/FixedByteMVMutableForwardIndex.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/FixedByteMVMutableForwardIndex.java
@@ -28,7 +28,6 @@ import
org.apache.pinot.segment.local.io.writer.impl.FixedByteSingleValueMultiCo
import org.apache.pinot.segment.spi.index.mutable.MutableForwardIndex;
import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
-import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,8 +93,6 @@ import org.slf4j.LoggerFactory;
public class FixedByteMVMutableForwardIndex implements MutableForwardIndex {
private static final Logger LOGGER =
LoggerFactory.getLogger(FixedByteMVMutableForwardIndex.class);
- /// number of columns is 1, column size is variable but less than
\_maxNumberOfMultiValuesPerRow
-
private static final int SIZE_OF_INT = 4;
private static final int NUM_COLS_IN_HEADER = 3;
@@ -110,16 +107,17 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
private final List<FixedByteSingleValueMultiColReader> _headerReaders = new
CopyOnWriteArrayList<>();
private final List<FixedByteSingleValueMultiColWriter> _dataWriters = new
ArrayList<>();
private final List<FixedByteSingleValueMultiColReader> _dataReaders = new
CopyOnWriteArrayList<>();
- private final int _headerSize;
- private final int _incrementalCapacity;
- private final int _columnSizeInBytes;
- private final int _maxNumberOfMultiValuesPerRow;
- private final int _rowCountPerChunk;
+
+ private final int _maxNumMultiValues;
+ private final int _numRowsPerChunk;
+ private final int _valueSizeInBytes;
private final PinotDataBufferMemoryManager _memoryManager;
private final String _context;
private final boolean _isDictionaryEncoded;
- private final FieldSpec.DataType _storedType;
- private final FieldSpec.DataType _dataType;
+ private final DataType _storedType;
+ private final DataType _dataType;
+ private final int _headerSize;
+ private final int _incrementalCapacity;
private FixedByteSingleValueMultiColWriter _curHeaderWriter;
private FixedByteSingleValueMultiColWriter _currentDataWriter;
@@ -128,33 +126,30 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
private int _prevRowLength = 0; // Number of values in the column for the
last row added.
private int _numValues = 0;
- public FixedByteMVMutableForwardIndex(int maxNumberOfMultiValuesPerRow, int
avgMultiValueCount, int rowCountPerChunk,
- int columnSizeInBytes, PinotDataBufferMemoryManager memoryManager,
String context, boolean isDictionaryEncoded,
- FieldSpec.DataType storedType) {
- this(maxNumberOfMultiValuesPerRow, avgMultiValueCount, rowCountPerChunk,
columnSizeInBytes, memoryManager, context,
+ public FixedByteMVMutableForwardIndex(int maxNumMultiValues, int
avgNumMultiValues, int numRowsPerChunk,
+ int valueSizeInBytes, PinotDataBufferMemoryManager memoryManager, String
context, boolean isDictionaryEncoded,
+ DataType storedType) {
+ this(maxNumMultiValues, avgNumMultiValues, numRowsPerChunk,
valueSizeInBytes, memoryManager, context,
isDictionaryEncoded, storedType, storedType);
}
- public FixedByteMVMutableForwardIndex(int maxNumberOfMultiValuesPerRow, int
avgMultiValueCount, int rowCountPerChunk,
- int columnSizeInBytes, PinotDataBufferMemoryManager memoryManager,
String context, boolean isDictionaryEncoded,
- FieldSpec.DataType storedType, FieldSpec.DataType dataType) {
+ public FixedByteMVMutableForwardIndex(int maxNumMultiValues, int
avgNumMultiValues, int numRowsPerChunk,
+ int valueSizeInBytes, PinotDataBufferMemoryManager memoryManager, String
context, boolean isDictionaryEncoded,
+ DataType storedType, DataType dataType) {
+ _maxNumMultiValues = maxNumMultiValues;
+ _numRowsPerChunk = numRowsPerChunk;
+ _valueSizeInBytes = valueSizeInBytes;
_memoryManager = memoryManager;
_context = context;
- int initialCapacity = Math.max(maxNumberOfMultiValuesPerRow,
rowCountPerChunk * avgMultiValueCount);
- int incrementalCapacity =
- Math.max(maxNumberOfMultiValuesPerRow, (int) (initialCapacity * 1.0f *
INCREMENT_PERCENTAGE / 100));
- _columnSizeInBytes = columnSizeInBytes;
- _maxNumberOfMultiValuesPerRow = maxNumberOfMultiValuesPerRow;
- _headerSize = rowCountPerChunk * SIZE_OF_INT * NUM_COLS_IN_HEADER;
- _rowCountPerChunk = rowCountPerChunk;
- addHeaderBuffer();
- //at least create space for million entries, which for INT translates into
4mb buffer
- _incrementalCapacity = incrementalCapacity;
- addDataBuffer(initialCapacity);
- //init(_rowCountPerChunk, _columnSizeInBytes,
_maxNumberOfMultiValuesPerRow, initialCapacity, _incrementalCapacity);
_isDictionaryEncoded = isDictionaryEncoded;
_storedType = storedType;
_dataType = dataType;
+
+ _headerSize = numRowsPerChunk * SIZE_OF_INT * NUM_COLS_IN_HEADER;
+ addHeaderBuffer();
+ int initialCapacity = Math.max(maxNumMultiValues, numRowsPerChunk *
avgNumMultiValues);
+ addDataBuffer(initialCapacity);
+ _incrementalCapacity = Math.max(maxNumMultiValues, (int) ((double)
initialCapacity * INCREMENT_PERCENTAGE / 100));
}
private void addHeaderBuffer() {
@@ -165,21 +160,21 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
_curHeaderWriter =
new FixedByteSingleValueMultiColWriter(headerBuffer, 3, new
int[]{SIZE_OF_INT, SIZE_OF_INT, SIZE_OF_INT});
_headerWriters.add(_curHeaderWriter);
- _headerReaders.add(new FixedByteSingleValueMultiColReader(headerBuffer,
_rowCountPerChunk,
+ _headerReaders.add(new FixedByteSingleValueMultiColReader(headerBuffer,
_numRowsPerChunk,
new int[]{SIZE_OF_INT, SIZE_OF_INT, SIZE_OF_INT}));
}
- /// This method automatically computes the space needed based on the
\_columnSizeInBytes
+ /// This method automatically computes the space needed based on the
\_valueSizeInBytes
/// @param rowCapacity Additional capacity to be added in terms of number of
rows
private void addDataBuffer(int rowCapacity) {
try {
- long size = (long) rowCapacity * (long) _columnSizeInBytes;
+ long size = (long) rowCapacity * (long) _valueSizeInBytes;
LOGGER.info("Allocating data buffer of size {} for column {}", size,
_context);
// NOTE: PinotDataBuffer is tracked in PinotDataBufferMemoryManager. No
need to track and close inside the class.
PinotDataBuffer dataBuffer = _memoryManager.allocate(size, _context);
- _currentDataWriter = new FixedByteSingleValueMultiColWriter(dataBuffer,
1, new int[]{_columnSizeInBytes});
+ _currentDataWriter = new FixedByteSingleValueMultiColWriter(dataBuffer,
1, new int[]{_valueSizeInBytes});
_dataWriters.add(_currentDataWriter);
- _dataReaders.add(new FixedByteSingleValueMultiColReader(dataBuffer,
rowCapacity, new int[]{_columnSizeInBytes}));
+ _dataReaders.add(new FixedByteSingleValueMultiColReader(dataBuffer,
rowCapacity, new int[]{_valueSizeInBytes}));
//update the capacity
_currentCapacity = rowCapacity;
} catch (Exception e) {
@@ -189,7 +184,7 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
}
private void writeIntoHeader(int row, int dataWriterIndex, int startIndex,
int length) {
- if (row >= _headerWriters.size() * _rowCountPerChunk) {
+ if (row >= _headerWriters.size() * _numRowsPerChunk) {
addHeaderBuffer();
}
_curHeaderWriter.setInt(getRowInCurrentHeader(row), 0, dataWriterIndex);
@@ -197,22 +192,19 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
_curHeaderWriter.setInt(getRowInCurrentHeader(row), 2, length);
}
- // TODO Use powers of two for _rowCountPerChunk to optimize computation for
the
+ // TODO Use powers of two for _numRowsPerChunk to optimize computation for
the
// methods below. Or, assert that the input values to the class are powers
of two. TBD.
private FixedByteSingleValueMultiColReader getCurrentReader(int row) {
- return _headerReaders.get(row / _rowCountPerChunk);
+ return _headerReaders.get(row / _numRowsPerChunk);
}
private int getRowInCurrentHeader(int row) {
- return row % _rowCountPerChunk;
+ return row % _numRowsPerChunk;
}
private int updateHeader(int row, int numValues) {
- if (numValues > _maxNumberOfMultiValuesPerRow) {
- Preconditions.checkArgument(numValues <= _maxNumberOfMultiValuesPerRow,
- "Row %s has %s multi-values, exceeding the maximum of %s", row,
numValues,
- _maxNumberOfMultiValuesPerRow);
- }
+ Preconditions.checkArgument(numValues <= _maxNumMultiValues,
+ "Row %s has %s multi-values, exceeding the maximum of %s", row,
numValues, _maxNumMultiValues);
_numValues += numValues;
int newStartIndex = _prevRowStartIndex + _prevRowLength;
if (newStartIndex + numValues > _currentCapacity) {
@@ -227,17 +219,6 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
return newStartIndex;
}
- /// Returns the maximum number of values allowed in a single row.
- public int getMaxNumberOfMultiValuesPerRow() {
- return _maxNumberOfMultiValuesPerRow;
- }
-
- public int getMaxChunkCapacity() {
- // The incremental capacity will be >= the initial capacity and (the way
the code is currently written) will be
- // the largest the buffer could ever get.
- return _incrementalCapacity;
- }
-
@Override
public boolean isDictionaryEncoded() {
return _isDictionaryEncoded;
@@ -255,12 +236,12 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
@Override
public int getLengthOfShortestElement() {
- return _columnSizeInBytes;
+ return _valueSizeInBytes;
}
@Override
public int getLengthOfLongestElement() {
- return _columnSizeInBytes;
+ return _valueSizeInBytes;
}
@Override
@@ -475,9 +456,9 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
int newStartIndex = updateHeader(docId, values.length);
for (int i = 0; i < values.length; i++) {
byte[] value = values[i];
- if (value.length != _columnSizeInBytes) {
+ if (value.length != _valueSizeInBytes) {
throw new IllegalArgumentException(
- "Expected fixed-width bytes value of length: " +
_columnSizeInBytes + ", got: " + value.length);
+ "Expected fixed-width bytes value of length: " + _valueSizeInBytes
+ ", got: " + value.length);
}
_currentDataWriter.setBytes(newStartIndex + i, 0, value);
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java
index 33089d6ec06..47d1ea02c2e 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java
@@ -72,9 +72,6 @@ public class ForwardIndexType extends
AbstractIndexType<ForwardIndexConfig, Forw
private static final Logger LOGGER =
LoggerFactory.getLogger(ForwardIndexType.class);
public static final String INDEX_DISPLAY_NAME = "forward";
- // For multi-valued column, forward-index.
- // Default maximum number of multi-values per row. Some indexes, such as
vectors, configure a different row limit.
- public static final int MAX_MULTI_VALUES_PER_ROW = 1000;
private static final int
NODICT_VARIABLE_WIDTH_ESTIMATED_AVERAGE_VALUE_LENGTH_DEFAULT = 100;
private static final int
NODICT_VARIABLE_WIDTH_ESTIMATED_NUMBER_OF_VALUES_DEFAULT = 100_000;
//@formatter:off
@@ -368,8 +365,6 @@ public class ForwardIndexType extends
AbstractIndexType<ForwardIndexConfig, Forw
FieldSpec.DataType storedType = dataType.getStoredType();
int fixedLengthBytes = context.getFixedLengthBytes();
boolean isSingleValue = context.getFieldSpec().isSingleValueField();
- int maxNumMultiValuesPerRow = context.getMaxNumMultiValuesPerRowOverride()
> 0
- ? context.getMaxNumMultiValuesPerRowOverride() :
MAX_MULTI_VALUES_PER_ROW;
if (!context.hasDictionary()) {
if (isSingleValue) {
String allocationContext =
@@ -411,7 +406,7 @@ public class ForwardIndexType extends
AbstractIndexType<ForwardIndexConfig, Forw
IndexUtil.buildAllocationContext(context.getSegmentName(),
context.getFieldSpec().getName(),
V1Constants.Indexes.RAW_MV_FORWARD_INDEX_FILE_EXTENSION);
// TODO: Start with a smaller capacity on
FixedByteMVForwardIndexReaderWriter and let it expand
- return new FixedByteMVMutableForwardIndex(maxNumMultiValuesPerRow,
context.getAvgNumMultiValues(),
+ return new
FixedByteMVMutableForwardIndex(context.getMaxNumMultiValues(),
context.getAvgNumMultiValues(),
context.getCapacity(), dataType.size(),
context.getMemoryManager(), allocationContext, false, storedType,
dataType);
}
@@ -425,7 +420,7 @@ public class ForwardIndexType extends
AbstractIndexType<ForwardIndexConfig, Forw
String allocationContext =
IndexUtil.buildAllocationContext(segmentName, column,
V1Constants.Indexes.UNSORTED_MV_FORWARD_INDEX_FILE_EXTENSION);
// TODO: Start with a smaller capacity on
FixedByteMVForwardIndexReaderWriter and let it expand
- return new FixedByteMVMutableForwardIndex(maxNumMultiValuesPerRow,
context.getAvgNumMultiValues(),
+ return new
FixedByteMVMutableForwardIndex(context.getMaxNumMultiValues(),
context.getAvgNumMultiValues(),
context.getCapacity(), Integer.BYTES, context.getMemoryManager(),
allocationContext, true,
FieldSpec.DataType.INT);
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplMVLengthValidationTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplNumMultiValuesValidationTest.java
similarity index 55%
rename from
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplMVLengthValidationTest.java
rename to
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplNumMultiValuesValidationTest.java
index 9af323b19f3..a802aee6404 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplMVLengthValidationTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplNumMultiValuesValidationTest.java
@@ -40,35 +40,32 @@ import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.UpsertConfig;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.PrimaryKey;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
-/// Verifies mutable-segment enforcement of the fixed-byte multi-value row
limit. Each test owns its mutable segment,
-/// so no state is shared between test invocations.
-public class MutableSegmentImplMVLengthValidationTest implements
PinotBuffersAfterClassCheckRule {
+/// Verifies that the mutable segment enforces the maximum number of values
per fixed-byte multi-value entry. Each test
+/// owns its mutable segment, so no state is shared between test invocations.
+public class MutableSegmentImplNumMultiValuesValidationTest implements
PinotBuffersAfterClassCheckRule {
private static final String PARTITION_COLUMN = "partitionColumn";
private static final String PRIMARY_KEY_COLUMN = "primaryKey";
private static final String COMPARISON_COLUMN = "comparisonColumn";
private static final String MV_COLUMN = "mvColumn";
- private static final int MAX_MULTI_VALUES_PER_ROW = 1000;
+ private static final int MAX_NUM_MULTI_VALUES = 1000;
private static final int SMALL_VECTOR_DIMENSION = 768;
private static final int VECTOR_DIMENSION = 1536;
@@ -80,8 +77,8 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
@Test
public void testRejectOversizedMultiValueRowBeforeWrites()
throws Exception {
- Schema schema = new
Schema.SchemaBuilder().setSchemaName("mvLengthValidation")
- .addMultiValueDimension(MV_COLUMN, FieldSpec.DataType.INT)
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("numMultiValuesValidation")
+ .addMultiValueDimension(MV_COLUMN, DataType.INT)
.build();
MutableSegmentImpl mutableSegment =
MutableSegmentImplTestUtils.createMutableSegmentImpl(schema);
try {
@@ -89,23 +86,21 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
mutableSegment.index(createRow(firstValues), null);
DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
- FixedByteMVMutableForwardIndex forwardIndex =
- (FixedByteMVMutableForwardIndex) dataSource.getForwardIndex();
- Assert.assertTrue(forwardIndex.getMaxChunkCapacity() >
MAX_MULTI_VALUES_PER_ROW + 1);
-
- UnsupportedOperationException exception =
Assert.expectThrows(UnsupportedOperationException.class,
- () ->
mutableSegment.index(createRow(createValues(MAX_MULTI_VALUES_PER_ROW + 1, 10)),
null));
- Assert.assertTrue(exception.getMessage().contains(MV_COLUMN));
-
Assert.assertTrue(exception.getMessage().contains(Integer.toString(MAX_MULTI_VALUES_PER_ROW
+ 1)));
- Assert.assertTrue(exception.getMessage().contains("exceeding the maximum
of " + MAX_MULTI_VALUES_PER_ROW),
+ FixedByteMVMutableForwardIndex forwardIndex =
(FixedByteMVMutableForwardIndex) dataSource.getForwardIndex();
+
+ IllegalStateException exception =
expectThrows(IllegalStateException.class,
+ () ->
mutableSegment.index(createRow(createValues(MAX_NUM_MULTI_VALUES + 1, 10)),
null));
+ assertTrue(exception.getMessage().contains(MV_COLUMN));
+
assertTrue(exception.getMessage().contains(Integer.toString(MAX_NUM_MULTI_VALUES
+ 1)));
+ assertTrue(exception.getMessage().contains("exceeds the maximum allowed:
" + MAX_NUM_MULTI_VALUES),
exception.getMessage());
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 1);
+ assertEquals(mutableSegment.getNumDocsIndexed(), 1);
assertValues(dataSource, forwardIndex, 0, firstValues);
- Object[] maxLengthValues = createValues(MAX_MULTI_VALUES_PER_ROW, 10000);
- mutableSegment.index(createRow(maxLengthValues), null);
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 2);
- assertValues(dataSource, forwardIndex, 1, maxLengthValues);
+ Object[] valuesAtLimit = createValues(MAX_NUM_MULTI_VALUES, 10000);
+ mutableSegment.index(createRow(valuesAtLimit), null);
+ assertEquals(mutableSegment.getNumDocsIndexed(), 2);
+ assertValues(dataSource, forwardIndex, 1, valuesAtLimit);
} finally {
mutableSegment.destroy();
}
@@ -114,55 +109,52 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
@Test
public void testAcceptVectorDimensionAboveDefaultMultiValueLimit()
throws Exception {
- Schema schema = new
Schema.SchemaBuilder().setSchemaName("vectorMvLengthValidation")
- .addMultiValueDimension(MV_COLUMN, FieldSpec.DataType.FLOAT)
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("vectorNumMultiValuesValidation")
+ .addMultiValueDimension(MV_COLUMN, DataType.FLOAT)
.build();
- VectorIndexConfig vectorIndexConfig = new VectorIndexConfig(false, "HNSW",
VECTOR_DIMENSION, 1,
- VectorIndexConfig.VectorDistanceFunction.COSINE,
- Map.of("vectorIndexType", "HNSW", "vectorDimension",
Integer.toString(VECTOR_DIMENSION), "commitDocs", "1"));
- MutableSegmentImpl mutableSegment =
MutableSegmentImplTestUtils.createMutableSegmentImplWithVectorIndexConfigs(
- schema, Set.of(MV_COLUMN), Set.of(), Set.of(), Map.of(MV_COLUMN,
vectorIndexConfig), null);
+ VectorIndexConfig vectorIndexConfig =
+ new VectorIndexConfig(false, "HNSW", VECTOR_DIMENSION, 1,
VectorIndexConfig.VectorDistanceFunction.COSINE,
+ Map.of("vectorIndexType", "HNSW", "vectorDimension",
Integer.toString(VECTOR_DIMENSION), "commitDocs",
+ "1"));
+ MutableSegmentImpl mutableSegment =
+
MutableSegmentImplTestUtils.createMutableSegmentImplWithVectorIndexConfigs(schema,
Set.of(MV_COLUMN), Set.of(),
+ Set.of(), Map.of(MV_COLUMN, vectorIndexConfig), null);
try {
Object[] vector = createFloatValues(VECTOR_DIMENSION);
mutableSegment.index(createRow(vector), null);
DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
- FixedByteMVMutableForwardIndex forwardIndex =
- (FixedByteMVMutableForwardIndex) dataSource.getForwardIndex();
+ FixedByteMVMutableForwardIndex forwardIndex =
(FixedByteMVMutableForwardIndex) dataSource.getForwardIndex();
float[] queryVector = toPrimitiveFloatArray(vector);
- Assert.assertEquals(forwardIndex.getMaxNumberOfMultiValuesPerRow(),
VECTOR_DIMENSION);
- Assert.assertEquals(forwardIndex.getFloatMV(0), queryVector);
- Assert.assertNotNull(dataSource.getVectorIndex());
- Assert.assertEquals(dataSource.getVectorIndex().getDocIds(queryVector,
1).toArray(), new int[]{0});
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 1);
+ assertEquals(forwardIndex.getFloatMV(0), queryVector);
+ assertNotNull(dataSource.getVectorIndex());
+ assertEquals(dataSource.getVectorIndex().getDocIds(queryVector,
1).toArray(), new int[]{0});
+ assertEquals(mutableSegment.getNumDocsIndexed(), 1);
} finally {
mutableSegment.destroy();
}
}
@Test
- public void testRejectVectorAboveConfiguredDimensionBeforeWrites()
- throws Exception {
- Schema schema = new
Schema.SchemaBuilder().setSchemaName("smallVectorMvLengthValidation")
- .addMultiValueDimension(MV_COLUMN, FieldSpec.DataType.FLOAT)
+ public void testRejectVectorAboveConfiguredDimensionBeforeWrites() {
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("smallVectorNumMultiValuesValidation")
+ .addMultiValueDimension(MV_COLUMN, DataType.FLOAT)
.build();
- VectorIndexConfig vectorIndexConfig = new VectorIndexConfig(false, "HNSW",
SMALL_VECTOR_DIMENSION, 1,
- VectorIndexConfig.VectorDistanceFunction.COSINE,
- Map.of("vectorIndexType", "HNSW", "vectorDimension",
Integer.toString(SMALL_VECTOR_DIMENSION),
- "commitDocs", "1"));
- MutableSegmentImpl mutableSegment =
MutableSegmentImplTestUtils.createMutableSegmentImplWithVectorIndexConfigs(
- schema, Set.of(MV_COLUMN), Set.of(), Set.of(), Map.of(MV_COLUMN,
vectorIndexConfig), null);
+ VectorIndexConfig vectorIndexConfig =
+ new VectorIndexConfig(false, "HNSW", SMALL_VECTOR_DIMENSION, 1,
VectorIndexConfig.VectorDistanceFunction.COSINE,
+ Map.of("vectorIndexType", "HNSW", "vectorDimension",
Integer.toString(SMALL_VECTOR_DIMENSION), "commitDocs",
+ "1"));
+ MutableSegmentImpl mutableSegment =
+
MutableSegmentImplTestUtils.createMutableSegmentImplWithVectorIndexConfigs(schema,
Set.of(MV_COLUMN), Set.of(),
+ Set.of(), Map.of(MV_COLUMN, vectorIndexConfig), null);
try {
- Assert.expectThrows(UnsupportedOperationException.class,
+ expectThrows(IllegalStateException.class,
() ->
mutableSegment.index(createRow(createFloatValues(SMALL_VECTOR_DIMENSION + 1)),
null));
DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
- FixedByteMVMutableForwardIndex forwardIndex =
- (FixedByteMVMutableForwardIndex) dataSource.getForwardIndex();
- Assert.assertEquals(forwardIndex.getMaxNumberOfMultiValuesPerRow(),
SMALL_VECTOR_DIMENSION);
- Assert.assertEquals(dataSource.getDataSourceMetadata().getNumValues(),
0);
- Assert.assertTrue(dataSource.getVectorIndex().getDocIds(new
float[SMALL_VECTOR_DIMENSION], 1).isEmpty());
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 0);
+ assertEquals(dataSource.getDataSourceMetadata().getNumValues(), 0);
+ assertTrue(dataSource.getVectorIndex().getDocIds(new
float[SMALL_VECTOR_DIMENSION], 1).isEmpty());
+ assertEquals(mutableSegment.getNumDocsIndexed(), 0);
} finally {
mutableSegment.destroy();
}
@@ -171,33 +163,32 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
@Test
public void testRejectOversizedMultiValueRowBeforeDedupUpdate()
throws Exception {
- Schema schema = new
Schema.SchemaBuilder().setSchemaName("dedupMvLengthValidation")
- .addSingleValueDimension(PRIMARY_KEY_COLUMN, FieldSpec.DataType.STRING)
- .addMultiValueDimension(MV_COLUMN, FieldSpec.DataType.INT)
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("dedupNumMultiValuesValidation")
+ .addSingleValueDimension(PRIMARY_KEY_COLUMN, DataType.STRING)
+ .addMultiValueDimension(MV_COLUMN, DataType.INT)
.setPrimaryKeyColumns(List.of(PRIMARY_KEY_COLUMN))
.build();
PartitionDedupMetadataManager dedupMetadataManager =
mock(PartitionDedupMetadataManager.class);
when(dedupMetadataManager.getContext()).thenReturn(mock(DedupContext.class));
Set<PrimaryKey> seenPrimaryKeys = new HashSet<>();
-
when(dedupMetadataManager.checkRecordPresentOrUpdate(any(DedupRecordInfo.class),
any()))
- .thenAnswer(invocation -> {
- DedupRecordInfo recordInfo = invocation.getArgument(0);
- return !seenPrimaryKeys.add(recordInfo.getPrimaryKey());
- });
+
when(dedupMetadataManager.checkRecordPresentOrUpdate(any(DedupRecordInfo.class),
any())).thenAnswer(invocation -> {
+ DedupRecordInfo recordInfo = invocation.getArgument(0);
+ return !seenPrimaryKeys.add(recordInfo.getPrimaryKey());
+ });
MutableSegmentImpl mutableSegment =
MutableSegmentImplTestUtils.createMutableSegmentImpl(schema, false,
null, null, dedupMetadataManager);
try {
String primaryKey = "same-key";
- Assert.expectThrows(UnsupportedOperationException.class, () ->
mutableSegment.index(
- createDedupRow(primaryKey, createValues(MAX_MULTI_VALUES_PER_ROW +
1, 0)), null));
- Assert.assertTrue(seenPrimaryKeys.isEmpty());
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 0);
+ expectThrows(IllegalStateException.class,
+ () -> mutableSegment.index(createDedupRow(primaryKey,
createValues(MAX_NUM_MULTI_VALUES + 1, 0)), null));
+ assertTrue(seenPrimaryKeys.isEmpty());
+ assertEquals(mutableSegment.getNumDocsIndexed(), 0);
Object[] validValues = createValues(3, 10000);
mutableSegment.index(createDedupRow(primaryKey, validValues), null);
- Assert.assertEquals(seenPrimaryKeys.size(), 1);
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 1);
+ assertEquals(seenPrimaryKeys.size(), 1);
+ assertEquals(mutableSegment.getNumDocsIndexed(), 1);
DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
assertValues(dataSource, (FixedByteMVMutableForwardIndex)
dataSource.getForwardIndex(), 0, validValues);
@@ -209,33 +200,34 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
@Test(dataProvider = "collectionMergeStrategies")
public void
testRejectOversizedMultiValueRowProducedByPartialUpsert(UpsertConfig.Strategy
strategy)
throws Exception {
- Schema schema = new
Schema.SchemaBuilder().setSchemaName("partialUpsertMvLengthValidation")
- .addSingleValueDimension(PRIMARY_KEY_COLUMN, FieldSpec.DataType.STRING)
- .addMultiValueDimension(MV_COLUMN, FieldSpec.DataType.INT)
- .addDateTime(COMPARISON_COLUMN, FieldSpec.DataType.LONG,
"1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("partialUpsertNumMultiValuesValidation")
+ .addSingleValueDimension(PRIMARY_KEY_COLUMN, DataType.STRING)
+ .addMultiValueDimension(MV_COLUMN, DataType.INT)
+ .addDateTime(COMPARISON_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
.setPrimaryKeyColumns(List.of(PRIMARY_KEY_COLUMN))
.build();
UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.PARTIAL);
upsertConfig.setComparisonColumns(List.of(COMPARISON_COLUMN));
upsertConfig.setPartialUpsertStrategies(Map.of(MV_COLUMN, strategy));
- TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME)
- .setTableName("partialUpsertMvLengthValidation")
- .setTimeColumnName(COMPARISON_COLUMN)
- .setUpsertConfig(upsertConfig)
- .setNullHandlingEnabled(true)
- .build();
- TableUpsertMetadataManager tableUpsertMetadataManager =
TableUpsertMetadataManagerFactory.create(
- new PinotConfiguration(), tableConfig, schema,
mock(TableDataManager.class), null);
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.REALTIME).setTableName("partialUpsertNumMultiValuesValidation")
+ .setTimeColumnName(COMPARISON_COLUMN)
+ .setUpsertConfig(upsertConfig)
+ .setNullHandlingEnabled(true)
+ .build();
+ TableUpsertMetadataManager tableUpsertMetadataManager =
+ TableUpsertMetadataManagerFactory.create(new PinotConfiguration(),
tableConfig, schema,
+ mock(TableDataManager.class), null);
PartitionUpsertMetadataManager upsertMetadataManager =
spy(tableUpsertMetadataManager.getOrCreatePartitionManager(0));
- MutableSegmentImpl mutableSegment =
MutableSegmentImplTestUtils.createMutableSegmentImpl(schema, true,
- COMPARISON_COLUMN, upsertMetadataManager, null);
+ MutableSegmentImpl mutableSegment =
+ MutableSegmentImplTestUtils.createMutableSegmentImpl(schema, true,
COMPARISON_COLUMN, upsertMetadataManager,
+ null);
try {
String primaryKey = "same-key";
- Assert.expectThrows(UnsupportedOperationException.class,
- () -> mutableSegment.index(
- createUpsertRow(primaryKey, 0L,
createValues(MAX_MULTI_VALUES_PER_ROW + 1, 0)), null));
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 0);
+ expectThrows(IllegalStateException.class,
+ () -> mutableSegment.index(createUpsertRow(primaryKey, 0L,
createValues(MAX_NUM_MULTI_VALUES + 1, 0)), null));
+ assertEquals(mutableSegment.getNumDocsIndexed(), 0);
verify(upsertMetadataManager,
times(0)).updateRecord(any(GenericRow.class), any(RecordInfo.class));
verify(upsertMetadataManager, times(0)).addRecord(eq(mutableSegment),
any(RecordInfo.class));
@@ -243,29 +235,28 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
mutableSegment.index(createUpsertRow(primaryKey, 1L, firstValues), null);
DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
- FixedByteMVMutableForwardIndex forwardIndex =
- (FixedByteMVMutableForwardIndex) dataSource.getForwardIndex();
+ FixedByteMVMutableForwardIndex forwardIndex =
(FixedByteMVMutableForwardIndex) dataSource.getForwardIndex();
int cardinalityBeforeRejection = dataSource.getDictionary().length();
int numValuesBeforeRejection =
dataSource.getDataSourceMetadata().getNumValues();
ImmutableRoaringBitmap validDocIds =
mutableSegment.getValidDocIds().getMutableRoaringBitmap();
- Assert.assertEquals(validDocIds.toArray(), new int[]{0});
+ assertEquals(validDocIds.toArray(), new int[]{0});
verify(upsertMetadataManager, times(1)).addRecord(eq(mutableSegment),
any(RecordInfo.class));
- Assert.expectThrows(UnsupportedOperationException.class,
+ expectThrows(IllegalStateException.class,
() -> mutableSegment.index(createUpsertRow(primaryKey, 2L,
createValues(500, 600)), null));
DataSource dataSourceAfterRejection =
mutableSegment.getDataSource(MV_COLUMN);
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 1);
- Assert.assertEquals(dataSourceAfterRejection.getDictionary().length(),
cardinalityBeforeRejection);
-
Assert.assertEquals(dataSourceAfterRejection.getDataSourceMetadata().getNumValues(),
numValuesBeforeRejection);
-
Assert.assertEquals(mutableSegment.getValidDocIds().getMutableRoaringBitmap().toArray(),
new int[]{0});
+ assertEquals(mutableSegment.getNumDocsIndexed(), 1);
+ assertEquals(dataSourceAfterRejection.getDictionary().length(),
cardinalityBeforeRejection);
+
assertEquals(dataSourceAfterRejection.getDataSourceMetadata().getNumValues(),
numValuesBeforeRejection);
+
assertEquals(mutableSegment.getValidDocIds().getMutableRoaringBitmap().toArray(),
new int[]{0});
verify(upsertMetadataManager, times(1)).addRecord(eq(mutableSegment),
any(RecordInfo.class));
assertValues(dataSourceAfterRejection, forwardIndex, 0, firstValues);
Object[] validUpdate = createValues(3, 600);
mutableSegment.index(createUpsertRow(primaryKey, 3L, validUpdate), null);
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 2);
-
Assert.assertEquals(mutableSegment.getValidDocIds().getMutableRoaringBitmap().toArray(),
new int[]{1});
+ assertEquals(mutableSegment.getNumDocsIndexed(), 2);
+
assertEquals(mutableSegment.getValidDocIds().getMutableRoaringBitmap().toArray(),
new int[]{1});
verify(upsertMetadataManager, times(2)).addRecord(eq(mutableSegment),
any(RecordInfo.class));
assertValues(mutableSegment.getDataSource(MV_COLUMN), forwardIndex, 1,
createValues(603, 0));
} finally {
@@ -278,25 +269,23 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
@Test
public void testRejectOversizedMultiValueRowBeforePartitionTracking()
throws Exception {
- Schema schema = new
Schema.SchemaBuilder().setSchemaName("partitionMvLengthValidation")
- .addSingleValueDimension(PARTITION_COLUMN, FieldSpec.DataType.INT)
- .addMultiValueDimension(MV_COLUMN, FieldSpec.DataType.INT)
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("partitionNumMultiValuesValidation")
+ .addSingleValueDimension(PARTITION_COLUMN, DataType.INT)
+ .addMultiValueDimension(MV_COLUMN, DataType.INT)
.build();
MutableSegmentImpl mutableSegment =
MutableSegmentImplTestUtils.createMutableSegmentImpl(schema, PARTITION_COLUMN,
new ModuloPartitionFunction(4, null), 0, false);
try {
-
Assert.assertEquals(mutableSegment.getDataSource(PARTITION_COLUMN).getDataSourceMetadata().getPartitions(),
- Set.of(0));
- Assert.expectThrows(UnsupportedOperationException.class,
- () -> mutableSegment.index(createPartitionedRow(1,
createValues(MAX_MULTI_VALUES_PER_ROW + 1, 0)), null));
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 0);
-
Assert.assertEquals(mutableSegment.getDataSource(PARTITION_COLUMN).getDataSourceMetadata().getPartitions(),
- Set.of(0));
+
assertEquals(mutableSegment.getDataSource(PARTITION_COLUMN).getDataSourceMetadata().getPartitions(),
Set.of(0));
+ expectThrows(IllegalStateException.class,
+ () -> mutableSegment.index(createPartitionedRow(1,
createValues(MAX_NUM_MULTI_VALUES + 1, 0)), null));
+ assertEquals(mutableSegment.getNumDocsIndexed(), 0);
+
assertEquals(mutableSegment.getDataSource(PARTITION_COLUMN).getDataSourceMetadata().getPartitions(),
Set.of(0));
Object[] validValues = createValues(3, 0);
mutableSegment.index(createPartitionedRow(1, validValues), null);
- Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 1);
-
Assert.assertEquals(mutableSegment.getDataSource(PARTITION_COLUMN).getDataSourceMetadata().getPartitions(),
+ assertEquals(mutableSegment.getNumDocsIndexed(), 1);
+
assertEquals(mutableSegment.getDataSource(PARTITION_COLUMN).getDataSourceMetadata().getPartitions(),
Set.of(0, 1));
DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
assertValues(dataSource, (FixedByteMVMutableForwardIndex)
dataSource.getForwardIndex(), 0, validValues);
@@ -308,8 +297,7 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
@DataProvider(name = "collectionMergeStrategies")
private static Object[][] collectionMergeStrategies() {
return new Object[][]{
- {UpsertConfig.Strategy.APPEND},
- {UpsertConfig.Strategy.UNION}
+ {UpsertConfig.Strategy.APPEND}, {UpsertConfig.Strategy.UNION}
};
}
@@ -337,17 +325,17 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
return row;
}
- private static Object[] createValues(int length, int offset) {
- Object[] values = new Object[length];
- for (int i = 0; i < length; i++) {
+ private static Object[] createValues(int numValues, int offset) {
+ Object[] values = new Object[numValues];
+ for (int i = 0; i < numValues; i++) {
values[i] = offset + i;
}
return values;
}
- private static Object[] createFloatValues(int length) {
- Object[] values = new Object[length];
- for (int i = 0; i < length; i++) {
+ private static Object[] createFloatValues(int numValues) {
+ Object[] values = new Object[numValues];
+ for (int i = 0; i < numValues; i++) {
values[i] = (float) i;
}
return values;
@@ -364,11 +352,11 @@ public class MutableSegmentImplMVLengthValidationTest
implements PinotBuffersAft
private static void assertValues(DataSource dataSource,
FixedByteMVMutableForwardIndex forwardIndex, int docId,
Object[] expectedValues) {
Dictionary dictionary = dataSource.getDictionary();
- Assert.assertNotNull(dictionary);
+ assertNotNull(dictionary);
int[] dictIds = forwardIndex.getDictIdMV(docId);
- Assert.assertEquals(dictIds.length, expectedValues.length);
+ assertEquals(dictIds.length, expectedValues.length);
for (int i = 0; i < expectedValues.length; i++) {
- Assert.assertEquals(dictionary.get(dictIds[i]), expectedValues[i]);
+ assertEquals(dictionary.get(dictIds[i]), expectedValues[i]);
}
}
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/mutable/FixedByteMVMutableForwardIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/mutable/FixedByteMVMutableForwardIndexTest.java
index dabf1f95ace..ac6a9ca95b5 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/mutable/FixedByteMVMutableForwardIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/mutable/FixedByteMVMutableForwardIndexTest.java
@@ -74,13 +74,13 @@ public class FixedByteMVMutableForwardIndexTest implements
PinotBuffersAfterClas
@Test
public void testRejectMultiValuesExceedingMaxPerRow()
throws Exception {
- int maxNumberOfMultiValuesPerRow = 5;
+ int maxNumMultiValues = 5;
FixedByteMVMutableForwardIndex readerWriter =
- new FixedByteMVMutableForwardIndex(maxNumberOfMultiValuesPerRow, 2,
10, Integer.BYTES, _memoryManager,
+ new FixedByteMVMutableForwardIndex(maxNumMultiValues, 2, 10,
Integer.BYTES, _memoryManager,
"RejectMultiValuesExceedingMaxPerRow", true,
FieldSpec.DataType.INT);
try {
Assert.expectThrows(IllegalArgumentException.class,
- () -> readerWriter.setIntMV(0, new int[maxNumberOfMultiValuesPerRow
+ 1]));
+ () -> readerWriter.setIntMV(0, new int[maxNumMultiValues + 1]));
Assert.assertEquals(getNumValues(readerWriter), 0);
Assert.assertEquals(readerWriter.getNumValuesMV(0), 0);
} finally {
@@ -91,9 +91,9 @@ public class FixedByteMVMutableForwardIndexTest implements
PinotBuffersAfterClas
@Test
public void testAcceptMultiValuesAtMaxPerRow()
throws Exception {
- int maxNumberOfMultiValuesPerRow = 5;
+ int maxNumMultiValues = 5;
FixedByteMVMutableForwardIndex readerWriter =
- new FixedByteMVMutableForwardIndex(maxNumberOfMultiValuesPerRow, 2,
10, Integer.BYTES, _memoryManager,
+ new FixedByteMVMutableForwardIndex(maxNumMultiValues, 2, 10,
Integer.BYTES, _memoryManager,
"AcceptMultiValuesAtMaxPerRow", true, FieldSpec.DataType.INT);
try {
int[] values = new int[]{1, 2, 3, 4, 5};
@@ -108,24 +108,24 @@ public class FixedByteMVMutableForwardIndexTest
implements PinotBuffersAfterClas
throws IOException {
FixedByteMVMutableForwardIndex readerWriter;
int rows = 1000;
- int columnSizeInBytes = Integer.BYTES;
- int maxNumberOfMultiValuesPerRow = 2000;
+ int valueSizeInBytes = Integer.BYTES;
+ int maxNumMultiValues = 2000;
readerWriter =
- new FixedByteMVMutableForwardIndex(maxNumberOfMultiValuesPerRow, 2,
rows / 2, columnSizeInBytes, _memoryManager,
+ new FixedByteMVMutableForwardIndex(maxNumMultiValues, 2, rows / 2,
valueSizeInBytes, _memoryManager,
"IntArray", isDictionaryEncoded, FieldSpec.DataType.INT);
int valuesAdded = 0;
Random r = new Random(seed);
int[][] data = new int[rows][];
for (int i = 0; i < rows; i++) {
- data[i] = new int[r.nextInt(maxNumberOfMultiValuesPerRow)];
+ data[i] = new int[r.nextInt(maxNumMultiValues)];
for (int j = 0; j < data[i].length; j++) {
data[i][j] = r.nextInt();
}
readerWriter.setIntMV(i, data[i]);
valuesAdded += data[i].length;
}
- int[] ret = new int[maxNumberOfMultiValuesPerRow];
+ int[] ret = new int[maxNumMultiValues];
for (int i = 0; i < rows; i++) {
int length = readerWriter.getIntMV(i, ret);
Assert.assertEquals(data[i].length, length, "Failed with seed=" + seed);
@@ -139,11 +139,11 @@ public class FixedByteMVMutableForwardIndexTest
implements PinotBuffersAfterClas
throws IOException {
FixedByteMVMutableForwardIndex readerWriter;
int rows = 1000;
- int columnSizeInBytes = Integer.BYTES;
+ int valueSizeInBytes = Integer.BYTES;
// Keep the rowsPerChunk as a multiple of multiValuesPerRow to check the
cases when both data and header buffers
// transition to new ones
readerWriter = new FixedByteMVMutableForwardIndex(multiValuesPerRow,
multiValuesPerRow, multiValuesPerRow * 2,
- columnSizeInBytes, _memoryManager, "IntArrayFixedSize",
isDictionaryEncoded, FieldSpec.DataType.INT);
+ valueSizeInBytes, _memoryManager, "IntArrayFixedSize",
isDictionaryEncoded, FieldSpec.DataType.INT);
int valuesAdded = 0;
Random r = new Random(seed);
@@ -169,19 +169,19 @@ public class FixedByteMVMutableForwardIndexTest
implements PinotBuffersAfterClas
public void testWithZeroSize(long seed, boolean isDictionaryEncoded)
throws IOException {
FixedByteMVMutableForwardIndex readerWriter;
- final int maxNumberOfMultiValuesPerRow = 5;
+ final int maxNumMultiValues = 5;
int rows = 1000;
- int columnSizeInBytes = Integer.BYTES;
+ int valueSizeInBytes = Integer.BYTES;
Random r = new Random(seed);
readerWriter =
- new FixedByteMVMutableForwardIndex(maxNumberOfMultiValuesPerRow, 3,
r.nextInt(rows) + 1, columnSizeInBytes,
+ new FixedByteMVMutableForwardIndex(maxNumMultiValues, 3,
r.nextInt(rows) + 1, valueSizeInBytes,
_memoryManager, "ZeroSize", isDictionaryEncoded,
FieldSpec.DataType.INT);
int valuesAdded = 0;
int[][] data = new int[rows][];
for (int i = 0; i < rows; i++) {
if (r.nextInt() > 0) {
- data[i] = new int[r.nextInt(maxNumberOfMultiValuesPerRow)];
+ data[i] = new int[r.nextInt(maxNumMultiValues)];
for (int j = 0; j < data[i].length; j++) {
data[i][j] = r.nextInt();
}
@@ -193,7 +193,7 @@ public class FixedByteMVMutableForwardIndexTest implements
PinotBuffersAfterClas
valuesAdded += data[i].length;
}
}
- int[] ret = new int[maxNumberOfMultiValuesPerRow];
+ int[] ret = new int[maxNumMultiValues];
for (int i = 0; i < rows; i++) {
int length = readerWriter.getIntMV(i, ret);
Assert.assertEquals(data[i].length, length, "Failed with seed=" + seed);
@@ -204,11 +204,11 @@ public class FixedByteMVMutableForwardIndexTest
implements PinotBuffersAfterClas
}
private FixedByteMVMutableForwardIndex createReaderWriter(FieldSpec.DataType
dataType, Random r, int rows,
- int maxNumberOfMultiValuesPerRow, boolean isDictionaryEncoded) {
- final int avgMultiValueCount = r.nextInt(maxNumberOfMultiValuesPerRow) + 1;
- final int rowCountPerChunk = r.nextInt(rows) + 1;
+ int maxNumMultiValues, boolean isDictionaryEncoded) {
+ final int avgNumMultiValues = r.nextInt(maxNumMultiValues) + 1;
+ final int numRowsPerChunk = r.nextInt(rows) + 1;
- return new FixedByteMVMutableForwardIndex(maxNumberOfMultiValuesPerRow,
avgMultiValueCount, rowCountPerChunk,
+ return new FixedByteMVMutableForwardIndex(maxNumMultiValues,
avgNumMultiValues, numRowsPerChunk,
dataType.size(), _memoryManager, "ReaderWriter", isDictionaryEncoded,
dataType);
}
@@ -229,15 +229,15 @@ public class FixedByteMVMutableForwardIndexTest
implements PinotBuffersAfterClas
final long seed = generateSeed();
Random r = new Random(seed);
int rows = 1000;
- final int maxNumberOfMultiValuesPerRow = r.nextInt(100) + 1;
+ final int maxNumMultiValues = r.nextInt(100) + 1;
FixedByteMVMutableForwardIndex readerWriter =
- createReaderWriter(FieldSpec.DataType.LONG, r, rows,
maxNumberOfMultiValuesPerRow, isDictionaryEncoded);
+ createReaderWriter(FieldSpec.DataType.LONG, r, rows,
maxNumMultiValues, isDictionaryEncoded);
int valuesAdded = 0;
long[][] data = new long[rows][];
for (int i = 0; i < rows; i++) {
if (r.nextInt() > 0) {
- data[i] = new long[r.nextInt(maxNumberOfMultiValuesPerRow)];
+ data[i] = new long[r.nextInt(maxNumMultiValues)];
for (int j = 0; j < data[i].length; j++) {
data[i][j] = r.nextLong();
}
@@ -249,7 +249,7 @@ public class FixedByteMVMutableForwardIndexTest implements
PinotBuffersAfterClas
valuesAdded += data[i].length;
}
}
- long[] ret = new long[maxNumberOfMultiValuesPerRow];
+ long[] ret = new long[maxNumMultiValues];
for (int i = 0; i < rows; i++) {
int length = readerWriter.getLongMV(i, ret);
Assert.assertEquals(data[i].length, length, "Failed with seed=" + seed);
@@ -271,15 +271,15 @@ public class FixedByteMVMutableForwardIndexTest
implements PinotBuffersAfterClas
final long seed = generateSeed();
Random r = new Random(seed);
int rows = 1000;
- final int maxNumberOfMultiValuesPerRow = r.nextInt(100) + 1;
+ final int maxNumMultiValues = r.nextInt(100) + 1;
FixedByteMVMutableForwardIndex readerWriter =
- createReaderWriter(FieldSpec.DataType.FLOAT, r, rows,
maxNumberOfMultiValuesPerRow, isDictoinaryEncoded);
+ createReaderWriter(FieldSpec.DataType.FLOAT, r, rows,
maxNumMultiValues, isDictoinaryEncoded);
int valuesAdded = 0;
float[][] data = new float[rows][];
for (int i = 0; i < rows; i++) {
if (r.nextInt() > 0) {
- data[i] = new float[r.nextInt(maxNumberOfMultiValuesPerRow)];
+ data[i] = new float[r.nextInt(maxNumMultiValues)];
for (int j = 0; j < data[i].length; j++) {
data[i][j] = r.nextFloat();
}
@@ -291,7 +291,7 @@ public class FixedByteMVMutableForwardIndexTest implements
PinotBuffersAfterClas
valuesAdded += data[i].length;
}
}
- float[] ret = new float[maxNumberOfMultiValuesPerRow];
+ float[] ret = new float[maxNumMultiValues];
for (int i = 0; i < rows; i++) {
int length = readerWriter.getFloatMV(i, ret);
Assert.assertEquals(data[i].length, length, "Failed with seed=" + seed);
@@ -313,15 +313,15 @@ public class FixedByteMVMutableForwardIndexTest
implements PinotBuffersAfterClas
final long seed = generateSeed();
Random r = new Random(seed);
int rows = 1000;
- final int maxNumberOfMultiValuesPerRow = r.nextInt(100) + 1;
+ final int maxNumMultiValues = r.nextInt(100) + 1;
FixedByteMVMutableForwardIndex readerWriter =
- createReaderWriter(FieldSpec.DataType.DOUBLE, r, rows,
maxNumberOfMultiValuesPerRow, isDictonaryEncoded);
+ createReaderWriter(FieldSpec.DataType.DOUBLE, r, rows,
maxNumMultiValues, isDictonaryEncoded);
int valuesAdded = 0;
double[][] data = new double[rows][];
for (int i = 0; i < rows; i++) {
if (r.nextInt() > 0) {
- data[i] = new double[r.nextInt(maxNumberOfMultiValuesPerRow)];
+ data[i] = new double[r.nextInt(maxNumMultiValues)];
for (int j = 0; j < data[i].length; j++) {
data[i][j] = r.nextDouble();
}
@@ -333,7 +333,7 @@ public class FixedByteMVMutableForwardIndexTest implements
PinotBuffersAfterClas
valuesAdded += data[i].length;
}
}
- double[] ret = new double[maxNumberOfMultiValuesPerRow];
+ double[] ret = new double[maxNumMultiValues];
for (int i = 0; i < rows; i++) {
int length = readerWriter.getDoubleMV(i, ret);
Assert.assertEquals(data[i].length, length, "Failed with seed=" + seed);
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java
index 5bf316768e1..5a1fea5f3d6 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java
@@ -37,6 +37,11 @@ import org.apache.pinot.spi.utils.DataSizeUtils;
public class ForwardIndexConfig extends IndexConfig {
+ // TODO: Make the maximum number of multi-values configurable, and also
enforce it when creating the immutable forward
+ // index, which currently accepts any number of values per entry.
+ /// Default maximum number of values in a single multi-value entry of the
mutable forward index.
+ public static final int DEFAULT_MAX_NUM_MULTI_VALUES = 1000;
+
private static int _defaultRawIndexWriterVersion = 4;
private static String _defaultTargetMaxChunkSize = "1MB";
private static int _defaultTargetMaxChunkSizeBytes = 1024 * 1024;
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/mutable/provider/MutableIndexContext.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/mutable/provider/MutableIndexContext.java
index 010baead912..6eba8daf7c1 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/mutable/provider/MutableIndexContext.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/mutable/provider/MutableIndexContext.java
@@ -20,34 +20,28 @@ package org.apache.pinot.segment.spi.index.mutable.provider;
import java.io.File;
import java.util.Objects;
+import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
import org.apache.pinot.spi.data.FieldSpec;
public class MutableIndexContext {
- private final int _capacity;
private final FieldSpec _fieldSpec;
private final int _fixedLengthBytes;
private final boolean _hasDictionary;
+ private final String _segmentName;
+ private final PinotDataBufferMemoryManager _memoryManager;
+ private final int _capacity;
private final boolean _offHeap;
private final int _estimatedColSize;
private final int _estimatedCardinality;
+ private final int _maxNumMultiValues;
private final int _avgNumMultiValues;
- private final int _maxNumMultiValuesPerRowOverride;
- private final String _segmentName;
- private final PinotDataBufferMemoryManager _memoryManager;
private final File _consumerDir;
public MutableIndexContext(FieldSpec fieldSpec, int fixedLengthBytes,
boolean hasDictionary, String segmentName,
PinotDataBufferMemoryManager memoryManager, int capacity, boolean
offHeap, int estimatedColSize,
- int estimatedCardinality, int avgNumMultiValues, File consumerDir) {
- this(fieldSpec, fixedLengthBytes, hasDictionary, segmentName,
memoryManager, capacity, offHeap, estimatedColSize,
- estimatedCardinality, avgNumMultiValues, 0, consumerDir);
- }
-
- public MutableIndexContext(FieldSpec fieldSpec, int fixedLengthBytes,
boolean hasDictionary, String segmentName,
- PinotDataBufferMemoryManager memoryManager, int capacity, boolean
offHeap, int estimatedColSize,
- int estimatedCardinality, int avgNumMultiValues, int
maxNumMultiValuesPerRowOverride, File consumerDir) {
+ int estimatedCardinality, int maxNumMultiValues, int avgNumMultiValues,
File consumerDir) {
_fieldSpec = fieldSpec;
_fixedLengthBytes = fixedLengthBytes;
_hasDictionary = hasDictionary;
@@ -57,8 +51,8 @@ public class MutableIndexContext {
_offHeap = offHeap;
_estimatedColSize = estimatedColSize;
_estimatedCardinality = estimatedCardinality;
+ _maxNumMultiValues = maxNumMultiValues;
_avgNumMultiValues = avgNumMultiValues;
- _maxNumMultiValuesPerRowOverride = maxNumMultiValuesPerRowOverride;
_consumerDir = consumerDir;
}
@@ -98,13 +92,12 @@ public class MutableIndexContext {
return _estimatedCardinality;
}
- public int getAvgNumMultiValues() {
- return _avgNumMultiValues;
+ public int getMaxNumMultiValues() {
+ return _maxNumMultiValues;
}
- /// Returns the configured maximum number of values for one MV row, or 0
when the index default should be used.
- public int getMaxNumMultiValuesPerRowOverride() {
- return _maxNumMultiValuesPerRowOverride;
+ public int getAvgNumMultiValues() {
+ return _avgNumMultiValues;
}
public File getConsumerDir() {
@@ -125,8 +118,8 @@ public class MutableIndexContext {
private PinotDataBufferMemoryManager _memoryManager;
private int _estimatedColSize;
private int _estimatedCardinality;
+ private int _maxNumMultiValues =
ForwardIndexConfig.DEFAULT_MAX_NUM_MULTI_VALUES;
private int _avgNumMultiValues;
- private int _maxNumMultiValuesPerRowOverride;
private File _consumerDir;
public Builder withMemoryManager(PinotDataBufferMemoryManager
memoryManager) {
@@ -169,13 +162,13 @@ public class MutableIndexContext {
return this;
}
- public Builder withAvgNumMultiValues(int avgNumMultiValues) {
- _avgNumMultiValues = avgNumMultiValues;
+ public Builder withMaxNumMultiValues(int maxNumMultiValues) {
+ _maxNumMultiValues = maxNumMultiValues;
return this;
}
- public Builder withMaxNumMultiValuesPerRowOverride(int
maxNumMultiValuesPerRowOverride) {
- _maxNumMultiValuesPerRowOverride = maxNumMultiValuesPerRowOverride;
+ public Builder withAvgNumMultiValues(int avgNumMultiValues) {
+ _avgNumMultiValues = avgNumMultiValues;
return this;
}
@@ -192,7 +185,7 @@ public class MutableIndexContext {
public MutableIndexContext build() {
return new MutableIndexContext(Objects.requireNonNull(_fieldSpec),
_fixedLengthBytes, _hasDictionary,
Objects.requireNonNull(_segmentName),
Objects.requireNonNull(_memoryManager), _capacity, _offHeap,
- _estimatedColSize, _estimatedCardinality, _avgNumMultiValues,
_maxNumMultiValuesPerRowOverride, _consumerDir);
+ _estimatedColSize, _estimatedCardinality, _maxNumMultiValues,
_avgNumMultiValues, _consumerDir);
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
