Jackie-Jiang commented on a change in pull request #7885: URL: https://github.com/apache/pinot/pull/7885#discussion_r769145870
########## File path: pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/TextIndexCreatorProvider.java ########## @@ -0,0 +1,37 @@ +/** + * 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.spi.creator; + +import java.io.IOException; +import org.apache.pinot.segment.spi.index.creator.TextIndexCreator; + + +public interface TextIndexCreatorProvider { + + /** + * Creates a {@see TextIndexCreator} from information about index creation. + * This allows a plugin to pattern match index creation information to select + * an appropriate implementation. + * @param context context about the index creation. + * @return a {@see ForwardIndexCreator} + * @throws IOException whenever something goes wrong matching or constructing the creator + */ + TextIndexCreator newFSTIndexCreator(IndexCreationContext context) Review comment: (major) the `newTextIndexCreator()` method is missing ########## File path: pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/DefaultIndexCreatorProvider.java ########## @@ -0,0 +1,249 @@ +/** + * 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.segment.creator.impl; + +import com.google.common.base.Preconditions; +import java.io.File; +import java.io.IOException; +import java.util.Map; +import java.util.Objects; +import org.apache.pinot.segment.local.io.writer.impl.BaseChunkSVForwardIndexWriter; +import org.apache.pinot.segment.local.segment.creator.impl.fwd.MultiValueFixedByteRawIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.fwd.MultiValueUnsortedForwardIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.fwd.MultiValueVarByteRawIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueFixedByteRawIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueSortedForwardIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueUnsortedForwardIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueVarByteRawIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.inv.OffHeapBitmapInvertedIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.inv.OnHeapBitmapInvertedIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.inv.geospatial.OffHeapH3IndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.inv.geospatial.OnHeapH3IndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.inv.json.OffHeapJsonIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.inv.json.OnHeapJsonIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.inv.text.LuceneFSTIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.text.LuceneTextIndexCreator; +import org.apache.pinot.segment.local.utils.nativefst.NativeFSTIndexCreator; +import org.apache.pinot.segment.spi.compression.ChunkCompressionType; +import org.apache.pinot.segment.spi.creator.IndexCreationContext; +import org.apache.pinot.segment.spi.creator.IndexCreatorProvider; +import org.apache.pinot.segment.spi.index.creator.DictionaryBasedInvertedIndexCreator; +import org.apache.pinot.segment.spi.index.creator.ForwardIndexCreator; +import org.apache.pinot.segment.spi.index.creator.GeoSpatialIndexCreator; +import org.apache.pinot.segment.spi.index.creator.JsonIndexCreator; +import org.apache.pinot.segment.spi.index.creator.TextIndexCreator; +import org.apache.pinot.segment.spi.index.reader.H3IndexResolution; +import org.apache.pinot.spi.config.table.FSTType; +import org.apache.pinot.spi.config.table.FieldConfig; +import org.apache.pinot.spi.data.FieldSpec; + + +public final class DefaultIndexCreatorProvider implements IndexCreatorProvider { + + @Override + public ForwardIndexCreator newForwardIndexCreator(IndexCreationContext context) + throws Exception { + if (context.isRawColumn()) { + boolean deriveNumDocsPerChunk = + shouldDeriveNumDocsPerChunk(context.getFieldSpec().getName(), context.getColumnProperties()); + int writerVersion = rawIndexWriterVersion(context.getFieldSpec().getName(), context.getColumnProperties()); + if (context.getFieldSpec().isSingleValueField()) { + return getRawIndexCreatorForSVColumn(context.getIndexDir(), context.getChunkCompressionType(), + context.getFieldSpec().getName(), + context.getFieldSpec().getDataType().getStoredType(), context.getTotalDocs(), + context.getLengthOfLongestEntry(), deriveNumDocsPerChunk, writerVersion); + } else { + return getRawIndexCreatorForMVColumn(context.getIndexDir(), context.getChunkCompressionType(), + context.getFieldSpec().getName(), context.getFieldSpec().getDataType().getStoredType(), + context.getTotalDocs(), + context.getMaxNumberOfMultiValueElements(), deriveNumDocsPerChunk, writerVersion, + context.getMaxRowLengthInBytes()); + } + } else { + if (context.getFieldSpec().isSingleValueField()) { + if (context.isSorted()) { + return new SingleValueSortedForwardIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), + context.getDistinctValueCount()); + } else { + return new SingleValueUnsortedForwardIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), + context.getDistinctValueCount(), + context.getTotalDocs()); + } + } else { + return new MultiValueUnsortedForwardIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), + context.getDistinctValueCount(), + context.getTotalDocs(), + context.getTotalNumberOfEntries()); + } + } + } + + @Override + public DictionaryBasedInvertedIndexCreator newInvertedIndexCreator(IndexCreationContext context) + throws IOException { + if (context.isOnHeap()) { + return new OnHeapBitmapInvertedIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), + context.getDistinctValueCount()); + } else { + return new OffHeapBitmapInvertedIndexCreator(context.getIndexDir(), context.getFieldSpec(), + context.getDistinctValueCount(), context.getTotalDocs(), + context.getTotalNumberOfEntries()); + } + } + + @Override + public JsonIndexCreator newJsonIndexCreator(IndexCreationContext context) + throws IOException { + Preconditions.checkState(context.getFieldSpec().isSingleValueField(), + "Json index is currently only supported on single-value columns"); + Preconditions.checkState(context.getFieldSpec().getDataType().getStoredType() == FieldSpec.DataType.STRING, + "Json index is currently only supported on STRING columns"); + return context.isOnHeap() ? new OnHeapJsonIndexCreator(context.getIndexDir(), + context.getFieldSpec().getName()) + : new OffHeapJsonIndexCreator(context.getIndexDir(), context.getFieldSpec().getName()); + } + + @Override + public TextIndexCreator newTextIndexCreator(IndexCreationContext context) + throws IOException { + // Initialize text index creator + Preconditions.checkState(context.getFieldSpec().getDataType().getStoredType() == FieldSpec.DataType.STRING, + "Text index is currently only supported on STRING type columns"); + return new LuceneTextIndexCreator(context.getFieldSpec().getName(), context.getIndexDir(), + true /* commitOnClose */); + } + + @Override + public TextIndexCreator newFSTIndexCreator(IndexCreationContext context) + throws IOException { + Preconditions.checkState(context.getFieldSpec().isSingleValueField(), + "FST index is currently only supported on single-value columns"); + Preconditions.checkState(context.getFieldSpec().getDataType().getStoredType() == FieldSpec.DataType.STRING, + "FST index is currently only supported on STRING type columns"); + Preconditions.checkState(!context.isRawColumn(), + "FST index is currently only supported on dictionary-encoded columns"); + String[] sortedValues = (String[]) context.getSortedUniqueElementsArray(); + if (context.getFstType() == FSTType.NATIVE) { + return new NativeFSTIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), sortedValues); + } else { + return new LuceneFSTIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), sortedValues); + } + } + + @Override + public GeoSpatialIndexCreator newGeoSpatialIndexCreator(IndexCreationContext context) + throws IOException { + Preconditions.checkState(context.getFieldSpec().isSingleValueField(), + "H3 index is currently only supported on single-value columns"); + Preconditions.checkState(context.getFieldSpec().getDataType().getStoredType() == FieldSpec.DataType.BYTES, + "H3 index is currently only supported on BYTES columns"); + H3IndexResolution resolution = Objects.requireNonNull(context.getH3IndexConfig()).getResolution(); + return context.isOnHeap() ? new OnHeapH3IndexCreator(context.getIndexDir(), + context.getFieldSpec().getName(), resolution) + : new OffHeapH3IndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), resolution); + } + + public static boolean shouldDeriveNumDocsPerChunk(String columnName, + Map<String, Map<String, String>> columnProperties) { + if (columnProperties != null) { + Map<String, String> properties = columnProperties.get(columnName); + return properties != null && Boolean.parseBoolean( + properties.get(FieldConfig.DERIVE_NUM_DOCS_PER_CHUNK_RAW_INDEX_KEY)); + } + return false; + } + + public static int rawIndexWriterVersion(String columnName, Map<String, Map<String, String>> columnProperties) { Review comment: Seems also missed in a branch ^^ ########## File path: pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java ########## @@ -0,0 +1,261 @@ +/** + * 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.spi.creator; + +import java.io.File; +import java.util.Map; +import java.util.Objects; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import org.apache.pinot.segment.spi.ColumnMetadata; +import org.apache.pinot.segment.spi.compression.ChunkCompressionType; +import org.apache.pinot.segment.spi.index.creator.H3IndexConfig; +import org.apache.pinot.spi.config.table.BloomFilterConfig; +import org.apache.pinot.spi.config.table.FSTType; +import org.apache.pinot.spi.data.FieldSpec; + + +/** + * Provides parameters for constructing indexes via {@see IndexCreatorProvider}. + * The responsibility for ensuring that the correct parameters for a particular + * index type lies with the caller. + */ +public class IndexCreationContext { + + public static final class Builder { + private File _indexDir; + private Map<String, Map<String, String>> _columnProperties; + private FSTType _fstType; + private H3IndexConfig _h3IndexConfig; + private int _lengthOfLongestEntry; + private int _maxNumberOfMultiValueElements; + private int _maxRowLengthInBytes; + private Object _sortedUniqueElementsArray; + private boolean _onHeap = false; + private ChunkCompressionType _chunkCompressionType; + public BloomFilterConfig _bloomFilterConfig; + private int _rangeIndexVersion; + private ColumnMetadata _columnMetadata; Review comment: I feel we should not include `ColumnMetadata` within the `IndexCreationContext`. Conceptually column metadata should be available only after the indexes are created. Suggest putting the required properties such as field spec, min/max values as individual fields in the `IndexCreationContext` ########## File path: pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java ########## @@ -179,126 +167,71 @@ public void init(SegmentGeneratorConfig segmentCreationSpec, SegmentIndexCreatio } String columnName = fieldSpec.getName(); - DataType storedType = fieldSpec.getDataType().getStoredType(); - ColumnIndexCreationInfo indexCreationInfo = indexCreationInfoMap.get(columnName); - Preconditions.checkNotNull(indexCreationInfo, "Missing index creation info for column: %s", columnName); - boolean dictEnabledColumn = createDictionaryForColumn(indexCreationInfo, segmentCreationSpec, fieldSpec); - + ColumnIndexCreationInfo columnIndexCreationInfo = indexCreationInfoMap.get(columnName); + Preconditions.checkNotNull(columnIndexCreationInfo, "Missing index creation info for column: %s", columnName); + boolean dictEnabledColumn = createDictionaryForColumn(columnIndexCreationInfo, segmentCreationSpec, fieldSpec); + Preconditions.checkState(dictEnabledColumn || !invertedIndexColumns.contains(columnName), + "Cannot create inverted index for raw index column: %s", columnName); + ColumnMetadata columnMetadata = ColumnMetadataImpl.builder() + .setCardinality(columnIndexCreationInfo.getDistinctValueCount()) + .setHasDictionary(dictEnabledColumn) + .setFieldSpec(fieldSpec) + .setTotalDocs(segmentIndexCreationInfo.getTotalDocs()) + .setTotalNumberOfEntries(columnIndexCreationInfo.getTotalNumberOfEntries()) + .setSorted(columnIndexCreationInfo.isSorted()) + .build(); + IndexCreationContext context = IndexCreationContext.builder() + .withIndexDir(_indexDir) + .withColumnMetadata(columnMetadata) + .withSegmentGeneratorConfig(segmentCreationSpec) + .withColumnIndexCreationInfo(columnIndexCreationInfo) + .withFSTType(_config.getFSTIndexType()) + .withH3IndexConfig(h3IndexConfigs.get(columnName)) + .withCompressionType(dictEnabledColumn ? null : getColumnCompressionType(segmentCreationSpec, fieldSpec)) + .build(); + // Initialize forward index creator + _forwardIndexCreatorMap.put(columnName, _indexCreatorProvider.newForwardIndexCreator(context)); + + // Initialize inverted index creator; skip creating inverted index if sorted + if (invertedIndexColumns.contains(columnName) && !columnIndexCreationInfo.isSorted()) { + _invertedIndexCreatorMap.put(columnName, _indexCreatorProvider.newInvertedIndexCreator(context)); + } if (dictEnabledColumn) { // Create dictionary-encoded index - // Initialize dictionary creator SegmentDictionaryCreator dictionaryCreator = - new SegmentDictionaryCreator(indexCreationInfo.getSortedUniqueElementsArray(), fieldSpec, _indexDir, - indexCreationInfo.isUseVarLengthDictionary()); + new SegmentDictionaryCreator(columnIndexCreationInfo.getSortedUniqueElementsArray(), fieldSpec, _indexDir, + columnIndexCreationInfo.isUseVarLengthDictionary()); _dictionaryCreatorMap.put(columnName, dictionaryCreator); - // Create dictionary try { dictionaryCreator.build(); } catch (Exception e) { LOGGER.error("Error building dictionary for field: {}, cardinality: {}, number of bytes per entry: {}", - fieldSpec.getName(), indexCreationInfo.getDistinctValueCount(), dictionaryCreator.getNumBytesPerEntry()); + fieldSpec.getName(), columnIndexCreationInfo.getDistinctValueCount(), + dictionaryCreator.getNumBytesPerEntry()); throw e; } - - // Initialize forward index creator - int cardinality = indexCreationInfo.getDistinctValueCount(); - if (fieldSpec.isSingleValueField()) { - if (indexCreationInfo.isSorted()) { - _forwardIndexCreatorMap.put(columnName, - new SingleValueSortedForwardIndexCreator(_indexDir, columnName, cardinality)); - } else { - _forwardIndexCreatorMap.put(columnName, - new SingleValueUnsortedForwardIndexCreator(_indexDir, columnName, cardinality, _totalDocs)); - } - } else { - _forwardIndexCreatorMap.put(columnName, - new MultiValueUnsortedForwardIndexCreator(_indexDir, columnName, cardinality, _totalDocs, - indexCreationInfo.getTotalNumberOfEntries())); - } - - // Initialize inverted index creator; skip creating inverted index if sorted - if (invertedIndexColumns.contains(columnName) && !indexCreationInfo.isSorted()) { - if (segmentCreationSpec.isOnHeap()) { - _invertedIndexCreatorMap.put(columnName, - new OnHeapBitmapInvertedIndexCreator(_indexDir, columnName, cardinality)); - } else { - _invertedIndexCreatorMap.put(columnName, - new OffHeapBitmapInvertedIndexCreator(_indexDir, fieldSpec, cardinality, _totalDocs, - indexCreationInfo.getTotalNumberOfEntries())); - } - } - } else { - // Create raw index - Preconditions.checkState(!invertedIndexColumns.contains(columnName), - "Cannot create inverted index for raw index column: %s", columnName); - - ChunkCompressionType compressionType = getColumnCompressionType(segmentCreationSpec, fieldSpec); - - // Initialize forward index creator - boolean deriveNumDocsPerChunk = - shouldDeriveNumDocsPerChunk(columnName, segmentCreationSpec.getColumnProperties()); - int writerVersion = rawIndexWriterVersion(columnName, segmentCreationSpec.getColumnProperties()); - if (fieldSpec.isSingleValueField()) { - _forwardIndexCreatorMap.put(columnName, - getRawIndexCreatorForSVColumn(_indexDir, compressionType, columnName, storedType, _totalDocs, - indexCreationInfo.getLengthOfLongestEntry(), deriveNumDocsPerChunk, writerVersion)); - } else { - _forwardIndexCreatorMap.put(columnName, - getRawIndexCreatorForMVColumn(_indexDir, compressionType, columnName, storedType, _totalDocs, - indexCreationInfo.getMaxNumberOfMultiValueElements(), deriveNumDocsPerChunk, writerVersion, - indexCreationInfo.getMaxRowLengthInBytes())); - } } if (textIndexColumns.contains(columnName)) { // Initialize text index creator - Preconditions.checkState(storedType == DataType.STRING, + Preconditions.checkState(fieldSpec.getDataType().getStoredType() == FieldSpec.DataType.STRING, "Text index is currently only supported on STRING type columns"); _textIndexCreatorMap.put(columnName, new LuceneTextIndexCreator(columnName, _indexDir, true /* commitOnClose */)); Review comment: Any specific reason why this one is not included in the index creator interface? -- 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]
