xiangfu0 commented on code in PR #19477:
URL: https://github.com/apache/pinot/pull/19477#discussion_r4102160680
##########
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:
Addressed in 4ed98b81c1. `FilePerIndexDirectory` now serializes its buffer
cache (`getBuffer` / `newBuffer` / `removeIndex` / `close` are synchronized on
the directory), `SegmentDirectory.Reader#getIndexFor` states the
concurrent-read requirement lazy mode puts on implementations, and
`FilePerIndexDirectoryTest#testConcurrentGetBufferMapsEachIndexOnce` maps eight
columns from eight threads and asserts one buffer per column. Without the
synchronization that test fails (two distinct buffers for `col0`); with it the
after-method buffer check also confirms nothing escaped `close()`.
##########
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:
pinot-docs is a separate repository, so this stays a follow-up to file
alongside the merge: the key and its default, the first-access failure
trade-off from the description, and (after 4ed98b81c1) that v1/v2 segments are
covered because `FilePerIndexDirectory` serializes its buffer cache while
external `SegmentDirectory.Reader` implementations must be safe for concurrent
reads before the flag is turned on.
--
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]
