diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/JsonIndexHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/JsonIndexHandler.java index 567aed0a0a74..180f923783c3 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/JsonIndexHandler.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/JsonIndexHandler.java @@ -23,6 +23,8 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import javax.annotation.Nullable; +import org.apache.commons.configuration2.PropertiesConfiguration; import org.apache.commons.io.FileUtils; import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType; import org.apache.pinot.segment.local.segment.index.loader.BaseIndexHandler; @@ -40,10 +42,14 @@ import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader; import org.apache.pinot.segment.spi.index.reader.ForwardIndexReaderContext; import org.apache.pinot.segment.spi.store.SegmentDirectory; +import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils; import org.apache.pinot.spi.config.table.JsonIndexConfig; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.data.FieldSpec.DataType; import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.env.CommonsConfigurationUtils; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.MapUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -65,12 +71,43 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) { String segmentName = _segmentDirectory.getSegmentMetadata().getName(); Set columnsToAddIdx = new HashSet<>(_jsonIndexConfigs.keySet()); Set existingColumns = segmentReader.toSegmentDirectory().getColumnsWithIndex(StandardIndexes.json()); - // Check if any existing index need to be removed. + // Load metadata properties once to avoid repeated disk reads when checking multiple columns. + PropertiesConfiguration properties = loadMetadataProperties(); + // Check if any existing index need to be removed or if the config changed. for (String column : existingColumns) { if (!columnsToAddIdx.remove(column)) { LOGGER.info("Need to remove existing json index from segment: {}, column: {}", segmentName, column); return true; } + // Column exists in both existing indexes and config; check if config changed. + // When properties != null but storedConfig is null the index predates config persistence (legacy segment): + // only backfill (rebuild) when the forward index is present or reconstructable. If the forward index is + // absent (e.g. forwardIndexDisabled=true with no dictionary/inverted pair), preserve the existing JSON + // index untouched — createForwardIndexIfNeeded() would fail with no dictionary/inverted index available. + JsonIndexConfig currentConfig = _jsonIndexConfigs.get(column); + JsonIndexConfig storedConfig = readStoredJsonIndexConfig(column, properties); + if (properties != null) { + // configChanged: stored config exists and differs from current — rebuild regardless of forward + // index presence (same pre-existing behaviour; createForwardIndexIfNeeded will fail if the + // forward index is truly unrecoverable, which requires an explicit segment refresh). + boolean configChanged = storedConfig != null && !storedConfig.equals(currentConfig); + // legacyBackfill: stored config is absent (either the index predates config persistence, or + // the stored value failed to deserialize and readStoredJsonIndexConfig returned null). Only + // backfill when the forward index is present or reconstructable; if it is absent + // (forwardIndexDisabled=true with no dictionary/inverted pair) createForwardIndexIfNeeded() + // would throw, so preserve the existing JSON index instead. + boolean legacyBackfill = storedConfig == null && isForwardIndexAvailable(segmentReader, column); + if (configChanged) { + LOGGER.info("Need to rebuild json index for segment: {}, column: {} due to config change", segmentName, + column); + return true; + } + if (legacyBackfill) { + LOGGER.info("Need to rebuild json index for segment: {}, column: {} to backfill stored config " + + "(legacy segment or unreadable stored config)", segmentName, column); + return true; + } + } } // Check if any new index need to be added. for (String column : columnsToAddIdx) { @@ -90,21 +127,118 @@ public void updateIndices(SegmentDirectory.Writer segmentWriter) String segmentName = _segmentDirectory.getSegmentMetadata().getName(); Set columnsToAddIdx = new HashSet<>(_jsonIndexConfigs.keySet()); Set existingColumns = segmentWriter.toSegmentDirectory().getColumnsWithIndex(StandardIndexes.json()); + // Load metadata properties once; track whether we need to save changes. + PropertiesConfiguration properties = loadMetadataProperties(); + boolean metadataModified = false; for (String column : existingColumns) { if (!columnsToAddIdx.remove(column)) { LOGGER.info("Removing existing json index from segment: {}, column: {}", segmentName, column); segmentWriter.removeIndex(column, StandardIndexes.json()); LOGGER.info("Removed existing json index from segment: {}, column: {}", segmentName, column); + } else { + // Column exists in both existing indexes and config; check if config changed and rebuild if needed. + // When properties != null but storedConfig is null the index predates config persistence (legacy segment): + // only backfill (rebuild) when the forward index is present or reconstructable. If the forward index is + // absent (e.g. forwardIndexDisabled=true with no dictionary/inverted pair), preserve the existing JSON + // index untouched — createForwardIndexIfNeeded() would fail with no dictionary/inverted index available. + JsonIndexConfig currentConfig = _jsonIndexConfigs.get(column); + JsonIndexConfig storedConfig = readStoredJsonIndexConfig(column, properties); + if (properties != null) { + // configChanged: stored config exists and differs — rebuild (see needUpdateIndices for rationale). + boolean configChanged = storedConfig != null && !storedConfig.equals(currentConfig); + // legacyBackfill: absent/unreadable stored config; only rebuild when forward index is recoverable. + boolean legacyBackfill = storedConfig == null && isForwardIndexAvailable(segmentWriter, column); + if (configChanged) { + LOGGER.info("Rebuilding json index for segment: {}, column: {} due to config change", segmentName, column); + segmentWriter.removeIndex(column, StandardIndexes.json()); + columnsToAddIdx.add(column); + } else if (legacyBackfill) { + LOGGER.info("Rebuilding json index for segment: {}, column: {} to backfill stored config " + + "(legacy segment or unreadable stored config)", segmentName, column); + segmentWriter.removeIndex(column, StandardIndexes.json()); + columnsToAddIdx.add(column); + } + } } } for (String column : columnsToAddIdx) { ColumnMetadata columnMetadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column); if (shouldCreateJsonIndex(columnMetadata)) { createJsonIndexForColumn(segmentWriter, columnMetadata); + if (properties != null) { + setStoredJsonIndexConfig(column, _jsonIndexConfigs.get(column), properties); + metadataModified = true; + } + } + } + if (metadataModified && properties != null) { + SegmentMetadataUtils.savePropertiesConfiguration(properties, + _segmentDirectory.getSegmentMetadata().getIndexDir()); + } + } + + /// Loads the segment {@code metadata.properties} file, or returns {@code null} if unavailable (in-memory segments). + @Nullable + private PropertiesConfiguration loadMetadataProperties() { + try { + return SegmentMetadataUtils.getPropertiesConfiguration(_segmentDirectory.getSegmentMetadata()); + } catch (Exception e) { + LOGGER.warn("Cannot load segment metadata properties for segment: {}, config-change detection disabled", + _segmentDirectory.getSegmentMetadata().getName(), e); + return null; + } + } + + /// Reads the stored {@link JsonIndexConfig} from the given {@code properties}, or returns {@code null} if not + /// present (e.g., index was built before config persistence was added). + @Nullable + private static JsonIndexConfig readStoredJsonIndexConfig(String columnName, + @Nullable PropertiesConfiguration properties) { + if (properties == null) { + return null; + } + try { + String key = V1Constants.MetadataKeys.Column.getKeyFor(columnName, "jsonIndexConfig"); + String escaped = properties.getString(key, null); + if (escaped == null) { + return null; + } + String serialized = CommonsConfigurationUtils.recoverSpecialCharacterInPropertyValue(escaped); + return JsonUtils.stringToObject(serialized, JsonIndexConfig.class); + } catch (Exception e) { + LOGGER.warn("Failed to read stored json index config for column: {}, treating as legacy (no stored config)", + columnName, e); + return null; + } + } + + /// Writes the {@link JsonIndexConfig} for {@code columnName} into {@code properties} (without saving to disk). + private static void setStoredJsonIndexConfig(String columnName, JsonIndexConfig config, + PropertiesConfiguration properties) { + try { + String serialized = JsonUtils.objectToString(config); + String escaped = CommonsConfigurationUtils.replaceSpecialCharacterInPropertyValue(serialized); + if (escaped == null) { + LOGGER.warn("Cannot persist json index config for column: {} due to unsupported characters", columnName); + return; } + String key = V1Constants.MetadataKeys.Column.getKeyFor(columnName, "jsonIndexConfig"); + properties.setProperty(key, escaped); + } catch (Exception e) { + LOGGER.warn("Failed to serialize json index config for column: {}, config changes will not trigger rebuild", + columnName, e); } } + /// Returns {@code true} if the forward index for {@code column} is already present in the segment or can be + /// reconstructed from the dictionary and inverted index (the path used by + /// {@link BaseIndexHandler#createForwardIndexIfNeeded}). + private static boolean isForwardIndexAvailable(SegmentDirectory.Reader segmentReader, String column) { + return segmentReader.hasIndexFor(column, StandardIndexes.forward()) + || (segmentReader.hasIndexFor(column, StandardIndexes.dictionary()) + && segmentReader.hasIndexFor(column, StandardIndexes.inverted())); + } + private boolean shouldCreateJsonIndex(ColumnMetadata columnMetadata) { return columnMetadata != null; } @@ -186,8 +320,13 @@ private void handleNonDictionaryBasedColumn(SegmentDirectory.Writer segmentWrite ForwardIndexReaderContext readerContext = forwardIndexReader.createContext(); JsonIndexCreator jsonIndexCreator = StandardIndexes.json().createIndexCreator(context, config)) { int numDocs = columnMetadata.getTotalDocs(); + boolean isMapType = columnMetadata.getDataType() == DataType.MAP; for (int i = 0; i < numDocs; i++) { - jsonIndexCreator.add(forwardIndexReader.getString(i, readerContext)); + // MAP columns store binary-encoded data; getString() returns garbled bytes. + // Use getMap() + MapUtils.toString() to produce the JSON string the index creator expects. + String value = isMapType ? MapUtils.toString(forwardIndexReader.getMap(i, readerContext)) + : forwardIndexReader.getString(i, readerContext); + jsonIndexCreator.add(value); } jsonIndexCreator.seal(); } diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/JsonIndexHandlerTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/JsonIndexHandlerTest.java new file mode 100644 index 000000000000..166fb7cb6758 --- /dev/null +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/JsonIndexHandlerTest.java @@ -0,0 +1,240 @@ +/** + * 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.loader.invertedindex; + +import java.io.File; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import org.apache.commons.configuration2.PropertiesConfiguration; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.segment.spi.V1Constants; +import org.apache.pinot.segment.spi.index.FieldIndexConfigs; +import org.apache.pinot.segment.spi.index.StandardIndexes; +import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl; +import org.apache.pinot.segment.spi.store.SegmentDirectory; +import org.apache.pinot.spi.config.table.JsonIndexConfig; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.env.CommonsConfigurationUtils; +import org.apache.pinot.spi.utils.JsonUtils; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + + +/** + * Unit tests for {@link JsonIndexHandler} JSON index config-change detection. + * + *

When a JSON index is created, the {@link JsonIndexConfig} is persisted into {@code metadata.properties} under + * {@code column..jsonIndexConfig}. On subsequent segment reloads, {@link JsonIndexHandler#needUpdateIndices} + * compares the stored config with the current table config; if they differ the index is rebuilt. If no stored config + * is present (index built before config persistence was added), the segment is treated as a legacy segment and a + * one-time rebuild is triggered to backfill the stored config and catch any config drift. + */ +public class JsonIndexHandlerTest { + private static final String COLUMN = "myJsonCol"; + + private File _indexDir; + + @BeforeMethod + public void setUp() + throws Exception { + _indexDir = new File(FileUtils.getTempDirectory(), "json-index-handler-test-" + System.nanoTime()); + assertTrue(_indexDir.mkdirs()); + } + + @AfterMethod + public void tearDown() { + FileUtils.deleteQuietly(_indexDir); + } + + @Test + public void testNeedUpdateReturnsFalseWhenConfigUnchanged() + throws Exception { + JsonIndexConfig config = new JsonIndexConfig(); + storeConfig(COLUMN, config); + + JsonIndexHandler handler = createHandler(COLUMN, config); + SegmentDirectory.Reader reader = mockReaderWithIndexOn(COLUMN); + + assertFalse(handler.needUpdateIndices(reader), + "No rebuild expected when stored config equals current config"); + } + + @Test + public void testNeedUpdateReturnsTrueWhenConfigChanged() + throws Exception { + // Store a default config (maxLevels = -1) + JsonIndexConfig storedConfig = new JsonIndexConfig(); + storeConfig(COLUMN, storedConfig); + + // Current config has maxLevels = 3 + JsonIndexConfig currentConfig = new JsonIndexConfig(); + currentConfig.setMaxLevels(3); + + JsonIndexHandler handler = createHandler(COLUMN, currentConfig); + SegmentDirectory.Reader reader = mockReaderWithIndexOn(COLUMN); + + assertTrue(handler.needUpdateIndices(reader), + "Rebuild expected when maxLevels changed from default to 3"); + } + + @Test + public void testNeedUpdateReturnsTrueWhenExcludeArrayChanges() + throws Exception { + JsonIndexConfig storedConfig = new JsonIndexConfig(); + storeConfig(COLUMN, storedConfig); + + JsonIndexConfig currentConfig = new JsonIndexConfig(); + currentConfig.setExcludeArray(true); + + JsonIndexHandler handler = createHandler(COLUMN, currentConfig); + SegmentDirectory.Reader reader = mockReaderWithIndexOn(COLUMN); + + assertTrue(handler.needUpdateIndices(reader), + "Rebuild expected when excludeArray changed from false to true"); + } + + @Test + public void testNeedUpdateReturnsTrueWhenMaxBytesSizeChanges() + throws Exception { + JsonIndexConfig storedConfig = new JsonIndexConfig(); + storeConfig(COLUMN, storedConfig); + + JsonIndexConfig currentConfig = new JsonIndexConfig(); + currentConfig.setMaxBytesSize(1024L); + + JsonIndexHandler handler = createHandler(COLUMN, currentConfig); + SegmentDirectory.Reader reader = mockReaderWithIndexOn(COLUMN); + + assertTrue(handler.needUpdateIndices(reader), + "Rebuild expected when maxBytesSize changed"); + } + + @Test + public void testNeedUpdateReturnsTrueWhenNoStoredConfigAndForwardIndexAvailable() + throws Exception { + // No stored config — index was built before config persistence was added (legacy segment). + // Forward index is present, so a one-time backfill rebuild is safe and expected. + createEmptyMetadataFile(); + + JsonIndexConfig currentConfig = new JsonIndexConfig(); + JsonIndexHandler handler = createHandler(COLUMN, currentConfig); + SegmentDirectory.Reader reader = mockReaderWithIndexOn(COLUMN, /* hasForwardIndex= */ true); + + assertTrue(handler.needUpdateIndices(reader), + "Rebuild expected for legacy segments with no stored config when forward index is available"); + } + + @Test + public void testNeedUpdateReturnsFalseWhenNoStoredConfigButForwardIndexAbsent() + throws Exception { + // No stored config (legacy segment) AND no forward index (forwardIndexDisabled=true with no + // dictionary/inverted pair). createForwardIndexIfNeeded() would fail in this case, so the + // existing JSON index must be preserved without triggering a rebuild. + createEmptyMetadataFile(); + + JsonIndexConfig currentConfig = new JsonIndexConfig(); + JsonIndexHandler handler = createHandler(COLUMN, currentConfig); + SegmentDirectory.Reader reader = mockReaderWithIndexOn(COLUMN, /* hasForwardIndex= */ false, + /* hasDictionary= */ false, /* hasInverted= */ false); + + assertFalse(handler.needUpdateIndices(reader), + "No rebuild expected for legacy segments when forward index is absent (forwardIndexDisabled)"); + } + + @Test + public void testNeedUpdateReturnsTrueWhenNoStoredConfigAndForwardIndexReconstructable() + throws Exception { + // No stored config (legacy segment) AND forward index is absent but can be reconstructed from + // dictionary + inverted index. The rebuild is safe and should be triggered for backfill. + createEmptyMetadataFile(); + + JsonIndexConfig currentConfig = new JsonIndexConfig(); + JsonIndexHandler handler = createHandler(COLUMN, currentConfig); + SegmentDirectory.Reader reader = mockReaderWithIndexOn(COLUMN, /* hasForwardIndex= */ false, + /* hasDictionary= */ true, /* hasInverted= */ true); + + assertTrue(handler.needUpdateIndices(reader), + "Rebuild expected when forward index is reconstructable via dictionary + inverted index"); + } + + // --- helpers --- + + private void storeConfig(String columnName, JsonIndexConfig config) + throws Exception { + File metadataFile = new File(_indexDir, V1Constants.MetadataKeys.METADATA_FILE_NAME); + PropertiesConfiguration properties = CommonsConfigurationUtils.fromFile(metadataFile); + String key = V1Constants.MetadataKeys.Column.getKeyFor(columnName, "jsonIndexConfig"); + String serialized = JsonUtils.objectToString(config); + String escaped = CommonsConfigurationUtils.replaceSpecialCharacterInPropertyValue(serialized); + properties.setProperty(key, escaped); + CommonsConfigurationUtils.saveToFile(properties, metadataFile); + } + + private void createEmptyMetadataFile() + throws Exception { + File metadataFile = new File(_indexDir, V1Constants.MetadataKeys.METADATA_FILE_NAME); + PropertiesConfiguration properties = CommonsConfigurationUtils.fromFile(metadataFile); + CommonsConfigurationUtils.saveToFile(properties, metadataFile); + } + + private JsonIndexHandler createHandler(String columnName, JsonIndexConfig config) { + FieldIndexConfigs fieldIndexConfigs = + new FieldIndexConfigs.Builder().add(StandardIndexes.json(), config).build(); + SegmentMetadataImpl segmentMetadata = mock(SegmentMetadataImpl.class); + when(segmentMetadata.getName()).thenReturn("testSegment"); + when(segmentMetadata.getIndexDir()).thenReturn(_indexDir); + when(segmentMetadata.getTotalDocs()).thenReturn(1); + when(segmentMetadata.getAllColumns()).thenReturn(new TreeSet<>(Set.of(columnName))); + SegmentDirectory segmentDirectory = mock(SegmentDirectory.class); + when(segmentDirectory.getSegmentMetadata()).thenReturn(segmentMetadata); + when(segmentDirectory.getColumnsWithIndex(StandardIndexes.json())).thenReturn(Set.of(columnName)); + return new JsonIndexHandler(segmentDirectory, Map.of(columnName, fieldIndexConfigs), + mock(TableConfig.class), mock(Schema.class)); + } + + private SegmentDirectory.Reader mockReaderWithIndexOn(String columnName) { + return mockReaderWithIndexOn(columnName, /* hasForwardIndex= */ true, /* hasDictionary= */ false, + /* hasInverted= */ false); + } + + private SegmentDirectory.Reader mockReaderWithIndexOn(String columnName, boolean hasForwardIndex) { + return mockReaderWithIndexOn(columnName, hasForwardIndex, /* hasDictionary= */ false, + /* hasInverted= */ false); + } + + private SegmentDirectory.Reader mockReaderWithIndexOn(String columnName, boolean hasForwardIndex, + boolean hasDictionary, boolean hasInverted) { + SegmentDirectory segDir = mock(SegmentDirectory.class); + when(segDir.getColumnsWithIndex(StandardIndexes.json())).thenReturn(Set.of(columnName)); + SegmentDirectory.Reader reader = mock(SegmentDirectory.Reader.class); + when(reader.toSegmentDirectory()).thenReturn(segDir); + when(reader.hasIndexFor(columnName, StandardIndexes.forward())).thenReturn(hasForwardIndex); + when(reader.hasIndexFor(columnName, StandardIndexes.dictionary())).thenReturn(hasDictionary); + when(reader.hasIndexFor(columnName, StandardIndexes.inverted())).thenReturn(hasInverted); + return reader; + } +}