xiangfu0 commented on code in PR #19477:
URL: https://github.com/apache/pinot/pull/19477#discussion_r4102164314
##########
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:
Addressed in 4ed98b81c1:
`testDestroyWaitsForInFlightMaterializationAndClosesIt` blocks
`createIndexContainer(a)` on a latch, calls `destroy()` from another thread,
asserts it has not returned (`TimeoutException` on a 200 ms `get`) and that
nothing is closed yet, releases the latch, then asserts the container was
closed exactly once, the directory closed, and a later
`getDataSourceNullable("b")` throws without touching the materializer.
##########
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:
Partly addressed in 4ed98b81c1. `FstIndexConfig` has no setting beyond
enabled/disabled, so the equality it inherits from `IndexConfig` is already
complete — no change there. `OpenStructIndexConfig` now defines value equality
over every setting
(`OpenStructIndexConfigTest#testEqualityCoversEverySetting`), the materializer
Javadoc no longer relies on the reader factory ignoring the config, and
`ColumnMaterializerTest#testOpenStructConfigsThatDifferAreNotCollapsed` pins
that two OPEN_STRUCT columns with different dense keys keep their own configs.
##########
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:
Addressed in 4ed98b81c1. The lazy path cannot take the eager order (the
materializer holds the text reader and the star-tree provider needs the
materializer), so the star-tree construction is wrapped: on failure the
multi-column text reader and every container materialized so far are closed,
close failures are attached as suppressed exceptions, and the original error is
rethrown.
##########
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:
Already covered by the restack: the flag lives in `ImmutableState`
(`_immutableState._lazyColumnMaterialization`) and `copyWithSegmentTier` goes
through the copy constructor, so a tier copy carries it.
--
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]
