xiangfu0 commented on code in PR #19998:
URL: https://github.com/apache/pinot/pull/19998#discussion_r4368364872
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java:
##########
@@ -1410,7 +1419,20 @@ protected void removeDocId(IndexSegment segment, int
docId) {
protected void trackUpdatedSegmentsSinceLastSnapshot(IndexSegment segment) {
if (_enableSnapshot && segment instanceof ImmutableSegmentImpl
immutableSegment) {
- _updatedSegmentsSinceLastSnapshot.add(immutableSegment);
+ markSegmentUpdated(immutableSegment);
+ }
+ }
+
+ private void markSegmentUpdated(ImmutableSegmentImpl segment) {
+ // Publish the marker before reading metadata so a concurrent snapshot
cannot classify this segment as unchanged.
+ if (_updatedSegmentsSinceLastSnapshot.putIfAbsent(segment,
TTL_WATERMARK_NOT_SET) != null
Review Comment:
Nice fix for the mapped-metadata lifetime: capturing the max comparison
value eagerly under the read lock avoids reading `getMaxValue()` from an
unmapped buffer after the segment is destroyed. Worth confirming the intent: if
`tryAcquireReadLock()` fails here the entry stays `TTL_WATERMARK_NOT_SET`,
which caps the persisted watermark to -inf for the round. That's the
safe/conservative behavior for a segment being torn down, but since
`putIfAbsent` short-circuits re-marks, a segment first marked while momentarily
locked won't get its real max captured on a later mark in the same round —
acceptable given it only lowers the watermark, just flagging the subtlety.
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java:
##########
@@ -1101,11 +1108,21 @@ private File createSegmentTarFile(TableDataManager
tableDataManager, String segm
File segmentTarFile =
org.apache.pinot.common.utils.FileUtils.concatAndValidateFile(segmentTarUploadDir,
tableNameWithType + "_" + segmentName + "_" + UUID.randomUUID() +
TarCompressionUtils.TAR_GZ_FILE_EXTENSION,
"Invalid table / segment name: %s, %s", tableNameWithType,
segmentName);
- TarCompressionUtils.createCompressedTarFile(new
File(tableDataManager.getTableDataDir(), segmentName),
- segmentTarFile);
+ createSegmentTransferTarFile(new File(tableDataManager.getTableDataDir(),
segmentName), segmentTarFile);
return segmentTarFile;
}
+ private static void createSegmentTransferTarFile(File segmentDir, File
segmentTarFile)
+ throws IOException {
+ TarCompressionUtils.createFilteredCompressedTarFile(segmentDir,
segmentTarFile, file -> {
+ if (!segmentDir.equals(file.getParentFile())) {
+ return true;
+ }
+ // Local serving artifacts are rebuilt from metadata.properties on the
receiving server.
+ return
!SegmentMetadataImpl.isMappedMetadataSidecarFileName(file.getName());
Review Comment:
Good — this is the key compat guard: the sidecar (and its
`.segment_metadata.*.tmp` temps) is stripped from both the peer-download and
upload transfer tars, so other servers/controllers never receive it and rebuild
locally. Confirmed the predicate only rejects direct children of `segmentDir`,
which is exactly where the sidecar lives (top level, even for v3).
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/MappedSegmentMetadataStore.java:
##########
@@ -0,0 +1,929 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.spi.index.metadata;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Interner;
+import com.google.common.collect.Interners;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.AtomicMoveNotSupportedException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32C;
+import javax.annotation.Nullable;
+import org.apache.commons.configuration2.Configuration;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.configuration2.ex.ConfigurationException;
+import org.apache.pinot.segment.spi.ColumnMetadata;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.creator.SegmentVersion;
+import org.apache.pinot.segment.spi.index.IndexService;
+import org.apache.pinot.segment.spi.index.IndexType;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.apache.pinot.segment.spi.partition.PartitionFunction;
+import org.apache.pinot.segment.spi.store.ColumnIndexUtils;
+import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
+import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils;
+import org.apache.pinot.spi.config.table.FieldConfig.EncodingType;
+import org.apache.pinot.spi.data.ComplexFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.env.CommonsConfigurationUtils;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.ColumnNameInterner;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.ReadMode;
+
+/// Versioned, file-backed metadata for a sealed segment. The source
`metadata.properties` remains the canonical
+/// representation for old readers and preprocessing.
+///
+/// The big-endian file starts with a 64-byte header containing the magic,
format version, total length, source
+/// `metadata.properties` and `index_map` length/CRC fingerprints, registered
index IDs, payload CRC,
+/// segment-property blob reference,
+/// column count, and record-table bounds. Sorted columns occupy fixed
180-byte records, followed by variable blobs;
+/// each blob reference is an absolute 32-bit offset and length. Numeric
column statistics are read directly from the
+/// backing buffer. The column layout retains names and reusable [FieldSpec]
references on heap.
+///
+/// A stale or corrupt sidecar is rebuilt from the canonical source and
replaced locally, using atomic rename where
+/// supported. Change `VERSION` whenever the encoding or the values derived
from canonical metadata change; old
+/// versions are rebuilt rather than migrated.
+final class MappedSegmentMetadataStore implements Closeable {
+ static final String FILE_NAME = "segment_metadata.bin";
+ private static final String TEMP_FILE_PREFIX = ".segment_metadata.";
+ private static final String TEMP_FILE_SUFFIX = ".tmp";
+
+ static boolean isSidecarFileName(String fileName) {
+ return FILE_NAME.equals(fileName)
+ || (fileName.startsWith(TEMP_FILE_PREFIX) &&
fileName.endsWith(TEMP_FILE_SUFFIX));
+ }
+
+ private static final int MAGIC = 0x504d4431; // PMD1
+ private static final int VERSION = 5;
+ private static final int HEADER_SIZE = 64;
+ private static final int RECORD_SIZE = 180;
+
+ static final int NAME = 0;
+ static final int FIELD_SPEC = 8;
+ static final int TOTAL_DOCS = 16;
+ static final int CARDINALITY = 20;
+ static final int FLAGS = 24;
+ static final int ENCODING = 28;
+ static final int SHORTEST = 32;
+ static final int LONGEST = 36;
+ static final int TOTAL_ENTRIES = 40;
+ static final int MAX_MV = 44;
+ static final int MAX_ROW = 48;
+ static final int BITS = 52;
+ static final int MIN_TAG = 56;
+ static final int MIN_RAW = 60;
+ static final int MIN_BLOB = 68;
+ static final int MAX_TAG = 76;
+ static final int MAX_RAW = 80;
+ static final int MAX_BLOB = 88;
+ static final int PARTITION_FUNCTION = 96;
+ static final int PARTITIONS = 104;
+ static final int PARENT = 112;
+ static final int SPARSE_KEYS = 120;
+ static final int SPARSE_MULTI_VALUE_KEYS = 128;
+ static final int SPARSE_KEY_TYPES = 136;
+ static final int RAW_UNCOMPRESSED_SIZE = 144;
+ static final int DICTIONARY_UNCOMPRESSED_SIZE = 152;
+ static final int COMPRESSION_TYPE = 160;
+ static final int INDEXES = 168;
+ static final int INDEX_COUNT = 176;
+
+ static final int HAS_DICTIONARY = 1;
+ static final int SORTED = 1 << 1;
+ static final int NON_NULL = 1 << 2;
+ static final int INVALID_MIN_MAX = 1 << 3;
+ static final int ASCII = 1 << 4;
+ static final int AUTO_GENERATED = 1 << 5;
+
+ private static final int NO_VALUE = 0;
+ private static final int INT_VALUE = 1;
+ private static final int LONG_VALUE = 2;
+ private static final int FLOAT_VALUE = 3;
+ private static final int DOUBLE_VALUE = 4;
+ private static final int DECIMAL_VALUE = 5;
+ private static final int STRING_VALUE = 6;
+ private static final int BYTES_VALUE = 7;
+
+ private static final Interner<FieldSpec> FIELD_SPECS =
Interners.newWeakInterner();
+ private static final Interner<ColumnLayout> COLUMN_LAYOUTS =
Interners.newWeakInterner();
+
+ private final PinotDataBuffer _buffer;
+ private final ColumnLayout _columns;
+ private final int _tableOffset;
+ private final int _segmentConfigOffset;
+ private final int _segmentConfigLength;
+
+ private MappedSegmentMetadataStore(PinotDataBuffer buffer, ColumnLayout
columns, int tableOffset,
+ int segmentConfigOffset, int segmentConfigLength) {
+ _buffer = buffer;
+ _columns = columns;
+ _tableOffset = tableOffset;
+ _segmentConfigOffset = segmentConfigOffset;
+ _segmentConfigLength = segmentConfigLength;
+ }
+
+ static MappedSegmentMetadataStore open(File indexDir, ReadMode readMode)
+ throws IOException, ConfigurationException {
+ File source = SegmentDirectoryPaths.findMetadataFile(indexDir);
+ if (source == null) {
+ throw new IOException("Missing metadata.properties in " + indexDir);
+ }
+ File sidecar = new File(indexDir, FILE_NAME);
+ boolean rebuilt = false;
+ for (int attempt = 0; attempt < 3; attempt++) {
+ IndexService indexService = IndexService.getInstance();
+ Fingerprint propertiesFingerprint = Fingerprint.of(source);
+ Fingerprint indexMapFingerprint = indexMapFingerprint(indexDir,
indexService);
+ // The validated heap copy is transient; the store retains only the
buffer and shared column layout.
+ byte[] bytes = readValid(sidecar, propertiesFingerprint,
indexMapFingerprint);
+ if (bytes == null) {
+ writeAtomically(sidecar, build(indexDir, propertiesFingerprint,
indexMapFingerprint, indexService));
+ rebuilt = true;
+ bytes = readValid(sidecar, propertiesFingerprint, indexMapFingerprint);
+ if (bytes == null) {
+ continue;
+ }
+ }
+ PinotDataBuffer buffer = readMode == ReadMode.heap
+ ? PinotDataBuffer.loadFile(sidecar, 0, bytes.length,
ByteOrder.BIG_ENDIAN, "segment metadata")
+ : PinotDataBuffer.mapFile(sidecar, true, 0, bytes.length,
ByteOrder.BIG_ENDIAN, "segment metadata");
+ boolean keepBuffer = false;
+ try {
+ // Another loader may atomically replace the sidecar between
validation and opening it. Check the exact
+ // buffer bytes and canonical inputs before publishing a view. The
header alone cannot prove the bytes match.
+ if (!matchesBuffer(buffer, bytes) ||
!propertiesFingerprint.sameAs(Fingerprint.of(source))
+ || indexService != IndexService.getInstance()
+ || !indexMapFingerprint.sameAs(indexMapFingerprint(indexDir,
indexService))) {
+ continue;
+ }
+ ByteBuffer header = ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN);
+ int count = header.getInt(52);
+ int tableOffset = header.getInt(56);
+ String[] names = new String[count];
+ FieldSpec[] fieldSpecs = new FieldSpec[count];
+ for (int i = 0; i < count; i++) {
+ int record = tableOffset + i * RECORD_SIZE;
+ names[i] = ColumnNameInterner.intern(readString(bytes,
header.getInt(record + NAME),
+ header.getInt(record + NAME + 4)));
+ FieldSpec parsed = JsonUtils.stringToObject(readString(bytes,
header.getInt(record + FIELD_SPEC),
+ header.getInt(record + FIELD_SPEC + 4)), FieldSpec.class);
+ fieldSpecs[i] = parsed instanceof ComplexFieldSpec ? parsed :
FIELD_SPECS.intern(parsed);
+ }
+ ColumnLayout columns = COLUMN_LAYOUTS.intern(new ColumnLayout(names,
fieldSpecs));
+ MappedSegmentMetadataStore store = new
MappedSegmentMetadataStore(buffer, columns, tableOffset,
+ header.getInt(44), header.getInt(48));
+ keepBuffer = true;
+ return store;
+ } catch (IOException | RuntimeException e) {
+ if (!rebuilt) {
+ // A checksum-valid older sidecar can still contain FieldSpec JSON
this reader cannot parse. Rebuild it
+ // once from canonical metadata instead of falling back to eager on
every load.
+ writeAtomically(sidecar, build(indexDir, propertiesFingerprint,
indexMapFingerprint, indexService));
+ rebuilt = true;
+ continue;
+ }
+ throw new IOException("Invalid field specs in mapped segment metadata:
" + sidecar, e);
+ } finally {
+ if (!keepBuffer) {
+ buffer.close();
+ }
+ }
+ }
+ throw new IOException("Metadata changed while opening segment: " +
indexDir);
+ }
+
+ private static boolean matchesBuffer(PinotDataBuffer buffer, byte[] bytes) {
+ if (buffer.size() != bytes.length) {
+ return false;
+ }
+ byte[] block = new byte[Math.min(8192, bytes.length)];
+ 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);
+ }
+
+ int getInt(int ordinal, int offset) {
+ return _buffer.getInt((long) _tableOffset + (long) ordinal * RECORD_SIZE +
offset);
+ }
+
+ long getLong(int ordinal, int offset) {
+ 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
+ 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);
+ }
+ }
+
+ PropertiesConfiguration segmentConfiguration()
+ throws IOException, ConfigurationException {
+ 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 void close()
+ throws IOException {
+ _buffer.close();
+ }
+
+ 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()) {
+ json.put("defaultNullValue", fieldSpec.getDefaultNullValueString());
+ }
+ // TimeFieldSpec omits the discriminator, including when nested in a
ComplexFieldSpec.
+ json.put("fieldType", fieldSpec.getFieldType().toString());
+ if (fieldSpec instanceof ComplexFieldSpec) {
+ ObjectNode children = (ObjectNode) json.get("childFieldSpecs");
+ for (Map.Entry<String, FieldSpec> entry : ((ComplexFieldSpec)
fieldSpec).getChildFieldSpecs().entrySet()) {
+ children.set(entry.getKey(), fieldSpecJson(entry.getValue()));
+ }
+ }
+ return json;
+ }
+
+ private static byte[] encodeSegmentConfiguration(PropertiesConfiguration
config)
+ throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (DataOutputStream output = new DataOutputStream(bytes)) {
+ List<String> keys = CommonsConfigurationUtils.getKeys(config).stream()
+ .filter(key ->
!key.startsWith(V1Constants.MetadataKeys.Column.COLUMN_PROPS_KEY_PREFIX))
+ .sorted().collect(Collectors.toList());
+ output.writeInt(keys.size());
+ for (String key : keys) {
+ writeString(output, key);
+ Object value = config.getProperty(key);
+ if (value instanceof List<?>) {
+ List<?> values = (List<?>) value;
+ output.writeInt(values.size());
+ for (Object entry : values) {
+ writeString(output, String.valueOf(entry));
+ }
+ } else {
+ output.writeInt(1);
+ writeString(output, String.valueOf(value));
+ }
+ }
+ }
+ return bytes.toByteArray();
+ }
+
+ private static byte[] encodePartitionFunction(String column,
PropertiesConfiguration metadata)
+ throws IOException {
+ String prefix = V1Constants.MetadataKeys.Column.COLUMN_PROPS_KEY_PREFIX +
column + ".";
+ String name = metadata.getString(prefix +
V1Constants.MetadataKeys.Column.PARTITION_FUNCTION);
+ int numPartitions = metadata.getInt(prefix +
V1Constants.MetadataKeys.Column.NUM_PARTITIONS);
+ Configuration functionConfig = metadata.subset(prefix +
V1Constants.MetadataKeys.Column.PARTITION_FUNCTION_CONFIG);
+ Map<String, String> config = new HashMap<>();
+ functionConfig.forEach((key, value) -> config.put(key,
+ value instanceof List ? ((List<?>)
value).stream().map(String::valueOf).collect(Collectors.joining(","))
+ : value.toString()));
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (DataOutputStream output = new DataOutputStream(bytes)) {
+ writeString(output, name);
+ output.writeInt(numPartitions);
+ output.writeInt(config.size());
+ for (String key :
config.keySet().stream().sorted().collect(Collectors.toList())) {
+ writeString(output, key);
+ writeString(output, config.get(key));
+ }
+ }
+ return bytes.toByteArray();
+ }
+
+ private static byte[] encodePartitions(Set<Integer> partitions)
+ throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (DataOutputStream output = new DataOutputStream(bytes)) {
+ output.writeInt(partitions.size());
+ for (int partition :
partitions.stream().sorted().collect(Collectors.toList())) {
+ output.writeInt(partition);
+ }
+ }
+ return bytes.toByteArray();
+ }
+
+ private static byte[] encodeIndexSizes(List<IndexSize> sizes)
+ throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (DataOutputStream output = new DataOutputStream(bytes)) {
+ for (IndexSize size : sizes) {
+ writeString(output, size._type);
+ output.writeLong(size._size);
+ }
+ }
+ return bytes.toByteArray();
+ }
+
+ private static Map<String, List<IndexSize>> readIndexSizes(File indexDir,
IndexService indexService)
+ throws ConfigurationException {
+ File indexMap = findIndexMap(indexDir);
+ if (indexMap == null) {
+ return Map.of();
+ }
+ // Match SegmentMetadataImpl: index sizes come only from the local v3
index_map.
+ PropertiesConfiguration config =
CommonsConfigurationUtils.fromFile(indexMap);
+ Map<String, List<IndexSize>> result = new HashMap<>();
+ for (String key : CommonsConfigurationUtils.getKeys(config)) {
+ try {
+ String[] parsed = ColumnIndexUtils.parseIndexMapKeys(key,
indexDir.getPath());
+ if (ColumnIndexUtils.MAP_KEY_NAME_SIZE.equals(parsed[2])) {
+ long size = config.getLong(key);
+ if (size < 0 || size > (1L << 48) - 1) {
+ continue;
+ }
+ IndexType<?, ?, ?> type = indexService.get(parsed[1]);
+ result.computeIfAbsent(parsed[0], ignored -> new ArrayList<>())
+ .add(new IndexSize(type.getId(), size));
+ }
+ } catch (Exception ignored) {
+ // Match the tolerant index_map reporting path in SegmentMetadataImpl.
+ }
+ }
+ return result;
+ }
+
+ @Nullable
+ private static File findIndexMap(File indexDir) {
+ File v3 = new File(new File(indexDir, "v3"),
V1Constants.INDEX_MAP_FILE_NAME);
+ return v3.isFile() ? v3 : null;
+ }
+
+ private static Fingerprint indexMapFingerprint(File indexDir, IndexService
indexService)
+ throws IOException {
+ Fingerprint file = Fingerprint.of(findIndexMap(indexDir));
+ // A newly registered index can turn a previously unknown index_map entry
into an available size.
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (DataOutputStream output = new DataOutputStream(bytes)) {
+ output.writeInt(file._crc);
+ for (IndexType<?, ?, ?> type : indexService.getAllIndexes()) {
+ writeString(output, type.getId());
+ }
+ }
+ byte[] encoded = bytes.toByteArray();
+ return new Fingerprint(file._length, crc(encoded, 0, encoded.length));
+ }
+
+ private static void writeValue(ByteBuffer target, int tagOffset, int
rawOffset, int blobOffset,
+ @Nullable Comparable<?> value, BlobWriter blobs) {
+ int tag = NO_VALUE;
+ long raw = 0;
+ BlobRef blob = BlobRef.ABSENT;
+ if (value instanceof Integer) {
+ tag = INT_VALUE;
+ raw = (Integer) value;
+ } else if (value instanceof Long) {
+ tag = LONG_VALUE;
+ raw = (Long) value;
+ } else if (value instanceof Float) {
+ tag = FLOAT_VALUE;
+ raw = Float.floatToRawIntBits((Float) value);
+ } else if (value instanceof Double) {
+ tag = DOUBLE_VALUE;
+ raw = Double.doubleToRawLongBits((Double) value);
+ } else if (value instanceof BigDecimal) {
+ tag = DECIMAL_VALUE;
+ blob = blobs.add(value.toString());
+ } else if (value instanceof String) {
+ tag = STRING_VALUE;
+ blob = blobs.add((String) value);
+ } else if (value instanceof ByteArray) {
+ tag = BYTES_VALUE;
+ blob = blobs.add(((ByteArray) value).getBytes());
+ } else if (value != null) {
+ throw new IllegalArgumentException("Unsupported metadata value: " +
value.getClass());
+ }
+ target.putInt(tagOffset, tag);
+ target.putLong(rawOffset, raw);
+ putRef(target, blobOffset, blob);
+ }
+
+ private static void putRef(ByteBuffer target, int offset, BlobRef ref) {
+ target.putInt(offset, ref._offset);
+ target.putInt(offset + 4, ref._length);
+ }
+
+ @Nullable
+ private static byte[] readValid(File file, Fingerprint properties,
Fingerprint indexMap)
+ throws IOException {
+ if (!file.isFile() || file.length() < HEADER_SIZE || file.length() >
Integer.MAX_VALUE) {
+ return null;
+ }
+ byte[] bytes = Files.readAllBytes(file.toPath());
+ try {
+ validate(bytes, properties, indexMap);
+ return bytes;
+ } catch (RuntimeException e) {
+ return null;
+ }
+ }
+
+ private static void validate(byte[] bytes, Fingerprint properties,
Fingerprint indexMap) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN);
+ if (buffer.getInt(0) != MAGIC || buffer.getInt(4) != VERSION ||
buffer.getLong(8) != bytes.length
+ || buffer.getLong(16) != properties._length || buffer.getInt(24) !=
properties._crc
+ || buffer.getLong(28) != indexMap._length || buffer.getInt(36) !=
indexMap._crc
+ || buffer.getInt(40) != sidecarCrc(bytes)) {
+ throw new IllegalArgumentException("Invalid metadata header");
+ }
+ int count = buffer.getInt(52);
+ int table = buffer.getInt(56);
+ int blobStart = buffer.getInt(60);
+ if (count < 0 || table != HEADER_SIZE || (long) table + (long) count *
RECORD_SIZE != blobStart
+ || blobStart > bytes.length) {
+ throw new IllegalArgumentException("Invalid column table bounds");
+ }
+ validateBlob(buffer, buffer.getInt(44), buffer.getInt(48), blobStart,
bytes.length, false);
+ validateSegmentConfiguration(bytes, buffer.getInt(44), buffer.getInt(48));
+ String prior = null;
+ for (int i = 0; i < count; i++) {
+ int record = table + i * RECORD_SIZE;
+ for (int field : new int[]{NAME, FIELD_SPEC, MIN_BLOB, MAX_BLOB,
PARTITION_FUNCTION, PARTITIONS, PARENT,
+ SPARSE_KEYS, SPARSE_MULTI_VALUE_KEYS, SPARSE_KEY_TYPES,
COMPRESSION_TYPE, INDEXES}) {
+ validateBlob(buffer, buffer.getInt(record + field),
buffer.getInt(record + field + 4), blobStart,
+ bytes.length, field != NAME && field != FIELD_SPEC);
+ }
+ int nameOffset = buffer.getInt(record + NAME);
+ int nameLength = buffer.getInt(record + NAME + 4);
+ String name = readString(bytes, nameOffset, nameLength);
+ if (prior != null && prior.compareTo(name) >= 0) {
+ throw new IllegalArgumentException("Column names are not sorted and
unique");
+ }
+ prior = name;
+ int minTag = buffer.getInt(record + MIN_TAG);
+ int maxTag = buffer.getInt(record + MAX_TAG);
+ int encoding = buffer.getInt(record + ENCODING);
+ if (minTag < NO_VALUE || minTag > BYTES_VALUE || maxTag < NO_VALUE ||
maxTag > BYTES_VALUE
+ || encoding < 0 || encoding > 2 || buffer.getInt(record +
INDEX_COUNT) < 0) {
+ throw new IllegalArgumentException("Invalid column value tag or index
count");
+ }
+ validateValueBlob(buffer, record + MIN_BLOB, minTag);
+ validateValueBlob(buffer, record + MAX_BLOB, maxTag);
+ validateIndexSizes(bytes, buffer.getInt(record + INDEXES),
buffer.getInt(record + INDEXES + 4),
+ buffer.getInt(record + INDEX_COUNT));
+ }
+ }
+
+ private static void validateValueBlob(ByteBuffer buffer, int field, int tag)
{
+ boolean usesBlob = tag == DECIMAL_VALUE || tag == STRING_VALUE || tag ==
BYTES_VALUE;
+ int offset = buffer.getInt(field);
+ int length = buffer.getInt(field + 4);
+ if (usesBlob ? offset < 0 : offset != -1 || length != 0) {
+ throw new IllegalArgumentException("Invalid typed value blob");
+ }
+ }
+
+ private static void validateSegmentConfiguration(byte[] bytes, int offset,
int length) {
+ try (DataInputStream input = new DataInputStream(new
ByteArrayInputStream(bytes, offset, length))) {
+ int count = input.readInt();
+ if (count < 0 || count > length / 8) {
+ throw new IllegalArgumentException("Invalid segment property count");
+ }
+ for (int i = 0; i < count; i++) {
+ readString(input);
+ int valueCount = input.readInt();
+ if (valueCount < 0 || valueCount > input.available() / 4) {
+ throw new IllegalArgumentException("Invalid segment property value
count");
+ }
+ for (int j = 0; j < valueCount; j++) {
+ readString(input);
+ }
+ }
+ if (input.available() != 0) {
+ throw new IllegalArgumentException("Trailing segment property bytes");
+ }
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Invalid segment properties", e);
+ }
+ }
+
+ private static void validateBlob(ByteBuffer buffer, int offset, int length,
int blobStart, int total,
+ boolean optional) {
+ if (optional && offset == -1 && length == 0) {
+ return;
+ }
+ if (offset < blobStart || length < 0 || (long) offset + length > total) {
+ throw new IllegalArgumentException("Invalid metadata blob bounds");
+ }
+ }
+
+ private static void validateIndexSizes(byte[] bytes, int offset, int length,
int count) {
+ if (count == 0) {
+ if (offset != -1 || length != 0) {
+ throw new IllegalArgumentException("Invalid empty index-size list");
+ }
+ return;
+ }
+ ByteBuffer buffer = ByteBuffer.wrap(bytes, offset,
length).order(ByteOrder.BIG_ENDIAN);
+ int cursor = offset;
+ int end = offset + length;
+ for (int i = 0; i < count; i++) {
+ if (cursor + 4 > end) {
+ throw new IllegalArgumentException("Truncated index-size name");
+ }
+ int nameLength = buffer.getInt(cursor);
+ cursor += 4;
+ if (nameLength < 0 || (long) cursor + nameLength + 8 > end) {
+ throw new IllegalArgumentException("Invalid index-size entry");
+ }
+ String type = new String(bytes, cursor, nameLength,
StandardCharsets.UTF_8);
+ if (IndexService.getInstance().get(type) == null) {
+ throw new IllegalArgumentException("Unknown index type: " + type);
+ }
+ cursor += nameLength + 8;
+ }
+ if (cursor != end) {
+ throw new IllegalArgumentException("Trailing index-size bytes");
+ }
+ }
+
+ static String readString(DataInputStream input)
+ throws IOException {
+ int length = input.readInt();
+ if (length < 0 || length > input.available()) {
+ throw new IOException("Invalid string length: " + length);
+ }
+ byte[] bytes = new byte[length];
+ input.readFully(bytes);
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+
+ private static String readString(byte[] bytes, int offset, int length) {
+ return new String(bytes, offset, length, StandardCharsets.UTF_8);
+ }
+
+ private static void writeString(DataOutputStream output, String string)
+ throws IOException {
+ byte[] bytes = string.getBytes(StandardCharsets.UTF_8);
+ output.writeInt(bytes.length);
+ output.write(bytes);
+ }
+
+ private static int crc(byte[] bytes, int offset, int length) {
+ CRC32C crc = new CRC32C();
+ crc.update(bytes, offset, length);
+ return (int) crc.getValue();
+ }
+
+ private static int sidecarCrc(byte[] bytes) {
+ CRC32C crc = new CRC32C();
+ crc.update(bytes, 0, 40);
+ for (int i = 0; i < Integer.BYTES; i++) {
+ crc.update(0);
+ }
+ crc.update(bytes, 44, bytes.length - 44);
+ return (int) crc.getValue();
+ }
+
+ private static void writeAtomically(File sidecar, byte[] bytes)
+ throws IOException {
+ Path path = sidecar.toPath();
+ Path temp = Files.createTempFile(path.getParent(), TEMP_FILE_PREFIX,
TEMP_FILE_SUFFIX);
+ try {
+ Files.write(temp, bytes);
+ try {
+ Files.move(temp, path, StandardCopyOption.ATOMIC_MOVE,
StandardCopyOption.REPLACE_EXISTING);
+ } catch (AtomicMoveNotSupportedException e) {
+ Files.move(temp, path, StandardCopyOption.REPLACE_EXISTING);
+ }
+ } finally {
+ Files.deleteIfExists(temp);
Review Comment:
`writeAtomically` cleans its own temp on success/failure, but on a hard JVM
crash between create and rename an orphan `.segment_metadata.<rand>.tmp` can
linger. It's excluded from transfer tars via `isSidecarFileName` and matches no
index pattern, so it's benign — just noting there's no active reaper for
orphans.
--
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]
