This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/data-3221-10-physical-column-names in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 2676ba99227009a6afab87cfd55afa7fce9df111 Author: Xiang Fu <[email protected]> AuthorDate: Sun Sep 6 16:05:58 2026 -0700 DATA-3221 (11): list physical columns without building the segment schema Two things the preprocess does on every segment load asked the segment metadata for its schema, and since the schema is derived and then cached, each one pinned a per-segment `Schema` for the segment's whole life: - `ForwardIndexHandler#computeOperations` needs the set of physical column names, and - `ColumnMinMaxValueGenerator` needs the columns its mode selects (the default mode is `ALL`, so this runs on every load). Both questions are answered by the column metadata the schema is itself derived from. `SegmentMetadata#getPhysicalColumnNames()` walks the column metadata for the first, falling back to the schema for a segment that holds no column metadata (a CONSUMING one), and the min/max generator now selects straight off each column's field spec. On a server measured with 13.6k loaded segments, the schemas built here were ~144 MB of tree entries and list slots, all of it a second copy of data the column metadata already holds. Co-Authored-By: Claude Opus 5 <[email protected]> --- .../segment/index/loader/ForwardIndexHandler.java | 2 +- .../ColumnMinMaxValueGenerator.java | 70 +++++++++------------- .../segment/index/SegmentMetadataImplTest.java | 35 +++++++++++ .../apache/pinot/segment/spi/SegmentMetadata.java | 22 +++++++ 4 files changed, 86 insertions(+), 43 deletions(-) diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java index 9c53152cdee..d9a5a93fa50 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java @@ -248,7 +248,7 @@ public class ForwardIndexHandler extends BaseIndexHandler { } Map<String, List<Operation>> columnOperationsMap = new HashMap<>(); - Set<String> existingAllColumns = segmentMetadata.getSchema().getPhysicalColumnNames(); + Set<String> existingAllColumns = segmentMetadata.getPhysicalColumnNames(); Set<String> existingDictColumns = _segmentDirectory.getColumnsWithIndex(StandardIndexes.dictionary()); Set<String> existingForwardIndexColumns = _segmentDirectory.getColumnsWithIndex(StandardIndexes.forward()); Set<String> existingInvertedIndexColumns = diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/columnminmaxvalue/ColumnMinMaxValueGenerator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/columnminmaxvalue/ColumnMinMaxValueGenerator.java index ae4b99f8479..3ea90dc46cf 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/columnminmaxvalue/ColumnMinMaxValueGenerator.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/columnminmaxvalue/ColumnMinMaxValueGenerator.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.math.BigDecimal; import java.util.ArrayList; import java.util.List; +import java.util.function.Consumer; import org.apache.commons.configuration2.PropertiesConfiguration; import org.apache.commons.lang3.Strings; import org.apache.pinot.segment.local.segment.creator.impl.SegmentColumnarIndexCreator; @@ -44,7 +45,6 @@ import org.apache.pinot.segment.spi.memory.PinotDataBuffer; import org.apache.pinot.segment.spi.store.SegmentDirectory; import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils; import org.apache.pinot.spi.data.FieldSpec; -import org.apache.pinot.spi.data.Schema; import org.apache.pinot.spi.utils.ByteArray; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -75,11 +75,11 @@ public class ColumnMinMaxValueGenerator { /// Returns the list of columns that need min/max values to be updated public List<String> columnMinMaxValueUpdates() { List<String> columns = new ArrayList<>(); - for (String column : getColumnsToAddMinMaxValue()) { - if (needAddColumnMinMaxValueForColumn(column)) { - columns.add(column); + forEachSelectedColumn(columnMetadata -> { + if (needAddColumnMinMaxValueForColumn(columnMetadata)) { + columns.add(columnMetadata.getColumnName()); } - } + }); return columns; } @@ -87,53 +87,40 @@ public class ColumnMinMaxValueGenerator { throws Exception { Preconditions.checkState(_columnMinMaxValueGeneratorMode != ColumnMinMaxValueGeneratorMode.NONE); _segmentProperties = SegmentMetadataUtils.getPropertiesConfiguration(_segmentMetadata); - for (String column : getColumnsToAddMinMaxValue()) { - addColumnMinMaxValueForColumn(column); - } + forEachSelectedColumn(this::addColumnMinMaxValueForColumn); if (_minMaxValueAdded) { SegmentMetadataUtils.savePropertiesConfiguration(_segmentProperties, _segmentMetadata.getIndexDir()); } } - private List<String> getColumnsToAddMinMaxValue() { - Schema schema = _segmentMetadata.getSchema(); - List<String> columnsToAddMinMaxValue = new ArrayList<>(); + /// Runs `action` on every column the generator mode selects. + /// + /// The selection reads the field specs off the column metadata rather than off `_segmentMetadata.getSchema()`, + /// which is the same data (the schema is derived from the column metadata) but costs a `Schema` per segment. This + /// runs on every segment load — the default mode is `ALL` — so a schema built here would be cached for the + /// segment's whole life, and a server holding tens of thousands of wide segments would keep one per segment. + private void forEachSelectedColumn(Consumer<ColumnMetadata> action) { + for (ColumnMetadata columnMetadata : _segmentMetadata.getAllColumnMetadata()) { + FieldSpec fieldSpec = columnMetadata.getFieldSpec(); + if (!fieldSpec.isVirtualColumn() && isSelected(fieldSpec.getFieldType())) { + action.accept(columnMetadata); + } + } + } - // mode ALL - use all columns - // mode NON_METRIC - use all dimensions and time columns - // mode TIME - use only time columns + /// Whether the generator mode covers the given field type: `ALL` takes every column, `NON_METRIC` every column but + /// the metrics, `TIME` only the time columns. + private boolean isSelected(FieldSpec.FieldType fieldType) { switch (_columnMinMaxValueGeneratorMode) { case ALL: - for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) { - if (!fieldSpec.isVirtualColumn()) { - columnsToAddMinMaxValue.add(fieldSpec.getName()); - } - } - break; + return true; case NON_METRIC: - for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) { - if (!fieldSpec.isVirtualColumn() && fieldSpec.getFieldType() != FieldSpec.FieldType.METRIC) { - columnsToAddMinMaxValue.add(fieldSpec.getName()); - } - } - break; + return fieldType != FieldSpec.FieldType.METRIC; case TIME: - for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) { - if (!fieldSpec.isVirtualColumn() && (fieldSpec.getFieldType() == FieldSpec.FieldType.TIME - || fieldSpec.getFieldType() == FieldSpec.FieldType.DATE_TIME)) { - columnsToAddMinMaxValue.add(fieldSpec.getName()); - } - } - break; + return fieldType == FieldSpec.FieldType.TIME || fieldType == FieldSpec.FieldType.DATE_TIME; default: throw new IllegalStateException("Unsupported generator mode: " + _columnMinMaxValueGeneratorMode); } - - return columnsToAddMinMaxValue; - } - - private boolean needAddColumnMinMaxValueForColumn(String columnName) { - return needAddColumnMinMaxValueForColumn(_segmentMetadata.getColumnMetadataFor(columnName)); } private boolean needAddColumnMinMaxValueForColumn(ColumnMetadata columnMetadata) { @@ -141,8 +128,7 @@ public class ColumnMinMaxValueGenerator { && !columnMetadata.isMinMaxValueInvalid(); } - private void addColumnMinMaxValueForColumn(String columnName) { - ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(columnName); + private void addColumnMinMaxValueForColumn(ColumnMetadata columnMetadata) { if (!needAddColumnMinMaxValueForColumn(columnMetadata)) { return; } @@ -155,7 +141,7 @@ public class ColumnMinMaxValueGenerator { _minMaxValueAdded = true; } catch (Exception e) { LOGGER.error("Caught exception while generating min/max value for column: {} in segment: {}, continuing without " - + "persisting them", columnName, _segmentMetadata.getName(), e); + + "persisting them", columnMetadata.getColumnName(), _segmentMetadata.getName(), e); } } diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java index 31ede949bbe..af97763a2be 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/SegmentMetadataImplTest.java @@ -40,7 +40,10 @@ import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoa import org.apache.pinot.segment.local.segment.creator.SegmentTestUtils; import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; import org.apache.pinot.segment.local.segment.index.converter.SegmentV1V2ToV3FormatConverter; +import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig; +import org.apache.pinot.segment.local.segment.index.loader.SegmentPreProcessor; import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader; +import org.apache.pinot.segment.local.segment.store.SegmentLocalFSDirectory; import org.apache.pinot.segment.local.segment.virtualcolumn.VirtualColumnProviderFactory; import org.apache.pinot.segment.spi.ColumnMetadata; import org.apache.pinot.segment.spi.ImmutableSegment; @@ -51,6 +54,7 @@ import org.apache.pinot.segment.spi.index.StandardIndexes; import org.apache.pinot.segment.spi.index.metadata.ColumnMetadataImpl; import org.apache.pinot.segment.spi.index.metadata.EmptyColumnMetadata; 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.table.FieldConfig; import org.apache.pinot.spi.config.table.OpenStructIndexConfig; @@ -332,6 +336,37 @@ public class SegmentMetadataImplTest { assertEquals(SegmentMetadataImpl.getNumSchemaMaterializations(), materializations + 1); } + /// The preprocess that runs on every segment load asks the forward-index handler which physical columns exist. That + /// question must not build the per-segment schema: doing so once per segment pins one [Schema] per loaded segment + /// for its whole life, which on a server holding tens of thousands of wide segments is hundreds of megabytes. + @Test + public void testPreprocessDoesNotBuildTheSegmentSchema() + throws Exception { + // The forward-index handler skips segments older than v3, so the preprocess only reaches it on a v3 segment. + new SegmentV1V2ToV3FormatConverter().convert(_segmentDirectory); + + long materializations = SegmentMetadataImpl.getNumSchemaMaterializations(); + SegmentMetadataImpl metadata = new SegmentMetadataImpl(_segmentDirectory); + Set<String> physical = metadata.getPhysicalColumnNames(); + assertFalse(metadata.isSchemaMaterialized(), "listing physical columns must not build the segment schema"); + assertEquals(SegmentMetadataImpl.getNumSchemaMaterializations(), materializations); + assertEquals(physical, metadata.getSchema().getPhysicalColumnNames(), + "the derived names must equal what the schema reports"); + assertFalse(physical.contains(BuiltInVirtualColumn.DOCID)); + + TableConfig tableConfig = + new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").setTimeColumnName("daysSinceEpoch").build(); + IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(tableConfig, metadata.getSchema()); + indexLoadingConfig.setReadMode(ReadMode.mmap); + long beforePreprocess = SegmentMetadataImpl.getNumSchemaMaterializations(); + try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(_segmentDirectory, ReadMode.mmap); + SegmentPreProcessor preProcessor = new SegmentPreProcessor(segmentDirectory, indexLoadingConfig)) { + preProcessor.process(); + } + assertEquals(SegmentMetadataImpl.getNumSchemaMaterializations(), beforePreprocess, + "segment preprocess must not build any segment schema"); + } + /// Loading a segment registers the built-in virtual columns in the column metadata, so the schema derived afterwards /// includes them exactly as the schema the loader used to build eagerly did, while neither the load nor serving the /// segment (column listings, data sources, the metadata JSON) builds any schema. diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentMetadata.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentMetadata.java index 142bf688dca..a2caae4105e 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentMetadata.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentMetadata.java @@ -19,6 +19,7 @@ package org.apache.pinot.segment.spi; import com.fasterxml.jackson.databind.JsonNode; +import com.google.common.collect.Sets; import java.io.File; import java.util.Collection; import java.util.List; @@ -33,6 +34,7 @@ import org.apache.pinot.segment.spi.creator.SegmentVersion; import org.apache.pinot.segment.spi.index.multicolumntext.MultiColumnTextMetadata; import org.apache.pinot.segment.spi.index.startree.StarTreeV2Metadata; import org.apache.pinot.spi.annotations.InterfaceAudience; +import org.apache.pinot.spi.data.FieldSpec; import org.apache.pinot.spi.data.Schema; import org.joda.time.Duration; import org.joda.time.Interval; @@ -161,6 +163,26 @@ public interface SegmentMetadata { return getColumnMetadataMap().get(column); } + /// The names of the physical (non-virtual) columns, i.e. `getSchema().getPhysicalColumnNames()` without building + /// the schema. Segment load runs this once per segment (the forward-index handler asks which columns exist), and on + /// a server holding tens of thousands of wide segments a schema built there would be cached for the segment's whole + /// life: one [Schema] per segment, each with a tree entry and two list slots per column. A segment that holds no + /// column metadata (a CONSUMING one) still answers from its schema, which it was constructed with. + default Set<String> getPhysicalColumnNames() { + Collection<ColumnMetadata> columnMetadata = getAllColumnMetadata(); + if (columnMetadata.isEmpty()) { + return getSchema().getPhysicalColumnNames(); + } + Set<String> physicalColumnNames = Sets.newHashSetWithExpectedSize(columnMetadata.size()); + for (ColumnMetadata metadata : columnMetadata) { + FieldSpec fieldSpec = metadata.getFieldSpec(); + if (!fieldSpec.isVirtualColumn()) { + physicalColumnNames.add(fieldSpec.getName()); + } + } + return physicalColumnNames; + } + /// Registers the metadata of a column, replacing any metadata already registered under the same name. An /// implementation that holds no column metadata (a CONSUMING segment) may reject this. default void addColumnMetadata(String column, ColumnMetadata columnMetadata) { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
