Jackie-Jiang commented on code in PR #19998: URL: https://github.com/apache/pinot/pull/19998#discussion_r4394692785
########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedSegmentMetadataStore.java: ########## @@ -0,0 +1,943 @@ +/** + * 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 boolean _closed; + + 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)]; + for (int offset = 0; offset < bytes.length; offset += block.length) { + int length = Math.min(block.length, bytes.length - offset); + buffer.copyTo(offset, block, 0, length); + for (int i = 0; i < length; i++) { + if (block[i] != bytes[offset + i]) { + return false; + } + } + } + return true; + } + + int size() { + return _columns._names.length; + } + + String name(int ordinal) { + return _columns._names[ordinal]; + } + + int ordinal(String column) { + return Arrays.binarySearch(_columns._names, column); + } + + synchronized int getInt(int ordinal, int offset) { + ensureOpen(); + return _buffer.getInt((long) _tableOffset + (long) ordinal * RECORD_SIZE + offset); + } + + synchronized long getLong(int ordinal, int offset) { + ensureOpen(); + return _buffer.getLong((long) _tableOffset + (long) ordinal * RECORD_SIZE + offset); + } + + boolean flag(int ordinal, int flag) { + return (getInt(ordinal, FLAGS) & flag) != 0; + } + + @Nullable + String string(int ordinal, int field) { + byte[] bytes = blob(ordinal, field); + return bytes != null ? new String(bytes, StandardCharsets.UTF_8) : null; + } + + @Nullable + synchronized byte[] blob(int ordinal, int field) { + int offset = getInt(ordinal, field); + if (offset < 0) { + return null; + } + int length = getInt(ordinal, field + 4); + byte[] bytes = new byte[length]; + _buffer.copyTo(offset, bytes); + return bytes; + } + + FieldSpec fieldSpec(int ordinal) { + return _columns._fieldSpecs[ordinal]; + } + + @Nullable + Comparable<?> value(int ordinal, boolean minimum) { + int tag = getInt(ordinal, minimum ? MIN_TAG : MAX_TAG); + long raw = getLong(ordinal, minimum ? MIN_RAW : MAX_RAW); + int blobField = minimum ? MIN_BLOB : MAX_BLOB; + switch (tag) { + case NO_VALUE: + return null; + case INT_VALUE: + return (int) raw; + case LONG_VALUE: + return raw; + case FLOAT_VALUE: + return Float.intBitsToFloat((int) raw); + case DOUBLE_VALUE: + return Double.longBitsToDouble(raw); + case DECIMAL_VALUE: + return new BigDecimal(string(ordinal, blobField)); + case STRING_VALUE: + return string(ordinal, blobField); + case BYTES_VALUE: + return new ByteArray(blob(ordinal, blobField)); + default: + throw new IllegalStateException("Invalid value tag: " + tag); + } + } + + synchronized PropertiesConfiguration segmentConfiguration() + throws IOException, ConfigurationException { + ensureOpen(); + byte[] bytes = new byte[_segmentConfigLength]; + _buffer.copyTo(_segmentConfigOffset, bytes); + // Values were already split by the source parser. A second delimiter pass would change escaped commas. + PropertiesConfiguration config = CommonsConfigurationUtils.fromInputStream(null, false, null); + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes))) { + int count = input.readInt(); + for (int i = 0; i < count; i++) { + String key = readString(input); + int valueCount = input.readInt(); + if (valueCount == 1) { + config.setProperty(key, readString(input)); + } else { + List<String> values = new ArrayList<>(valueCount); + for (int j = 0; j < valueCount; j++) { + values.add(readString(input)); + } + config.setProperty(key, values); + } + } + } + return config; + } + + @Override + public synchronized void close() + throws IOException { + if (_closed) { + return; + } + _closed = true; + _buffer.close(); + } + + private void ensureOpen() { + if (_closed) { + throw new IllegalStateException("Mapped segment metadata is closed"); + } + } + + private static byte[] build(File indexDir, Fingerprint propertiesFingerprint, Fingerprint indexMapFingerprint, + IndexService indexService) + throws IOException, ConfigurationException { + SegmentMetadataImpl source = new SegmentMetadataImpl(indexDir); + PropertiesConfiguration config = SegmentMetadataUtils.getPropertiesConfiguration(indexDir); + List<String> columns = new ArrayList<>(source.getColumnMetadataMap().keySet()); + columns.sort(Comparator.naturalOrder()); + if ((long) columns.size() * RECORD_SIZE + HEADER_SIZE > Integer.MAX_VALUE) { + throw new IOException("Too many columns for mapped metadata: " + columns.size()); + } + int blobBase = HEADER_SIZE + columns.size() * RECORD_SIZE; + BlobWriter blobs = new BlobWriter(blobBase); + BlobRef segmentConfig = blobs.add(encodeSegmentConfiguration(config)); + // EmptyColumnMetadata does not load index sizes even if a stale index_map is still present. + Map<String, List<IndexSize>> indexSizes = source.getTotalDocs() > 0 + && source.getVersion() == SegmentVersion.v3 ? readIndexSizes(indexDir, indexService) : Map.of(); + byte[] records = new byte[columns.size() * RECORD_SIZE]; + ByteBuffer recordBuffer = ByteBuffer.wrap(records).order(ByteOrder.BIG_ENDIAN); + for (int ordinal = 0; ordinal < columns.size(); ordinal++) { + String column = columns.get(ordinal); + ColumnMetadata metadata = source.getColumnMetadataFor(column); + int offset = ordinal * RECORD_SIZE; + putRef(recordBuffer, offset + NAME, blobs.add(column)); + putRef(recordBuffer, offset + FIELD_SPEC, blobs.add(fieldSpecJson(metadata.getFieldSpec()).toString())); + recordBuffer.putInt(offset + TOTAL_DOCS, metadata.getTotalDocs()); + recordBuffer.putInt(offset + CARDINALITY, metadata.getCardinality()); + int flags = (metadata.hasDictionary() ? HAS_DICTIONARY : 0) | (metadata.isSorted() ? SORTED : 0) + | (metadata.isNonNull() ? NON_NULL : 0) | (metadata.isMinMaxValueInvalid() ? INVALID_MIN_MAX : 0) + | (metadata.isAscii() ? ASCII : 0) | (metadata.isAutoGenerated() ? AUTO_GENERATED : 0); + recordBuffer.putInt(offset + FLAGS, flags); + EncodingType encoding = metadata.getForwardIndexEncoding(); + recordBuffer.putInt(offset + ENCODING, encoding == null ? 0 : encoding == EncodingType.RAW ? 1 : 2); + recordBuffer.putInt(offset + SHORTEST, metadata.getLengthOfShortestElement()); + recordBuffer.putInt(offset + LONGEST, metadata.getLengthOfLongestElement()); + recordBuffer.putInt(offset + TOTAL_ENTRIES, metadata.getTotalNumberOfEntries()); + recordBuffer.putInt(offset + MAX_MV, metadata.getMaxNumberOfMultiValues()); + recordBuffer.putInt(offset + MAX_ROW, metadata.getMaxRowLengthInBytes()); + recordBuffer.putInt(offset + BITS, metadata.getBitsPerElement()); + writeValue(recordBuffer, offset + MIN_TAG, offset + MIN_RAW, offset + MIN_BLOB, metadata.getMinValue(), blobs); + writeValue(recordBuffer, offset + MAX_TAG, offset + MAX_RAW, offset + MAX_BLOB, metadata.getMaxValue(), blobs); + PartitionFunction partitionFunction = metadata.getPartitionFunction(); + putRef(recordBuffer, offset + PARTITION_FUNCTION, + partitionFunction != null ? blobs.add(encodePartitionFunction(column, config)) : BlobRef.ABSENT); + putRef(recordBuffer, offset + PARTITIONS, + metadata.getPartitions() != null ? blobs.add(encodePartitions(metadata.getPartitions())) : BlobRef.ABSENT); + if (metadata instanceof ColumnMetadataImpl) { + ColumnMetadataImpl impl = (ColumnMetadataImpl) metadata; + putRef(recordBuffer, offset + PARENT, blobs.add(impl.getParentColumn())); + putRef(recordBuffer, offset + SPARSE_KEYS, blobs.addJson(impl.getSparseKeys())); + putRef(recordBuffer, offset + SPARSE_MULTI_VALUE_KEYS, blobs.addJson(impl.getSparseMultiValueKeys())); + putRef(recordBuffer, offset + SPARSE_KEY_TYPES, blobs.addJson(impl.getSparseKeyTypes())); + } else { + putRef(recordBuffer, offset + PARENT, BlobRef.ABSENT); + putRef(recordBuffer, offset + SPARSE_KEYS, BlobRef.ABSENT); + putRef(recordBuffer, offset + SPARSE_MULTI_VALUE_KEYS, BlobRef.ABSENT); + putRef(recordBuffer, offset + SPARSE_KEY_TYPES, BlobRef.ABSENT); + } + recordBuffer.putLong(offset + RAW_UNCOMPRESSED_SIZE, + metadata.getRawForwardIndexUncompressedValueSizeInBytes()); + recordBuffer.putLong(offset + DICTIONARY_UNCOMPRESSED_SIZE, + metadata.getDictionaryEncodedUncompressedValueSizeInBytes()); + putRef(recordBuffer, offset + COMPRESSION_TYPE, + blobs.add(metadata.getRawForwardIndexChunkCompressionType() != null + ? metadata.getRawForwardIndexChunkCompressionType().name() : null)); + List<IndexSize> sizes = indexSizes.getOrDefault(column, List.of()); + recordBuffer.putInt(offset + INDEX_COUNT, sizes.size()); + putRef(recordBuffer, offset + INDEXES, sizes.isEmpty() ? BlobRef.ABSENT : blobs.add(encodeIndexSizes(sizes))); + } + byte[] blobBytes = blobs.toByteArray(); + long totalSize = (long) HEADER_SIZE + records.length + blobBytes.length; + if (totalSize > Integer.MAX_VALUE) { + throw new IOException("Mapped metadata exceeds 2 GiB: " + totalSize); + } + byte[] bytes = new byte[(int) totalSize]; + ByteBuffer header = ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN); + header.putInt(0, MAGIC); + header.putInt(4, VERSION); + header.putLong(8, totalSize); + header.putLong(16, propertiesFingerprint._length); + header.putInt(24, propertiesFingerprint._crc); + header.putLong(28, indexMapFingerprint._length); + header.putInt(36, indexMapFingerprint._crc); + header.putInt(44, segmentConfig._offset); + header.putInt(48, segmentConfig._length); + header.putInt(52, columns.size()); + header.putInt(56, HEADER_SIZE); + header.putInt(60, blobBase); + System.arraycopy(records, 0, bytes, HEADER_SIZE, records.length); + System.arraycopy(blobBytes, 0, bytes, blobBase, blobBytes.length); + header.putInt(40, sidecarCrc(bytes)); + return bytes; + } + + private static ObjectNode fieldSpecJson(FieldSpec fieldSpec) { + ObjectNode json = fieldSpec.toJsonObject(); + JsonNode defaultNullValue = json.get("defaultNullValue"); + if (defaultNullValue != null && defaultNullValue.isNumber()) { Review Comment: [P1] Preserve TIMESTAMP null defaults as epoch milliseconds The numeric-only normalization misses TIMESTAMP defaults: FieldSpec.toJsonObject() emits them via Timestamp.toString(), losing the timezone/offset. This changes the value even when the writer and reader use the same timezone. An isolated JDK 25 reproduction in America/Los_Angeles gives: ``` original = 1730622600000 encoded = 2024-11-03 01:30:00.0 decoded = 1730626200000 // +1 hour at the DST overlap ``` The canonical properties retain the epoch value. With column null handling and an exact range index, a predicate whose upper bound lies between these timestamps includes the stored null sentinel, but RangeIndexBasedFilterOperator.matchesDefaultNullValue() checks the shifted sentinel and fails to subtract the null documents. Reopening the same sidecar under another JVM timezone also changes nonzero timestamp defaults without invalidating its fingerprints. Please encode TIMESTAMP defaults as epoch-millisecond strings, including nested FieldSpecs, and add the DST-overlap and changed-timezone round-trip cases. If existing v5 caches have already been deployed, invalidate them through a version change; otherwise include this fix in the initial format before merge. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java: ########## @@ -107,13 +109,17 @@ public abstract class BasePartitionUpsertMetadataManager implements PartitionUps // Basically, it's possible that some segments in the table partition are not tracked here, as their upsert metadata // is not managed by the manager currently. protected final Set<IndexSegment> _trackedSegments = ConcurrentHashMap.newKeySet(); - // Track all the immutable segments where changes took place since last snapshot was taken. - // Note: the set can be updated by multiple Helix task threads and the consuming thread concurrently. To not miss - // segments updated while a snapshot is being taken, the snapshot flow removes the exact segments it has persisted - // instead of clearing the set, and it retains only the tracked segments at the end of each snapshot round to avoid - // keeping stale segment object references around. + // Track all the immutable segments where changes took place since last snapshot was taken, along with their maximum + // comparison values. The value must remain readable if a segment is destroyed while the snapshot pass is capping + // the TTL watermark. Helix task threads and the consuming thread can update the map concurrently, so the snapshot + // flow removes only the exact segments it persisted and retains tracked segments at the end of each round. // Skip mutableSegments as only immutable segments are for taking snapshots. - protected final Set<ImmutableSegmentImpl> _updatedSegmentsSinceLastSnapshot = ConcurrentHashMap.newKeySet(); + protected final ConcurrentMap<ImmutableSegmentImpl, Double> _updatedSegmentsSinceLastSnapshot = Review Comment: Compatibility: preserve the protected snapshot-set field's binary type Changing this protected field from Set to ConcurrentMap changes its JVM field descriptor. A subclass compiled against the existing Set field will get NoSuchFieldError with this version, even for unchanged isEmpty()/size() calls. StarTree's RocksDBPartitionUpsertMetadataManager is a concrete consumer of both operations, including during shutdown snapshot persistence. This change applies even when mapped metadata is disabled. Current StarTree source still compiles when rebuilt. I did not find a documented binary-stability guarantee for protected pinot-segment-local fields, so this is conditional on reusing existing extension jars, rather than a claim that every rolling upgrade fails. Please preserve the protected Set contract with separate backing state, or explicitly document and validate the requirement to rebuild dependent extensions. The current binary-compatibility workflow checks pinot-spi and pinot-segment-spi, not this module. ########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedSegmentMetadataStore.java: ########## @@ -0,0 +1,943 @@ +/** + * 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 boolean _closed; + + 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)]; + for (int offset = 0; offset < bytes.length; offset += block.length) { + int length = Math.min(block.length, bytes.length - offset); + buffer.copyTo(offset, block, 0, length); + for (int i = 0; i < length; i++) { + if (block[i] != bytes[offset + i]) { + return false; + } + } + } + return true; + } + + int size() { + return _columns._names.length; + } + + String name(int ordinal) { + return _columns._names[ordinal]; + } + + int ordinal(String column) { + return Arrays.binarySearch(_columns._names, column); + } + + synchronized int getInt(int ordinal, int offset) { + ensureOpen(); + return _buffer.getInt((long) _tableOffset + (long) ordinal * RECORD_SIZE + offset); + } + + synchronized long getLong(int ordinal, int offset) { + ensureOpen(); + return _buffer.getLong((long) _tableOffset + (long) ordinal * RECORD_SIZE + offset); + } + + boolean flag(int ordinal, int flag) { + return (getInt(ordinal, FLAGS) & flag) != 0; + } + + @Nullable + String string(int ordinal, int field) { + byte[] bytes = blob(ordinal, field); + return bytes != null ? new String(bytes, StandardCharsets.UTF_8) : null; + } + + @Nullable + synchronized byte[] blob(int ordinal, int field) { + int offset = getInt(ordinal, field); + if (offset < 0) { + return null; + } + int length = getInt(ordinal, field + 4); + byte[] bytes = new byte[length]; + _buffer.copyTo(offset, bytes); + return bytes; + } + + FieldSpec fieldSpec(int ordinal) { + return _columns._fieldSpecs[ordinal]; + } + + @Nullable + Comparable<?> value(int ordinal, boolean minimum) { + int tag = getInt(ordinal, minimum ? MIN_TAG : MAX_TAG); + long raw = getLong(ordinal, minimum ? MIN_RAW : MAX_RAW); + int blobField = minimum ? MIN_BLOB : MAX_BLOB; + switch (tag) { + case NO_VALUE: + return null; + case INT_VALUE: + return (int) raw; + case LONG_VALUE: + return raw; + case FLOAT_VALUE: + return Float.intBitsToFloat((int) raw); + case DOUBLE_VALUE: + return Double.longBitsToDouble(raw); + case DECIMAL_VALUE: + return new BigDecimal(string(ordinal, blobField)); + case STRING_VALUE: + return string(ordinal, blobField); + case BYTES_VALUE: + return new ByteArray(blob(ordinal, blobField)); + default: + throw new IllegalStateException("Invalid value tag: " + tag); + } + } + + synchronized PropertiesConfiguration segmentConfiguration() + throws IOException, ConfigurationException { + ensureOpen(); + byte[] bytes = new byte[_segmentConfigLength]; + _buffer.copyTo(_segmentConfigOffset, bytes); + // Values were already split by the source parser. A second delimiter pass would change escaped commas. + PropertiesConfiguration config = CommonsConfigurationUtils.fromInputStream(null, false, null); + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes))) { + int count = input.readInt(); + for (int i = 0; i < count; i++) { + String key = readString(input); + int valueCount = input.readInt(); + if (valueCount == 1) { + config.setProperty(key, readString(input)); + } else { + List<String> values = new ArrayList<>(valueCount); + for (int j = 0; j < valueCount; j++) { + values.add(readString(input)); + } + config.setProperty(key, values); + } + } + } + return config; + } + + @Override + public synchronized void close() + throws IOException { + if (_closed) { + return; + } + _closed = true; + _buffer.close(); + } + + private void ensureOpen() { + if (_closed) { + throw new IllegalStateException("Mapped segment metadata is closed"); + } + } + + private static byte[] build(File indexDir, Fingerprint propertiesFingerprint, Fingerprint indexMapFingerprint, + IndexService indexService) + throws IOException, ConfigurationException { + SegmentMetadataImpl source = new SegmentMetadataImpl(indexDir); Review Comment: Performance validation: benchmark the complete serving load path The warm metadata-only load probe does not exercise the full serving path. SegmentLocalFSDirectory first constructs eager SegmentMetadataImpl, and ImmutableSegmentLoader later swaps it for mapped metadata. Opening a valid sidecar then fingerprints the canonical files before and after opening, reads/validates the sidecar, compares the opened bytes, and parses the FieldSpecs. Cold generation adds further parsing here. Please measure full serving startup/readiness time, peak heap, GC, and RSS for both cold and reused sidecars, including realistic load concurrency. The reported retained-heap reduction is useful, but it does not establish an improvement in those startup costs. The PR description acknowledges this limitation; I would keep it explicit in rollout criteria. ########## 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. Buffer-backed reads after close throw. +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()); Review Comment: Performance validation: include warmed index-size caches in heap measurements This permanently caches an IndexSize object and a freshly decoded String for each index, whereas eager ColumnMetadataImpl keeps packed type/size values in a primitive LongArrayList. These caches are reached through normal compression-summary polling: SegmentStatusChecker requests AGGREGATE_SUMMARY, and SegmentCompressionStatsReader calls getIndexSizeFor() for the columns it visits. Please include compression-summary polling and min/max pruning before measuring steady-state retained heap. Consider canonicalizing IDs to IndexService's existing strings and using compact primitive cached entries. I am not asserting an overall heap regression without measurement, but the unwarmed metadata/serving-wrapper probes do not establish the retained heap after these normal accesses. ########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataImpl.java: ########## @@ -142,6 +159,89 @@ public SegmentMetadataImpl(File indexDir) } } + /// Opens the versioned, file-backed metadata for a sealed segment. A legacy segment is converted locally on first + /// open; subsequent opens read the binary file and do not build retained per-column metadata objects. + public static SegmentMetadataImpl loadMapped(File indexDir) + throws IOException, ConfigurationException { + return loadMapped(indexDir, ReadMode.mmap); + } + + /// Uses a direct buffer for `heap` mode and a file mapping for `mmap` mode. + public static SegmentMetadataImpl loadMapped(File indexDir, ReadMode readMode) + throws IOException, ConfigurationException { + return new SegmentMetadataImpl(indexDir, true, readMode); + } + + /// Identifies the local mapped metadata sidecar and its incomplete temporary files. + public static boolean isMappedMetadataSidecarFileName(String fileName) { + return MappedSegmentMetadataStore.isSidecarFileName(fileName); + } + + /// Mapped constructor for segment-directory implementations that subclass [SegmentMetadataImpl]. + protected SegmentMetadataImpl(File indexDir, boolean mapped) + throws IOException, ConfigurationException { + this(indexDir, mapped, ReadMode.mmap); + } + + protected SegmentMetadataImpl(File indexDir, boolean mapped, ReadMode readMode) Review Comment: Design/scope: separate binary backing from schema and column-directory optimizations MappedSegmentMetadataStore.open() already implements the desired check/rebuild/open flow. Most of the additional complexity here comes from combining eager, mapped, and consuming representations in one class, plus removing the retained Schema/TreeMap, adding lazy wrappers, and managing removed-column/virtual-column overlays. This introduces nullable backing fields, repeated _mappedStore branches, and getter behavior changes such as rebuilding getColumnMetadataMap() and getAllColumns(). Some representation and lifetime changes are necessary for the heap-reduction goal. However, moving column statistics off heap does not inherently require all of the schema/map-container optimizations in the same change. Please isolate the mapped implementation/backing strategy, keep legacy parsing for conversion/preprocessing and stream/realtime callers, and give the finalized sealed-segment path one consistent metadata representation. Preserve accessor contracts and costs where practical; split additional container-retention optimizations if they require broad caller changes without separate performance evidence. As a small related API simplification, these protected constructors take a mapped boolean but reject false; a constructor distinguished by ReadMode or an explicit factory avoids an invalid option. Loader lifetime fixes, transfer exclusions, and upsert safety fixes remain necessary integration work and should not be removed as unrelated scope. ########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataImpl.java: ########## @@ -483,21 +599,106 @@ public String getEndOffset() { @Override public TreeMap<String, ColumnMetadata> getColumnMetadataMap() { - return _columnMetadataMap; + if (_mappedStore == null) { + return _columnMetadataMap; + } + // Compatibility API for metadata/admin callers. The serving path uses column names and direct lookup instead. + TreeMap<String, ColumnMetadata> result = new TreeMap<>(); + for (String column : getPhysicalColumnNames()) { + result.put(column, getColumnMetadataFor(column)); + } + result.putAll(_virtualColumnMetadata); + return result; + } + + /// Returns the physical column names without materializing a metadata map or Schema on mapped segments. + public Set<String> getPhysicalColumnNames() { + return _mappedStore != null ? _physicalColumnNames : _schema.getPhysicalColumnNames(); + } + + @Override + public NavigableSet<String> getAllColumns() { + if (_mappedStore == null) { + return getSchema().getColumnNames(); + } + NavigableSet<String> columns = new TreeSet<>(getPhysicalColumnNames()); Review Comment: [P2] Avoid rebuilding the column-name tree for every partial upsert For mapped segments this creates one tree node per column and performs O(C log C) construction on every call. There is an existing per-record caller: ConcurrentMapPartitionUpsertMetadataManager.doUpdateRecord() -> PartialUpsertHandler.merge() -> PartialUpsertColumnarMerger.merge() -> LazyRow.getColumnNames() -> ImmutableSegmentImpl.getColumnNames(). It reaches this path whenever the previous record is in a mapped sealed segment. Previously Schema.getColumnNames() returned its existing navigable key-set view. This adds full-schema allocation and sorting to each affected update, particularly costly for the wide schemas targeted by the heap optimization. Please reuse a stable column-name view or cache the set once load-time removals and virtual-column additions are complete. I have not measured an end-to-end throughput delta. ########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedSegmentMetadataStore.java: ########## @@ -0,0 +1,943 @@ +/** + * 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 boolean _closed; + + 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)]; + for (int offset = 0; offset < bytes.length; offset += block.length) { + int length = Math.min(block.length, bytes.length - offset); + buffer.copyTo(offset, block, 0, length); + for (int i = 0; i < length; i++) { + if (block[i] != bytes[offset + i]) { + return false; + } + } + } + return true; + } + + int size() { + return _columns._names.length; + } + + String name(int ordinal) { + return _columns._names[ordinal]; + } + + int ordinal(String column) { + return Arrays.binarySearch(_columns._names, column); + } + + synchronized int getInt(int ordinal, int offset) { Review Comment: Performance validation: measure the segment-wide scalar-read monitor Every scalar read now synchronizes on the same store object, across all columns and concurrent queries for that segment. ImmutableDataSource delegates its scalar metadata accessors to these methods. Scan iterators cache cardinality and MV widths during construction, so I did not find per-row query-scan locking, and the cached bounds/partition paths avoid repeated decoding. Nevertheless, this introduces a shared contention point for concurrent short queries. Please include a representative concurrent-query comparison on the current implementation, covering queries that hit the same segments, before treating the retained-heap results as sufficient serving-performance evidence. This is an unmeasured risk, not a claim of a demonstrated latency regression. Preserve the close/read safety guarantee if the locking approach changes. ########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedSegmentMetadataStore.java: ########## @@ -0,0 +1,943 @@ +/** + * 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; Review Comment: Minor: start the newly introduced format at version 1 Since this PR introduces segment_metadata.bin, VERSION = 5 appears to encode development iterations as public format history. Unless versions 1-5 have already produced deployed caches that need explicit invalidation, please define the final initial format as version 1 and remove the v4/v5-specific history from comments and tests. Keep the unsupported-version rebuild test using a deliberately different version. This is a clarity suggestion, not a decoding defect. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/SegmentLocalFSDirectory.java: ########## @@ -286,15 +310,24 @@ private synchronized void loadData() @Override public void close() throws IOException { - _segmentLock.close(); - synchronized (this) { - if (_columnIndexDirectory != null) { - _columnIndexDirectory.close(); - _columnIndexDirectory = null; + try { + try { + _segmentLock.close(); + } finally { + synchronized (this) { + if (_columnIndexDirectory != null) { + _columnIndexDirectory.close(); + _columnIndexDirectory = null; + } + if (_starTreeIndexReader != null) { + _starTreeIndexReader.close(); + _starTreeIndexReader = null; + } + } } - if (_starTreeIndexReader != null) { - _starTreeIndexReader.close(); - _starTreeIndexReader = null; + } finally { + if (_segmentMetadata != null) { + _segmentMetadata.close(); Review Comment: Compatibility: metadata-returning instance APIs release ownership before callers can use the result Closing mapped metadata here changes the lifetime of objects returned by HelixInstanceDataManager.getSegmentMetadata() and getAllSegmentsMetadata(). Both methods acquire segment references, return the live SegmentMetadata objects, and release the references in finally before the caller can use the returned metadata. If a segment is then unloaded, a caller's getColumnMetadataFor(...).getCardinality() or toJson() now throws IllegalStateException; the previous heap-backed metadata remained usable. The new requirement that callers hold the owning segment reference cannot be satisfied through those APIs because they do not return that reference. Please make these APIs return detached metadata or keep ownership for the duration of a callback/explicit handle. I found no active OSS or checked-out StarTree callers, so this is an API-contract concern, not a demonstrated production query failure. The fix should retain safe buffer cleanup. ########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedSegmentMetadataStore.java: ########## @@ -0,0 +1,943 @@ +/** + * 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 boolean _closed; + + 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)]; + for (int offset = 0; offset < bytes.length; offset += block.length) { + int length = Math.min(block.length, bytes.length - offset); + buffer.copyTo(offset, block, 0, length); + for (int i = 0; i < length; i++) { + if (block[i] != bytes[offset + i]) { + return false; + } + } + } + return true; + } + + int size() { + return _columns._names.length; + } + + String name(int ordinal) { + return _columns._names[ordinal]; + } + + int ordinal(String column) { + return Arrays.binarySearch(_columns._names, column); + } + + synchronized int getInt(int ordinal, int offset) { + ensureOpen(); + return _buffer.getInt((long) _tableOffset + (long) ordinal * RECORD_SIZE + offset); + } + + synchronized long getLong(int ordinal, int offset) { + ensureOpen(); + return _buffer.getLong((long) _tableOffset + (long) ordinal * RECORD_SIZE + offset); + } + + boolean flag(int ordinal, int flag) { + return (getInt(ordinal, FLAGS) & flag) != 0; + } + + @Nullable + String string(int ordinal, int field) { + byte[] bytes = blob(ordinal, field); + return bytes != null ? new String(bytes, StandardCharsets.UTF_8) : null; + } + + @Nullable + synchronized byte[] blob(int ordinal, int field) { + int offset = getInt(ordinal, field); + if (offset < 0) { + return null; + } + int length = getInt(ordinal, field + 4); + byte[] bytes = new byte[length]; + _buffer.copyTo(offset, bytes); + return bytes; + } + + FieldSpec fieldSpec(int ordinal) { + return _columns._fieldSpecs[ordinal]; + } + + @Nullable + Comparable<?> value(int ordinal, boolean minimum) { + int tag = getInt(ordinal, minimum ? MIN_TAG : MAX_TAG); + long raw = getLong(ordinal, minimum ? MIN_RAW : MAX_RAW); + int blobField = minimum ? MIN_BLOB : MAX_BLOB; + switch (tag) { + case NO_VALUE: + return null; + case INT_VALUE: + return (int) raw; + case LONG_VALUE: + return raw; + case FLOAT_VALUE: + return Float.intBitsToFloat((int) raw); + case DOUBLE_VALUE: + return Double.longBitsToDouble(raw); + case DECIMAL_VALUE: + return new BigDecimal(string(ordinal, blobField)); + case STRING_VALUE: + return string(ordinal, blobField); + case BYTES_VALUE: + return new ByteArray(blob(ordinal, blobField)); + default: + throw new IllegalStateException("Invalid value tag: " + tag); + } + } + + synchronized PropertiesConfiguration segmentConfiguration() + throws IOException, ConfigurationException { + ensureOpen(); + byte[] bytes = new byte[_segmentConfigLength]; + _buffer.copyTo(_segmentConfigOffset, bytes); + // Values were already split by the source parser. A second delimiter pass would change escaped commas. + PropertiesConfiguration config = CommonsConfigurationUtils.fromInputStream(null, false, null); + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes))) { + int count = input.readInt(); + for (int i = 0; i < count; i++) { + String key = readString(input); + int valueCount = input.readInt(); + if (valueCount == 1) { + config.setProperty(key, readString(input)); + } else { + List<String> values = new ArrayList<>(valueCount); + for (int j = 0; j < valueCount; j++) { + values.add(readString(input)); + } + config.setProperty(key, values); + } + } + } + return config; + } + + @Override + public synchronized void close() + throws IOException { + if (_closed) { + return; + } + _closed = true; + _buffer.close(); + } + + private void ensureOpen() { + if (_closed) { + throw new IllegalStateException("Mapped segment metadata is closed"); + } + } + + private static byte[] build(File indexDir, Fingerprint propertiesFingerprint, Fingerprint indexMapFingerprint, + IndexService indexService) + throws IOException, ConfigurationException { + SegmentMetadataImpl source = new SegmentMetadataImpl(indexDir); + PropertiesConfiguration config = SegmentMetadataUtils.getPropertiesConfiguration(indexDir); + List<String> columns = new ArrayList<>(source.getColumnMetadataMap().keySet()); + columns.sort(Comparator.naturalOrder()); + if ((long) columns.size() * RECORD_SIZE + HEADER_SIZE > Integer.MAX_VALUE) { + throw new IOException("Too many columns for mapped metadata: " + columns.size()); + } + int blobBase = HEADER_SIZE + columns.size() * RECORD_SIZE; + BlobWriter blobs = new BlobWriter(blobBase); + BlobRef segmentConfig = blobs.add(encodeSegmentConfiguration(config)); + // EmptyColumnMetadata does not load index sizes even if a stale index_map is still present. + Map<String, List<IndexSize>> indexSizes = source.getTotalDocs() > 0 Review Comment: Simplification: serialize index sizes from the already parsed metadata The SegmentMetadataImpl constructed above has already parsed metadata.properties and, for nonempty v3 segments, index_map. This code reads the properties again and calls readIndexSizes(), duplicating the eager index-map parser and its unknown-index/invalid-size tolerance policy. Please encode the existing ColumnMetadata index entries using getNumIndexes(), getIndexType(), and getIndexSize(), or share a single parsed source representation. This removes redundant work and avoids maintaining two parsers that must remain semantically equivalent whenever the canonical metadata logic changes. -- 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]
