raghavyadav01 commented on code in PR #18643: URL: https://github.com/apache/pinot/pull/18643#discussion_r3574843837
########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java: ########## @@ -0,0 +1,538 @@ +/** + * 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.openstruct; + +import com.google.common.base.Utf8; +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import javax.annotation.Nullable; +import org.apache.commons.configuration2.PropertiesConfiguration; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; +import org.apache.pinot.segment.local.segment.creator.impl.BaseSegmentCreator; +import org.apache.pinot.segment.local.segment.creator.impl.SegmentDictionaryCreator; +import org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueVarByteRawIndexCreator; +import org.apache.pinot.segment.local.segment.creator.impl.nullvalue.NullValueVectorCreator; +import org.apache.pinot.segment.local.segment.creator.impl.stats.AbstractColumnStatisticsCollector; +import org.apache.pinot.segment.local.segment.creator.impl.stats.StatsCollectorUtil; +import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType; +import org.apache.pinot.segment.local.segment.index.openstruct.OpenStructSupportedIndexes; +import org.apache.pinot.segment.spi.V1Constants; +import org.apache.pinot.segment.spi.compression.ChunkCompressionType; +import org.apache.pinot.segment.spi.creator.IndexCreationContext; +import org.apache.pinot.segment.spi.index.DictionaryIndexConfig; +import org.apache.pinot.segment.spi.index.FieldIndexConfigs; +import org.apache.pinot.segment.spi.index.FieldIndexConfigsUtil; +import org.apache.pinot.segment.spi.index.ForwardIndexConfig; +import org.apache.pinot.segment.spi.index.IndexCreator; +import org.apache.pinot.segment.spi.index.IndexService; +import org.apache.pinot.segment.spi.index.IndexType; +import org.apache.pinot.segment.spi.index.StandardIndexes; +import org.apache.pinot.segment.spi.index.creator.ColumnarOpenStructIndexCreator; +import org.apache.pinot.spi.config.table.FieldConfig; +import org.apache.pinot.spi.config.table.IndexConfig; +import org.apache.pinot.spi.config.table.IndexingConfig; +import org.apache.pinot.spi.config.table.OpenStructIndexConfig; +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.OpenStructTypeInference; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.PinotDataType; +import org.roaringbitmap.RoaringBitmap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Splits an OPEN_STRUCT column into per-key materialized columns using standard Pinot index +/// creators. Dense keys become independent virtual columns; remaining keys go into a single +/// synthetic JSON column for sparse storage. +/// +/// Lifecycle: instantiated by `BaseSegmentCreator` for OPEN_STRUCT columns. Receives +/// per-doc `Map<String, Object>` values via [#add(Map, int)], accumulates in memory, +/// then on [#seal()] writes per-key column files using standard creators. +public class OpenStructColumnSplitter implements ColumnarOpenStructIndexCreator { + + private static final Logger LOGGER = LoggerFactory.getLogger(OpenStructColumnSplitter.class); + + private final File _indexDir; + private final String _columnName; + private final Map<String, FieldSpec> _childFieldSpecs; + private final OpenStructIndexConfig _config; + private final int _maxDenseKeys; + + // Per-key accumulation + private final Map<String, RoaringBitmap> _presenceBitmaps = new HashMap<>(); + private final Map<String, List<Object>> _values = new HashMap<>(); + private final Map<String, DataType> _inferredTypes = new HashMap<>(); + private int _numDocs; + + // Resolved at seal time + @Nullable + private Set<String> _resolvedDenseKeys; + private final Map<String, PropertiesConfiguration> _materializedColumnMetadata = new LinkedHashMap<>(); + + public OpenStructColumnSplitter(File indexDir, String columnName, FieldSpec fieldSpec, + OpenStructIndexConfig config) { + _indexDir = indexDir; + _columnName = columnName; + _config = config; + _maxDenseKeys = config.getMaxDenseKeys(); + + Map<String, FieldSpec> childFieldSpecs = null; + if (fieldSpec instanceof ComplexFieldSpec) { + ComplexFieldSpec complexSpec = (ComplexFieldSpec) fieldSpec; + childFieldSpecs = complexSpec.getChildFieldSpecs(); + } + _childFieldSpecs = childFieldSpecs != null ? new HashMap<>(childFieldSpecs) : new HashMap<>(); + } + + @Override + public void add(Object value, int docId) + throws IOException { + if (value instanceof Map) { + @SuppressWarnings("unchecked") + Map<String, Object> map = (Map<String, Object>) value; + addMap(map); + } else { + addMap(null); + } + } + + @Override + public void add(Map<String, Object> openStructValue, int docId) + throws IOException { + addMap(openStructValue); + } + + @Override + public void add(Object[] values, @Nullable int[] dictIds) + throws IOException { + throw new UnsupportedOperationException("OPEN_STRUCT index is single-value only"); + } + + /// Returns the resolved dense-key set after [#seal()] or [#classify()]. + /// Returns an empty set before resolution. + public Set<String> getResolvedDenseKeys() { + return _resolvedDenseKeys != null ? Collections.unmodifiableSet(_resolvedDenseKeys) : Set.of(); + } + + /// Resolves dense vs sparse keys without writing any files. Exposed for testing and for callers + /// that need the classification independent of file output. [#seal()] calls this internally. + public Set<String> classify() { + if (_resolvedDenseKeys != null) { + return _resolvedDenseKeys; + } + if (_numDocs == 0 || _presenceBitmaps.isEmpty()) { + _resolvedDenseKeys = new LinkedHashSet<>(); + return _resolvedDenseKeys; + } + List<String> allKeys = new ArrayList<>(_presenceBitmaps.keySet()); + allKeys.sort((a, b) -> { + double fillA = (double) _presenceBitmaps.get(a).getCardinality() / _numDocs; + double fillB = (double) _presenceBitmaps.get(b).getCardinality() / _numDocs; + int cmp = Double.compare(fillB, fillA); + return cmp != 0 ? cmp : a.compareTo(b); + }); + + double minFillRate = _config.getDenseKeyMinFillRate(); + _resolvedDenseKeys = new LinkedHashSet<>(); + + Set<String> configuredDenseKeys = _config.getDenseKeys(); + for (String key : configuredDenseKeys) { + if (_presenceBitmaps.containsKey(key) && (_maxDenseKeys < 0 || _resolvedDenseKeys.size() < _maxDenseKeys)) { + _resolvedDenseKeys.add(key); + } + } + + for (String key : allKeys) { + if (_resolvedDenseKeys.contains(key)) { + continue; + } + double fillRate = (double) _presenceBitmaps.get(key).getCardinality() / _numDocs; Review Comment: A key's type is locked to the first value seen (`computeIfAbsent`/first-row inference) here and in `MutableOpenStructIndex.index`. A later value that fails coercion to that type is dropped — `bitmap.remove()` + `continue`, only a `LOGGER.warn`. For a *dense* key this is quiet correctness loss on heterogeneous JSON (`{"x": 5}` then `{"x": "hello"}`). Is that intended for dense keys too, and should we surface it via a metric/counter rather than just a log line? -- 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]
