xiangfu0 commented on code in PR #19477:
URL: https://github.com/apache/pinot/pull/19477#discussion_r4091516952
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java:
##########
@@ -174,6 +216,117 @@ public ImmutableSegmentImpl(
this(segmentDirectory, segmentMetadata, columnIndexContainerMap,
starTreeIndexContainer, null);
}
+ /// Creates a segment that materializes its physical columns lazily through
`columnMaterializer`.
+ ///
+ /// `materializedIndexContainers` holds the containers created at load
(built-in virtual columns and star-tree
+ /// dimensions) and becomes the registry of every container created
afterwards, so that [#destroy()] closes exactly
+ /// the materialized ones. The columns already in it get their data source
now, as in the eager mode.
+ ImmutableSegmentImpl(SegmentDirectory segmentDirectory, SegmentMetadataImpl
segmentMetadata,
+ ColumnMaterializer columnMaterializer, ConcurrentMap<String,
ColumnIndexContainer> materializedIndexContainers,
+ @Nullable StarTreeIndexContainer starTreeIndexContainer,
+ @Nullable MultiColumnLuceneTextIndexReader multiColumnTextIndex) {
+ _segmentDirectory = segmentDirectory;
+ _segmentMetadata = segmentMetadata;
+ _indexContainerMap = materializedIndexContainers;
+ _starTreeIndexContainer = starTreeIndexContainer;
+ _multiColumnTextIndex = multiColumnTextIndex;
+ _columnMaterializer = columnMaterializer;
+ _openStructChildren = groupOpenStructChildren(segmentMetadata);
+ _materializationLock = new ReentrantReadWriteLock();
+ _dataSources = new ConcurrentHashMap<>();
+ for (String column : materializedIndexContainers.keySet()) {
+ materializeDataSource(column);
+ }
+ }
+
+ /// Groups the materialized OPEN_STRUCT child columns under their parent,
keeping only the parents the segment schema
+ /// declares as complex (the same rule the eager constructor applies).
+ @Nullable
+ private static Map<String, List<String>>
groupOpenStructChildren(SegmentMetadataImpl segmentMetadata) {
+ Map<String, List<String>> children = null;
+ for (Map.Entry<String, ColumnMetadata> entry :
segmentMetadata.getColumnMetadataMap().entrySet()) {
+ if (entry.getValue() instanceof ColumnMetadataImpl impl &&
impl.isMaterializedChild()) {
+ if (children == null) {
+ children = new HashMap<>();
+ }
+ children.computeIfAbsent(impl.getParentColumn(), k -> new
ArrayList<>()).add(entry.getKey());
+ }
+ }
+ if (children == null) {
+ return null;
+ }
+ Schema schema = segmentMetadata.getSchema();
+ children.keySet()
+ .removeIf(parent -> !(schema != null && schema.getFieldSpecFor(parent)
instanceof ComplexFieldSpec));
+ return children.isEmpty() ? null : children;
+ }
+
+ /// Lazy mode: returns the data source of the column, creating it on first
access, or `null` when the segment has no
+ /// such column. OPEN_STRUCT child columns are reachable only through their
parent, as in the eager mode.
+ @Nullable
+ private DataSource materializeDataSource(String column) {
+ ColumnMetadata columnMetadata =
_segmentMetadata.getColumnMetadataMap().get(column);
+ boolean openStructParent = _openStructChildren != null &&
_openStructChildren.containsKey(column);
+ if (!openStructParent && (columnMetadata == null ||
isMaterializedChild(columnMetadata))) {
+ return null;
+ }
+ Lock lock = _materializationLock.readLock();
+ lock.lock();
+ try {
+ checkNotDestroyed(column);
+ // Single flight per column: the mapping function runs at most once per
column and leaves no mapping when it
+ // fails. It never reads this map again (creating the children of an
OPEN_STRUCT parent goes through
+ // _indexContainerMap only), which computeIfAbsent forbids.
+ return _dataSources.computeIfAbsent(column,
+ k -> openStructParent ? createOpenStructDataSource(k) :
createDataSource(k, columnMetadata));
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private DataSource createDataSource(String column, ColumnMetadata
columnMetadata) {
+ ColumnIndexContainer container = materializedIndexContainer(column,
columnMetadata);
+ return columnMetadata.getFieldSpec().getDataType() ==
FieldSpec.DataType.MAP
+ ? new ImmutableMapDataSource(columnMetadata, container) : new
ImmutableDataSource(columnMetadata, container);
+ }
+
+ private DataSource createOpenStructDataSource(String parent) {
+ Map<String, ColumnMetadata> columnMetadataMap =
_segmentMetadata.getColumnMetadataMap();
+ Map<String, DataSource> denseChildren = new HashMap<>();
+ DataSource sparseChild = null;
+ for (String child : _openStructChildren.get(parent)) {
+ ColumnMetadata childMetadata = columnMetadataMap.get(child);
+ DataSource childDataSource =
+ new ImmutableDataSource(childMetadata,
materializedIndexContainer(child, childMetadata));
+ if (OpenStructNaming.isSparseColumn(child)) {
+ sparseChild = childDataSource;
+ } else {
+ denseChildren.put(OpenStructNaming.parseKey(child), childDataSource);
+ }
+ }
+ ComplexFieldSpec fieldSpec = (ComplexFieldSpec)
_segmentMetadata.getSchema().getFieldSpecFor(parent);
+ List<String> sparseKeys =
+ columnMetadataMap.get(parent) instanceof ColumnMetadataImpl impl ?
impl.getSparseKeys() : null;
+ return new ImmutableOpenStructDataSource(fieldSpec, denseChildren,
sparseChild, _segmentMetadata.getTotalDocs(),
+ sparseKeys);
+ }
+
+ /// Lazy mode: returns the index container of the column, creating and
registering it on first access. The mapping
+ /// function opens the column's index readers while it holds the map's bin
lock, so a slow open (e.g. an on-heap
+ /// dictionary) can briefly stall the first access to an unrelated column in
the same bin.
+ private ColumnIndexContainer materializedIndexContainer(String column,
ColumnMetadata columnMetadata) {
+ return _indexContainerMap.computeIfAbsent(column, k ->
_columnMaterializer.createIndexContainer(columnMetadata));
Review Comment:
**MAJOR [BUG-RACE]:** this moves `SegmentDirectory.Reader#getIndexFor` from
the single loader thread onto query threads, and distinct columns land in
distinct bins, so two query threads run `PhysicalColumnIndexContainer`
constructors for different columns of one segment concurrently.
For **v1/v2 segments** `SegmentLocalFSDirectory.loadData()` picks
`FilePerIndexDirectory`, whose `_indexBuffers` is a plain `new HashMap<>()`
(`FilePerIndexDirectory.java:52`) and whose `getReadBufferFor` does
`containsKey` / `mapForReads` / `put` with no synchronization (`:144-159`); the
`getIndexForColumn` path in `SegmentLocalFSDirectory` (`:302-306`) takes no
lock either. Nothing calls `getBuffer` concurrently on one directory today
(eager load is single-threaded, preprocess handlers are sequential), so this PR
introduces the first concurrent use. Concurrent first accesses to two columns
can lose an entry (that `PinotDataBuffer` is then never released by `close()` →
mmap leak on every destroy/reload) or leave the table inconsistent. v3 is safe:
`SingleFileIndexDirectory.checkAndGetIndexBuffer` only reads `_columnEntries`
populated at construction.
Not CRITICAL because it needs the opt-in flag plus a non-default segment
format, and the bytes each thread reads are still correct.
Fix, either: (a) make `FilePerIndexDirectory.getReadBufferFor` /
`getWriteBufferFor` safe for concurrent callers
(`ConcurrentHashMap.computeIfAbsent` wrapping the `IOException`, or
`synchronized`); or (b) fall back to eager in
`ImmutableSegmentLoader.load(SegmentDirectory, …)` when
`segmentMetadata.getVersion() != v3` and document it. Either way, state on the
`SegmentDirectory.Reader` SPI (`SegmentDirectory.java:163-176`, which has no
thread-safety contract today) that lazy mode requires `getIndexFor` /
`hasIndexFor` to be safe under concurrent calls on distinct columns — external
tiered readers must meet that too.
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImplTest.java:
##########
@@ -86,4 +125,287 @@ private static ImmutableSegmentImpl
createSegment(SegmentDirectory segmentDirect
when(segmentMetadata.getColumnMetadataMap()).thenReturn(new TreeMap<>());
return new ImmutableSegmentImpl(segmentDirectory, segmentMetadata,
Map.of(), null);
}
+
+ /// The eager (flag off) mode is untouched: every data source exists from
construction over the given containers.
+ @Test
+ public void testEagerModeCreatesDataSourcesAtConstruction() {
+ ColumnMetadataImpl a = columnMetadata(intColumn("a"), null);
+ ColumnIndexContainer containerA = mock(ColumnIndexContainer.class);
+ ImmutableSegmentImpl segment =
+ new ImmutableSegmentImpl(mock(SegmentDirectory.class),
segmentMetadata(schema(a), a), Map.of("a", containerA),
+ null);
+
+ assertSame(segment.getDataSourceNullable("a").getIndexContainer(),
containerA);
+ assertNull(segment.getDataSourceNullable("unknown"));
+ }
+
+ @Test
+ public void testLazyModeCreatesNothingAtConstruction()
+ throws Exception {
+ ColumnMetadataImpl a = columnMetadata(intColumn("a"), null);
+ ColumnMetadataImpl b = columnMetadata(intColumn("b"), null);
+ ColumnMaterializer materializer = mock(ColumnMaterializer.class);
+ SegmentDirectory segmentDirectory = mock(SegmentDirectory.class);
+ ImmutableSegmentImpl segment = lazySegment(segmentDirectory, schema(a, b),
materializer, a, b);
+
+ verifyNoInteractions(materializer);
+ // Column listings come from the metadata schema and never materialize
anything
+ assertEquals(segment.getColumnNames(), Set.of("a", "b"));
+ assertEquals(segment.getPhysicalColumnNames(), Set.of("a", "b"));
+ // Neither does asking for a column the segment does not have
+ assertNull(segment.getDataSourceNullable("unknown"));
+ assertThrows(NullPointerException.class, () -> segment.getIndex("unknown",
StandardIndexes.forward()));
+ verifyNoInteractions(materializer);
+
+ segment.destroy();
+ verify(segmentDirectory).close();
+ }
+
+ @Test
+ public void testLazyModeMaterializesEachColumnOnceUnderConcurrentAccess()
+ throws Exception {
+ ColumnMetadataImpl a = columnMetadata(intColumn("a"), null);
+ ColumnIndexContainer containerA = mock(ColumnIndexContainer.class);
+ ColumnMaterializer materializer = mock(ColumnMaterializer.class);
+ AtomicInteger creations = new AtomicInteger();
+ when(materializer.createIndexContainer(a)).thenAnswer(invocation -> {
+ creations.incrementAndGet();
+ // Widen the window in which every other caller must wait for this
creation instead of starting its own
+ Thread.sleep(50);
+ return containerA;
+ });
+ ImmutableSegmentImpl segment = lazySegment(mock(SegmentDirectory.class),
schema(a), materializer, a);
+
+ int numCallers = 16;
+ ExecutorService executor = Executors.newFixedThreadPool(numCallers);
+ DataSource first;
+ try {
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<DataSource>> futures = new ArrayList<>(numCallers);
+ for (int i = 0; i < numCallers; i++) {
+ futures.add(executor.submit(() -> {
+ start.await();
+ return segment.getDataSourceNullable("a");
+ }));
+ }
+ start.countDown();
+ first = futures.get(0).get();
+ for (Future<DataSource> future : futures) {
+ assertSame(future.get(), first);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+
+ assertNotNull(first);
+ assertSame(first.getIndexContainer(), containerA);
+ assertEquals(creations.get(), 1);
+ verify(materializer, times(1)).createIndexContainer(a);
+ }
+
+ @Test
+ public void
testDestroyClosesOnlyMaterializedContainersAndRefusesLaterMaterialization()
Review Comment:
**MAJOR [testing]:** the only reason `_materializationLock` exists is the
destroy-vs-in-flight-materialization interleaving, and this test never
exercises it: column `a` is materialized to completion before `destroy()`, so
the write-lock wait never runs. Dropping the lock, or flipping `_destroyed`
before the container is registered, would still pass; the concurrency test
above covers single-flight only. I verified the ordering by reading (read lock
across `checkNotDestroyed` + `computeIfAbsent`; write lock excludes all
readers), but this is exactly the property that needs a guard.
Suggested shape: block `createIndexContainer(a)` on a latch inside the
mocked `ColumnMaterializer`, call `destroy()` from another thread, assert it
has not returned while blocked, release the latch, then assert the container
was closed exactly once and `getDataSourceNullable("b")` throws
`IllegalStateException`.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java:
##########
@@ -282,6 +277,58 @@ public static ImmutableSegment load(SegmentDirectory
segmentDirectory, IndexLoad
return segment;
}
+ /// Lazy counterpart of the load above (see [ImmutableSegmentImpl]): no
per-column container is created here. The
+ /// built-in virtual columns keep their eager containers, the star-tree
dimensions are materialized now because the
+ /// star-tree shares their dictionaries, and every other physical column
waits for its first access. The
+ /// [ColumnMaterializer] snapshots the per-column index configs before the
virtual columns are added to the metadata,
+ /// so it covers exactly the physical columns.
+ private static ImmutableSegmentImpl loadWithLazyColumns(SegmentDirectory
segmentDirectory,
+ SegmentDirectory.Reader segmentReader, SegmentMetadataImpl
segmentMetadata,
+ IndexLoadingConfig indexLoadingConfig)
+ throws IOException {
+ Map<String, ColumnMetadata> columnMetadataMap =
segmentMetadata.getColumnMetadataMap();
+ MultiColumnLuceneTextIndexReader mcTextReader = null;
+ Set<String> mcTextColumns = Set.of();
+ if (segmentReader.hasMultiColumnTextIndex()) {
+ mcTextReader = new MultiColumnLuceneTextIndexReader(segmentMetadata);
Review Comment:
**MINOR (resource leak on failure):** the lazy path creates
`MultiColumnLuceneTextIndexReader` before the star-tree container, while the
eager path does the reverse (`:257-270`). If `new StarTreeIndexContainer(...)`
throws here, the Lucene file handles leak: `segmentDirectory.close()` in the
file-load catch (`:160-165`) does not close `mcTextReader`. Create the
star-tree first (matching eager) or close `mcTextReader` on failure.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java:
##########
@@ -45,200 +44,65 @@
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;
-/// Index loading config with shared table-level state and segment-local
mutable overrides.
+/// Table level index loading config.
public class IndexLoadingConfig {
private static final int DEFAULT_REALTIME_AVG_MULTI_VALUE_COUNT = 2;
public static final String READ_MODE_KEY = "readMode";
- private final ImmutableState _immutableState;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final TableConfig _tableConfig;
+ private final Schema _schema;
- // Mutable config and segment-specific overrides.
- @Nullable
- private ReadMode _readModeOverride;
- @Nullable
- private SegmentVersion _segmentVersionOverride;
+ // These fields can be modified after initialization
+ // TODO: Revisit them
+ private ReadMode _readMode = ReadMode.DEFAULT_MODE;
+ private SegmentVersion _segmentVersion;
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();
- }
-
- 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());
- }
- }
-
- _instanceId = instanceId;
- _readMode = readMode;
- _segmentVersion = segmentVersion;
- _isRealtimeOffHeapAllocation = isRealtimeOffHeapAllocation;
- _isDirectRealtimeOffHeapAllocation = isDirectRealtimeOffHeapAllocation;
- _realtimeAvgMultiValueCount = realtimeAvgMultiValueCount;
- _segmentStoreURI = segmentStoreURI;
- _segmentDirectoryLoader = segmentDirectoryLoader;
- _instanceTierConfigs = instanceTierConfigs;
- _sortedColumns = sortedColumns;
- _columnMinMaxValueGeneratorMode = columnMinMaxValueGeneratorMode;
- _hasOpenStructColumns = hasOpenStructColumns;
- }
- }
-
- /// 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 ResolvedIndexState withIndexConfigsByColName(Map<String,
FieldIndexConfigs> indexConfigsByColName) {
- return new ResolvedIndexState(_enableDynamicStarTreeCreation,
_starTreeIndexConfigs, _enableDefaultStarTree,
- indexConfigsByColName, _skipSegmentPreprocess,
_multiColTextIndexConfig);
- }
- }
+ // 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;
+ private boolean _lazyColumnMaterialization;
Review Comment:
**MINOR (restack):** after rebasing onto #19571, `IndexLoadingConfig` is
`ImmutableState` + `ResolvedIndexState` with `copyWithSegmentTier`. This flag
must move into `ImmutableState` (sourced from the instance config at
construction) and be carried by `copyWithSegmentTier`, or a tier copy silently
loads eagerly. The mutable field + `setLazyColumnMaterialization` setter
pattern here will not survive the merge as-is.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ColumnMaterializer.java:
##########
@@ -0,0 +1,125 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.indexsegment.immutable;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import javax.annotation.Nullable;
+import
org.apache.pinot.segment.local.segment.index.column.PhysicalColumnIndexContainer;
+import
org.apache.pinot.segment.local.segment.index.readers.text.MultiColumnLuceneTextIndexReader;
+import org.apache.pinot.segment.spi.ColumnMetadata;
+import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
+import org.apache.pinot.segment.spi.index.column.ColumnIndexContainer;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+
+
+/// Creates the [ColumnIndexContainer] of a physical column of an
[ImmutableSegmentImpl] on demand.
+///
+/// [ImmutableSegmentLoader] builds one per segment when lazy column
materialization is on, instead of a
+/// [PhysicalColumnIndexContainer] per column at load. It retains only what
creating a container later needs: the
+/// segment reader (held for the segment's lifetime anyway), the
forward-index-only flag, the shared multi-column text
+/// index reader and, per column, the [FieldIndexConfigs] that were in effect
at load.
+///
+/// The per-column configs are snapshotted at construction because the loading
config they come from is mutable and
+/// shared: its map is replaced whenever the config is refreshed and mutated
in place while OPEN_STRUCT child configs
+/// are resolved. The snapshot is compacted so that a wide segment retains
close to nothing per column, which is the
+/// point of materializing lazily: configs that are equal by value collapse to
one instance, the most common one
+/// becomes the implicit default, and only the columns that differ from it
keep an entry (keyed by the column-name
+/// strings the segment metadata already holds). A column absent from the
loading config maps to
+/// [FieldIndexConfigs#EMPTY], exactly what the eager path hands to the
container. Collapsing relies on the value
+/// equality of the index configs; a config type that inherits the
enabled/disabled-only equality of `IndexConfig`
Review Comment:
**MINOR:** `OpenStructIndexConfig` is not the only one — `FstIndexConfig`
also has no `equals`/`hashCode` (those two are the only `extends IndexConfig`
types in `src/main` without overrides). Safe today only because
`FstIndexType.ReaderFactory` ignores the config (`FstIndexType.java:159-176`),
so collapse-by-value rests on an unguarded invariant. Suggest adding
`equals`/`hashCode` to both, plus a test that every config class reachable from
`IndexService.getAllIndexes()` overrides `equals`.
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManagerConfig.java:
##########
@@ -107,6 +107,11 @@ public class HelixInstanceDataManagerConfig implements
InstanceDataManagerConfig
public static final String DISABLE_DIMENSION_TABLE_PRELOAD =
"disable.dimension.table.preload";
private static final boolean DEFAULT_DISABLE_DIMENSION_TABLE_PRELOAD = false;
+ // Whether to create the index container and data source of a physical
column on first access instead of for every
+ // column at segment load. Off by default. See
InstanceDataManagerConfig#isLazyColumnMaterialization().
+ public static final String LAZY_COLUMN_MATERIALIZATION =
"segment.lazy.column.materialization";
Review Comment:
**MINOR (docs):** new public server config key
`pinot.server.instance.segment.lazy.column.materialization` (default off) needs
a pinot-docs entry, including the trade-off already stated in the description
(index errors on never-queried columns surface at first access, on the query
thread) and, once fixed, the v1/v2 caveat from the race finding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
