This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch release-1.6.0-rc in repository https://gitbox.apache.org/repos/asf/pinot.git
commit e4061a89c34b41cb482adddecf9020c9b8ab32ad Author: Xiang Fu <[email protected]> AuthorDate: Wed Sep 23 17:24:39 2026 -0700 Reuse processed table state across segment loads (#19571) * Reuse normalized table schemas from the existing table manager cache * Use cached table config and schema for ordinary segment loads * Cache processed index loading settings for ordinary segment loads * Share cached loading config on read-only segment load paths * Derive segment-local index loading wrappers * Separate immutable index loading state * Separate resolved index state from segment overrides * add regression test * Fix default-tier reload config race * Expose cached loading config to subclasses * Avoid invalidating unchanged index state --------- Co-authored-by: J-HowHuang <[email protected]> (cherry picked from commit 257a0e7df2cee4b26cf89c417eef13f71047f1c4) --- .../core/data/manager/BaseTableDataManager.java | 22 +- .../manager/offline/OfflineTableDataManager.java | 3 +- .../manager/realtime/RealtimeTableDataManager.java | 12 +- .../data/manager/BaseTableDataManagerTest.java | 152 +++++++ .../immutable/ImmutableSegmentLoader.java | 5 +- .../segment/index/loader/IndexLoadingConfig.java | 438 +++++++++++++-------- .../index/loader/IndexLoadingConfigTest.java | 62 ++- 7 files changed, 511 insertions(+), 183 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java index 4da4cf0d6ef..800b59caf7a 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java @@ -176,6 +176,7 @@ public abstract class BaseTableDataManager implements TableDataManager { // Caches the latest TableConfig and Schema pair. The cache should not be modified. protected volatile Pair<TableConfig, Schema> _cachedTableConfigAndSchema; + protected volatile IndexLoadingConfig _cachedIndexLoadingConfig; protected volatile boolean _shutDown; protected volatile boolean _isDeleted; @@ -437,6 +438,21 @@ public abstract class BaseTableDataManager implements TableDataManager { return indexLoadingConfig; } + /// Returns the shared processed config for ordinary loads. Segment-specific settings are applied with the + /// `with...` methods on [IndexLoadingConfig], which share its processed table-level state. + /// Explicit reloads and config/schema refresh messages still use [#fetchIndexLoadingConfig()]. + protected IndexLoadingConfig getCachedIndexLoadingConfig() { + Pair<TableConfig, Schema> cached = _cachedTableConfigAndSchema; + IndexLoadingConfig indexLoadingConfig = _cachedIndexLoadingConfig; + if (indexLoadingConfig == null || indexLoadingConfig.getTableConfig() != cached.getLeft() + || indexLoadingConfig.getSchema() != cached.getRight()) { + indexLoadingConfig = new IndexLoadingConfig(_instanceDataManagerConfig, cached.getLeft(), cached.getRight()); + indexLoadingConfig.setTableDataDir(_tableDataDir); + _cachedIndexLoadingConfig = indexLoadingConfig; + } + return indexLoadingConfig; + } + @Override public Pair<TableConfig, Schema> getCachedTableConfigAndSchema() { return _cachedTableConfigAndSchema; @@ -444,6 +460,10 @@ public abstract class BaseTableDataManager implements TableDataManager { @Override public void updateCachedTableConfigAndSchema(TableConfig tableConfig, Schema schema) { + // Normalize before publishing so segment loads never add timestamp fields to a shared schema. + if (schema != null) { + TimestampIndexUtils.applyTimestampIndex(tableConfig, schema); + } _cachedTableConfigAndSchema = Pair.of(tableConfig, schema); } @@ -1100,7 +1120,7 @@ public abstract class BaseTableDataManager implements TableDataManager { SegmentMetadata localMetadata, boolean forceDownload) throws Exception { String segmentTier = getSegmentCurrentTier(segmentName); - indexLoadingConfig.setSegmentTier(segmentTier); + indexLoadingConfig = indexLoadingConfig.copyWithSegmentTier(segmentTier); indexLoadingConfig.setTableDataDir(_tableDataDir); File indexDir = getSegmentDataDir(segmentName, segmentTier, indexLoadingConfig.getTableConfig()); _segmentReloadSemaphore.acquire(segmentName, _logger); diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java index 8717408446f..74ed8c44e81 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java @@ -71,8 +71,7 @@ public class OfflineTableDataManager extends BaseTableDataManager { protected void doAddOnlineSegment(String segmentName) throws Exception { SegmentZKMetadata zkMetadata = fetchZKMetadata(segmentName); - IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig(); - indexLoadingConfig.setSegmentTier(zkMetadata.getTier()); + IndexLoadingConfig indexLoadingConfig = getCachedIndexLoadingConfig().withSegmentTier(zkMetadata.getTier()); handleUpsertPreload(zkMetadata, indexLoadingConfig); SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName); if (segmentDataManager == null) { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java index 934f14a9447..6970d66d105 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java @@ -470,8 +470,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager { SegmentZKMetadata zkMetadata = fetchZKMetadata(segmentName); Preconditions.checkState(zkMetadata.getStatus() != Status.IN_PROGRESS, "Segment: %s of table: %s is not committed, cannot make it ONLINE", segmentName, _tableNameWithType); - IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig(); - indexLoadingConfig.setSegmentTier(zkMetadata.getTier()); + IndexLoadingConfig indexLoadingConfig = getCachedIndexLoadingConfig().withSegmentTier(zkMetadata.getTier()); handleSegmentPreload(zkMetadata, indexLoadingConfig); SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName); if (segmentDataManager == null) { @@ -541,7 +540,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager { _logger.warn("Segment: {} is already completed, skipping adding it as CONSUMING segment", segmentName); return; } - IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig(); + IndexLoadingConfig indexLoadingConfig = getCachedIndexLoadingConfig(); handleSegmentPreload(zkMetadata, indexLoadingConfig); SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName); if (segmentDataManager != null) { @@ -752,9 +751,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager { String segmentName = zkMetadata.getSegmentName(); _logger.info("Downloading and replacing CONSUMING segment: {} with committed one", segmentName); File indexDir = downloadSegment(zkMetadata); - // Get a new index loading config with latest table config and schema to load the segment - IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig(); - indexLoadingConfig.setSegmentTier(zkMetadata.getTier()); + IndexLoadingConfig indexLoadingConfig = getCachedIndexLoadingConfig().withSegmentTier(zkMetadata.getTier()); addSegment(ImmutableSegmentLoader.load(indexDir, indexLoadingConfig, _segmentOperationsThrottlerSet, zkMetadata), zkMetadata); _ingestionDelayTracker.markPartitionForVerification(segmentName); @@ -774,8 +771,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager { throws Exception { _logger.info("Replacing CONSUMING segment: {} with the one sealed locally", segmentName); File indexDir = new File(_indexDir, segmentName); - // Get a new index loading config with latest table config and schema to load the segment - IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig(); + IndexLoadingConfig indexLoadingConfig = getCachedIndexLoadingConfig(); ImmutableSegment immutableSegment = ImmutableSegmentLoader.load(indexDir, indexLoadingConfig, _segmentOperationsThrottlerSet, zkMetadata); diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java index 5a5b7de8bdc..649968e87c8 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java @@ -30,16 +30,20 @@ import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import org.apache.commons.io.FileUtils; import org.apache.helix.HelixManager; +import org.apache.pinot.common.metadata.ZKMetadataProvider; import org.apache.pinot.common.metadata.segment.SegmentZKMetadata; import org.apache.pinot.common.metrics.ServerMetrics; import org.apache.pinot.common.tier.TierFactory; import org.apache.pinot.common.utils.TarCompressionUtils; import org.apache.pinot.common.utils.fetcher.BaseSegmentFetcher; import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory; +import org.apache.pinot.common.utils.helix.FakePropertyStore; import org.apache.pinot.core.data.manager.offline.ImmutableSegmentDataManager; import org.apache.pinot.core.data.manager.offline.OfflineTableDataManager; import org.apache.pinot.segment.local.data.manager.SegmentDataManager; @@ -57,13 +61,17 @@ import org.apache.pinot.segment.spi.SegmentMetadata; import org.apache.pinot.segment.spi.V1Constants; import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; import org.apache.pinot.segment.spi.creator.SegmentVersion; +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.segment.spi.store.SegmentDirectoryPaths; import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig; +import org.apache.pinot.spi.config.table.FieldConfig; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.config.table.TableType; import org.apache.pinot.spi.config.table.TierConfig; +import org.apache.pinot.spi.config.table.TimestampConfig; +import org.apache.pinot.spi.config.table.TimestampIndexGranularity; import org.apache.pinot.spi.crypt.PinotCrypter; import org.apache.pinot.spi.crypt.PinotCrypterFactory; import org.apache.pinot.spi.data.FieldSpec.DataType; @@ -88,6 +96,7 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import static org.testng.Assert.*; @@ -279,6 +288,28 @@ public class BaseTableDataManagerTest { assertEquals(new SegmentMetadataImpl(tierDataDir).getTotalDocs(), 5); } + /// Regression test for https://github.com/apache/pinot/issues/18164: the table-level config passed into reload is + /// shared by all segments, so reloading one segment must not set its tier on the shared config. + @Test + public void testReloadSegmentFromDefaultTierDoesNotMutateSharedIndexLoadingConfig() + throws Exception { + // The current tier is null while the target tier is coolTier, exercising both tier mutations in the reload path. + SegmentZKMetadata zkMetadata = createRawSegment(SegmentVersion.v3, 5); + zkMetadata.setTier(TIER_NAME); + SegmentMetadata localMetadata = mock(SegmentMetadata.class); + when(localMetadata.getCrc()).thenReturn(0L); + + ImmutableSegmentDataManager segmentDataManager = createImmutableSegmentDataManager(SEGMENT_NAME, localMetadata); + BaseTableDataManager tableDataManager = spy(createTableManager()); + tableDataManager.registerSegment(SEGMENT_NAME, segmentDataManager); + seedZKMetadata(tableDataManager, SEGMENT_NAME, zkMetadata); + + IndexLoadingConfig sharedConfig = new IndexLoadingConfig(DEFAULT_TABLE_CONFIG, SCHEMA); + tableDataManager.reloadSegment(segmentDataManager, sharedConfig, true); + assertNull(sharedConfig.getSegmentTier()); + assertNull(sharedConfig.getTableDataDir()); + } + @Test public void testReloadSegmentConvertVersion() throws Exception { @@ -1069,6 +1100,127 @@ public class BaseTableDataManagerTest { verify(segmentDirectory, times(1)).onSegmentAdded(); } + @Test + public void testGetCachedIndexLoadingConfigReusesSchemaAndRefreshes() { + TableConfig table = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build(); + Schema original = createSchemaReuseSchema(); + BaseTableDataManager manager = createSchemaReuseManager(table, original); + IndexLoadingConfig first = manager.getCachedIndexLoadingConfig(); + IndexLoadingConfig second = manager.getCachedIndexLoadingConfig(); + assertSame(first.getSchema(), original); + assertSame(second.getSchema(), original); + assertSame(first.getTableConfig(), second.getTableConfig()); + assertSame(first, second); + verifyNoInteractions(manager._propertyStore); + + Schema changed = createSchemaReuseSchema(); + changed.getFieldSpecFor("id").setDefaultNullValue(-2); + ZKMetadataProvider.setSchema(manager._propertyStore, changed); + // Explicit refreshes still fetch the latest schema even without a separate refresh message. + Schema refreshed = manager.fetchIndexLoadingConfig().getSchema(); + assertNotSame(refreshed, original); + assertEquals(refreshed.getFieldSpecFor("id").getDefaultNullValue(), -2); + assertEquals(original.getFieldSpecFor("id").getDefaultNullValue(), -1); + assertSame(manager.getCachedTableConfigAndSchema().getRight(), refreshed); + assertSame(manager.getCachedIndexLoadingConfig().getSchema(), refreshed); + + manager.updateCachedTableConfigAndSchema(table, changed); + assertSame(manager.getCachedIndexLoadingConfig().getSchema(), changed); + TableConfig indexedTable = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME) + .setInvertedIndexColumns(List.of("id")).build(); + manager.updateCachedTableConfigAndSchema(indexedTable, changed); + IndexLoadingConfig indexed = manager.getCachedIndexLoadingConfig(); + assertTrue(indexed.getFieldIndexConfig("id").getConfig(StandardIndexes.inverted()).isEnabled()); + assertFalse(first.getFieldIndexConfig("id").getConfig(StandardIndexes.inverted()).isEnabled()); + assertSame(indexed.getFieldIndexConfig("id"), manager.getCachedIndexLoadingConfig().getFieldIndexConfig("id")); + } + + @Test + public void testGetCachedIndexLoadingConfigDerivesSegmentTier() { + TableConfig table = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build(); + Schema schema = createSchemaReuseSchema(); + BaseTableDataManager manager = createSchemaReuseManager(table, schema); + IndexLoadingConfig shared = manager.getCachedIndexLoadingConfig(); + IndexLoadingConfig tierConfig = shared.withSegmentTier("cold"); + assertNotSame(tierConfig, shared); + assertEquals(tierConfig.getSegmentTier(), "cold"); + assertNull(shared.getSegmentTier()); + assertSame(manager.getCachedIndexLoadingConfig(), shared); + assertSame(shared.withSegmentTier(null), shared); + verifyNoInteractions(manager._propertyStore); + } + + @Test + public void testGetCachedIndexLoadingConfigNormalizesTimestampBeforeReuse() { + BaseTableDataManager manager = + createSchemaReuseManager(createTimestampTable(TimestampIndexGranularity.DAY), createSchemaReuseSchema()); + Schema first = manager.getCachedIndexLoadingConfig().getSchema(); + IndexLoadingConfig second = manager.getCachedIndexLoadingConfig(); + assertSame(second.getSchema(), first); + assertTrue(first.hasColumn("$ts$DAY")); + assertTrue(second.getFieldIndexConfigByColName().get("$ts$DAY").getConfig(StandardIndexes.range()).isEnabled()); + assertEquals(second.getTableConfig().getIndexingConfig().getRangeIndexColumns(), List.of("$ts$DAY")); + assertEquals(second.getTableConfig().getIngestionConfig().getTransformConfigs().size(), 1); + + ZKMetadataProvider.setTableConfig(manager._propertyStore, createTimestampTable(TimestampIndexGranularity.HOUR)); + manager.onTableConfigOrSchemaRefresh(); + Schema changed = manager.getCachedIndexLoadingConfig().getSchema(); + assertNotSame(changed, first); + assertTrue(changed.hasColumn("$ts$HOUR")); + assertFalse(changed.hasColumn("$ts$DAY")); + assertFalse(first.hasColumn("$ts$HOUR")); + } + + @Test + public void testGetCachedIndexLoadingConfigConcurrentlyReusesCachedSchema() + throws Exception { + BaseTableDataManager manager = + createSchemaReuseManager(createTimestampTable(TimestampIndexGranularity.DAY), createSchemaReuseSchema()); + IndexLoadingConfig shared = manager.getCachedIndexLoadingConfig(); + ExecutorService executor = Executors.newFixedThreadPool(8); + CountDownLatch start = new CountDownLatch(1); + try { + List<Future<IndexLoadingConfig>> results = new ArrayList<>(); + for (int i = 0; i < 32; i++) { + results.add(executor.submit(() -> { + assertTrue(start.await(10, TimeUnit.SECONDS)); + return manager.getCachedIndexLoadingConfig(); + })); + } + start.countDown(); + for (Future<IndexLoadingConfig> result : results) { + assertSame(result.get(10, TimeUnit.SECONDS), shared); + } + verifyNoInteractions(manager._propertyStore); + } finally { + start.countDown(); + executor.shutdownNow(); + } + } + + private static BaseTableDataManager createSchemaReuseManager(TableConfig table, Schema schema) { + BaseTableDataManager manager = new OfflineTableDataManager(); + manager._propertyStore = new FakePropertyStore(); + manager._tableNameWithType = OFFLINE_TABLE_NAME; + ZKMetadataProvider.setTableConfig(manager._propertyStore, table); + ZKMetadataProvider.setSchema(manager._propertyStore, schema); + manager.updateCachedTableConfigAndSchema(table, schema); + manager._propertyStore = spy(manager._propertyStore); + return manager; + } + + private static Schema createSchemaReuseSchema() { + return new Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME) + .addSingleValueDimension("id", DataType.INT, -1) + .addDateTime("ts", DataType.TIMESTAMP, "TIMESTAMP", "1:MILLISECONDS").build(); + } + + private static TableConfig createTimestampTable(TimestampIndexGranularity granularity) { + return new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME) + .setFieldConfigList(List.of(new FieldConfig.Builder("ts") + .withTimestampConfig(new TimestampConfig(List.of(granularity))).build())).build(); + } + protected BaseTableDataManager createTableManager() { return createTableManager(createDefaultInstanceDataManagerConfig()); } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java index 724bd04ef69..b5a57289138 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java @@ -127,6 +127,7 @@ public class ImmutableSegmentLoader { if (segmentMetadata.getTotalDocs() == 0) { return new EmptyIndexSegment(segmentMetadata); } + indexLoadingConfig = indexLoadingConfig.withOpenStructChildConfigs(segmentMetadata); String segmentName = segmentMetadata.getName(); SegmentDirectoryLoaderContext segmentLoaderContext = new SegmentDirectoryLoaderContext.Builder() .setReadMode(indexLoadingConfig.getReadMode()) @@ -171,6 +172,7 @@ public class ImmutableSegmentLoader { SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir); if (segmentMetadata.getTotalDocs() > 0) { + indexLoadingConfig = indexLoadingConfig.withOpenStructChildConfigs(segmentMetadata); if (segmentOperationsThrottlerSet != null) { segmentOperationsThrottlerSet.getSegmentAllIndexPreprocessThrottler().acquire(); } @@ -205,6 +207,7 @@ public class ImmutableSegmentLoader { // mirroring the non-empty ImmutableSegmentImpl path. return new EmptyIndexSegment(segmentMetadata, segmentDirectory); } + indexLoadingConfig = indexLoadingConfig.withOpenStructChildConfigs(segmentMetadata); // Remove columns not in schema from the metadata Map<String, ColumnMetadata> columnMetadataMap = segmentMetadata.getColumnMetadataMap(); @@ -229,7 +232,7 @@ public class ImmutableSegmentLoader { } } } else { - indexLoadingConfig.addKnownColumns(columnMetadataMap.keySet()); + indexLoadingConfig = indexLoadingConfig.withKnownColumns(columnMetadataMap.keySet()); } SegmentDirectory.Reader segmentReader = segmentDirectory.createReader(); diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java index 3782f3abbf8..e68480ace43 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java @@ -24,6 +24,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import javax.annotation.Nullable; import org.apache.commons.lang3.StringUtils; @@ -44,64 +45,200 @@ import org.apache.pinot.spi.config.table.MultiColumnTextIndexConfig; import org.apache.pinot.spi.config.table.OpenStructIndexConfig; import org.apache.pinot.spi.config.table.StarTreeIndexConfig; import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.data.ComplexFieldSpec; 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.data.OpenStructNaming; import org.apache.pinot.spi.data.Schema; import org.apache.pinot.spi.utils.ReadMode; import org.apache.pinot.spi.utils.TimestampIndexUtils; -/// Table level index loading config. +/// Index loading config with shared table-level state and segment-local mutable overrides. public class IndexLoadingConfig { private static final int DEFAULT_REALTIME_AVG_MULTI_VALUE_COUNT = 2; public static final String READ_MODE_KEY = "readMode"; - private final InstanceDataManagerConfig _instanceDataManagerConfig; - private final TableConfig _tableConfig; - private final Schema _schema; + private final ImmutableState _immutableState; - // These fields can be modified after initialization - // TODO: Revisit them - private ReadMode _readMode = ReadMode.DEFAULT_MODE; - private SegmentVersion _segmentVersion; + // Mutable config and segment-specific overrides. + @Nullable + private ReadMode _readModeOverride; + @Nullable + private SegmentVersion _segmentVersionOverride; private String _segmentTier; private Set<String> _knownColumns; private String _tableDataDir; private boolean _errorOnColumnBuildFailure; private boolean _forwardIndexOnly; + private ResolvedIndexState _resolvedIndexState; + + /// Immutable table-level state shared by derived segment configs. + private static final class ImmutableState { + @Nullable + private final InstanceDataManagerConfig _instanceDataManagerConfig; + @Nullable + private final TableConfig _tableConfig; + @Nullable + private final Schema _schema; + private final ReadMode _readMode; + @Nullable + private final SegmentVersion _segmentVersion; + @Nullable + private final String _instanceId; + private final boolean _isRealtimeOffHeapAllocation; + private final boolean _isDirectRealtimeOffHeapAllocation; + private final int _realtimeAvgMultiValueCount; + @Nullable + private final String _segmentStoreURI; + @Nullable + private final String _segmentDirectoryLoader; + @Nullable + private final Map<String, Map<String, String>> _instanceTierConfigs; + private final List<String> _sortedColumns; + private final ColumnMinMaxValueGeneratorMode _columnMinMaxValueGeneratorMode; + private final boolean _hasOpenStructColumns; + + private ImmutableState(@Nullable InstanceDataManagerConfig instanceDataManagerConfig, + @Nullable TableConfig tableConfig, @Nullable Schema schema) { + _instanceDataManagerConfig = instanceDataManagerConfig; + _tableConfig = tableConfig; + _schema = schema; + + String instanceId = null; + boolean isRealtimeOffHeapAllocation = false; + boolean isDirectRealtimeOffHeapAllocation = false; + int realtimeAvgMultiValueCount = DEFAULT_REALTIME_AVG_MULTI_VALUE_COUNT; + ReadMode readMode = ReadMode.DEFAULT_MODE; + SegmentVersion segmentVersion = null; + String segmentStoreURI = null; + String segmentDirectoryLoader = null; + Map<String, Map<String, String>> instanceTierConfigs = null; + if (instanceDataManagerConfig != null) { + ReadMode instanceReadMode = instanceDataManagerConfig.getReadMode(); + if (instanceReadMode != null) { + readMode = instanceReadMode; + } + String instanceSegmentVersion = instanceDataManagerConfig.getSegmentFormatVersion(); + if (instanceSegmentVersion != null) { + segmentVersion = SegmentVersion.valueOf(instanceSegmentVersion.toLowerCase()); + } + instanceId = instanceDataManagerConfig.getInstanceId(); + isRealtimeOffHeapAllocation = instanceDataManagerConfig.isRealtimeOffHeapAllocation(); + isDirectRealtimeOffHeapAllocation = instanceDataManagerConfig.isDirectRealtimeOffHeapAllocation(); + String avgMultiValueCount = instanceDataManagerConfig.getAvgMultiValueCount(); + if (avgMultiValueCount != null) { + realtimeAvgMultiValueCount = Integer.parseInt(avgMultiValueCount); + } + segmentStoreURI = instanceDataManagerConfig.getSegmentStoreUri(); + segmentDirectoryLoader = instanceDataManagerConfig.getSegmentDirectoryLoader(); + Map<String, Map<String, String>> tierConfigs = instanceDataManagerConfig.getTierConfigs(); + instanceTierConfigs = tierConfigs != null ? tierConfigs : Map.of(); + } - // Initialized by instance data manager config - private String _instanceId; - private boolean _isRealtimeOffHeapAllocation; - private boolean _isDirectRealtimeOffHeapAllocation; - private int _realtimeAvgMultiValueCount = DEFAULT_REALTIME_AVG_MULTI_VALUE_COUNT; - private String _segmentStoreURI; - private String _segmentDirectoryLoader; - private Map<String, Map<String, String>> _instanceTierConfigs; + List<String> sortedColumns = List.of(); + ColumnMinMaxValueGeneratorMode columnMinMaxValueGeneratorMode = ColumnMinMaxValueGeneratorMode.DEFAULT_MODE; + boolean hasOpenStructColumns = false; + if (tableConfig != null) { + if (schema != null) { + TimestampIndexUtils.applyTimestampIndex(tableConfig, schema); + for (ComplexFieldSpec fieldSpec : schema.getComplexFieldSpecs()) { + if (fieldSpec.getDataType() == DataType.OPEN_STRUCT) { + hasOpenStructColumns = true; + break; + } + } + } + IndexingConfig indexingConfig = tableConfig.getIndexingConfig(); + String tableReadMode = indexingConfig.getLoadMode(); + if (tableReadMode != null) { + readMode = ReadMode.getEnum(tableReadMode); + } + String tableSegmentVersion = indexingConfig.getSegmentFormatVersion(); + if (tableSegmentVersion != null) { + segmentVersion = SegmentVersion.valueOf(tableSegmentVersion.toLowerCase()); + } + List<String> tableSortedColumns = indexingConfig.getSortedColumn(); + if (tableSortedColumns != null) { + sortedColumns = tableSortedColumns; + } + String generatorMode = indexingConfig.getColumnMinMaxValueGeneratorMode(); + if (generatorMode != null) { + columnMinMaxValueGeneratorMode = ColumnMinMaxValueGeneratorMode.valueOf(generatorMode.toUpperCase()); + } + } - // Initialized by table config and schema - private List<String> _sortedColumns = List.of(); - private ColumnMinMaxValueGeneratorMode _columnMinMaxValueGeneratorMode = ColumnMinMaxValueGeneratorMode.DEFAULT_MODE; - private boolean _enableDynamicStarTreeCreation; - private List<StarTreeIndexConfig> _starTreeIndexConfigs; - private boolean _enableDefaultStarTree; - private Map<String, FieldIndexConfigs> _indexConfigsByColName = new HashMap<>(); - private boolean _skipSegmentPreprocess; + _instanceId = instanceId; + _readMode = readMode; + _segmentVersion = segmentVersion; + _isRealtimeOffHeapAllocation = isRealtimeOffHeapAllocation; + _isDirectRealtimeOffHeapAllocation = isDirectRealtimeOffHeapAllocation; + _realtimeAvgMultiValueCount = realtimeAvgMultiValueCount; + _segmentStoreURI = segmentStoreURI; + _segmentDirectoryLoader = segmentDirectoryLoader; + _instanceTierConfigs = instanceTierConfigs; + _sortedColumns = sortedColumns; + _columnMinMaxValueGeneratorMode = columnMinMaxValueGeneratorMode; + _hasOpenStructColumns = hasOpenStructColumns; + } + } - private boolean _dirty = true; + /// Index settings resolved from the table config, segment tier, schema, and known segment columns. + private static final class ResolvedIndexState { + private static final ResolvedIndexState EMPTY = + new ResolvedIndexState(false, null, false, Map.of(), false, null); + + private final boolean _enableDynamicStarTreeCreation; + @Nullable + private final List<StarTreeIndexConfig> _starTreeIndexConfigs; + private final boolean _enableDefaultStarTree; + private final Map<String, FieldIndexConfigs> _indexConfigsByColName; + private final boolean _skipSegmentPreprocess; + @Nullable + private final MultiColumnTextIndexConfig _multiColTextIndexConfig; + + private ResolvedIndexState(boolean enableDynamicStarTreeCreation, + @Nullable List<StarTreeIndexConfig> starTreeIndexConfigs, boolean enableDefaultStarTree, + Map<String, FieldIndexConfigs> indexConfigsByColName, boolean skipSegmentPreprocess, + @Nullable MultiColumnTextIndexConfig multiColTextIndexConfig) { + _enableDynamicStarTreeCreation = enableDynamicStarTreeCreation; + _starTreeIndexConfigs = starTreeIndexConfigs; + _enableDefaultStarTree = enableDefaultStarTree; + _indexConfigsByColName = indexConfigsByColName; + _skipSegmentPreprocess = skipSegmentPreprocess; + _multiColTextIndexConfig = multiColTextIndexConfig; + } - private MultiColumnTextIndexConfig _multiColTextIndexConfig; + private ResolvedIndexState withIndexConfigsByColName(Map<String, FieldIndexConfigs> indexConfigsByColName) { + return new ResolvedIndexState(_enableDynamicStarTreeCreation, _starTreeIndexConfigs, _enableDefaultStarTree, + indexConfigsByColName, _skipSegmentPreprocess, _multiColTextIndexConfig); + } + } /// NOTE: This step might modify the passed in table config and schema. /// /// TODO: Revisit the init handling. Currently it doesn't apply tiered config override public IndexLoadingConfig(@Nullable InstanceDataManagerConfig instanceDataManagerConfig, @Nullable TableConfig tableConfig, @Nullable Schema schema) { - _instanceDataManagerConfig = instanceDataManagerConfig; - _tableConfig = tableConfig; - _schema = schema; - init(); + _immutableState = new ImmutableState(instanceDataManagerConfig, tableConfig, schema); + if (tableConfig != null) { + refreshIndexConfigs(); + } + } + + /// Creates a segment-local wrapper around the already processed table-level config. The wrapper shares the + /// table config, schema, and resolved index configs until a segment-specific override requires a local copy. + private IndexLoadingConfig(IndexLoadingConfig source) { + _immutableState = source._immutableState; + _readModeOverride = source._readModeOverride; + _segmentVersionOverride = source._segmentVersionOverride; + _segmentTier = source._segmentTier; + _knownColumns = source._knownColumns; + _tableDataDir = source._tableDataDir; + _errorOnColumnBuildFailure = source._errorOnColumnBuildFailure; + _forwardIndexOnly = source._forwardIndexOnly; + _resolvedIndexState = source._resolvedIndexState; } @VisibleForTesting @@ -121,88 +258,22 @@ public class IndexLoadingConfig { @Nullable public InstanceDataManagerConfig getInstanceDataManagerConfig() { - return _instanceDataManagerConfig; + return _immutableState._instanceDataManagerConfig; } @Nullable public TableConfig getTableConfig() { - return _tableConfig; + return _immutableState._tableConfig; } @Nullable public Schema getSchema() { - return _schema; - } - - private void init() { - if (_instanceDataManagerConfig != null) { - extractFromInstanceConfig(); - } - if (_tableConfig != null) { - extractFromTableConfigAndSchema(); - } - } - - private void extractFromInstanceConfig() { - _instanceId = _instanceDataManagerConfig.getInstanceId(); - - ReadMode instanceReadMode = _instanceDataManagerConfig.getReadMode(); - if (instanceReadMode != null) { - _readMode = instanceReadMode; - } - - String instanceSegmentVersion = _instanceDataManagerConfig.getSegmentFormatVersion(); - if (instanceSegmentVersion != null) { - _segmentVersion = SegmentVersion.valueOf(instanceSegmentVersion.toLowerCase()); - } - - _isRealtimeOffHeapAllocation = _instanceDataManagerConfig.isRealtimeOffHeapAllocation(); - _isDirectRealtimeOffHeapAllocation = _instanceDataManagerConfig.isDirectRealtimeOffHeapAllocation(); - - String avgMultiValueCount = _instanceDataManagerConfig.getAvgMultiValueCount(); - if (avgMultiValueCount != null) { - _realtimeAvgMultiValueCount = Integer.parseInt(avgMultiValueCount); - } - _segmentStoreURI = _instanceDataManagerConfig.getSegmentStoreUri(); - _segmentDirectoryLoader = _instanceDataManagerConfig.getSegmentDirectoryLoader(); - - Map<String, Map<String, String>> tierConfigs = _instanceDataManagerConfig.getTierConfigs(); - _instanceTierConfigs = tierConfigs != null ? tierConfigs : Map.of(); - } - - private void extractFromTableConfigAndSchema() { - if (_schema != null) { - TimestampIndexUtils.applyTimestampIndex(_tableConfig, _schema); - } - - IndexingConfig indexingConfig = _tableConfig.getIndexingConfig(); - String tableReadMode = indexingConfig.getLoadMode(); - if (tableReadMode != null) { - _readMode = ReadMode.getEnum(tableReadMode); - } - - List<String> sortedColumns = indexingConfig.getSortedColumn(); - if (sortedColumns != null) { - _sortedColumns = sortedColumns; - } - - String tableSegmentVersion = indexingConfig.getSegmentFormatVersion(); - if (tableSegmentVersion != null) { - _segmentVersion = SegmentVersion.valueOf(tableSegmentVersion.toLowerCase()); - } - - String columnMinMaxValueGeneratorMode = indexingConfig.getColumnMinMaxValueGeneratorMode(); - if (columnMinMaxValueGeneratorMode != null) { - _columnMinMaxValueGeneratorMode = - ColumnMinMaxValueGeneratorMode.valueOf(columnMinMaxValueGeneratorMode.toUpperCase()); - } - - refreshIndexConfigs(); + return _immutableState._schema; } public void refreshIndexConfigs() { - if (_tableConfig == null) { - _dirty = false; + if (_immutableState._tableConfig == null) { + _resolvedIndexState = ResolvedIndexState.EMPTY; return; } // Accessing the index configs for single-column index is handled by IndexType.getConfig() as defined in index-spi. @@ -210,112 +281,105 @@ public class IndexLoadingConfig { // specific index configs transparently. TableConfig tableConfig = getTableConfigWithTierOverwrites(); Schema schema = inferSchema(); - _indexConfigsByColName = FieldIndexConfigsUtil.createIndexConfigsByColName(tableConfig, schema); + Map<String, FieldIndexConfigs> indexConfigsByColName = + FieldIndexConfigsUtil.createIndexConfigsByColName(tableConfig, schema); // Accessing the StarTree index configs is not handled by IndexType.getConfig(), so we manually update them. IndexingConfig indexingConfig = tableConfig.getIndexingConfig(); - _enableDynamicStarTreeCreation = indexingConfig.isEnableDynamicStarTreeCreation(); - _starTreeIndexConfigs = indexingConfig.getStarTreeIndexConfigs(); - _enableDefaultStarTree = indexingConfig.isEnableDefaultStarTree(); - _multiColTextIndexConfig = indexingConfig.getMultiColumnTextIndexConfig(); - _skipSegmentPreprocess = indexingConfig.isSkipSegmentPreprocess(); - _dirty = false; + _resolvedIndexState = new ResolvedIndexState(indexingConfig.isEnableDynamicStarTreeCreation(), + indexingConfig.getStarTreeIndexConfigs(), indexingConfig.isEnableDefaultStarTree(), indexConfigsByColName, + indexingConfig.isSkipSegmentPreprocess(), indexingConfig.getMultiColumnTextIndexConfig()); + } + + private ResolvedIndexState getResolvedIndexState() { + if (_resolvedIndexState == null) { + refreshIndexConfigs(); + } + return _resolvedIndexState; } private TableConfig getTableConfigWithTierOverwrites() { - return (_segmentTier == null || _tableConfig == null) ? _tableConfig - : TableConfigUtils.overwriteTableConfigForTier(_tableConfig, _segmentTier); + return _segmentTier == null || _immutableState._tableConfig == null ? _immutableState._tableConfig + : TableConfigUtils.overwriteTableConfigForTier(_immutableState._tableConfig, _segmentTier); } private Schema inferSchema() { - if (_schema != null) { - return _schema; + if (_immutableState._schema != null) { + return _immutableState._schema; } Schema schema = new Schema(); for (String column : getAllKnownColumns()) { - schema.addField(new DimensionFieldSpec(column, FieldSpec.DataType.STRING, true)); + schema.addField(new DimensionFieldSpec(column, DataType.STRING, true)); } return schema; } public ReadMode getReadMode() { - return _readMode; + return _readModeOverride != null ? _readModeOverride : _immutableState._readMode; } public void setReadMode(ReadMode readMode) { - _readMode = readMode; + _readModeOverride = readMode; } public List<String> getSortedColumns() { - return unmodifiable(_sortedColumns); + return unmodifiable(_immutableState._sortedColumns); } public boolean isEnableDynamicStarTreeCreation() { - if (_dirty) { - refreshIndexConfigs(); - } - return _enableDynamicStarTreeCreation; + return getResolvedIndexState()._enableDynamicStarTreeCreation; } @Nullable public List<StarTreeIndexConfig> getStarTreeIndexConfigs() { - if (_dirty) { - refreshIndexConfigs(); - } - return unmodifiable(_starTreeIndexConfigs); + return unmodifiable(getResolvedIndexState()._starTreeIndexConfigs); } @Nullable public MultiColumnTextIndexConfig getMultiColTextIndexConfig() { - if (_dirty) { - refreshIndexConfigs(); - } - return _multiColTextIndexConfig; + return getResolvedIndexState()._multiColTextIndexConfig; } public boolean isEnableDefaultStarTree() { - if (_dirty) { - refreshIndexConfigs(); - } - return _enableDefaultStarTree; + return getResolvedIndexState()._enableDefaultStarTree; } @Nullable public SegmentVersion getSegmentVersion() { - return _segmentVersion; + return _segmentVersionOverride != null ? _segmentVersionOverride : _immutableState._segmentVersion; } /// For tests only. public void setSegmentVersion(SegmentVersion segmentVersion) { - _segmentVersion = segmentVersion; + _segmentVersionOverride = segmentVersion; } public boolean isRealtimeOffHeapAllocation() { - return _isRealtimeOffHeapAllocation; + return _immutableState._isRealtimeOffHeapAllocation; } public boolean isDirectRealtimeOffHeapAllocation() { - return _isDirectRealtimeOffHeapAllocation; + return _immutableState._isDirectRealtimeOffHeapAllocation; } public ColumnMinMaxValueGeneratorMode getColumnMinMaxValueGeneratorMode() { - return _columnMinMaxValueGeneratorMode; + return _immutableState._columnMinMaxValueGeneratorMode; } public String getSegmentStoreURI() { - return _segmentStoreURI; + return _immutableState._segmentStoreURI; } public int getRealtimeAvgMultiValueCount() { - return _realtimeAvgMultiValueCount; + return _immutableState._realtimeAvgMultiValueCount; } public String getSegmentDirectoryLoader() { - return StringUtils.isNotBlank(_segmentDirectoryLoader) ? _segmentDirectoryLoader + return StringUtils.isNotBlank(_immutableState._segmentDirectoryLoader) ? _immutableState._segmentDirectoryLoader : SegmentDirectoryLoaderRegistry.DEFAULT_SEGMENT_DIRECTORY_LOADER_NAME; } public String getInstanceId() { - return _instanceId; + return _immutableState._instanceId; } public String getSegmentTier() { @@ -323,8 +387,26 @@ public class IndexLoadingConfig { } public void setSegmentTier(String segmentTier) { + if (Objects.equals(_segmentTier, segmentTier)) { + return; + } _segmentTier = segmentTier; - _dirty = true; + _resolvedIndexState = null; + } + + public IndexLoadingConfig withSegmentTier(@Nullable String segmentTier) { + if (Objects.equals(_segmentTier, segmentTier)) { + return this; + } + return copyWithSegmentTier(segmentTier); + } + + /// Creates a segment-local mutable wrapper with the given tier. This always creates a wrapper, even when the tier is + /// unchanged, so callers can safely apply additional segment-specific overrides. + public IndexLoadingConfig copyWithSegmentTier(@Nullable String segmentTier) { + IndexLoadingConfig derived = new IndexLoadingConfig(this); + derived.setSegmentTier(segmentTier); + return derived; } public String getTableDataDir() { @@ -352,25 +434,16 @@ public class IndexLoadingConfig { } public boolean isSkipSegmentPreprocess() { - if (_dirty) { - refreshIndexConfigs(); - } - return _skipSegmentPreprocess; + return getResolvedIndexState()._skipSegmentPreprocess; } @Nullable public FieldIndexConfigs getFieldIndexConfig(String columnName) { - if (_indexConfigsByColName == null || _dirty) { - refreshIndexConfigs(); - } - return _indexConfigsByColName.get(columnName); + return getResolvedIndexState()._indexConfigsByColName.get(columnName); } public Map<String, FieldIndexConfigs> getFieldIndexConfigByColName() { - if (_indexConfigsByColName == null || _dirty) { - refreshIndexConfigs(); - } - return unmodifiable(_indexConfigsByColName); + return unmodifiable(getResolvedIndexState()._indexConfigsByColName); } /// Returns a subset of the columns on the table. @@ -379,10 +452,11 @@ public class IndexLoadingConfig { /// tries its bests to get the columns from other attributes like [#getTableConfig()], which may also not be /// defined or may not be complete. private Set<String> getAllKnownColumns() { - assert _tableConfig != null && _schema == null; + assert _immutableState._tableConfig != null && _immutableState._schema == null; if (_knownColumns == null) { - Set<String> knownColumns = _tableConfig.getIndexingConfig().getAllReferencedColumns(); - List<FieldConfig> fieldConfigs = _tableConfig.getFieldConfigList(); + Set<String> knownColumns = + new HashSet<>(_immutableState._tableConfig.getIndexingConfig().getAllReferencedColumns()); + List<FieldConfig> fieldConfigs = _immutableState._tableConfig.getFieldConfigList(); if (fieldConfigs != null) { for (FieldConfig fieldConfig : fieldConfigs) { knownColumns.add(fieldConfig.getName()); @@ -394,7 +468,7 @@ public class IndexLoadingConfig { } public Map<String, Map<String, String>> getInstanceTierConfigs() { - return unmodifiable(_instanceTierConfigs); + return unmodifiable(_immutableState._instanceTierConfigs); } private <E> List<E> unmodifiable(List<E> list) { @@ -409,20 +483,23 @@ public class IndexLoadingConfig { return map == null ? null : Collections.unmodifiableMap(map); } - public void addOpenStructChildConfigs(SegmentMetadataImpl segmentMetadata) { - if (_indexConfigsByColName == null || _dirty) { - refreshIndexConfigs(); + public IndexLoadingConfig withOpenStructChildConfigs(SegmentMetadataImpl segmentMetadata) { + if (!_immutableState._hasOpenStructColumns) { + return this; } + ResolvedIndexState resolvedIndexState = getResolvedIndexState(); + Map<String, FieldIndexConfigs> indexConfigsByColName = resolvedIndexState._indexConfigsByColName; + Map<String, FieldIndexConfigs> updatedConfigs = null; for (Map.Entry<String, ColumnMetadata> entry : segmentMetadata.getColumnMetadataMap().entrySet()) { String childColumn = entry.getKey(); - if (!childColumn.contains(OpenStructNaming.SEPARATOR) || _indexConfigsByColName.containsKey(childColumn)) { + if (!childColumn.contains(OpenStructNaming.SEPARATOR) || indexConfigsByColName.containsKey(childColumn)) { continue; } if (OpenStructNaming.isSparseColumn(childColumn)) { continue; } String parentColumn = OpenStructNaming.parseParentColumn(childColumn); - FieldIndexConfigs parentConfigs = _indexConfigsByColName.get(parentColumn); + FieldIndexConfigs parentConfigs = indexConfigsByColName.get(parentColumn); if (parentConfigs == null) { continue; } @@ -442,16 +519,39 @@ public class IndexLoadingConfig { FieldIndexConfigsUtil.fromFieldConfig(keyFieldConfig, childFieldSpec)) .add(StandardIndexes.inverted(), enableInverted ? IndexConfig.ENABLED : IndexConfig.DISABLED) .build(); - _indexConfigsByColName.put(childColumn, childConfigs); + if (updatedConfigs == null) { + updatedConfigs = new HashMap<>(indexConfigsByColName); + } + updatedConfigs.put(childColumn, childConfigs); + } + if (updatedConfigs == null) { + return this; + } + IndexLoadingConfig derived = new IndexLoadingConfig(this); + derived._resolvedIndexState = resolvedIndexState.withIndexConfigsByColName(updatedConfigs); + return derived; + } + + public void addOpenStructChildConfigs(SegmentMetadataImpl segmentMetadata) { + _resolvedIndexState = withOpenStructChildConfigs(segmentMetadata)._resolvedIndexState; + } + + public IndexLoadingConfig withKnownColumns(Set<String> columns) { + if (_knownColumns != null && _knownColumns.containsAll(columns)) { + return this; } + IndexLoadingConfig derived = new IndexLoadingConfig(this); + derived.addKnownColumns(columns); + return derived; } public void addKnownColumns(Set<String> columns) { - if (_knownColumns == null) { - _knownColumns = new HashSet<>(columns); - } else { - _knownColumns.addAll(columns); + if (_knownColumns != null && _knownColumns.containsAll(columns)) { + return; } - _dirty = true; + Set<String> knownColumns = _knownColumns != null ? new HashSet<>(_knownColumns) : new HashSet<>(); + knownColumns.addAll(columns); + _knownColumns = knownColumns; + _resolvedIndexState = null; } } diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfigTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfigTest.java index 4edb1c383cf..4b246e98afc 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfigTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfigTest.java @@ -23,18 +23,26 @@ import java.io.IOException; import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import org.apache.pinot.segment.spi.ColumnMetadata; import org.apache.pinot.segment.spi.index.FieldIndexConfigs; import org.apache.pinot.segment.spi.index.ForwardIndexConfig; import org.apache.pinot.segment.spi.index.StandardIndexes; +import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl; import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig; import org.apache.pinot.spi.config.table.FieldConfig; import org.apache.pinot.spi.config.table.StarTreeIndexConfig; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.ComplexFieldSpec; +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.data.Schema; import org.apache.pinot.spi.env.PinotConfiguration; import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.ReadMode; import org.apache.pinot.spi.utils.builder.TableConfigBuilder; import org.testng.annotations.Test; @@ -46,6 +54,24 @@ import static org.testng.Assert.*; public class IndexLoadingConfigTest { private static final String TABLE_NAME = "table01"; + @Test + public void testReadModePrecedenceAndOverrideIsolation() { + InstanceDataManagerConfig instanceConfig = mock(InstanceDataManagerConfig.class); + when(instanceConfig.getReadMode()).thenReturn(ReadMode.heap); + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) + .setLoadMode("MMAP").build(); + + IndexLoadingConfig base = new IndexLoadingConfig(instanceConfig, tableConfig, null); + assertEquals(base.getReadMode(), ReadMode.mmap); + + IndexLoadingConfig derived = base.withSegmentTier("coldTier"); + derived.setReadMode(ReadMode.heap); + assertEquals(derived.getReadMode(), ReadMode.heap); + assertEquals(base.getReadMode(), ReadMode.mmap); + + assertEquals(new IndexLoadingConfig(instanceConfig, null, null).getReadMode(), ReadMode.heap); + } + @Test public void testCalculateIndexConfigsWithoutTierOverwrites() throws IOException { @@ -141,8 +167,10 @@ public class IndexLoadingConfigTest { .setStarTreeIndexConfigs(List.of(stIdxCfg)) .setTierOverwrites(JsonUtils.stringToJsonNode("{\"coldTier\": {\"starTreeIndexConfigs\": []}}")) .setFieldConfigList(Arrays.asList(col1Cfg, col2Cfg)).build(); - IndexLoadingConfig ilc = new IndexLoadingConfig(idmCfg, tableConfig, schema); - ilc.setSegmentTier("coldTier"); + IndexLoadingConfig base = new IndexLoadingConfig(idmCfg, tableConfig, schema); + IndexLoadingConfig ilc = base.withSegmentTier("coldTier"); + assertSame(base.withSegmentTier(null), base); + assertNull(base.getSegmentTier()); // Check index configs for coldTier assertEquals(ilc.getStarTreeIndexConfigs().size(), 0); Map<String, FieldIndexConfigs> allFieldCfgs = ilc.getFieldIndexConfigByColName(); @@ -154,6 +182,36 @@ public class IndexLoadingConfigTest { assertFalse(fieldCfgs.getConfig(StandardIndexes.inverted()).isEnabled()); assertFalse(fieldCfgs.getConfig(StandardIndexes.bloomFilter()).isEnabled()); assertFalse(fieldCfgs.getConfig(StandardIndexes.dictionary()).isEnabled()); + assertEquals(base.getStarTreeIndexConfigs().size(), 1); + assertTrue(base.getFieldIndexConfig("col1").getConfig(StandardIndexes.inverted()).isEnabled()); + assertTrue(base.getFieldIndexConfig("col2").getConfig(StandardIndexes.dictionary()).isEnabled()); + } + + @Test + public void testDerivedConfigIsolatesSegmentColumns() + throws IOException { + Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME) + .addField(new ComplexFieldSpec("event", DataType.OPEN_STRUCT, true, Map.of())).build(); + FieldConfig fieldConfig = JsonUtils.stringToObject( + "{\"name\":\"event\",\"indexes\":{\"open_struct\":{}}}", FieldConfig.class); + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) + .setFieldConfigList(List.of(fieldConfig)).build(); + IndexLoadingConfig base = new IndexLoadingConfig(tableConfig, schema); + ColumnMetadata child = mock(ColumnMetadata.class); + when(child.getFieldSpec()).thenReturn(new DimensionFieldSpec("event$key", DataType.INT, true)); + SegmentMetadataImpl metadata = mock(SegmentMetadataImpl.class); + when(metadata.getColumnMetadataMap()).thenReturn(new TreeMap<>(Map.of("event$key", child))); + + IndexLoadingConfig derived = base.withOpenStructChildConfigs(metadata); + assertNotSame(derived, base); + assertNotNull(derived.getFieldIndexConfig("event$key")); + assertNull(base.getFieldIndexConfig("event$key")); + + TableConfig schemaLessTable = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + IndexLoadingConfig schemaLess = new IndexLoadingConfig(schemaLessTable, null); + derived = schemaLess.withKnownColumns(Set.of("segmentColumn")); + assertNotNull(derived.getFieldIndexConfig("segmentColumn")); + assertNull(schemaLess.getFieldIndexConfig("segmentColumn")); } @Test --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
