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 9fd510fe81f Enforce multi-value row limits before mutable-segment
writes (#19399)
9fd510fe81f is described below
commit 9fd510fe81f00ffdd8c3bceb07dfebcae3fd5aa9
Author: deepinsight coder <[email protected]>
AuthorDate: Tue Sep 1 15:05:41 2026 -0700
Enforce multi-value row limits before mutable-segment writes (#19399)
---
.../indexsegment/mutable/MutableSegmentImpl.java | 162 +++++----
.../forward/FixedByteMVMutableForwardIndex.java | 12 +-
.../segment/index/forward/ForwardIndexType.java | 8 +-
.../MutableSegmentImplMVLengthValidationTest.java | 374 +++++++++++++++++++++
.../FixedByteMVMutableForwardIndexTest.java | 33 ++
.../mutable/provider/MutableIndexContext.java | 22 +-
6 files changed, 546 insertions(+), 65 deletions(-)
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 681c546d418..6f93fad11e2 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
@@ -166,6 +166,7 @@ public class MutableSegmentImpl implements MutableSegment {
private final File _consumerDir;
private final Map<String, IndexContainer> _indexContainerMap = new
HashMap<>();
+ private final MultiValueRowLimit[] _multiValueRowLimits;
private final IdMap<FixedIntArray> _recordIdMap;
private final int _numKeyColumns;
// Cache the physical (non-virtual) field specs
@@ -177,6 +178,7 @@ public class MutableSegmentImpl implements MutableSegment {
private final PartitionDedupMetadataManager _partitionDedupMetadataManager;
private final String _dedupTimeColumn;
private final PartitionUpsertMetadataManager _partitionUpsertMetadataManager;
+ private final boolean _isPartialUpsert;
private final List<String> _upsertComparisonColumns;
private final String _deleteRecordColumn;
private final boolean _upsertDropOutOfOrderRecord;
@@ -309,6 +311,7 @@ public class MutableSegmentImpl implements MutableSegment {
// Initialize for each column
boolean hasColumnWithReuseMutableTextIndex = false;
+ List<MultiValueRowLimit> multiValueRowLimits = new ArrayList<>();
for (FieldSpec fieldSpec : _physicalFieldSpecs) {
String column = fieldSpec.getName();
@@ -325,6 +328,7 @@ public class MutableSegmentImpl implements MutableSegment {
FieldIndexConfigs indexConfigs =
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()
@@ -337,6 +341,8 @@ public class MutableSegmentImpl implements MutableSegment {
.withEstimatedCardinality(_statsHistory.getEstimatedCardinality(column))
.withEstimatedColSize(_statsHistory.getEstimatedAvgColSize(column))
.withAvgNumMultiValues(_statsHistory.getEstimatedAvgColSize(column))
+ .withMaxNumMultiValuesPerRowOverride(
+ vectorIndexConfig.isEnabled() ?
vectorIndexConfig.getVectorDimension() : 0)
.withConsumerDir(_consumerDir)
.withFixedLengthBytes(fixedByteSize).build();
@@ -393,7 +399,7 @@ public class MutableSegmentImpl implements MutableSegment {
}
Map<IndexType, MutableIndex> mutableIndexes =
- new MutableIndexes(indexConfigs.getConfig(StandardIndexes.vector()));
+ new MutableIndexes(vectorIndexConfig);
for (IndexType<?, ?, ?> indexType :
IndexService.getInstance().getAllIndexes()) {
if (!specialIndexes.contains(indexType)) {
addMutableIndex(mutableIndexes, indexType, context, indexConfigs);
@@ -405,6 +411,14 @@ 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
@@ -439,6 +453,7 @@ public class MutableSegmentImpl implements MutableSegment {
nullValueVector, sourceColumn, valueAggregator));
}
_hasColumnWithReuseMutableTextIndex = hasColumnWithReuseMutableTextIndex;
+ _multiValueRowLimits =
multiValueRowLimits.toArray(MultiValueRowLimit[]::new);
_partitionDedupMetadataManager = config.getPartitionDedupMetadataManager();
_dedupTimeColumn =
@@ -450,6 +465,7 @@ public class MutableSegmentImpl implements MutableSegment {
Preconditions.checkState(!isAggregateMetricsEnabled(),
"Metrics aggregation and upsert cannot be enabled together");
UpsertContext upsertContext =
_partitionUpsertMetadataManager.getContext();
+ _isPartialUpsert = upsertContext.getUpsertMode() ==
UpsertConfig.Mode.PARTIAL;
_upsertComparisonColumns = upsertContext.getComparisonColumns();
_deleteRecordColumn = upsertContext.getDeleteRecordColumn();
_upsertDropOutOfOrderRecord = upsertContext.isDropOutOfOrderRecord();
@@ -462,6 +478,7 @@ public class MutableSegmentImpl implements MutableSegment {
_queryableDocIds = null;
}
} else {
+ _isPartialUpsert = false;
_upsertComparisonColumns = null;
_deleteRecordColumn = null;
_upsertDropOutOfOrderRecord = false;
@@ -614,6 +631,9 @@ public class MutableSegmentImpl implements MutableSegment {
@Override
public boolean index(GenericRow row, @Nullable StreamMessageMetadata
metadata)
throws IOException {
+ IndexContainer mismatchedPartitionIndexContainer = null;
+ String mismatchedPartitionValue = null;
+ int mismatchedPartition = -1;
if (_partitionColumn != null) {
Object value = row.getValue(_partitionColumn);
Preconditions.checkState(value != null, "Failed to find value for
partition column: %s", _partitionColumn);
@@ -628,37 +648,25 @@ public class MutableSegmentImpl implements MutableSegment
{
updateIndexedAndIngestionTime(metadata);
return true;
}
- if (indexContainer._partitions.add(partition)) {
- // for every partition other than mainPartitionId, log a warning once
- _logger.warn("Found new partition: {} from partition column: {},
value: {}", partition, _partitionColumn,
- stringValue);
- }
- }
- }
-
- if (isDedupEnabled()) {
- DedupRecordInfo dedupRecordInfo = getDedupRecordInfo(row);
- if
(_partitionDedupMetadataManager.checkRecordPresentOrUpdate(dedupRecordInfo,
this)) {
- if (_serverMetrics != null) {
- _serverMetrics.addMeteredTableValue(_realtimeTableName,
ServerMeter.REALTIME_DEDUP_DROPPED, 1);
- }
- updateIndexedAndIngestionTime(metadata);
- return true;
+ mismatchedPartitionIndexContainer = indexContainer;
+ mismatchedPartitionValue = stringValue;
+ mismatchedPartition = partition;
}
}
- // Validate the length of each multi-value to ensure it can be properly
stored in the underlying forward index.
- // If the length of any MV column exceeds the capacity of a chunk in the
forward index, an exception is thrown.
- // If an exception is not thrown, it leads to a mismatch in the number of
values in the MV column compared to
- // other columns when sealing the segment (due to the overflow), causing
the sealing process to fail.
- // NOTE: We must do this before we index a single column to avoid
partially indexing the row
- validateLengthOfMVColumns(row);
-
- boolean canTakeMore;
int numDocsIndexed = _numDocsIndexed;
if (isUpsertEnabled()) {
+ // Validate the incoming row before partial-upsert strategies can copy
or expand oversized MV values.
+ validateLengthOfMVColumns(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);
+ }
+ trackMismatchedPartition(mismatchedPartitionIndexContainer,
mismatchedPartition, mismatchedPartitionValue);
+
+ boolean canTakeMore;
// NOTE: out-of-order records can not be dropped or marked when
consistent upsert view is enabled.
// Since Indexing the record and updation of _numDocsIndexed counter
happens before updating the upsert
// metadata, we wouldn't be able to actually drop or mark those records
as dropped. This order is important for
@@ -695,31 +703,58 @@ public class MutableSegmentImpl implements MutableSegment
{
canTakeMore = numDocsIndexed < _capacity;
_numDocsIndexed = numDocsIndexed;
}
- } else {
- // Update dictionary first
- updateDictionary(row);
-
- // If metrics aggregation is enabled and if the dimension values were
already seen, this will return existing
- // docId, else this will return a new docId.
- int docId = getOrCreateDocId();
-
- if (docId == numDocsIndexed) {
- // New row
- addNewRow(numDocsIndexed, row);
- // Update number of documents indexed at last to make the latest row
queryable
- canTakeMore = numDocsIndexed++ < _capacity;
- } else {
- assert isAggregateMetricsEnabled();
- aggregateMetrics(row, docId);
- canTakeMore = true;
+ updateIndexedAndIngestionTime(metadata);
+ return canTakeMore;
+ }
+
+ // Validate before dedup or partition tracking so a rejected row cannot
leave metadata state behind.
+ validateLengthOfMVColumns(row);
+ trackMismatchedPartition(mismatchedPartitionIndexContainer,
mismatchedPartition, mismatchedPartitionValue);
+
+ if (isDedupEnabled()) {
+ DedupRecordInfo dedupRecordInfo = getDedupRecordInfo(row);
+ if
(_partitionDedupMetadataManager.checkRecordPresentOrUpdate(dedupRecordInfo,
this)) {
+ if (_serverMetrics != null) {
+ _serverMetrics.addMeteredTableValue(_realtimeTableName,
ServerMeter.REALTIME_DEDUP_DROPPED, 1);
+ }
+ updateIndexedAndIngestionTime(metadata);
+ return true;
}
- _numDocsIndexed = numDocsIndexed;
}
+ // Update dictionary first
+ updateDictionary(row);
+
+ // If metrics aggregation is enabled and if the dimension values were
already seen, this will return existing
+ // docId, else this will return a new docId.
+ int docId = getOrCreateDocId();
+
+ boolean canTakeMore;
+ if (docId == numDocsIndexed) {
+ // New row
+ addNewRow(numDocsIndexed, row);
+ // Update number of documents indexed at last to make the latest row
queryable
+ canTakeMore = numDocsIndexed++ < _capacity;
+ } else {
+ assert isAggregateMetricsEnabled();
+ aggregateMetrics(row, docId);
+ canTakeMore = true;
+ }
+ _numDocsIndexed = numDocsIndexed;
+
updateIndexedAndIngestionTime(metadata);
return canTakeMore;
}
+ private void trackMismatchedPartition(@Nullable IndexContainer
indexContainer, int partition,
+ @Nullable String partitionValue) {
+ if (indexContainer != null && indexContainer._partitions.add(partition)) {
+ // for every partition other than mainPartitionId, log a warning once
+ _logger.warn("Found new partition: {} from partition column: {}, value:
{}", partition, _partitionColumn,
+ partitionValue);
+ }
+ }
+
private void updateIndexedAndIngestionTime(@Nullable StreamMessageMetadata
metadata) {
_lastIndexedTimeMs = System.currentTimeMillis();
if (metadata != null) {
@@ -790,27 +825,20 @@ public class MutableSegmentImpl implements MutableSegment
{
}
/// @param row
- /// @throws UnsupportedOperationException if the length of an MV column
would exceed the
- /// capacity of a chunk in the ForwardIndex
+ /// @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 (Map.Entry<String, IndexContainer> entry :
_indexContainerMap.entrySet()) {
- IndexContainer indexContainer = entry.getValue();
- FieldSpec fieldSpec = indexContainer._fieldSpec;
- MutableIndex forwardIndex =
indexContainer._mutableIndexes.get(StandardIndexes.forward());
- if (fieldSpec.isSingleValueField() || !(forwardIndex instanceof
FixedByteMVMutableForwardIndex)) {
+ 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[]) row.getValue(entry.getKey());
- // Note that max chunk capacity is derived from
"FixedByteMVMutableForwardIndex._maxNumberOfMultiValuesPerRow"
- // which is set to "1000" in
"ForwardIndexType.MAX_MULTI_VALUES_PER_ROW". If the number of values in the
- // multi-value entry that we are attempting to ingest is greater than
the maximum accepted value, we throw an
- // UnsupportedOperationException.
- int maxChunkCapacity = ((FixedByteMVMutableForwardIndex)
forwardIndex).getMaxChunkCapacity();
- if (values.length > maxChunkCapacity) {
- throw new UnsupportedOperationException(
- "Length of MV column " + entry.getKey() + " is longer than
ForwardIndex's capacity per chunk.");
+ 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.");
}
}
}
@@ -970,6 +998,9 @@ 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);
}
@@ -1662,6 +1693,17 @@ 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;
+ }
+ }
+
private class IndexContainer implements Closeable {
final FieldSpec _fieldSpec;
final PartitionFunction _partitionFunction;
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 6fcac48ab2c..bf91fb2b06a 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
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.segment.local.realtime.impl.forward;
+import com.google.common.base.Preconditions;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -207,7 +208,11 @@ public class FixedByteMVMutableForwardIndex implements
MutableForwardIndex {
}
private int updateHeader(int row, int numValues) {
- assert (numValues <= _maxNumberOfMultiValuesPerRow);
+ if (numValues > _maxNumberOfMultiValuesPerRow) {
+ Preconditions.checkArgument(numValues <= _maxNumberOfMultiValuesPerRow,
+ "Row %s has %s multi-values, exceeding the maximum of %s", row,
numValues,
+ _maxNumberOfMultiValuesPerRow);
+ }
_numValues += numValues;
int newStartIndex = _prevRowStartIndex + _prevRowLength;
if (newStartIndex + numValues > _currentCapacity) {
@@ -222,6 +227,11 @@ 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.
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 7f4ae0a210b..33089d6ec06 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
@@ -73,7 +73,7 @@ public class ForwardIndexType extends
AbstractIndexType<ForwardIndexConfig, Forw
public static final String INDEX_DISPLAY_NAME = "forward";
// For multi-valued column, forward-index.
- // Maximum number of multi-values per row. We assert on this.
+ // 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;
@@ -368,6 +368,8 @@ 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 =
@@ -409,7 +411,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(MAX_MULTI_VALUES_PER_ROW,
context.getAvgNumMultiValues(),
+ return new FixedByteMVMutableForwardIndex(maxNumMultiValuesPerRow,
context.getAvgNumMultiValues(),
context.getCapacity(), dataType.size(),
context.getMemoryManager(), allocationContext, false, storedType,
dataType);
}
@@ -423,7 +425,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(MAX_MULTI_VALUES_PER_ROW,
context.getAvgNumMultiValues(),
+ return new FixedByteMVMutableForwardIndex(maxNumMultiValuesPerRow,
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/MutableSegmentImplMVLengthValidationTest.java
new file mode 100644
index 00000000000..9af323b19f3
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplMVLengthValidationTest.java
@@ -0,0 +1,374 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.indexsegment.mutable;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.common.partition.function.ModuloPartitionFunction;
+import org.apache.pinot.segment.local.PinotBuffersAfterClassCheckRule;
+import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.segment.local.dedup.DedupContext;
+import org.apache.pinot.segment.local.dedup.DedupRecordInfo;
+import org.apache.pinot.segment.local.dedup.PartitionDedupMetadataManager;
+import
org.apache.pinot.segment.local.realtime.impl.forward.FixedByteMVMutableForwardIndex;
+import
org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeLuceneTextIndexSearcherPool;
+import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
+import org.apache.pinot.segment.local.upsert.RecordInfo;
+import org.apache.pinot.segment.local.upsert.TableUpsertMetadataManager;
+import org.apache.pinot.segment.local.upsert.TableUpsertMetadataManagerFactory;
+import org.apache.pinot.segment.spi.datasource.DataSource;
+import org.apache.pinot.segment.spi.index.creator.VectorIndexConfig;
+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.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;
+
+
+/// 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 {
+ 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 SMALL_VECTOR_DIMENSION = 768;
+ private static final int VECTOR_DIMENSION = 1536;
+
+ @BeforeClass
+ public void setUpSearcherPool() {
+ RealtimeLuceneTextIndexSearcherPool.init(1);
+ }
+
+ @Test
+ public void testRejectOversizedMultiValueRowBeforeWrites()
+ throws Exception {
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("mvLengthValidation")
+ .addMultiValueDimension(MV_COLUMN, FieldSpec.DataType.INT)
+ .build();
+ MutableSegmentImpl mutableSegment =
MutableSegmentImplTestUtils.createMutableSegmentImpl(schema);
+ try {
+ Object[] firstValues = createValues(3, 0);
+ 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),
+ exception.getMessage());
+ Assert.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);
+ } finally {
+ mutableSegment.destroy();
+ }
+ }
+
+ @Test
+ public void testAcceptVectorDimensionAboveDefaultMultiValueLimit()
+ throws Exception {
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("vectorMvLengthValidation")
+ .addMultiValueDimension(MV_COLUMN, FieldSpec.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);
+ try {
+ Object[] vector = createFloatValues(VECTOR_DIMENSION);
+ mutableSegment.index(createRow(vector), null);
+
+ DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
+ 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);
+ } finally {
+ mutableSegment.destroy();
+ }
+ }
+
+ @Test
+ public void testRejectVectorAboveConfiguredDimensionBeforeWrites()
+ throws Exception {
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("smallVectorMvLengthValidation")
+ .addMultiValueDimension(MV_COLUMN, FieldSpec.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);
+ try {
+ Assert.expectThrows(UnsupportedOperationException.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);
+ } finally {
+ mutableSegment.destroy();
+ }
+ }
+
+ @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)
+ .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());
+ });
+
+ 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);
+
+ Object[] validValues = createValues(3, 10000);
+ mutableSegment.index(createDedupRow(primaryKey, validValues), null);
+ Assert.assertEquals(seenPrimaryKeys.size(), 1);
+ Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 1);
+
+ DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
+ assertValues(dataSource, (FixedByteMVMutableForwardIndex)
dataSource.getForwardIndex(), 0, validValues);
+ } finally {
+ mutableSegment.destroy();
+ }
+ }
+
+ @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")
+ .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);
+ PartitionUpsertMetadataManager upsertMetadataManager =
+ spy(tableUpsertMetadataManager.getOrCreatePartitionManager(0));
+ 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);
+ verify(upsertMetadataManager,
times(0)).updateRecord(any(GenericRow.class), any(RecordInfo.class));
+ verify(upsertMetadataManager, times(0)).addRecord(eq(mutableSegment),
any(RecordInfo.class));
+
+ Object[] firstValues = createValues(600, 0);
+ mutableSegment.index(createUpsertRow(primaryKey, 1L, firstValues), null);
+
+ DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
+ 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});
+ verify(upsertMetadataManager, times(1)).addRecord(eq(mutableSegment),
any(RecordInfo.class));
+
+ Assert.expectThrows(UnsupportedOperationException.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});
+ 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});
+ verify(upsertMetadataManager, times(2)).addRecord(eq(mutableSegment),
any(RecordInfo.class));
+ assertValues(mutableSegment.getDataSource(MV_COLUMN), forwardIndex, 1,
createValues(603, 0));
+ } finally {
+ mutableSegment.destroy();
+ upsertMetadataManager.stop();
+ upsertMetadataManager.close();
+ }
+ }
+
+ @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)
+ .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));
+
+ Object[] validValues = createValues(3, 0);
+ mutableSegment.index(createPartitionedRow(1, validValues), null);
+ Assert.assertEquals(mutableSegment.getNumDocsIndexed(), 1);
+
Assert.assertEquals(mutableSegment.getDataSource(PARTITION_COLUMN).getDataSourceMetadata().getPartitions(),
+ Set.of(0, 1));
+ DataSource dataSource = mutableSegment.getDataSource(MV_COLUMN);
+ assertValues(dataSource, (FixedByteMVMutableForwardIndex)
dataSource.getForwardIndex(), 0, validValues);
+ } finally {
+ mutableSegment.destroy();
+ }
+ }
+
+ @DataProvider(name = "collectionMergeStrategies")
+ private static Object[][] collectionMergeStrategies() {
+ return new Object[][]{
+ {UpsertConfig.Strategy.APPEND},
+ {UpsertConfig.Strategy.UNION}
+ };
+ }
+
+ private static GenericRow createRow(Object[] values) {
+ GenericRow row = new GenericRow();
+ row.putValue(MV_COLUMN, values);
+ return row;
+ }
+
+ private static GenericRow createPartitionedRow(int partitionValue, Object[]
values) {
+ GenericRow row = createRow(values);
+ row.putValue(PARTITION_COLUMN, partitionValue);
+ return row;
+ }
+
+ private static GenericRow createUpsertRow(String primaryKey, long
comparisonValue, Object[] values) {
+ GenericRow row = createDedupRow(primaryKey, values);
+ row.putValue(COMPARISON_COLUMN, comparisonValue);
+ return row;
+ }
+
+ private static GenericRow createDedupRow(String primaryKey, Object[] values)
{
+ GenericRow row = createRow(values);
+ row.putValue(PRIMARY_KEY_COLUMN, primaryKey);
+ return row;
+ }
+
+ private static Object[] createValues(int length, int offset) {
+ Object[] values = new Object[length];
+ for (int i = 0; i < length; 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++) {
+ values[i] = (float) i;
+ }
+ return values;
+ }
+
+ private static float[] toPrimitiveFloatArray(Object[] values) {
+ float[] result = new float[values.length];
+ for (int i = 0; i < values.length; i++) {
+ result[i] = (float) values[i];
+ }
+ return result;
+ }
+
+ private static void assertValues(DataSource dataSource,
FixedByteMVMutableForwardIndex forwardIndex, int docId,
+ Object[] expectedValues) {
+ Dictionary dictionary = dataSource.getDictionary();
+ Assert.assertNotNull(dictionary);
+ int[] dictIds = forwardIndex.getDictIdMV(docId);
+ Assert.assertEquals(dictIds.length, expectedValues.length);
+ for (int i = 0; i < expectedValues.length; i++) {
+ Assert.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 993c9ee7c36..dabf1f95ace 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
@@ -71,6 +71,39 @@ public class FixedByteMVMutableForwardIndexTest implements
PinotBuffersAfterClas
}
}
+ @Test
+ public void testRejectMultiValuesExceedingMaxPerRow()
+ throws Exception {
+ int maxNumberOfMultiValuesPerRow = 5;
+ FixedByteMVMutableForwardIndex readerWriter =
+ new FixedByteMVMutableForwardIndex(maxNumberOfMultiValuesPerRow, 2,
10, Integer.BYTES, _memoryManager,
+ "RejectMultiValuesExceedingMaxPerRow", true,
FieldSpec.DataType.INT);
+ try {
+ Assert.expectThrows(IllegalArgumentException.class,
+ () -> readerWriter.setIntMV(0, new int[maxNumberOfMultiValuesPerRow
+ 1]));
+ Assert.assertEquals(getNumValues(readerWriter), 0);
+ Assert.assertEquals(readerWriter.getNumValuesMV(0), 0);
+ } finally {
+ readerWriter.close();
+ }
+ }
+
+ @Test
+ public void testAcceptMultiValuesAtMaxPerRow()
+ throws Exception {
+ int maxNumberOfMultiValuesPerRow = 5;
+ FixedByteMVMutableForwardIndex readerWriter =
+ new FixedByteMVMutableForwardIndex(maxNumberOfMultiValuesPerRow, 2,
10, Integer.BYTES, _memoryManager,
+ "AcceptMultiValuesAtMaxPerRow", true, FieldSpec.DataType.INT);
+ try {
+ int[] values = new int[]{1, 2, 3, 4, 5};
+ readerWriter.setIntMV(0, values);
+ Assert.assertEquals(readerWriter.getIntMV(0), values);
+ } finally {
+ readerWriter.close();
+ }
+ }
+
public void testIntArray(final long seed, boolean isDictionaryEncoded)
throws IOException {
FixedByteMVMutableForwardIndex readerWriter;
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 57d71aa2788..010baead912 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
@@ -33,6 +33,7 @@ public class MutableIndexContext {
private final int _estimatedColSize;
private final int _estimatedCardinality;
private final int _avgNumMultiValues;
+ private final int _maxNumMultiValuesPerRowOverride;
private final String _segmentName;
private final PinotDataBufferMemoryManager _memoryManager;
private final File _consumerDir;
@@ -40,6 +41,13 @@ public class MutableIndexContext {
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) {
_fieldSpec = fieldSpec;
_fixedLengthBytes = fixedLengthBytes;
_hasDictionary = hasDictionary;
@@ -50,6 +58,7 @@ public class MutableIndexContext {
_estimatedColSize = estimatedColSize;
_estimatedCardinality = estimatedCardinality;
_avgNumMultiValues = avgNumMultiValues;
+ _maxNumMultiValuesPerRowOverride = maxNumMultiValuesPerRowOverride;
_consumerDir = consumerDir;
}
@@ -93,6 +102,11 @@ public class MutableIndexContext {
return _avgNumMultiValues;
}
+ /// 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 File getConsumerDir() {
return _consumerDir;
}
@@ -112,6 +126,7 @@ public class MutableIndexContext {
private int _estimatedColSize;
private int _estimatedCardinality;
private int _avgNumMultiValues;
+ private int _maxNumMultiValuesPerRowOverride;
private File _consumerDir;
public Builder withMemoryManager(PinotDataBufferMemoryManager
memoryManager) {
@@ -159,6 +174,11 @@ public class MutableIndexContext {
return this;
}
+ public Builder withMaxNumMultiValuesPerRowOverride(int
maxNumMultiValuesPerRowOverride) {
+ _maxNumMultiValuesPerRowOverride = maxNumMultiValuesPerRowOverride;
+ return this;
+ }
+
public Builder withConsumerDir(File consumerDir) {
_consumerDir = consumerDir;
return this;
@@ -172,7 +192,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,
_consumerDir);
+ _estimatedColSize, _estimatedCardinality, _avgNumMultiValues,
_maxNumMultiValuesPerRowOverride, _consumerDir);
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
