createDeserializerForL
@Override
public SegmentDictionaryCreator createIndexCreator(IndexCreationContext context, DictionaryIndexConfig indexConfig) {
boolean useVarLengthDictionary = shouldUseVarLengthDictionary(context, indexConfig);
- return new SegmentDictionaryCreator(context.getFieldSpec(), context.getIndexDir(), useVarLengthDictionary);
+ SegmentDictionaryCreator creator =
+ new SegmentDictionaryCreator(context.getFieldSpec(), context.getIndexDir(), useVarLengthDictionary);
+ configureOnHeapLookupMap(creator, context);
+ return creator;
+ }
+
+ /**
+ * For high-cardinality STRING/BYTES columns, configures the dictionary creator to skip the on-heap value->dictId
+ * map (and let the sorted unique-values array be released early) when the column's estimated on-heap footprint
+ * {@code (60 + maxLen) * cardinality} reaches the configured byte budget
+ * ({@link IndexCreationContext#getDictionaryOnHeapLookupMapMaxBytes()}); above the budget dictIds are resolved by
+ * binary-searching the mmap-backed dictionary buffer instead. No-op (default behavior) for numeric columns or when
+ * the budget is disabled ({@link Long#MAX_VALUE} or non-positive).
+ */
+ private static void configureOnHeapLookupMap(SegmentDictionaryCreator creator, IndexCreationContext context) {
+ long maxBytesBudget = context.getDictionaryOnHeapLookupMapMaxBytes();
+ if (maxBytesBudget <= 0 || maxBytesBudget == Long.MAX_VALUE) {
+ // Disabled: keep the on-heap map for all cardinalities (no behavior change).
+ return;
+ }
+ DataType storedType = context.getFieldSpec().getDataType().getStoredType();
+ if (storedType != DataType.STRING && storedType != DataType.BYTES) {
+ // Numeric dictionaries use cheap primitive maps and are never the OOM driver; leave them untouched.
+ return;
+ }
+ // Per-value on-heap footprint of the eliminated structures: 60 bytes overhead + the value's max length in bytes.
+ long perValueBytes =
+ APPROX_ON_HEAP_LOOKUP_BYTES_PER_VALUE_OVERHEAD + Math.max(context.getLengthOfLongestEntry(), 0);
+ // We want binary-search lookup exactly when perValueBytes * cardinality >= budget, i.e.
+ // cardinality >= ceil(budget / perValueBytes). The creator flips on a strict cardinality > threshold, so
+ // threshold = ceil(budget / perValueBytes) - 1. Computed without overflow (budget < Long.MAX_VALUE here).
+ long ceilCardinality = maxBytesBudget / perValueBytes + (maxBytesBudget % perValueBytes == 0 ? 0 : 1);
+ long thresholdCardinality = ceilCardinality - 1;
+ int clampedThreshold;
+ if (thresholdCardinality >= Integer.MAX_VALUE) {
+ // No int cardinality can exceed this, so the map is always kept (effectively disabled for this column).
+ clampedThreshold = Integer.MAX_VALUE;
+ } else if (thresholdCardinality < 0) {
+ clampedThreshold = 0;
+ } else {
+ clampedThreshold = (int) thresholdCardinality;
+ }
+ creator.setOnHeapLookupMapMaxCardinality(clampedThreshold);
}
public boolean shouldUseVarLengthDictionary(IndexCreationContext context, DictionaryIndexConfig indexConfig) {
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentDictionaryCreatorBinarySearchLookupTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentDictionaryCreatorBinarySearchLookupTest.java
new file mode 100644
index 0000000000..28cbfee189
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentDictionaryCreatorBinarySearchLookupTest.java
@@ -0,0 +1,255 @@
+/**
+ * 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.segment.creator.impl;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Random;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.PinotBuffersAfterMethodCheckRule;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/**
+ * Verifies that the binary-search dictId lookup path of {@link SegmentDictionaryCreator} (used for high-cardinality
+ * STRING/BYTES columns above the configured threshold) returns dictIds identical to the on-heap map path. The dictId of
+ * a value equals its position in the sorted dictionary, so both paths must agree with each other and with the index.
+ */
+public class SegmentDictionaryCreatorBinarySearchLookupTest implements PinotBuffersAfterMethodCheckRule {
+ private static final File TEMP_DIR =
+ new File(FileUtils.getTempDirectory(), "SegmentDictionaryCreatorBinarySearchLookupTest");
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ FileUtils.forceMkdir(TEMP_DIR);
+ }
+
+ @AfterClass
+ public void tearDown() {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ }
+
+ @DataProvider(name = "stringConfigs")
+ public Object[][] stringConfigs() {
+ // {cardinality, avgLength, useVarLengthDictionary}
+ return new Object[][]{
+ {1, 8, false}, {1, 8, true},
+ {2, 8, false}, {2, 8, true},
+ {16, 16, false}, {16, 16, true},
+ {1000, 32, false}, {1000, 32, true},
+ {5000, 64, false}, {5000, 64, true},
+ {10000, 4, false}, {10000, 4, true}
+ };
+ }
+
+ @DataProvider(name = "bytesConfigs")
+ public Object[][] bytesConfigs() {
+ return new Object[][]{
+ {1, 8, false}, {1, 8, true},
+ {16, 16, false}, {16, 16, true},
+ {1000, 32, false}, {1000, 32, true},
+ {5000, 12, false}, {5000, 12, true}
+ };
+ }
+
+ /**
+ * For STRING: builds the dictionary twice from the same sorted values — once in on-heap map mode, once forced into
+ * binary-search mode — and asserts indexOfSV/indexOfMV agree with each other and with the sorted position for every
+ * value.
+ */
+ @Test(dataProvider = "stringConfigs")
+ public void testStringMapVsBinarySearchEquivalence(int cardinality, int avgLength, boolean useVarLength)
+ throws Exception {
+ String[] sortedValues = generateSortedStrings(cardinality, avgLength);
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.STRING, true);
+
+ // Map mode: threshold above cardinality keeps the on-heap map.
+ try (SegmentDictionaryCreator mapCreator =
+ new SegmentDictionaryCreator(fieldSpec, dir("str_map"), useVarLength)) {
+ mapCreator.setOnHeapLookupMapMaxCardinality(Integer.MAX_VALUE);
+ mapCreator.build(sortedValues);
+ assertFalse(mapCreator.isUsingBinarySearchLookup());
+
+ // Binary-search mode: threshold 0 forces binary search (cardinality > 0 always).
+ try (SegmentDictionaryCreator bsCreator =
+ new SegmentDictionaryCreator(fieldSpec, dir("str_bs"), useVarLength)) {
+ bsCreator.setOnHeapLookupMapMaxCardinality(0);
+ bsCreator.build(sortedValues);
+ assertTrue(bsCreator.isUsingBinarySearchLookup());
+
+ for (int i = 0; i < cardinality; i++) {
+ String value = sortedValues[i];
+ assertEquals(mapCreator.indexOfSV(value), i, "map dictId mismatch at " + i);
+ assertEquals(bsCreator.indexOfSV(value), i, "binary-search dictId mismatch at " + i);
+ }
+ // Multi-value lookup over a shuffled subset.
+ Object[] mvValues = shuffledSubset(sortedValues);
+ int[] mapMv = mapCreator.indexOfMV(mvValues);
+ int[] bsMv = bsCreator.indexOfMV(mvValues);
+ assertEquals(bsMv, mapMv, "indexOfMV mismatch between map and binary-search modes");
+ }
+ }
+ }
+
+ /**
+ * For BYTES: same equivalence check as STRING.
+ */
+ @Test(dataProvider = "bytesConfigs")
+ public void testBytesMapVsBinarySearchEquivalence(int cardinality, int avgLength, boolean useVarLength)
+ throws Exception {
+ ByteArray[] sortedValues = generateSortedByteArrays(cardinality, avgLength);
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.BYTES, true);
+
+ try (SegmentDictionaryCreator mapCreator =
+ new SegmentDictionaryCreator(fieldSpec, dir("bytes_map"), useVarLength)) {
+ mapCreator.setOnHeapLookupMapMaxCardinality(Integer.MAX_VALUE);
+ mapCreator.build(sortedValues);
+ assertFalse(mapCreator.isUsingBinarySearchLookup());
+
+ try (SegmentDictionaryCreator bsCreator =
+ new SegmentDictionaryCreator(fieldSpec, dir("bytes_bs"), useVarLength)) {
+ bsCreator.setOnHeapLookupMapMaxCardinality(0);
+ bsCreator.build(sortedValues);
+ assertTrue(bsCreator.isUsingBinarySearchLookup());
+
+ for (int i = 0; i < cardinality; i++) {
+ byte[] value = sortedValues[i].getBytes();
+ assertEquals(mapCreator.indexOfSV(value), i, "map dictId mismatch at " + i);
+ assertEquals(bsCreator.indexOfSV(value), i, "binary-search dictId mismatch at " + i);
+ }
+ Object[] mvValues = new Object[Math.min(cardinality, 32)];
+ for (int i = 0; i < mvValues.length; i++) {
+ mvValues[i] = sortedValues[(i * 7) % cardinality].getBytes();
+ }
+ int[] mapMv = mapCreator.indexOfMV(mvValues);
+ int[] bsMv = bsCreator.indexOfMV(mvValues);
+ assertEquals(bsMv, mapMv, "indexOfMV mismatch between map and binary-search modes");
+ }
+ }
+ }
+
+ /**
+ * The mode is decided strictly by {@code cardinality > threshold}, evaluated per column against that column's own
+ * cardinality.
+ */
+ @Test
+ public void testThresholdBoundary()
+ throws Exception {
+ String[] sortedValues = generateSortedStrings(100, 16);
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.STRING, true);
+
+ // threshold == cardinality -> NOT greater -> map mode.
+ try (SegmentDictionaryCreator creator = new SegmentDictionaryCreator(fieldSpec, dir("boundary_eq"), false)) {
+ creator.setOnHeapLookupMapMaxCardinality(100);
+ creator.build(sortedValues);
+ assertFalse(creator.isUsingBinarySearchLookup(), "cardinality == threshold should stay on map");
+ }
+
+ // threshold == cardinality - 1 -> greater -> binary search.
+ try (SegmentDictionaryCreator creator = new SegmentDictionaryCreator(fieldSpec, dir("boundary_lt"), false)) {
+ creator.setOnHeapLookupMapMaxCardinality(99);
+ creator.build(sortedValues);
+ assertTrue(creator.isUsingBinarySearchLookup(), "cardinality > threshold should use binary search");
+ for (int i = 0; i < sortedValues.length; i++) {
+ assertEquals(creator.indexOfSV(sortedValues[i]), i);
+ }
+ }
+ }
+
+ /**
+ * Default threshold keeps existing behavior (on-heap map) regardless of cardinality.
+ */
+ @Test
+ public void testDefaultKeepsMapMode()
+ throws Exception {
+ String[] sortedValues = generateSortedStrings(1000, 16);
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.STRING, true);
+ try (SegmentDictionaryCreator creator = new SegmentDictionaryCreator(fieldSpec, dir("default_mode"), false)) {
+ creator.build(sortedValues);
+ assertFalse(creator.isUsingBinarySearchLookup(), "default must keep the on-heap map (no behavior change)");
+ for (int i = 0; i < sortedValues.length; i++) {
+ assertEquals(creator.indexOfSV(sortedValues[i]), i);
+ }
+ }
+ }
+
+ private File dir(String name)
+ throws Exception {
+ File dir = new File(TEMP_DIR, name);
+ FileUtils.deleteQuietly(dir);
+ FileUtils.forceMkdir(dir);
+ return dir;
+ }
+
+ private static String[] generateSortedStrings(int cardinality, int avgLength) {
+ Random random = new Random(cardinality * 31L + avgLength);
+ String[] values = new String[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ StringBuilder sb = new StringBuilder();
+ // Prefix with the index to guarantee uniqueness, then pad with random chars to reach ~avgLength.
+ sb.append(i).append('_');
+ while (sb.length() < avgLength) {
+ sb.append((char) ('a' + random.nextInt(26)));
+ }
+ values[i] = sb.toString();
+ }
+ Arrays.sort(values);
+ return values;
+ }
+
+ private static ByteArray[] generateSortedByteArrays(int cardinality, int avgLength) {
+ Random random = new Random(cardinality * 17L + avgLength);
+ ByteArray[] values = new ByteArray[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ int len = Math.max(4, avgLength);
+ byte[] bytes = new byte[len];
+ random.nextBytes(bytes);
+ // Encode the index into the leading bytes to guarantee uniqueness.
+ bytes[0] = (byte) (i >>> 24);
+ bytes[1] = (byte) (i >>> 16);
+ bytes[2] = (byte) (i >>> 8);
+ bytes[3] = (byte) i;
+ values[i] = new ByteArray(bytes);
+ }
+ Arrays.sort(values);
+ return values;
+ }
+
+ private static Object[] shuffledSubset(String[] sortedValues) {
+ int size = Math.min(sortedValues.length, 32);
+ Object[] subset = new Object[size];
+ for (int i = 0; i < size; i++) {
+ subset[i] = sortedValues[(i * 7) % sortedValues.length];
+ }
+ return subset;
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/ColumnStatisticsReleaseSortedValuesTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/ColumnStatisticsReleaseSortedValuesTest.java
new file mode 100644
index 0000000000..2a34b7c9dd
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/ColumnStatisticsReleaseSortedValuesTest.java
@@ -0,0 +1,141 @@
+/**
+ * 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.segment.creator.impl.stats;
+
+import org.apache.pinot.segment.spi.creator.ColumnIndexCreationInfo;
+import org.apache.pinot.segment.spi.creator.StatsCollectorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+
+
+/**
+ * Verifies {@link org.apache.pinot.segment.spi.creator.ColumnStatistics#releaseSortedValues()} (3b): after the sorted
+ * unique-values array is released to free heap, min/max/cardinality are still served from cached scalars, the array
+ * accessor returns null, and {@link ColumnIndexCreationInfo} keeps reporting the correct cardinality/min/max.
+ */
+public class ColumnStatisticsReleaseSortedValuesTest {
+ private static final TableConfig TABLE_CONFIG =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("test").build();
+
+ private StatsCollectorConfig statsConfig(String column, DataType dataType) {
+ Schema schema = new Schema();
+ schema.addField(new DimensionFieldSpec(column, dataType, true));
+ return new StatsCollectorConfig(TABLE_CONFIG, schema, null);
+ }
+
+ @Test
+ public void testStringReleaseKeepsMinMaxCardinality() {
+ StringColumnPreIndexStatsCollector collector =
+ new StringColumnPreIndexStatsCollector("col", statsConfig("col", DataType.STRING));
+ for (String v : new String[]{"banana", "apple", "cherry", "apple"}) {
+ collector.collect(v);
+ }
+ collector.seal();
+
+ // Before release: sorted array present, scalars derived from it.
+ assertNotNull(collector.getUniqueValuesSet());
+ assertEquals(collector.getCardinality(), 3);
+ assertEquals(collector.getMinValue(), "apple");
+ assertEquals(collector.getMaxValue(), "cherry");
+
+ collector.releaseSortedValues();
+
+ // After release: array gone, scalars served from cache.
+ assertNull(collector.getUniqueValuesSet(), "sorted array should be released");
+ assertEquals(collector.getCardinality(), 3);
+ assertEquals(collector.getMinValue(), "apple");
+ assertEquals(collector.getMaxValue(), "cherry");
+
+ // ColumnIndexCreationInfo must keep reporting the right values via the cached cardinality fallback.
+ ColumnIndexCreationInfo info = new ColumnIndexCreationInfo(collector, true, false, false, "null");
+ assertEquals(info.getDistinctValueCount(), 3);
+ assertEquals(info.getMin(), "apple");
+ assertEquals(info.getMax(), "cherry");
+ assertNull(info.getSortedUniqueElementsArray());
+
+ // Idempotent.
+ collector.releaseSortedValues();
+ assertEquals(collector.getCardinality(), 3);
+ }
+
+ @Test
+ public void testBytesReleaseKeepsMinMaxCardinality() {
+ BytesColumnPredIndexStatsCollector collector =
+ new BytesColumnPredIndexStatsCollector("col", statsConfig("col", DataType.BYTES));
+ byte[] a = {1, 2};
+ byte[] b = {1, 3};
+ byte[] c = {9, 0};
+ for (byte[] v : new byte[][]{b, a, c, a}) {
+ collector.collect(v);
+ }
+ collector.seal();
+
+ assertNotNull(collector.getUniqueValuesSet());
+ assertEquals(collector.getCardinality(), 3);
+ ByteArray min = collector.getMinValue();
+ ByteArray max = collector.getMaxValue();
+ assertEquals(min, new ByteArray(a));
+ assertEquals(max, new ByteArray(c));
+
+ collector.releaseSortedValues();
+
+ assertNull(collector.getUniqueValuesSet(), "sorted array should be released");
+ assertEquals(collector.getCardinality(), 3);
+ assertEquals(collector.getMinValue(), min);
+ assertEquals(collector.getMaxValue(), max);
+
+ ColumnIndexCreationInfo info = new ColumnIndexCreationInfo(collector, true, true, false, new byte[0]);
+ assertEquals(info.getDistinctValueCount(), 3);
+ assertEquals(info.getMin(), min);
+ assertEquals(info.getMax(), max);
+ assertNull(info.getSortedUniqueElementsArray());
+ }
+
+ /**
+ * Numeric collectors inherit the default no-op {@code releaseSortedValues()} (3b is scoped to STRING/BYTES), so their
+ * sorted array must remain intact after the call.
+ */
+ @Test
+ public void testNumericReleaseIsNoOp() {
+ IntColumnPreIndexStatsCollector collector =
+ new IntColumnPreIndexStatsCollector("col", statsConfig("col", DataType.INT));
+ for (int v : new int[]{5, 1, 9, 1}) {
+ collector.collect(v);
+ }
+ collector.seal();
+
+ collector.releaseSortedValues();
+
+ // Default no-op: array still present and intact.
+ assertNotNull(collector.getUniqueValuesSet(), "numeric collector should not release its array");
+ assertEquals(collector.getCardinality(), 3);
+ assertEquals(collector.getMinValue(), 1);
+ assertEquals(collector.getMaxValue(), 9);
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/SegmentGenerationWithDictionaryOnHeapLookupBudgetTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/SegmentGenerationWithDictionaryOnHeapLookupBudgetTest.java
new file mode 100644
index 0000000000..51633c7173
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/SegmentGenerationWithDictionaryOnHeapLookupBudgetTest.java
@@ -0,0 +1,359 @@
+/**
+ * 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.segment.index.creator;
+
+import java.io.File;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.PinotBuffersAfterMethodCheckRule;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentColumnarIndexCreator;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentDictionaryCreator;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
+import org.apache.pinot.segment.spi.ColumnMetadata;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+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.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+
+/**
+ * End-to-end test for the low-heap dictionary-creation path (3a binary-search dictId lookup + 3b early sorted-array
+ * release) driven by the {@code dictionaryOnHeapLookupMapMaxBytes} byte budget.
+ *
+ * The switch is scoped to STRING/BYTES and must apply to both single-value (SV) and multi-value (MV) columns, so
+ * this segment carries all four: SV STRING, MV STRING, SV BYTES, MV BYTES (plus a LONG metric that must be untouched).
+ *
+ *
The optimization is behavior-preserving by design: whether or not a column flips to binary-search lookup, the
+ * on-disk segment is identical -- so a plain "build a correct segment" check cannot tell the flipped path even ran.
+ * This test therefore builds the same data twice -- once with the budget set low enough to flip every high-cardinality
+ * STRING/BYTES column, once with the feature disabled (default) -- loads both, and asserts (a) every row round-trips
+ * correctly through the forward-index -> dictionary path for all column types, (b) min/max/cardinality metadata are
+ * correct, and (c) the two builds are identical (parity). {@link org.apache.pinot.segment.local.segment.index
+ * .dictionary.DictionaryIndexTypeOnHeapLookupBudgetTest} separately proves the chosen budget/cardinality actually flips
+ * STRING/BYTES SV and MV columns (and never numeric), so this correctness check is not vacuous.
+ */
+public class SegmentGenerationWithDictionaryOnHeapLookupBudgetTest implements PinotBuffersAfterMethodCheckRule {
+ private static final File TEMP_DIR =
+ new File(FileUtils.getTempDirectory(), "SegmentGenerationWithDictionaryOnHeapLookupBudgetTest");
+
+ private static final String TABLE_NAME = "testTable";
+ private static final String SV_STRING_COLUMN = "svString";
+ private static final String MV_STRING_COLUMN = "mvString";
+ private static final String SV_BYTES_COLUMN = "svBytes";
+ private static final String MV_BYTES_COLUMN = "mvBytes";
+ private static final String LONG_COLUMN = "longMetric";
+
+ // The in-scope (STRING/BYTES) dictionary-encoded columns that the switch may flip.
+ private static final String[] STRING_BYTES_COLUMNS =
+ {SV_STRING_COLUMN, MV_STRING_COLUMN, SV_BYTES_COLUMN, MV_BYTES_COLUMN};
+ // All dictionary-encoded columns plus the LONG metric, for full row round-trip.
+ private static final String[] ALL_COLUMNS =
+ {SV_STRING_COLUMN, MV_STRING_COLUMN, SV_BYTES_COLUMN, MV_BYTES_COLUMN, LONG_COLUMN};
+
+ private static final int NUM_ROWS = 5000;
+ private static final int BYTES_LENGTH = 16;
+ // All values distinct: SV columns -> cardinality NUM_ROWS; MV columns (2 distinct values/row) -> cardinality 2*ROWS.
+ private static final int SV_CARDINALITY = NUM_ROWS;
+ private static final int MV_CARDINALITY = 2 * NUM_ROWS;
+
+ // Must match DictionaryIndexType.APPROX_ON_HEAP_LOOKUP_BYTES_PER_VALUE_OVERHEAD.
+ private static final long PER_VALUE_OVERHEAD = 60L;
+ // Budget small enough that even the smallest cardinality column (60 * NUM_ROWS bytes) comfortably exceeds it, so
+ // every STRING/BYTES column -- SV and MV -- flips.
+ private static final long FLIP_BUDGET_BYTES = 100_000L;
+
+ private Schema _schema;
+ private List _rows;
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ FileUtils.forceMkdir(TEMP_DIR);
+ _schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension(SV_STRING_COLUMN, DataType.STRING)
+ .addMultiValueDimension(MV_STRING_COLUMN, DataType.STRING)
+ .addSingleValueDimension(SV_BYTES_COLUMN, DataType.BYTES)
+ .addMultiValueDimension(MV_BYTES_COLUMN, DataType.BYTES)
+ .addMetric(LONG_COLUMN, DataType.LONG)
+ .build();
+ _rows = generateRows();
+ }
+
+ @AfterClass
+ public void tearDown() {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ }
+
+ /**
+ * Guards the premise of this test: with the chosen parameters every high-cardinality STRING/BYTES column's estimated
+ * on-heap footprint must exceed {@link #FLIP_BUDGET_BYTES}, so the low-heap path is actually exercised by the ON
+ * build. Uses the 60-byte per-value overhead as a lower bound (ignoring maxLen), so this holds regardless of the
+ * exact value lengths.
+ */
+ @Test
+ public void testChosenParametersExceedGate() {
+ assertTrue(PER_VALUE_OVERHEAD * SV_CARDINALITY >= FLIP_BUDGET_BYTES,
+ "SV columns must flip: 60 * " + SV_CARDINALITY + " < budget " + FLIP_BUDGET_BYTES);
+ assertTrue(PER_VALUE_OVERHEAD * MV_CARDINALITY >= FLIP_BUDGET_BYTES,
+ "MV columns must flip: 60 * " + MV_CARDINALITY + " < budget " + FLIP_BUDGET_BYTES);
+ }
+
+ /**
+ * Builds a real segment with the budget set low enough to flip every high-cardinality STRING/BYTES column (SV and MV)
+ * to binary-search lookup, loads it, and asserts every row's values (all column types) and the SV STRING column's
+ * min/max/cardinality metadata are correct, plus the cardinality of the other dictionary-encoded columns.
+ */
+ @Test
+ public void testHighCardColumnsBinarySearchPathProducesCorrectSegment()
+ throws Exception {
+ SegmentIndexCreationDriverImpl driver = buildSegment("segment_on", FLIP_BUDGET_BYTES);
+ // Directly confirm the switch fired for every in-scope STRING/BYTES column in the real build.
+ assertSwitchFired(driver, true);
+ ImmutableSegment segment = ImmutableSegmentLoader.load(driver.getOutputDirectory(), ReadMode.mmap);
+ try {
+ assertRowsRoundTrip(segment);
+ assertMetadata(segment);
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ /**
+ * The out-of-the-box path (knob left at its {@code Long.MAX_VALUE} default) is a complete no-op: the switch fires for
+ * no column, and the segment is still built correctly (row round-trip + metadata). Confirms disabling means "no
+ * behavior change", observed directly via the fired flag rather than merely inferred.
+ */
+ @Test
+ public void testDefaultConfigIsNoOp()
+ throws Exception {
+ SegmentIndexCreationDriverImpl driver = buildSegmentWithDefaultConfig("segment_default");
+ assertSwitchFired(driver, false);
+ ImmutableSegment segment = ImmutableSegmentLoader.load(driver.getOutputDirectory(), ReadMode.mmap);
+ try {
+ assertRowsRoundTrip(segment);
+ assertMetadata(segment);
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ /**
+ * Parity: the segment built with the feature ON (STRING/BYTES columns flipped) must be identical -- row values and
+ * metadata -- to the segment built with the feature disabled (default budget). This is the core proof that the
+ * binary-search dictId encoding is correct for both SV and MV STRING/BYTES: any divergence in the forward index or
+ * dictionary would surface here.
+ */
+ @Test
+ public void testBinarySearchPathMatchesOnHeapMapPath()
+ throws Exception {
+ SegmentIndexCreationDriverImpl onDriver = buildSegment("parity_on", FLIP_BUDGET_BYTES);
+ SegmentIndexCreationDriverImpl offDriver = buildSegment("parity_off", /*disabled*/ Long.MAX_VALUE);
+ // Confirm the two builds genuinely took different paths: ON flipped every in-scope column, OFF flipped none.
+ assertSwitchFired(onDriver, true);
+ assertSwitchFired(offDriver, false);
+
+ ImmutableSegment onSegment = ImmutableSegmentLoader.load(onDriver.getOutputDirectory(), ReadMode.mmap);
+ ImmutableSegment offSegment = ImmutableSegmentLoader.load(offDriver.getOutputDirectory(), ReadMode.mmap);
+ try {
+ try (PinotSegmentRecordReader onReader = new PinotSegmentRecordReader();
+ PinotSegmentRecordReader offReader = new PinotSegmentRecordReader()) {
+ onReader.init(onSegment);
+ offReader.init(offSegment);
+ for (int i = 0; i < NUM_ROWS; i++) {
+ GenericRow onRow = onReader.next();
+ GenericRow offRow = offReader.next();
+ for (String column : ALL_COLUMNS) {
+ assertValueEquals(onRow.getValue(column), offRow.getValue(column),
+ column + " mismatch between ON and OFF builds at row " + i);
+ }
+ }
+ }
+
+ for (String column : STRING_BYTES_COLUMNS) {
+ ColumnMetadata onMeta = onSegment.getSegmentMetadata().getColumnMetadataFor(column);
+ ColumnMetadata offMeta = offSegment.getSegmentMetadata().getColumnMetadataFor(column);
+ assertEquals(onMeta.getCardinality(), offMeta.getCardinality(), column + " cardinality mismatch");
+ assertEquals(onMeta.getMinValue(), offMeta.getMinValue(), column + " min mismatch");
+ assertEquals(onMeta.getMaxValue(), offMeta.getMaxValue(), column + " max mismatch");
+ }
+ } finally {
+ onSegment.destroy();
+ offSegment.destroy();
+ }
+ }
+
+ private void assertRowsRoundTrip(ImmutableSegment segment)
+ throws Exception {
+ try (PinotSegmentRecordReader recordReader = new PinotSegmentRecordReader()) {
+ recordReader.init(segment);
+ for (int i = 0; i < NUM_ROWS; i++) {
+ GenericRow actual = recordReader.next();
+ GenericRow expected = _rows.get(i);
+ for (String column : ALL_COLUMNS) {
+ assertValueEquals(actual.getValue(column), expected.getValue(column),
+ column + " value mismatch at row " + i);
+ }
+ }
+ }
+ }
+
+ private void assertMetadata(ImmutableSegment segment) {
+ assertEquals(segment.getSegmentMetadata().getTotalDocs(), NUM_ROWS, "totalDocs mismatch");
+
+ ColumnMetadata svString = segment.getSegmentMetadata().getColumnMetadataFor(SV_STRING_COLUMN);
+ assertEquals(svString.getCardinality(), SV_CARDINALITY, "SV STRING cardinality mismatch");
+ assertEquals(svString.getMinValue(), svString(0), "SV STRING min mismatch");
+ assertEquals(svString.getMaxValue(), svString(NUM_ROWS - 1), "SV STRING max mismatch");
+
+ assertEquals(segment.getSegmentMetadata().getColumnMetadataFor(MV_STRING_COLUMN).getCardinality(), MV_CARDINALITY,
+ "MV STRING cardinality mismatch");
+ assertEquals(segment.getSegmentMetadata().getColumnMetadataFor(SV_BYTES_COLUMN).getCardinality(), SV_CARDINALITY,
+ "SV BYTES cardinality mismatch");
+ assertEquals(segment.getSegmentMetadata().getColumnMetadataFor(MV_BYTES_COLUMN).getCardinality(), MV_CARDINALITY,
+ "MV BYTES cardinality mismatch");
+ }
+
+ /**
+ * Value equality that handles multi-value columns (read back as {@code Object[]}) and BYTES columns (read back as
+ * {@code byte[]}), recursing into MV elements.
+ */
+ private static void assertValueEquals(Object actual, Object expected, String message) {
+ if (expected instanceof Object[]) {
+ Object[] expectedArray = (Object[]) expected;
+ Object[] actualArray = (Object[]) actual;
+ assertEquals(actualArray.length, expectedArray.length, message + " (MV length)");
+ for (int i = 0; i < expectedArray.length; i++) {
+ assertValueEquals(actualArray[i], expectedArray[i], message + " [" + i + "]");
+ }
+ } else if (expected instanceof byte[]) {
+ assertTrue(Arrays.equals((byte[]) actual, (byte[]) expected), message + " (bytes)");
+ } else {
+ assertEquals(actual, expected, message);
+ }
+ }
+
+ private SegmentIndexCreationDriverImpl buildSegment(String segmentName, long dictionaryOnHeapLookupMapMaxBytes)
+ throws Exception {
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
+ tableConfig.getIndexingConfig().setDictionaryOnHeapLookupMapMaxBytes(dictionaryOnHeapLookupMapMaxBytes);
+ return buildSegment(segmentName, tableConfig);
+ }
+
+ /** Builds with a completely unconfigured table (knob left at its default) -- the true out-of-the-box path. */
+ private SegmentIndexCreationDriverImpl buildSegmentWithDefaultConfig(String segmentName)
+ throws Exception {
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
+ return buildSegment(segmentName, tableConfig);
+ }
+
+ private SegmentIndexCreationDriverImpl buildSegment(String segmentName, TableConfig tableConfig)
+ throws Exception {
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig, _schema);
+ config.setOutDir(new File(TEMP_DIR, segmentName + "_out").getAbsolutePath());
+ config.setSegmentName(segmentName);
+
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(_rows));
+ driver.build();
+ return driver;
+ }
+
+ /**
+ * Directly observes, through the real build pipeline, whether each in-scope STRING/BYTES column's dictionary was
+ * built with the binary-search switch fired. The segment is byte-identical either way, so this reads the decision
+ * from the retained {@link SegmentDictionaryCreator#isUsingBinarySearchLookup()} on the driver's columnar index
+ * creator (reflection, since the driver doesn't expose it). This is what makes the correctness/parity checks
+ * non-vacuous: it proves the ON build actually flipped and the disabled build actually did not.
+ */
+ private static void assertSwitchFired(SegmentIndexCreationDriverImpl driver, boolean expectedFired)
+ throws Exception {
+ Field indexCreatorField = SegmentIndexCreationDriverImpl.class.getDeclaredField("_indexCreator");
+ indexCreatorField.setAccessible(true);
+ SegmentColumnarIndexCreator indexCreator = (SegmentColumnarIndexCreator) indexCreatorField.get(driver);
+
+ Field mapField = SegmentColumnarIndexCreator.class.getDeclaredField("_dictionaryCreatorMap");
+ mapField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map creators =
+ (Map) mapField.get(indexCreator);
+
+ for (String column : STRING_BYTES_COLUMNS) {
+ SegmentDictionaryCreator creator = creators.get(column);
+ assertNotNull(creator, "no dictionary creator retained for column " + column);
+ assertEquals(creator.isUsingBinarySearchLookup(), expectedFired,
+ column + " isUsingBinarySearchLookup expected " + expectedFired);
+ }
+ }
+
+ private List generateRows() {
+ List rows = new ArrayList<>(NUM_ROWS);
+ for (int i = 0; i < NUM_ROWS; i++) {
+ GenericRow row = new GenericRow();
+ // All values distinct and, for the SV STRING column, fixed-width and lexicographically ordered by index, so
+ // cardinalities and SV STRING min/max are deterministic. Rows are emitted in index order (doc order for unsorted
+ // columns), so the record reader reads them back in the same order.
+ row.putValue(SV_STRING_COLUMN, svString(i));
+ row.putValue(MV_STRING_COLUMN, new Object[]{mvString(i, 'a'), mvString(i, 'b')});
+ row.putValue(SV_BYTES_COLUMN, bytesOf(i, (byte) 0));
+ row.putValue(MV_BYTES_COLUMN, new Object[]{bytesOf(i, (byte) 1), bytesOf(i, (byte) 2)});
+ row.putValue(LONG_COLUMN, (long) i);
+ rows.add(row);
+ }
+ return rows;
+ }
+
+ private static String svString(int i) {
+ return String.format("val_%07d", i);
+ }
+
+ private static String mvString(int i, char suffix) {
+ return String.format("mv%c_%07d", suffix, i);
+ }
+
+ private static byte[] bytesOf(int i, byte marker) {
+ byte[] bytes = new byte[BYTES_LENGTH];
+ // Leading marker distinguishes the SV column from the two MV elements; the encoded index makes every value unique.
+ bytes[0] = marker;
+ bytes[1] = (byte) (i >>> 24);
+ bytes[2] = (byte) (i >>> 16);
+ bytes[3] = (byte) (i >>> 8);
+ bytes[4] = (byte) i;
+ return bytes;
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/dictionary/DictionaryIndexTypeOnHeapLookupBudgetTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/dictionary/DictionaryIndexTypeOnHeapLookupBudgetTest.java
new file mode 100644
index 0000000000..02c555c725
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/dictionary/DictionaryIndexTypeOnHeapLookupBudgetTest.java
@@ -0,0 +1,305 @@
+/**
+ * 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.segment.index.dictionary;
+
+import java.io.File;
+import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.Random;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.PinotBuffersAfterMethodCheckRule;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentDictionaryCreator;
+import org.apache.pinot.segment.spi.creator.IndexCreationContext;
+import org.apache.pinot.segment.spi.index.DictionaryIndexConfig;
+import org.apache.pinot.spi.config.table.IndexingConfig;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/**
+ * Verifies the byte-budget gate that {@link DictionaryIndexType#createIndexCreator(IndexCreationContext,
+ * DictionaryIndexConfig)} wires onto {@link SegmentDictionaryCreator} for high-cardinality STRING/BYTES columns.
+ *
+ * The feature (3a binary-search dictId lookup + 3b early sorted-array release) fires when a column's estimated
+ * on-heap footprint {@code (60 + maxLen) * cardinality} reaches the configured budget
+ * ({@link IndexingConfig#getDictionaryOnHeapLookupMapMaxBytes()}). {@code createIndexCreator} converts that byte budget
+ * into a per-column cardinality threshold and feeds it to the creator; the creator then flips on
+ * {@code cardinality > threshold} during {@link SegmentDictionaryCreator#build(Object)}. These tests assert the flip
+ * decision (observed via {@link SegmentDictionaryCreator#isUsingBinarySearchLookup()}) matches the byte-budget formula
+ * exactly at the boundary, is a no-op for numeric columns, and stays off when the budget is disabled -- so
+ * {@link SegmentGenerationWithDictionaryOnHeapLookupBudgetTest}'s end-to-end correctness check is not vacuous.
+ */
+public class DictionaryIndexTypeOnHeapLookupBudgetTest implements PinotBuffersAfterMethodCheckRule {
+ private static final File TEMP_DIR =
+ new File(FileUtils.getTempDirectory(), "DictionaryIndexTypeOnHeapLookupBudgetTest");
+
+ // Must match DictionaryIndexType.APPROX_ON_HEAP_LOOKUP_BYTES_PER_VALUE_OVERHEAD.
+ private static final long PER_VALUE_OVERHEAD = 60L;
+
+ private final DictionaryIndexType _indexType = new DictionaryIndexPlugin().getIndexType();
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ FileUtils.forceMkdir(TEMP_DIR);
+ }
+
+ @AfterClass
+ public void tearDown() {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ }
+
+ /**
+ * At a fixed (budget, maxLen), the column flips exactly at the smallest cardinality whose estimated footprint
+ * {@code (60 + maxLen) * cardinality} reaches the budget: one below stays on the on-heap map, at/above flips to
+ * binary-search lookup.
+ */
+ @DataProvider(name = "budgetConfigs")
+ public Object[][] budgetConfigs() {
+ // {budgetBytes, maxLen, singleValue}. boundaryCardinality is derived so both sides of the boundary are exercised.
+ // Each (budget, maxLen) is run for both single-value (SV) and multi-value (MV) columns: the gate keys off the
+ // stored type only, so STRING/BYTES must flip identically whether SV or MV.
+ long[][] budgets = {
+ {1000L, 40L}, // perValue 100 -> boundary cardinality 10
+ {1000L, 15L}, // perValue 75 -> boundary cardinality ceil(1000/75) = 14
+ {10_000L, 90L}, // perValue 150 -> boundary cardinality ceil(10000/150) = 67
+ {512L, 4L} // perValue 64 -> boundary cardinality 8
+ };
+ Object[][] configs = new Object[budgets.length * 2][];
+ int idx = 0;
+ for (long[] budget : budgets) {
+ configs[idx++] = new Object[]{budget[0], budget[1], true};
+ configs[idx++] = new Object[]{budget[0], budget[1], false};
+ }
+ return configs;
+ }
+
+ @Test(dataProvider = "budgetConfigs")
+ public void testStringBoundary(long budgetBytes, long maxLen, boolean singleValue)
+ throws Exception {
+ long perValue = PER_VALUE_OVERHEAD + maxLen;
+ int boundaryCardinality = (int) ((budgetBytes + perValue - 1) / perValue); // ceil(budget / perValue)
+
+ // One below the boundary: footprint < budget -> keep the on-heap map.
+ assertStringFlip((int) maxLen, budgetBytes, boundaryCardinality - 1, singleValue, false);
+ // At the boundary: footprint >= budget -> flip to binary-search lookup.
+ assertStringFlip((int) maxLen, budgetBytes, boundaryCardinality, singleValue, true);
+ // Comfortably above the boundary: still flipped.
+ assertStringFlip((int) maxLen, budgetBytes, boundaryCardinality + 5, singleValue, true);
+ }
+
+ @Test(dataProvider = "budgetConfigs")
+ public void testBytesBoundary(long budgetBytes, long maxLen, boolean singleValue)
+ throws Exception {
+ long perValue = PER_VALUE_OVERHEAD + maxLen;
+ int boundaryCardinality = (int) ((budgetBytes + perValue - 1) / perValue);
+
+ assertBytesFlip((int) maxLen, budgetBytes, boundaryCardinality - 1, singleValue, false);
+ assertBytesFlip((int) maxLen, budgetBytes, boundaryCardinality, singleValue, true);
+ }
+
+ /**
+ * The switch is scoped to STRING/BYTES: every other stored type (numeric primitives and BIG_DECIMAL) uses cheap maps
+ * and must never flip, regardless of the budget or single-value/multi-value, even at a tiny budget that would flip an
+ * equivalent STRING/BYTES column.
+ */
+ @Test
+ public void testNonStringBytesTypesNeverFlip()
+ throws Exception {
+ DataType[] types = {DataType.INT, DataType.LONG, DataType.FLOAT, DataType.DOUBLE, DataType.BIG_DECIMAL};
+ for (DataType type : types) {
+ for (boolean singleValue : new boolean[]{true, false}) {
+ // BIG_DECIMAL is a single-value-only stored type; skip the MV case for it.
+ if (type == DataType.BIG_DECIMAL && !singleValue) {
+ continue;
+ }
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", type, singleValue);
+ // Tiny budget (1 byte) that would flip an equivalent STRING/BYTES column of any cardinality.
+ IndexCreationContext context =
+ context(fieldSpec, dir("nonflip_" + type + "_" + (singleValue ? "sv" : "mv")), 8, /*budget*/ 1L, 1000);
+ try (SegmentDictionaryCreator creator = _indexType.createIndexCreator(context, DictionaryIndexConfig.DEFAULT)) {
+ creator.build(sortedValuesFor(type, 1000));
+ assertFalse(creator.isUsingBinarySearchLookup(),
+ type + (singleValue ? " (SV)" : " (MV)") + " must never flip to binary-search lookup");
+ }
+ }
+ }
+ }
+
+ /**
+ * The default budget ({@link Long#MAX_VALUE}) disables the optimization: the estimated footprint can never reach it,
+ * so a STRING column stays on the on-heap map regardless of cardinality (no behavior change).
+ */
+ @Test
+ public void testDisabledBudgetKeepsMapMode()
+ throws Exception {
+ // Out-of-the-box (unconfigured) tables must have the feature disabled -> no behavior change by default.
+ assertEquals(new IndexingConfig().getDictionaryOnHeapLookupMapMaxBytes(),
+ IndexingConfig.DEFAULT_DICTIONARY_ON_HEAP_LOOKUP_MAP_MAX_BYTES);
+ assertEquals(IndexingConfig.DEFAULT_DICTIONARY_ON_HEAP_LOOKUP_MAP_MAX_BYTES, Long.MAX_VALUE);
+
+ long disabledBudget = IndexingConfig.DEFAULT_DICTIONARY_ON_HEAP_LOOKUP_MAP_MAX_BYTES;
+ assertStringFlip(32, disabledBudget, 100_000, true, false);
+ // A non-positive budget is likewise treated as disabled.
+ assertStringFlip(32, 0L, 100_000, true, false);
+ assertStringFlip(32, -1L, 100_000, true, false);
+ }
+
+ private void assertStringFlip(int maxLen, long budgetBytes, int cardinality, boolean singleValue,
+ boolean expectedFlip)
+ throws Exception {
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.STRING, singleValue);
+ IndexCreationContext context =
+ context(fieldSpec, dir("str_" + budgetBytes + "_" + maxLen + "_" + cardinality + "_" + singleValue), maxLen,
+ budgetBytes, cardinality);
+ try (SegmentDictionaryCreator creator = _indexType.createIndexCreator(context, DictionaryIndexConfig.DEFAULT)) {
+ creator.build(generateSortedStrings(cardinality, maxLen));
+ assertFlip(creator, expectedFlip, "STRING" + (singleValue ? " (SV)" : " (MV)"), budgetBytes, maxLen, cardinality);
+ }
+ }
+
+ private void assertBytesFlip(int maxLen, long budgetBytes, int cardinality, boolean singleValue, boolean expectedFlip)
+ throws Exception {
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.BYTES, singleValue);
+ IndexCreationContext context =
+ context(fieldSpec, dir("bytes_" + budgetBytes + "_" + maxLen + "_" + cardinality + "_" + singleValue), maxLen,
+ budgetBytes, cardinality);
+ try (SegmentDictionaryCreator creator = _indexType.createIndexCreator(context, DictionaryIndexConfig.DEFAULT)) {
+ creator.build(generateSortedByteArrays(cardinality, maxLen));
+ assertFlip(creator, expectedFlip, "BYTES" + (singleValue ? " (SV)" : " (MV)"), budgetBytes, maxLen, cardinality);
+ }
+ }
+
+ private static void assertFlip(SegmentDictionaryCreator creator, boolean expectedFlip, String type, long budgetBytes,
+ int maxLen, int cardinality) {
+ String message = String.format(
+ "%s column with budget=%d, maxLen=%d, cardinality=%d: estimated footprint (60+%d)*%d=%d %s budget %d", type,
+ budgetBytes, maxLen, cardinality, maxLen, cardinality, (PER_VALUE_OVERHEAD + maxLen) * cardinality,
+ expectedFlip ? ">=" : "<", budgetBytes);
+ if (expectedFlip) {
+ assertTrue(creator.isUsingBinarySearchLookup(), "expected binary-search lookup: " + message);
+ } else {
+ assertFalse(creator.isUsingBinarySearchLookup(), "expected on-heap map: " + message);
+ }
+ }
+
+ private IndexCreationContext context(FieldSpec fieldSpec, File indexDir, int maxLen, long budgetBytes,
+ int cardinality) {
+ return IndexCreationContext.builder()
+ .withIndexDir(indexDir)
+ .withFieldSpec(fieldSpec)
+ .withLengthOfLongestEntry(maxLen)
+ .withCardinality(cardinality)
+ .withDictionaryOnHeapLookupMapMaxBytes(budgetBytes)
+ .build();
+ }
+
+ private File dir(String name)
+ throws Exception {
+ File dir = new File(TEMP_DIR, name);
+ FileUtils.deleteQuietly(dir);
+ FileUtils.forceMkdir(dir);
+ return dir;
+ }
+
+ private static Object sortedValuesFor(DataType type, int cardinality) {
+ switch (type) {
+ case INT: {
+ int[] values = new int[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ values[i] = i;
+ }
+ return values;
+ }
+ case LONG: {
+ long[] values = new long[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ values[i] = i;
+ }
+ return values;
+ }
+ case FLOAT: {
+ float[] values = new float[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ values[i] = i;
+ }
+ return values;
+ }
+ case DOUBLE: {
+ double[] values = new double[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ values[i] = i;
+ }
+ return values;
+ }
+ case BIG_DECIMAL: {
+ BigDecimal[] values = new BigDecimal[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ values[i] = BigDecimal.valueOf(i);
+ }
+ return values;
+ }
+ default:
+ throw new IllegalArgumentException("Unsupported non-STRING/BYTES type: " + type);
+ }
+ }
+
+ private static String[] generateSortedStrings(int cardinality, int maxLen) {
+ Random random = new Random(cardinality * 31L + maxLen);
+ String[] values = new String[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(i).append('_');
+ while (sb.length() < maxLen) {
+ sb.append((char) ('a' + random.nextInt(26)));
+ }
+ values[i] = sb.toString();
+ }
+ Arrays.sort(values);
+ return values;
+ }
+
+ private static ByteArray[] generateSortedByteArrays(int cardinality, int maxLen) {
+ Random random = new Random(cardinality * 17L + maxLen);
+ ByteArray[] values = new ByteArray[cardinality];
+ for (int i = 0; i < cardinality; i++) {
+ int len = Math.max(4, maxLen);
+ byte[] bytes = new byte[len];
+ random.nextBytes(bytes);
+ bytes[0] = (byte) (i >>> 24);
+ bytes[1] = (byte) (i >>> 16);
+ bytes[2] = (byte) (i >>> 8);
+ bytes[3] = (byte) i;
+ values[i] = new ByteArray(bytes);
+ }
+ Arrays.sort(values);
+ return values;
+ }
+}
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnIndexCreationInfo.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnIndexCreationInfo.java
index 9e47502098..ff613fd4bb 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnIndexCreationInfo.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnIndexCreationInfo.java
@@ -70,7 +70,10 @@ public Object getSortedUniqueElementsArray() {
public int getDistinctValueCount() {
Object uniqueValArray = _columnStatistics.getUniqueValuesSet();
if (uniqueValArray == null) {
- return Constants.UNKNOWN_CARDINALITY;
+ // The sorted unique-values array may have been released after dictionary build to free heap
+ // (see ColumnStatistics#releaseSortedValues()); in that case the cardinality is cached on the collector.
+ int cardinality = _columnStatistics.getCardinality();
+ return cardinality > 0 ? cardinality : Constants.UNKNOWN_CARDINALITY;
}
return ArrayUtils.getLength(uniqueValArray);
}
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnStatistics.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnStatistics.java
index e904fd0a9a..b4159f8325 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnStatistics.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnStatistics.java
@@ -40,10 +40,20 @@ public interface ColumnStatistics extends Serializable {
/**
*
- * @return An array of elements that has the unique values for this column, sorted order.
+ * @return An array of elements that has the unique values for this column, sorted order. May return {@code null}
+ * after {@link #releaseSortedValues()} has been called to free heap during segment creation.
*/
Object getUniqueValuesSet();
+ /**
+ * Releases the on-heap sorted unique-values array (the structure returned by {@link #getUniqueValuesSet()}) to free
+ * heap once it is no longer needed (i.e. after the dictionary and any value-array-consuming indexes are built).
+ * Implementations that support this must first cache anything still derived from the array — {@link #getMinValue()},
+ * {@link #getMaxValue()}, {@link #getCardinality()} — so those keep working afterwards. Default is a no-op.
+ */
+ default void releaseSortedValues() {
+ }
+
/**
*
* @return The number of unique values of this column.
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java
index 5d3a1a78ba..932f756521 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java
@@ -55,6 +55,15 @@ public interface IndexCreationContext {
int getTotalDocs();
+ /**
+ * Byte budget gating the low-heap dictionary-creation path for high-cardinality STRING/BYTES columns (see
+ * {@link IndexingConfig#getDictionaryOnHeapLookupMapMaxBytes()}). When a column's estimated on-heap footprint
+ * {@code (60 + maxLen) * cardinality} is {@code >=} this value, the dictionary creator skips the on-heap
+ * value->dictId map and resolves dictIds by binary-searching the mmap-backed dictionary buffer instead. Defaults
+ * to {@link Long#MAX_VALUE} (optimization disabled).
+ */
+ long getDictionaryOnHeapLookupMapMaxBytes();
+
boolean hasDictionary();
Comparable getMinValue();
@@ -123,6 +132,7 @@ final class Builder {
private int _cardinality;
private int _totalNumberOfEntries;
private int _totalDocs;
+ private long _dictionaryOnHeapLookupMapMaxBytes = IndexingConfig.DEFAULT_DICTIONARY_ON_HEAP_LOOKUP_MAP_MAX_BYTES;
private boolean _hasDictionary = true;
private Comparable _minValue;
private Comparable _maxValue;
@@ -226,6 +236,11 @@ public Builder withTotalDocs(int totalDocs) {
return this;
}
+ public Builder withDictionaryOnHeapLookupMapMaxBytes(long dictionaryOnHeapLookupMapMaxBytes) {
+ _dictionaryOnHeapLookupMapMaxBytes = dictionaryOnHeapLookupMapMaxBytes;
+ return this;
+ }
+
public Builder withDictionary(boolean hasDictionary) {
_hasDictionary = hasDictionary;
return this;
@@ -271,7 +286,7 @@ public Common build() {
_maxRowLengthInBytes, _onHeap, Objects.requireNonNull(_fieldSpec), _sorted, _cardinality,
_totalNumberOfEntries, _totalDocs, _hasDictionary, _minValue, _maxValue, _forwardIndexDisabled,
_sortedUniqueElementsArray, _optimizedDictionary, _fixedLength, _textCommitOnClose, _columnStatistics,
- _realtimeConversion, _consumerDir, _immutableToMutableIdMap);
+ _realtimeConversion, _consumerDir, _immutableToMutableIdMap, _dictionaryOnHeapLookupMapMaxBytes);
}
public Builder withSortedUniqueElementsArray(Object sortedUniqueElementsArray) {
@@ -308,6 +323,7 @@ final class Common implements IndexCreationContext {
private final boolean _realtimeConversion;
private final File _consumerDir;
private final int[] _immutableToMutableIdMap;
+ private final long _dictionaryOnHeapLookupMapMaxBytes;
public Common(File indexDir, int lengthOfLongestEntry,
int maxNumberOfMultiValueElements, int maxRowLengthInBytes, boolean onHeap,
@@ -315,7 +331,7 @@ public Common(File indexDir, int lengthOfLongestEntry,
int totalDocs, boolean hasDictionary, Comparable minValue, Comparable maxValue,
boolean forwardIndexDisabled, Object sortedUniqueElementsArray, boolean optimizeDictionary, boolean fixedLength,
boolean textCommitOnClose, ColumnStatistics columnStatistics, boolean realtimeConversion, File consumerDir,
- int[] immutableToMutableIdMap) {
+ int[] immutableToMutableIdMap, long dictionaryOnHeapLookupMapMaxBytes) {
_indexDir = indexDir;
_lengthOfLongestEntry = lengthOfLongestEntry;
_maxNumberOfMultiValueElements = maxNumberOfMultiValueElements;
@@ -338,6 +354,7 @@ public Common(File indexDir, int lengthOfLongestEntry,
_realtimeConversion = realtimeConversion;
_consumerDir = consumerDir;
_immutableToMutableIdMap = immutableToMutableIdMap;
+ _dictionaryOnHeapLookupMapMaxBytes = dictionaryOnHeapLookupMapMaxBytes;
}
public FieldSpec getFieldSpec() {
@@ -380,6 +397,11 @@ public int getTotalDocs() {
return _totalDocs;
}
+ @Override
+ public long getDictionaryOnHeapLookupMapMaxBytes() {
+ return _dictionaryOnHeapLookupMapMaxBytes;
+ }
+
public boolean hasDictionary() {
return _hasDictionary;
}
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentGeneratorConfig.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentGeneratorConfig.java
index 84614a52c4..1145fc87cd 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentGeneratorConfig.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentGeneratorConfig.java
@@ -111,6 +111,7 @@ public enum TimeColumnType {
private boolean _optimizeDictionaryType = false;
private double _noDictionarySizeRatioThreshold = IndexingConfig.DEFAULT_NO_DICTIONARY_SIZE_RATIO_THRESHOLD;
private Double _noDictionaryCardinalityRatioThreshold;
+ private long _dictionaryOnHeapLookupMapMaxBytes = IndexingConfig.DEFAULT_DICTIONARY_ON_HEAP_LOOKUP_MAP_MAX_BYTES;
private boolean _realtimeConversion = false;
// consumerDir contains data from the consuming segment, and is used during _realtimeConversion optimization
private File _consumerDir;
@@ -159,6 +160,7 @@ public SegmentGeneratorConfig(TableConfig tableConfig, Schema schema) {
_optimizeDictionaryType = indexingConfig.isOptimizeDictionaryType();
_noDictionarySizeRatioThreshold = indexingConfig.getNoDictionarySizeRatioThreshold();
_noDictionaryCardinalityRatioThreshold = indexingConfig.getNoDictionaryCardinalityRatioThreshold();
+ _dictionaryOnHeapLookupMapMaxBytes = indexingConfig.getDictionaryOnHeapLookupMapMaxBytes();
// Star-tree configs
setStarTreeIndexConfigs(indexingConfig.getStarTreeIndexConfigs());
@@ -663,6 +665,14 @@ public void setNoDictionaryCardinalityRatioThreshold(@Nullable Double noDictiona
_noDictionaryCardinalityRatioThreshold = noDictionaryCardinalityRatioThreshold;
}
+ public long getDictionaryOnHeapLookupMapMaxBytes() {
+ return _dictionaryOnHeapLookupMapMaxBytes;
+ }
+
+ public void setDictionaryOnHeapLookupMapMaxBytes(long dictionaryOnHeapLookupMapMaxBytes) {
+ _dictionaryOnHeapLookupMapMaxBytes = dictionaryOnHeapLookupMapMaxBytes;
+ }
+
public boolean isFailOnEmptySegment() {
return _failOnEmptySegment;
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java
index 9c5a8c3454..51c55472f4 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java
@@ -33,6 +33,12 @@ public class IndexingConfig extends BaseJsonConfig {
// Default ratio for overriding dictionary for fixed width columns
public static final double DEFAULT_NO_DICTIONARY_SIZE_RATIO_THRESHOLD = 0.85d;
+ // Default byte budget for the on-heap dictionary value->dictId lookup map during segment creation. Long.MAX_VALUE
+ // effectively disables the optimization: the estimated on-heap footprint (60 + maxLen) * cardinality can never reach
+ // it, so no column flips to binary-search lookup (default behavior unchanged). See
+ // _dictionaryOnHeapLookupMapMaxBytes.
+ public static final long DEFAULT_DICTIONARY_ON_HEAP_LOOKUP_MAP_MAX_BYTES = Long.MAX_VALUE;
+
/**
* This should be equal to the one specified in RangeIndexType.
*/
@@ -97,6 +103,22 @@ public class IndexingConfig extends BaseJsonConfig {
// then create a dictionary for the column. A value around 0.1 (10%) is a reasonable starting point
private Double _noDictionaryCardinalityRatioThreshold;
+ /**
+ * Byte budget that gates the low-heap dictionary-creation path for high-cardinality STRING/BYTES columns. During
+ * offline segment generation, a dictionary-encoded column normally keeps an on-heap value->dictId map plus the
+ * sorted unique-values array alive through the whole build; for high-cardinality columns this is a dominant heap
+ * cost and a frequent OOM source. When a column's estimated on-heap footprint {@code (60 + maxLen) * cardinality}
+ * (bytes; {@code maxLen} = the column's longest value in bytes, {@code cardinality} = number of distinct values) is
+ * {@code >=} this budget, the map is skipped and dictIds are resolved by binary-searching the (already off-heap,
+ * mmap-backed) dictionary buffer instead, and the sorted array is released early. Applies only to STRING/BYTES
+ * columns (numeric dictionaries use cheap primitive maps and are never the OOM driver). The value is a size in
+ * bytes; both sides of the comparison are byte sizes (e.g. cardinality 2,000,000 with maxLen 30 estimates
+ * {@code (60 + 30) * 2,000,000 ~= 172 MB}). Defaults to {@link Long#MAX_VALUE}, which effectively disables the
+ * optimization (no behavior change); set it to a smaller byte value to enable -- e.g. 134217728 for a 128 MB budget.
+ * Note a tiny value such as {@code 100} means 100 bytes and would flip essentially every column.
+ */
+ private long _dictionaryOnHeapLookupMapMaxBytes = DEFAULT_DICTIONARY_ON_HEAP_LOOKUP_MAP_MAX_BYTES;
+
// TODO: Add a new configuration related to the segment generation
private boolean _autoGeneratedInvertedIndex;
private String _segmentNameGeneratorType;
@@ -413,6 +435,14 @@ public void setNoDictionaryCardinalityRatioThreshold(Double noDictionaryCardinal
_noDictionaryCardinalityRatioThreshold = noDictionaryCardinalityRatioThreshold;
}
+ public long getDictionaryOnHeapLookupMapMaxBytes() {
+ return _dictionaryOnHeapLookupMapMaxBytes;
+ }
+
+ public void setDictionaryOnHeapLookupMapMaxBytes(long dictionaryOnHeapLookupMapMaxBytes) {
+ _dictionaryOnHeapLookupMapMaxBytes = dictionaryOnHeapLookupMapMaxBytes;
+ }
+
public String getSegmentNameGeneratorType() {
return _segmentNameGeneratorType;
}