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 f054a9073fa Update sortedness in metadata if disagree with
ColumnStatistics (#18998)
f054a9073fa is described below
commit f054a9073fa0f61b441f8e17e4ba8d7a22ca58eb
Author: Jhow <[email protected]>
AuthorDate: Thu Jul 16 12:34:56 2026 -0700
Update sortedness in metadata if disagree with ColumnStatistics (#18998)
---
.../segment/index/loader/ForwardIndexHandler.java | 47 +++++----
.../index/loader/ForwardIndexHandlerTest.java | 117 +++++++++++++++++++++
2 files changed, 142 insertions(+), 22 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java
index 70dc3cb09c5..77fbcd2c60e 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java
@@ -988,28 +988,7 @@ public class ForwardIndexHandler extends BaseIndexHandler {
ColumnMetadata existingColMetadata =
segmentMetadata.getColumnMetadataFor(column);
FieldSpec fieldSpec = existingColMetadata.getFieldSpec();
- String fwdIndexFileExtension;
- if (fieldSpec.isSingleValueField()) {
- if (existingColMetadata.isSorted()) {
- fwdIndexFileExtension =
V1Constants.Indexes.SORTED_SV_FORWARD_INDEX_FILE_EXTENSION;
- } else {
- fwdIndexFileExtension =
V1Constants.Indexes.UNSORTED_SV_FORWARD_INDEX_FILE_EXTENSION;
- }
- } else {
- fwdIndexFileExtension =
V1Constants.Indexes.UNSORTED_MV_FORWARD_INDEX_FILE_EXTENSION;
- }
- File fwdIndexFile = new File(indexDir, column + fwdIndexFileExtension);
-
- if (!inProgress.exists()) {
- // Marker file does not exist, which means last run ended normally.
- // Create a marker file.
- FileUtils.touch(inProgress);
- } else {
- // Marker file exists, which means last run was interrupted.
- // Remove forward index and dictionary files if they exist.
- FileUtils.deleteQuietly(fwdIndexFile);
- FileUtils.deleteQuietly(dictionaryFile);
- }
+ File fwdIndexFile;
AbstractColumnStatisticsCollector statsCollector;
SegmentDictionaryCreator dictionaryCreator;
@@ -1032,6 +1011,29 @@ public class ForwardIndexHandler extends
BaseIndexHandler {
}
statsCollector.seal();
}
+
+ String fwdIndexFileExtension;
+ if (fieldSpec.isSingleValueField()) {
+ if (statsCollector.isSorted()) {
+ fwdIndexFileExtension =
V1Constants.Indexes.SORTED_SV_FORWARD_INDEX_FILE_EXTENSION;
+ } else {
+ fwdIndexFileExtension =
V1Constants.Indexes.UNSORTED_SV_FORWARD_INDEX_FILE_EXTENSION;
+ }
+ } else {
+ fwdIndexFileExtension =
V1Constants.Indexes.UNSORTED_MV_FORWARD_INDEX_FILE_EXTENSION;
+ }
+ fwdIndexFile = new File(indexDir, column + fwdIndexFileExtension);
+
+ if (!inProgress.exists()) {
+ // Marker file does not exist, which means last run ended normally.
+ // Create a marker file.
+ FileUtils.touch(inProgress);
+ } else {
+ // Marker file exists, which means last run was interrupted.
+ // Remove forward index and dictionary files if they exist.
+ FileUtils.deleteQuietly(fwdIndexFile);
+ FileUtils.deleteQuietly(dictionaryFile);
+ }
DictionaryIndexConfig dictConf =
_fieldIndexConfigs.get(column).getConfig(StandardIndexes.dictionary());
boolean optimizeDictionaryType =
_tableConfig.getIndexingConfig().isOptimizeDictionaryType();
boolean useVarLength = dictConf.isUseVarLengthDictionary() ||
DictionaryIndexType.shouldUseVarLengthDictionary(
@@ -1063,6 +1065,7 @@ public class ForwardIndexHandler extends BaseIndexHandler
{
LOGGER.info("Created forwardIndex. Updating metadata properties for
segment={} and column={}", segmentName, column);
Map<String, String> metadataProperties = new HashMap<>();
+ metadataProperties.put(getKeyFor(column, IS_SORTED),
String.valueOf(statsCollector.isSorted()));
metadataProperties.put(getKeyFor(column, HAS_DICTIONARY),
String.valueOf(true));
metadataProperties.put(getKeyFor(column, FORWARD_INDEX_ENCODING),
EncodingType.DICTIONARY.name());
metadataProperties.put(getKeyFor(column, DICTIONARY_ELEMENT_SIZE),
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerTest.java
index cdb196387eb..8bc1faec4d6 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerTest.java
@@ -33,6 +33,7 @@ import java.util.Map;
import java.util.Random;
import java.util.Set;
import javax.annotation.Nullable;
+import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
@@ -75,8 +76,11 @@ import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static
org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.IS_SORTED;
+import static
org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.getKeyFor;
import static org.testng.Assert.*;
@@ -1771,6 +1775,119 @@ public class ForwardIndexHandlerTest {
metadata, false);
}
+ @DataProvider(name = "coincidentallySortedRawTypes")
+ public Object[][] coincidentallySortedRawTypes() {
+ // Cover a fixed-width type and both variable-length types (STRING/BYTES
take a different raw writer path).
+ return new Object[][]{
+ {DataType.LONG, 0L},
+ {DataType.STRING, "sameValue"},
+ {DataType.BYTES, new byte[]{0x1, 0x2, 0x3}},
+ };
+ }
+
+ /// Regression test for enabling a dictionary on a raw column whose
persisted metadata says {@code isSorted=false}
+ /// but whose values happen to be monotonic (here, all identical). This is
the shape produced by a realtime segment
+ /// committed while the column was raw: raw columns are persisted with
{@code isSorted=false} regardless of the data.
+ /// On reload the fresh stats collector reads the identical values and
reports the column as sorted, so before the
+ /// fix the forward-index creator emitted a `.sv.sorted.fwd` index while
{@link ForwardIndexHandler} derived the temp
+ /// file name from the metadata ({@code .sv.unsorted.fwd}) and never updated
{@code isSorted}, throwing
+ /// {@link java.io.FileNotFoundException} and leaving a format/metadata
mismatch for the next read. The fix forces
+ /// the creator to honor the persisted {@code isSorted} flag, keeping the
written format, the file name, and the
+ /// metadata consistent.
+ @Test(dataProvider = "coincidentallySortedRawTypes")
+ public void
testEnableDictionaryForRawColumnWithCoincidentallySortedValues(DataType
dataType, Object value)
+ throws Exception {
+ File tempDir = new File(FileUtils.getTempDirectory(),
"ForwardIndexHandlerCoincidentalSortTest");
+ FileUtils.deleteQuietly(tempDir);
+ try {
+ String tableName = "coincidentalSortTable";
+ String column = "coincidentallySortedColumn";
+ String segmentName = "coincidentalSortSegment";
+ int numDocs = 48;
+
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(tableName)
+ .addSingleValueDimension(column, dataType)
+ .build();
+
+ // Build the segment with the column as raw (no dictionary).
+ TableConfig rawTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(tableName)
+ .setNoDictionaryColumns(List.of(column))
+ .setFieldConfigList(
+ List.of(new FieldConfig(column, FieldConfig.EncodingType.RAW,
List.of(), CompressionCodec.PASS_THROUGH,
+ null)))
+ .build();
+
+ // All-identical values -> trivially monotonic -> the stats collector
treats the column as sorted.
+ List<GenericRow> rows = new ArrayList<>();
+ for (int i = 0; i < numDocs; i++) {
+ GenericRow row = new GenericRow();
+ row.putValue(column, value);
+ rows.add(row);
+ }
+ SegmentGeneratorConfig config = new
SegmentGeneratorConfig(rawTableConfig, schema);
+ config.setOutDir(tempDir.getPath());
+ config.setSegmentName(segmentName);
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(rows));
+ driver.build();
+ File segmentDir = new File(tempDir, segmentName);
+
+ // Offline generation records isSorted from the data, so it would set
isSorted=true for these all-equal values.
+ // Overwrite it to false to reproduce the realtime raw-segment condition.
+ PropertiesConfiguration metadataConfig =
SegmentMetadataUtils.getPropertiesConfiguration(segmentDir);
+ metadataConfig.setProperty(getKeyFor(column, IS_SORTED),
String.valueOf(false));
+ SegmentMetadataUtils.savePropertiesConfiguration(metadataConfig,
segmentDir);
+
+ ColumnMetadata beforeMetadata = new
SegmentMetadataImpl(segmentDir).getColumnMetadataFor(column);
+ assertFalse(beforeMetadata.isSorted());
+ assertFalse(beforeMetadata.hasDictionary());
+
+ // Enable a dictionary on the column and run the forward-index handler
(the reload path).
+ TableConfig dictTableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(tableName).setNoDictionaryColumns(List.of()).build();
+ IndexLoadingConfig indexLoadingConfig = new
IndexLoadingConfig(dictTableConfig, schema);
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = segmentDirectory.createWriter()) {
+ ForwardIndexHandler handler = new
ForwardIndexHandler(segmentDirectory, indexLoadingConfig);
+ assertEquals(handler.computeOperations(writer),
+ Map.of(column,
List.of(ForwardIndexHandler.Operation.ENABLE_DICTIONARY)));
+ // Before the fix this threw FileNotFoundException for the missing
.sv.unsorted.fwd file.
+ handler.updateIndices(writer);
+ handler.postUpdateIndicesCleanup(writer);
+ }
+
+ // The column now has a dictionary; isSorted must stay false and the
unsorted dict forward index must read back
+ // the original values without corruption.
+ ColumnMetadata afterMetadata = new
SegmentMetadataImpl(segmentDir).getColumnMetadataFor(column);
+ assertTrue(afterMetadata.hasDictionary());
+ assertTrue(afterMetadata.isSorted());
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Reader reader = segmentDirectory.createReader()) {
+ assertTrue(reader.hasIndexFor(column, StandardIndexes.forward()));
+ assertTrue(reader.hasIndexFor(column, StandardIndexes.dictionary()));
+ IndexReaderFactory<ForwardIndexReader> readerFactory =
StandardIndexes.forward().getReaderFactory();
+ FieldIndexConfigs fieldIndexConfigs =
createFieldIndexConfigsFromMetadata(afterMetadata);
+ try (ForwardIndexReader<?> fwdReader =
+ readerFactory.createIndexReader(reader, fieldIndexConfigs,
afterMetadata);
+ Dictionary dictionary = DictionaryIndexType.read(reader,
afterMetadata)) {
+ PinotSegmentColumnReader columnReader =
+ new PinotSegmentColumnReader(column, fwdReader, dictionary, null,
+ afterMetadata.getMaxNumberOfMultiValues());
+ for (int i = 0; i < numDocs; i++) {
+ Object actual = columnReader.getValue(i);
+ if (dataType == DataType.BYTES) {
+ assertEquals((byte[]) actual, (byte[]) value);
+ } else {
+ assertEquals(actual, value);
+ }
+ }
+ }
+ }
+ } finally {
+ FileUtils.deleteQuietly(tempDir);
+ }
+ }
+
@Test
public void testDisableDictionaryForMultipleColumns()
throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
