xiangfu0 commented on code in PR #19998: URL: https://github.com/apache/pinot/pull/19998#discussion_r4367388506
########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedSegmentMetadataStore.java: ########## @@ -0,0 +1,929 @@ +/** + * 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.index.metadata; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.common.collect.Interner; +import com.google.common.collect.Interners; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.Closeable; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.File; +import java.io.IOException; +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import java.nio.file.AtomicMoveNotSupportedException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.zip.CRC32C; +import javax.annotation.Nullable; +import org.apache.commons.configuration2.Configuration; +import org.apache.commons.configuration2.PropertiesConfiguration; +import org.apache.commons.configuration2.ex.ConfigurationException; +import org.apache.pinot.segment.spi.ColumnMetadata; +import org.apache.pinot.segment.spi.V1Constants; +import org.apache.pinot.segment.spi.creator.SegmentVersion; +import org.apache.pinot.segment.spi.index.IndexService; +import org.apache.pinot.segment.spi.index.IndexType; +import org.apache.pinot.segment.spi.memory.PinotDataBuffer; +import org.apache.pinot.segment.spi.partition.PartitionFunction; +import org.apache.pinot.segment.spi.store.ColumnIndexUtils; +import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths; +import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils; +import org.apache.pinot.spi.config.table.FieldConfig.EncodingType; +import org.apache.pinot.spi.data.ComplexFieldSpec; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.env.CommonsConfigurationUtils; +import org.apache.pinot.spi.utils.ByteArray; +import org.apache.pinot.spi.utils.ColumnNameInterner; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.ReadMode; + +/// Versioned, file-backed metadata for a sealed segment. The source `metadata.properties` remains the canonical +/// representation for old readers and preprocessing. +/// +/// The big-endian file starts with a 64-byte header containing the magic, format version, total length, source +/// `metadata.properties` and `index_map` length/CRC fingerprints, registered index IDs, payload CRC, +/// segment-property blob reference, +/// column count, and record-table bounds. Sorted columns occupy fixed 180-byte records, followed by variable blobs; +/// each blob reference is an absolute 32-bit offset and length. Numeric column statistics are read directly from the +/// backing buffer. The column layout retains names and reusable [FieldSpec] references on heap. +/// +/// A stale or corrupt sidecar is rebuilt from the canonical source and replaced locally, using atomic rename where +/// supported. Change `VERSION` whenever the encoding or the values derived from canonical metadata change; old +/// versions are rebuilt rather than migrated. +final class MappedSegmentMetadataStore implements Closeable { + static final String FILE_NAME = "segment_metadata.bin"; + private static final String TEMP_FILE_PREFIX = ".segment_metadata."; + private static final String TEMP_FILE_SUFFIX = ".tmp"; + + static boolean isSidecarFileName(String fileName) { + return FILE_NAME.equals(fileName) + || (fileName.startsWith(TEMP_FILE_PREFIX) && fileName.endsWith(TEMP_FILE_SUFFIX)); + } + + private static final int MAGIC = 0x504d4431; // PMD1 + private static final int VERSION = 5; + private static final int HEADER_SIZE = 64; + private static final int RECORD_SIZE = 180; + + static final int NAME = 0; + static final int FIELD_SPEC = 8; + static final int TOTAL_DOCS = 16; + static final int CARDINALITY = 20; + static final int FLAGS = 24; + static final int ENCODING = 28; + static final int SHORTEST = 32; + static final int LONGEST = 36; + static final int TOTAL_ENTRIES = 40; + static final int MAX_MV = 44; + static final int MAX_ROW = 48; + static final int BITS = 52; + static final int MIN_TAG = 56; + static final int MIN_RAW = 60; + static final int MIN_BLOB = 68; + static final int MAX_TAG = 76; + static final int MAX_RAW = 80; + static final int MAX_BLOB = 88; + static final int PARTITION_FUNCTION = 96; + static final int PARTITIONS = 104; + static final int PARENT = 112; + static final int SPARSE_KEYS = 120; + static final int SPARSE_MULTI_VALUE_KEYS = 128; + static final int SPARSE_KEY_TYPES = 136; + static final int RAW_UNCOMPRESSED_SIZE = 144; + static final int DICTIONARY_UNCOMPRESSED_SIZE = 152; + static final int COMPRESSION_TYPE = 160; + static final int INDEXES = 168; + static final int INDEX_COUNT = 176; + + static final int HAS_DICTIONARY = 1; + static final int SORTED = 1 << 1; + static final int NON_NULL = 1 << 2; + static final int INVALID_MIN_MAX = 1 << 3; + static final int ASCII = 1 << 4; + static final int AUTO_GENERATED = 1 << 5; + + private static final int NO_VALUE = 0; + private static final int INT_VALUE = 1; + private static final int LONG_VALUE = 2; + private static final int FLOAT_VALUE = 3; + private static final int DOUBLE_VALUE = 4; + private static final int DECIMAL_VALUE = 5; + private static final int STRING_VALUE = 6; + private static final int BYTES_VALUE = 7; + + private static final Interner<FieldSpec> FIELD_SPECS = Interners.newWeakInterner(); + private static final Interner<ColumnLayout> COLUMN_LAYOUTS = Interners.newWeakInterner(); + + private final PinotDataBuffer _buffer; + private final ColumnLayout _columns; + private final int _tableOffset; + private final int _segmentConfigOffset; + private final int _segmentConfigLength; + + private MappedSegmentMetadataStore(PinotDataBuffer buffer, ColumnLayout columns, int tableOffset, + int segmentConfigOffset, int segmentConfigLength) { + _buffer = buffer; + _columns = columns; + _tableOffset = tableOffset; + _segmentConfigOffset = segmentConfigOffset; + _segmentConfigLength = segmentConfigLength; + } + + static MappedSegmentMetadataStore open(File indexDir, ReadMode readMode) + throws IOException, ConfigurationException { + File source = SegmentDirectoryPaths.findMetadataFile(indexDir); + if (source == null) { + throw new IOException("Missing metadata.properties in " + indexDir); + } + File sidecar = new File(indexDir, FILE_NAME); + boolean rebuilt = false; + for (int attempt = 0; attempt < 3; attempt++) { + IndexService indexService = IndexService.getInstance(); + Fingerprint propertiesFingerprint = Fingerprint.of(source); + Fingerprint indexMapFingerprint = indexMapFingerprint(indexDir, indexService); + // The validated heap copy is transient; the store retains only the buffer and shared column layout. + byte[] bytes = readValid(sidecar, propertiesFingerprint, indexMapFingerprint); + if (bytes == null) { + writeAtomically(sidecar, build(indexDir, propertiesFingerprint, indexMapFingerprint, indexService)); + rebuilt = true; + bytes = readValid(sidecar, propertiesFingerprint, indexMapFingerprint); + if (bytes == null) { + continue; + } + } + PinotDataBuffer buffer = readMode == ReadMode.heap + ? PinotDataBuffer.loadFile(sidecar, 0, bytes.length, ByteOrder.BIG_ENDIAN, "segment metadata") + : PinotDataBuffer.mapFile(sidecar, true, 0, bytes.length, ByteOrder.BIG_ENDIAN, "segment metadata"); + boolean keepBuffer = false; + try { + // Another loader may atomically replace the sidecar between validation and opening it. Check the exact + // buffer bytes and canonical inputs before publishing a view. The header alone cannot prove the bytes match. + if (!matchesBuffer(buffer, bytes) || !propertiesFingerprint.sameAs(Fingerprint.of(source)) + || indexService != IndexService.getInstance() + || !indexMapFingerprint.sameAs(indexMapFingerprint(indexDir, indexService))) { + continue; + } + ByteBuffer header = ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN); + int count = header.getInt(52); + int tableOffset = header.getInt(56); + String[] names = new String[count]; + FieldSpec[] fieldSpecs = new FieldSpec[count]; + for (int i = 0; i < count; i++) { + int record = tableOffset + i * RECORD_SIZE; + names[i] = ColumnNameInterner.intern(readString(bytes, header.getInt(record + NAME), + header.getInt(record + NAME + 4))); + FieldSpec parsed = JsonUtils.stringToObject(readString(bytes, header.getInt(record + FIELD_SPEC), + header.getInt(record + FIELD_SPEC + 4)), FieldSpec.class); + fieldSpecs[i] = parsed instanceof ComplexFieldSpec ? parsed : FIELD_SPECS.intern(parsed); + } + ColumnLayout columns = COLUMN_LAYOUTS.intern(new ColumnLayout(names, fieldSpecs)); + MappedSegmentMetadataStore store = new MappedSegmentMetadataStore(buffer, columns, tableOffset, + header.getInt(44), header.getInt(48)); + keepBuffer = true; + return store; + } catch (IOException | RuntimeException e) { + if (!rebuilt) { + // A checksum-valid older sidecar can still contain FieldSpec JSON this reader cannot parse. Rebuild it + // once from canonical metadata instead of falling back to eager on every load. + writeAtomically(sidecar, build(indexDir, propertiesFingerprint, indexMapFingerprint, indexService)); + rebuilt = true; + continue; + } + throw new IOException("Invalid field specs in mapped segment metadata: " + sidecar, e); + } finally { + if (!keepBuffer) { + buffer.close(); + } + } + } + throw new IOException("Metadata changed while opening segment: " + indexDir); + } + + private static boolean matchesBuffer(PinotDataBuffer buffer, byte[] bytes) { + if (buffer.size() != bytes.length) { + return false; + } + byte[] block = new byte[Math.min(8192, bytes.length)]; Review Comment: Agreed. readValid() checks the transient bytes for integrity and matchesBuffer() scans the mapped bytes to guard an atomic replacement between validation and mapping. The transient copy is released after open; the goal here is lower retained heap for long-lived column metadata, while preserving that open-time correctness check. ########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedColumnMetadata.java: ########## @@ -0,0 +1,398 @@ +/** + * 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.index.metadata; + +import com.fasterxml.jackson.core.type.TypeReference; +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +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.IndexService; +import org.apache.pinot.segment.spi.index.IndexType; +import org.apache.pinot.segment.spi.partition.PartitionFunction; +import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory; +import org.apache.pinot.spi.config.table.FieldConfig.EncodingType; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.utils.JsonUtils; + +/// A small view of one fixed record in [MappedSegmentMetadataStore]. No per-column values are copied into this view. +/// Data-source metadata delegates to this live view. Serving callers must hold the owning segment's acquired +/// reference for the full read; the segment manager's reference count delays destruction and unmapping until readers +/// release it. Standalone callers must keep the owning mapped metadata open. Reads after close are unsafe. +public final class MappedColumnMetadata implements ColumnMetadata { + private final MappedSegmentMetadataStore _store; + private final int _ordinal; + @Nullable + private volatile PartitionFunction _partitionFunction; + private volatile boolean _partitionFunctionLoaded; + @Nullable + private volatile Set<Integer> _partitions; + private volatile boolean _partitionsLoaded; + @Nullable + private volatile Bounds _bounds; + @Nullable + private volatile IndexSize[] _indexes; + + private static final class Bounds { + @Nullable + private final Comparable<?> _minimum; + @Nullable + private final Comparable<?> _maximum; + + private Bounds(@Nullable Comparable<?> minimum, @Nullable Comparable<?> maximum) { + _minimum = minimum; + _maximum = maximum; + } + } + + MappedColumnMetadata(MappedSegmentMetadataStore store, int ordinal) { + _store = store; + _ordinal = ordinal; + } + + @Override + public FieldSpec getFieldSpec() { + // Field specs are shared across mapped segments; callers must treat the returned spec as read-only. + return _store.fieldSpec(_ordinal); + } + + @Override + public int getTotalDocs() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.TOTAL_DOCS); + } + + @Override + public int getCardinality() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.CARDINALITY); + } + + @Override + public boolean hasDictionary() { + return _store.flag(_ordinal, MappedSegmentMetadataStore.HAS_DICTIONARY); + } + + @Nullable + @Override + public EncodingType getForwardIndexEncoding() { + switch (_store.getInt(_ordinal, MappedSegmentMetadataStore.ENCODING)) { + case 0: + return null; + case 1: + return EncodingType.RAW; + case 2: + return EncodingType.DICTIONARY; + default: + throw new IllegalStateException("Invalid forward-index encoding"); + } + } + + @Override + public boolean isSorted() { + return _store.flag(_ordinal, MappedSegmentMetadataStore.SORTED); + } + + @Override + public boolean isNonNull() { + return _store.flag(_ordinal, MappedSegmentMetadataStore.NON_NULL); + } + + @Nullable + @Override + public Comparable<?> getMinValue() { + return bounds()._minimum; + } + + @Nullable + @Override + public Comparable<?> getMaxValue() { + return bounds()._maximum; + } + + private Bounds bounds() { + Bounds bounds = _bounds; + if (bounds == null) { + synchronized (this) { + bounds = _bounds; + if (bounds == null) { + bounds = new Bounds(_store.value(_ordinal, true), _store.value(_ordinal, false)); + _bounds = bounds; + } + } + } + return bounds; + } + + @Override + public boolean isMinMaxValueInvalid() { + return _store.flag(_ordinal, MappedSegmentMetadataStore.INVALID_MIN_MAX); + } + + @Override + public int getLengthOfShortestElement() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.SHORTEST); + } + + @Override + public int getLengthOfLongestElement() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.LONGEST); + } + + @Override + public boolean isAscii() { + return _store.flag(_ordinal, MappedSegmentMetadataStore.ASCII); + } + + @Override + public int getBitsPerElement() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.BITS); + } + + @Override + public int getTotalNumberOfEntries() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.TOTAL_ENTRIES); + } + + @Override + public int getMaxNumberOfMultiValues() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.MAX_MV); + } + + @Override + public int getMaxRowLengthInBytes() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.MAX_ROW); + } + + @Nullable + @Override + public PartitionFunction getPartitionFunction() { + if (!_partitionFunctionLoaded) { + synchronized (this) { + if (!_partitionFunctionLoaded) { + _partitionFunction = decodePartitionFunction(); + _partitionFunctionLoaded = true; + } + } + } + return _partitionFunction; + } + + @Nullable + private PartitionFunction decodePartitionFunction() { + byte[] bytes = _store.blob(_ordinal, MappedSegmentMetadataStore.PARTITION_FUNCTION); + if (bytes == null) { + return null; + } + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes))) { + String name = MappedSegmentMetadataStore.readString(input); + int numPartitions = input.readInt(); + int configSize = input.readInt(); + Map<String, String> config = configSize != 0 ? new HashMap<>(configSize) : null; + for (int i = 0; i < configSize; i++) { + config.put(MappedSegmentMetadataStore.readString(input), MappedSegmentMetadataStore.readString(input)); + } + return PartitionFunctionFactory.getPartitionFunction(name, numPartitions, config); + } catch (IOException e) { + throw new IllegalStateException("Cannot decode partition function for " + getColumnName(), e); + } + } + + @Nullable + @Override + public Set<Integer> getPartitions() { + if (!_partitionsLoaded) { + synchronized (this) { + if (!_partitionsLoaded) { + _partitions = decodePartitions(); + _partitionsLoaded = true; + } + } + } + return _partitions; + } + + @Nullable + private Set<Integer> decodePartitions() { + byte[] bytes = _store.blob(_ordinal, MappedSegmentMetadataStore.PARTITIONS); + if (bytes == null) { + return null; + } + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes))) { + int count = input.readInt(); + Set<Integer> partitions = new HashSet<>(count); + for (int i = 0; i < count; i++) { + partitions.add(input.readInt()); + } + return partitions; + } catch (IOException e) { + throw new IllegalStateException("Cannot decode partitions for " + getColumnName(), e); + } + } + + @Override + public boolean isAutoGenerated() { + return _store.flag(_ordinal, MappedSegmentMetadataStore.AUTO_GENERATED); + } + + @Override + public boolean isMaterializedChild() { + return getParentColumn() != null; + } + + @Nullable + @Override + public String getParentColumn() { + return _store.string(_ordinal, MappedSegmentMetadataStore.PARENT); + } + + @Nullable + @Override + public List<String> getSparseKeys() { + return readJson(MappedSegmentMetadataStore.SPARSE_KEYS, new TypeReference<List<String>>() { }); + } + + @Nullable + @Override + public Map<String, Integer> getSparseMultiValueKeys() { + return readJson(MappedSegmentMetadataStore.SPARSE_MULTI_VALUE_KEYS, + new TypeReference<Map<String, Integer>>() { }); + } + + @Nullable + @Override + public Map<String, DataType> getSparseKeyTypes() { + return readJson(MappedSegmentMetadataStore.SPARSE_KEY_TYPES, + new TypeReference<Map<String, DataType>>() { }); + } + + @Nullable + private <T> T readJson(int field, TypeReference<T> type) { + String json = _store.string(_ordinal, field); + if (json == null) { + return null; + } + try { + return JsonUtils.stringToObject(json, type); + } catch (IOException e) { + throw new IllegalStateException("Cannot decode OPEN_STRUCT metadata for " + getColumnName(), e); + } + } + + @Override + public int getNumIndexes() { + return _store.getInt(_ordinal, MappedSegmentMetadataStore.INDEX_COUNT); + } + + @Override + public short getIndexType(int position) { + return IndexService.getInstance().getNumericId(IndexService.getInstance().get(indexAt(position)._type)); + } + + @Override + public long getIndexSize(int position) { + return indexAt(position)._size; + } + + @Override + public long getIndexSizeFor(IndexType type) { + int count = getNumIndexes(); + for (int i = 0; i < count; i++) { + IndexSize entry = indexAt(i); + if (entry._type.equals(type.getId())) { + return entry._size; + } + } + return UNAVAILABLE; + } + + /// Kept for the segment-metadata REST JSON representation. + @SuppressWarnings("unused") + public Map<IndexType<?, ?, ?>, Long> getIndexSizeMap() { + Map<IndexType<?, ?, ?>, Long> sizes = new HashMap<>(); + int count = getNumIndexes(); + for (int i = 0; i < count; i++) { + IndexSize entry = indexAt(i); + sizes.put(IndexService.getInstance().get(entry._type), entry._size); + } + return sizes; + } + + private IndexSize indexAt(int position) { + IndexSize[] indexes = _indexes; + int count = indexes != null ? indexes.length : getNumIndexes(); + if (position < 0 || position >= count) { + throw new IndexOutOfBoundsException(position); + } + if (indexes == null) { + synchronized (this) { + indexes = _indexes; + if (indexes == null) { + byte[] bytes = _store.blob(_ordinal, MappedSegmentMetadataStore.INDEXES); + if (bytes == null) { + throw new IllegalStateException("Missing index sizes for " + getColumnName()); + } + indexes = new IndexSize[count]; + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes))) { + for (int i = 0; i < count; i++) { + indexes[i] = new IndexSize(MappedSegmentMetadataStore.readString(input), input.readLong()); + } + } catch (IOException e) { + throw new IllegalStateException("Cannot decode index sizes for " + getColumnName(), e); + } + _indexes = indexes; + } + } + } + return indexes[position]; + } + + private static final class IndexSize { + private final String _type; + private final long _size; + + private IndexSize(String type, long size) { + _type = type; + _size = size; + } + } + + @Override + public long getRawForwardIndexUncompressedValueSizeInBytes() { + return _store.getLong(_ordinal, MappedSegmentMetadataStore.RAW_UNCOMPRESSED_SIZE); + } + + @Nullable + @Override + public ChunkCompressionType getRawForwardIndexChunkCompressionType() { + String name = _store.string(_ordinal, MappedSegmentMetadataStore.COMPRESSION_TYPE); + return name != null ? ChunkCompressionType.valueOf(name) : null; Review Comment: The sidecar stores the enum name produced from canonical metadata.properties. ColumnMetadataImpl.parseCompressionType() also rejects an unknown name, so returning null only from the mapped getter would diverge from eager metadata. An enum rename or removal needs a canonical compatibility mapping and a sidecar VERSION bump, which rebuilds local sidecars. I am keeping the current strict behavior for this format. -- 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]
