xiangfu0 commented on code in PR #19998:
URL: https://github.com/apache/pinot/pull/19998#discussion_r4353604662
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/SegmentLocalFSDirectory.java:
##########
@@ -141,11 +141,29 @@ public SegmentMetadataImpl getSegmentMetadata() {
return _segmentMetadata;
}
+ /// Called only during preprocessing, before the segment is queryable;
replacing and closing metadata is not
+ /// synchronized with serving readers.
@Override
public void reloadMetadata()
throws Exception {
- _segmentMetadata = new SegmentMetadataImpl(_indexDir);
- _columnIndexDirectory.setSegmentMetadata(_segmentMetadata);
+ SegmentMetadataImpl metadata = new SegmentMetadataImpl(_indexDir);
+ _columnIndexDirectory.setSegmentMetadata(metadata);
+ SegmentMetadataImpl previous = _segmentMetadata;
+ _segmentMetadata = metadata;
+ previous.close();
+ }
+
+ /// Replaces mutable metadata with the mapped representation after
preprocessing and before the segment is queryable.
+ public void loadMappedMetadata()
+ throws IOException, ConfigurationException {
+ if (_segmentMetadata.isMapped()) {
+ return;
+ }
+ SegmentMetadataImpl metadata = SegmentMetadataImpl.loadMapped(_indexDir);
Review Comment:
Same point as in the loader: `loadMapped` will try to write the sidecar into
the index dir when it's missing/stale. Worth making a write failure non-fatal
(fall back to eager metadata) so a read-only or full directory doesn't take the
segment offline.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/SegmentMetadataImpl.java:
##########
@@ -142,6 +155,73 @@ 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 new SegmentMetadataImpl(indexDir, true);
+ }
+
+ /// Mapped constructor for segment-directory implementations that subclass
[SegmentMetadataImpl].
+ protected SegmentMetadataImpl(File indexDir, boolean mapped)
+ throws IOException, ConfigurationException {
+ if (!mapped) {
+ throw new IllegalArgumentException("Use the regular File constructor for
mutable metadata");
+ }
+ _indexDir = indexDir;
+ _columnMetadataMap = null;
+ _schema = null;
+ MappedSegmentMetadataStore store =
MappedSegmentMetadataStore.open(indexDir);
+ try {
+ PropertiesConfiguration segmentConfig = store.segmentConfiguration();
+ initSegmentFields(segmentConfig);
+ setTimeInfo(segmentConfig);
+ File creationMetaFile =
SegmentDirectoryPaths.findCreationMetaFile(indexDir);
+ if (creationMetaFile != null) {
+ loadCreationMeta(creationMetaFile);
+ }
+ _mappedStore = store;
+ _mappedColumnMetadata = new AtomicReferenceArray<>(store.size());
+ _physicalColumnNames = new AbstractSet<>() {
+ @Override
+ public int size() {
+ return store.size() - _removedColumns.size();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ return o instanceof String && store.ordinal((String) o) >= 0 &&
!_removedColumns.contains(o);
+ }
+
+ @Override
+ public Iterator<String> iterator() {
+ return new Iterator<>() {
+ private int _index;
+
+ @Override
+ public boolean hasNext() {
+ while (_index < store.size() &&
_removedColumns.contains(store.name(_index))) {
+ _index++;
+ }
+ return _index < store.size();
+ }
+
+ @Override
+ public String next() {
+ if (!hasNext()) {
+ throw new java.util.NoSuchElementException();
Review Comment:
Nit: `java.util.NoSuchElementException` is fully qualified here while the
rest of the file imports its types — an import would match the surrounding
style.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java:
##########
@@ -201,6 +201,10 @@ public static ImmutableSegment load(SegmentDirectory
segmentDirectory, IndexLoad
public static ImmutableSegment load(SegmentDirectory segmentDirectory,
IndexLoadingConfig indexLoadingConfig,
@Nullable Schema schema)
throws Exception {
+ // Preprocessing uses mutable metadata. Only the final immutable serving
path uses the mapped representation.
+ if (segmentDirectory instanceof SegmentLocalFSDirectory localDirectory) {
+ localDirectory.loadMappedMetadata();
Review Comment:
This runs on every local segment load with no opt-out. If `open()` has to
rebuild the sidecar and the index dir isn't writable (read-only mount, full
disk, transient IO), the write throws and the whole segment load fails — the
old eager path only needed read access to metadata.properties. Could we catch a
build/write failure here and fall back to the eager `SegmentMetadataImpl`,
and/or add an instance-level kill switch? That keeps a segment servable even
when the cache can't be written.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedSegmentMetadataStore.java:
##########
@@ -0,0 +1,882 @@
+/**
+ * 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.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.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;
+
+/// 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, 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
+/// mapping. 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 header, record, or blob encoding
changes; old versions are rebuilt rather
+/// than migrated.
+final class MappedSegmentMetadataStore implements Closeable {
+ static final String FILE_NAME = "segment_metadata.bin";
+
+ private static final int MAGIC = 0x504d4431; // PMD1
+ private static final int VERSION = 3;
+ private static final int HEADER_SIZE = 64;
+ private static final int RECORD_SIZE = 180;
+
+ static final int NAME = 0;
+ static final int FIELD_SPEC = 8;
+ static final int TOTAL_DOCS = 16;
+ static final int CARDINALITY = 20;
+ static final int FLAGS = 24;
+ static final int ENCODING = 28;
+ static final int SHORTEST = 32;
+ static final int LONGEST = 36;
+ static final int TOTAL_ENTRIES = 40;
+ static final int MAX_MV = 44;
+ static final int MAX_ROW = 48;
+ static final int BITS = 52;
+ static final int MIN_TAG = 56;
+ static final int MIN_RAW = 60;
+ static final int MIN_BLOB = 68;
+ static final int MAX_TAG = 76;
+ static final int MAX_RAW = 80;
+ static final int MAX_BLOB = 88;
+ static final int PARTITION_FUNCTION = 96;
+ static final int PARTITIONS = 104;
+ static final int PARENT = 112;
+ static final int SPARSE_KEYS = 120;
+ static final int SPARSE_MULTI_VALUE_KEYS = 128;
+ static final int SPARSE_KEY_TYPES = 136;
+ static final int RAW_UNCOMPRESSED_SIZE = 144;
+ static final int DICTIONARY_UNCOMPRESSED_SIZE = 152;
+ static final int COMPRESSION_TYPE = 160;
+ static final int INDEXES = 168;
+ static final int INDEX_COUNT = 176;
+
+ static final int HAS_DICTIONARY = 1;
+ static final int SORTED = 1 << 1;
+ static final int NON_NULL = 1 << 2;
+ static final int INVALID_MIN_MAX = 1 << 3;
+ static final int ASCII = 1 << 4;
+ static final int AUTO_GENERATED = 1 << 5;
+
+ private static final int NO_VALUE = 0;
+ private static final int INT_VALUE = 1;
+ private static final int LONG_VALUE = 2;
+ private static final int FLOAT_VALUE = 3;
+ private static final int DOUBLE_VALUE = 4;
+ private static final int DECIMAL_VALUE = 5;
+ private static final int STRING_VALUE = 6;
+ private static final int BYTES_VALUE = 7;
+
+ private static final Interner<FieldSpec> FIELD_SPECS =
Interners.newWeakInterner();
+ private static final Interner<ColumnLayout> COLUMN_LAYOUTS =
Interners.newWeakInterner();
+
+ private final PinotDataBuffer _buffer;
+ private final ColumnLayout _columns;
+ private final int _tableOffset;
+ private final int _segmentConfigOffset;
+ private final int _segmentConfigLength;
+
+ private MappedSegmentMetadataStore(PinotDataBuffer buffer, ColumnLayout
columns, int tableOffset,
+ int segmentConfigOffset, int segmentConfigLength) {
+ _buffer = buffer;
+ _columns = columns;
+ _tableOffset = tableOffset;
+ _segmentConfigOffset = segmentConfigOffset;
+ _segmentConfigLength = segmentConfigLength;
+ }
+
+ static MappedSegmentMetadataStore open(File indexDir)
+ 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);
+ for (int attempt = 0; attempt < 3; attempt++) {
+ Fingerprint propertiesFingerprint = Fingerprint.of(source);
+ Fingerprint indexMapFingerprint = Fingerprint.of(findIndexMap(indexDir));
+ // The validated heap copy is transient; the store retains only the
mapping and shared column layout.
+ byte[] bytes = readValid(sidecar, propertiesFingerprint,
indexMapFingerprint);
+ if (bytes == null) {
+ writeAtomically(sidecar, build(indexDir, propertiesFingerprint,
indexMapFingerprint));
+ bytes = readValid(sidecar, propertiesFingerprint, indexMapFingerprint);
+ if (bytes == null) {
+ continue;
+ }
+ }
+ PinotDataBuffer buffer = 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 mapFile(). Check the exact mapped
+ // bytes and the canonical inputs before publishing a view over this
mapping. The header alone cannot prove
+ // that the mapped payload is the same bytes we validated.
+ if (!matchesMapped(buffer, bytes) ||
!propertiesFingerprint.sameAs(Fingerprint.of(source))
+ ||
!indexMapFingerprint.sameAs(Fingerprint.of(findIndexMap(indexDir)))) {
+ 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);
Review Comment:
Minor: excluding ComplexFieldSpec from the interner means two
otherwise-identical segments that contain any MAP/OPEN_STRUCT column get
distinct identities, so the cross-segment ColumnLayout sharing (the heap win)
doesn't kick in for those tables. ComplexFieldSpec already has value-based
equals/hashCode, so a stable interning path for it would recover the savings if
complex columns are common.
--
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]
