RowIndexEntry.java
/*
* 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.cassandra.io.sstable.format.big;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import com.codahale.metrics.Histogram;
import org.apache.cassandra.config.DataStorageSpec;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ArrayClustering;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.MessageParams;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.RejectException;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.ISerializer;
import org.apache.cassandra.io.sstable.AbstractRowIndexEntry;
import org.apache.cassandra.io.sstable.IndexInfo;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.sstable.format.big.BigFormat.Components;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.FileHandle;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.io.util.TrackedDataInputPlus;
import org.apache.cassandra.metrics.DefaultNameFactory;
import org.apache.cassandra.metrics.MetricNameFactory;
import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.net.ParamType;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.vint.VIntCoding;
import org.github.jamm.Unmetered;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
/**
* Binary format of {@code RowIndexEntry} is defined as follows:
* {@code
* (long) position (64 bit long, vint encoded)
* (int) serialized size of data that follows (32 bit int, vint encoded)
* -- following for indexed entries only (so serialized size > 0)
* (int) int/Uint DeletionTime.localDeletionTime (uint if version >= "oa" (5.0))
* (long) DeletionTime.markedForDeletionAt
* (int) number of IndexInfo objects (32 bit int, vint encoded)
* (*) serialized IndexInfo objects, see below
* (*) offsets of serialized IndexInfo objects, since version "ma" (3.0)
* Each IndexInfo object's offset is relative to the first IndexInfo object.
* }
* <p>
* See {@link IndexInfo} for a description of the serialized format.
* </p>
*
* <p>
* For each partition, the layout of the index file looks like this:
* </p>
* <ol>
* <li>partition key - prefixed with {@code short} length</li>
* <li>serialized {@code RowIndexEntry} objects</li>
* </ol>
*
* <p>
* Generally, we distinguish between index entries that have <i>index
* samples</i> (list of {@link IndexInfo} objects) and those who don't.
* For each <i>portion</i> of data for a single partition in the data file,
* an index sample is created. The size of that <i>portion</i> is defined
* by {@link org.apache.cassandra.config.Config#column_index_size}.
* </p>
* <p>
* Index entries with less than 2 index samples, will just store the
* position in the data file.
* </p>
* <p>
* Note: legacy sstables for index entries are those sstable formats that
* do <i>not</i> have an offsets table to index samples ({@link IndexInfo}
* objects). These are those sstables created on Cassandra versions
* earlier than 3.0.
* </p>
* <p>
* For index entries with index samples we store the index samples
* ({@link IndexInfo} objects). The bigger the partition, the more
* index samples are created. Since a huge amount of index samples
* will "pollute" the heap and cause huge GC pressure, Cassandra 3.6
* (CASSANDRA-11206) distinguishes between index entries with an
* "acceptable" amount of index samples per partition and those
* with an "enormous" amount of index samples. The barrier
* is controlled by the configuration parameter
* {@link org.apache.cassandra.config.Config#column_index_cache_size}.
* Index entries with a total serialized size of index samples up to
* {@code column_index_cache_size} will be held in an array.
* Index entries exceeding that value will always be accessed from
* disk.
* </p>
* <p>
* This results in these classes:
* </p>
* <ul>
* <li>{@link RowIndexEntry} just stores the offset in the data file.</li>
* <li>{@link IndexedEntry} is for index entries with index samples
* and used for both current and legacy sstables, which do not exceed
* {@link org.apache.cassandra.config.Config#column_index_cache_size}.</li>
* <li>{@link ShallowIndexedEntry} is for index entries with index samples
* that exceed {@link org.apache.cassandra.config.Config#column_index_cache_size}
* for sstables with an offset table to the index samples.</li>
* </ul>
* <p>
* Since access to index samples on disk (obviously) requires some file
* reader, that functionality is encapsulated in implementations of
* {@link IndexInfoRetriever}. There is an implementation to access
* index samples of legacy sstables (without the offsets table),
* an implementation of access sstables with an offsets table.
* </p>
* <p>
* Until now (Cassandra 3.x), we still support reading from <i>legacy</i> sstables -
* i.e. sstables created by Cassandra < 3.0 (see {@link org.apache.cassandra.io.sstable.format.big.BigFormat}.
* </p>
*
*/
public class RowIndexEntry extends AbstractRowIndexEntry
{
private static final BigFormat FORMAT = BigFormat.getInstance();
private static final long EMPTY_SIZE = ObjectSizes.measure(new RowIndexEntry(0));
// constants for type of row-index-entry as serialized for saved-cache
static final int CACHE_NOT_INDEXED = 0;
static final int CACHE_INDEXED = 1;
static final int CACHE_INDEXED_SHALLOW = 2;
static final Histogram indexEntrySizeHistogram;
static final Histogram indexInfoCountHistogram;
static final Histogram indexInfoGetsHistogram;
static final Histogram indexInfoReadsHistogram;
static
{
MetricNameFactory factory = new DefaultNameFactory("Index", "RowIndexEntry");
indexEntrySizeHistogram = Metrics.histogram(factory.createMetricName("IndexedEntrySize"), false);
indexInfoCountHistogram = Metrics.histogram(factory.createMetricName("IndexInfoCount"), false);
indexInfoGetsHistogram = Metrics.histogram(factory.createMetricName("IndexInfoGets"), false);
indexInfoReadsHistogram = Metrics.histogram(factory.createMetricName("IndexInfoReads"), false);
}
public RowIndexEntry(long position)
{
super(position);
}
/**
* @return true if this index entry contains the row-level tombstone and column summary. Otherwise,
* caller should fetch these from the row header.
*/
@Override
public boolean isIndexed()
{
return blockCount() > 1;
}
public boolean indexOnHeap()
{
return false;
}
@Override
public DeletionTime deletionTime()
{
throw new UnsupportedOperationException();
}
@Override
public int blockCount()
{
return 0;
}
@Override
public BigFormat getSSTableFormat()
{
return FORMAT;
}
@Override
public long unsharedHeapSize()
{
return EMPTY_SIZE;
}
/**
* @param dataFilePosition position of the partition in the {@link Components.Types#DATA} file
* @param indexFilePosition position in the {@link Components.Types#PRIMARY_INDEX} of the {@link RowIndexEntry}
* @param deletionTime deletion time of {@link RowIndexEntry}
* @param headerLength deletion time of {@link RowIndexEntry}
* @param columnIndexCount number of {@link IndexInfo} entries in the {@link RowIndexEntry}
* @param indexedPartSize serialized size of all serialized {@link IndexInfo} objects and their offsets
* @param indexSamples list with IndexInfo offsets (if total serialized size is less than {@link org.apache.cassandra.config.Config#column_index_cache_size}
* @param offsets offsets of IndexInfo offsets
* @param idxInfoSerializer the {@link IndexInfo} serializer
*/
public static RowIndexEntry create(long dataFilePosition, long indexFilePosition,
DeletionTime deletionTime, long headerLength, int columnIndexCount,
int indexedPartSize,
List<IndexInfo> indexSamples, int[] offsets,
ISerializer<IndexInfo> idxInfoSerializer,
Version version)
{
// If the "partition building code" in BigTableWriter.append() via ColumnIndex returns a list
// of IndexInfo objects, which is the case if the serialized size is less than
// Config.column_index_cache_size, AND we have more than one IndexInfo object, we
// construct an IndexedEntry object. (note: indexSamples.size() and columnIndexCount have the same meaning)
if (indexSamples != null && indexSamples.size() > 1)
return new IndexedEntry(dataFilePosition, deletionTime, headerLength,
indexSamples.toArray(new IndexInfo[indexSamples.size()]), offsets,
indexedPartSize, idxInfoSerializer, version);
// Here we have to decide whether we have serialized IndexInfo objects that exceeds
// Config.column_index_cache_size (not exceeding case covered above).
// Such a "big" indexed-entry is represented as a shallow one.
if (columnIndexCount > 1)
return new ShallowIndexedEntry(dataFilePosition, indexFilePosition,
deletionTime, headerLength, columnIndexCount,
indexedPartSize, idxInfoSerializer, version);
// Last case is that there are no index samples.
return new RowIndexEntry(dataFilePosition);
}
public IndexInfoRetriever openWithIndex(FileHandle indexFile)
{
return null;
}
public interface IndexSerializer
{
void serialize(RowIndexEntry rie, DataOutputPlus out, ByteBuffer indexInfo) throws IOException;
RowIndexEntry deserialize(DataInputPlus in, long indexFilePosition) throws IOException;
default RowIndexEntry deserialize(RandomAccessReader reader) throws IOException
{
return deserialize(reader, reader.getFilePointer());
}
default RowIndexEntry deserialize(FileDataInput input) throws IOException
{
return deserialize(input, input.getFilePointer());
}
void serializeForCache(RowIndexEntry rie, DataOutputPlus out) throws IOException;
RowIndexEntry deserializeForCache(DataInputPlus in) throws IOException;
long deserializePositionAndSkip(DataInputPlus in) throws IOException;
ISerializer indexInfoSerializer();
}
public static final class Serializer implements IndexSerializer
{
private final IndexInfo.Serializer idxInfoSerializer;
private final Version version;
private final TableMetrics tableMetrics;
public Serializer(Version version, SerializationHeader header, TableMetrics tableMetrics)
{
this.idxInfoSerializer = IndexInfo.serializer(version, header);
this.version = version;
this.tableMetrics = tableMetrics;
}
@Override
public IndexInfo.Serializer indexInfoSerializer()
{
return idxInfoSerializer;
}
@Override
public void serialize(RowIndexEntry rie, DataOutputPlus out, ByteBuffer indexInfo) throws IOException
{
rie.serialize(out, indexInfo);
}
@Override
public void serializeForCache(RowIndexEntry rie, DataOutputPlus out) throws IOException
{
rie.serializeForCache(out);
}
@Override
public RowIndexEntry deserializeForCache(DataInputPlus in) throws IOException
{
long position = in.readUnsignedVInt();
switch (in.readByte())
{
case CACHE_NOT_INDEXED:
return new RowIndexEntry(position);
case CACHE_INDEXED:
return new IndexedEntry(position, in, idxInfoSerializer, version);
case CACHE_INDEXED_SHALLOW:
return new ShallowIndexedEntry(position, in, idxInfoSerializer, version);
default:
throw new AssertionError();
}
}
public static void skipForCache(DataInputPlus in, Version version) throws IOException
{
in.readUnsignedVInt();
switch (in.readByte())
{
case CACHE_NOT_INDEXED:
break;
case CACHE_INDEXED:
IndexedEntry.skipForCache(in, version);
break;
case CACHE_INDEXED_SHALLOW:
ShallowIndexedEntry.skipForCache(in, version);
break;
default:
assert false;
}
}
@Override
public RowIndexEntry deserialize(DataInputPlus in, long indexFilePosition) throws IOException
{
long position = in.readUnsignedVInt();
int size = in.readUnsignedVInt32();
if (size == 0)
{
return new RowIndexEntry(position);
}
else
{
long headerLength = in.readUnsignedVInt();
DeletionTime deletionTime = DeletionTime.getSerializer(version).deserialize(in);
int columnsIndexCount = in.readUnsignedVInt32();
checkSize(columnsIndexCount, size);
int indexedPartSize = size - serializedSize(deletionTime, headerLength, columnsIndexCount, version);
if (size <= DatabaseDescriptor.getColumnIndexCacheSize())
{
return new IndexedEntry(position, in, deletionTime, headerLength, columnsIndexCount,
idxInfoSerializer, indexedPartSize, version);
}
else
{
in.skipBytes(indexedPartSize);
return new ShallowIndexedEntry(position,
indexFilePosition,
deletionTime, headerLength, columnsIndexCount,
indexedPartSize, idxInfoSerializer, version);
}
}
}
private void checkSize(int entries, int bytes)
{
ReadCommand command = ReadCommand.getCommand();
if (command == null || SchemaConstants.isSystemKeyspace(command.metadata().keyspace) || !DatabaseDescriptor.getReadThresholdsEnabled())
return;
DataStorageSpec.LongBytesBound warnThreshold = DatabaseDescriptor.getRowIndexReadSizeWarnThreshold();
DataStorageSpec.LongBytesBound failThreshold = DatabaseDescriptor.getRowIndexReadSizeFailThreshold();
if (warnThreshold == null && failThreshold == null)
return;
long estimatedMemory = estimateMaterializedIndexSize(entries, bytes);
if (tableMetrics != null)
tableMetrics.rowIndexSize.update(estimatedMemory);
if (failThreshold != null && estimatedMemory > failThreshold.toBytes())
{
String msg = String.format("Query %s attempted to access a large RowIndexEntry estimated to be %d bytes " +
"in-memory (total entries: %d, total bytes: %d) but the max allowed is %s;" +
" query aborted (see row_index_read_size_fail_threshold)",
command.toCQLString(), estimatedMemory, entries, bytes, failThreshold);
MessageParams.remove(ParamType.ROW_INDEX_READ_SIZE_WARN);
MessageParams.add(ParamType.ROW_INDEX_READ_SIZE_FAIL, estimatedMemory);
throw new RowIndexEntryReadSizeTooLargeException(msg);
}
else if (warnThreshold != null && estimatedMemory > warnThreshold.toBytes())
{
// use addIfLarger rather than add as a previous partition may be larger than this one
Long current = MessageParams.get(ParamType.ROW_INDEX_READ_SIZE_WARN);
if (current == null || current.compareTo(estimatedMemory) < 0)
MessageParams.add(ParamType.ROW_INDEX_READ_SIZE_WARN, estimatedMemory);
}
}
private static long estimateMaterializedIndexSize(int entries, int bytes)
{
long overhead = IndexInfo.EMPTY_SIZE
+ ArrayClustering.EMPTY_SIZE
+ DeletionTime.EMPTY_SIZE;
return (overhead * entries) + bytes;
}
@Override
public long deserializePositionAndSkip(DataInputPlus in) throws IOException
{
long position = in.readUnsignedVInt();
int size = in.readUnsignedVInt32();
if (size > 0)
in.skipBytesFully(size);
return position;
}
/**
* Reads only the data 'position' of the index entry and returns it. Note that this left 'in' in the middle
* of reading an entry, so this is only useful if you know what you are doing and in most case 'deserialize'
* should be used instead.
*/
public static long readPosition(DataInputPlus in) throws IOException
{
return in.readUnsignedVInt();
}
public static void skip(DataInputPlus in, Version version) throws IOException
{
readPosition(in);
skipPromotedIndex(in);
}
private static void skipPromotedIndex(DataInputPlus in) throws IOException
{
int size = in.readUnsignedVInt32();
if (size <= 0)
return;
in.skipBytesFully(size);
}
}
private static int serializedSize(DeletionTime deletionTime, long headerLength, int columnIndexCount, Version version)
{
int dlt = (int) DeletionTime.getSerializer(version).serializedSize(deletionTime);
return TypeSizes.sizeofUnsignedVInt(headerLength)
+ dlt
+ TypeSizes.sizeofUnsignedVInt(columnIndexCount);
}
public void serialize(DataOutputPlus out, ByteBuffer indexInfo) throws IOException
{
out.writeUnsignedVInt(position);
out.writeUnsignedVInt32(0);
}
public void serializeForCache(DataOutputPlus out) throws IOException
{
out.writeUnsignedVInt(position);
out.writeByte(CACHE_NOT_INDEXED);
}
/**
* An entry in the row index for a row whose columns are indexed - used for both legacy and current formats.
*/
private static final class IndexedEntry extends RowIndexEntry
{
private static final long BASE_SIZE;
static
{
BASE_SIZE = ObjectSizes.measure(new IndexedEntry(0, DeletionTime.LIVE, 0, null, null, 0, null, BigFormat.getInstance().getLatestVersion()));
}
private final DeletionTime deletionTime;
private final long headerLength;
private final IndexInfo[] columnsIndex;
private final int[] offsets;
private final int indexedPartSize;
@Unmetered
private final ISerializer<IndexInfo> idxInfoSerializer;
private final Version version;
private IndexedEntry(long dataFilePosition, DeletionTime deletionTime, long headerLength,
IndexInfo[] columnsIndex, int[] offsets,
int indexedPartSize, ISerializer<IndexInfo> idxInfoSerializer, Version version)
{
super(dataFilePosition);
this.headerLength = headerLength;
this.deletionTime = deletionTime;
this.columnsIndex = columnsIndex;
this.offsets = offsets;
this.indexedPartSize = indexedPartSize;
this.idxInfoSerializer = idxInfoSerializer;
this.version = version;
}
private IndexedEntry(long dataFilePosition, DataInputPlus in,
DeletionTime deletionTime, long headerLength, int columnIndexCount,
IndexInfo.Serializer idxInfoSerializer, int indexedPartSize, Version version) throws IOException
{
super(dataFilePosition);
this.headerLength = headerLength;
this.deletionTime = deletionTime;
int columnsIndexCount = columnIndexCount;
this.columnsIndex = new IndexInfo[columnsIndexCount];
for (int i = 0; i < columnsIndexCount; i++)
this.columnsIndex[i] = idxInfoSerializer.deserialize(in);
this.offsets = new int[this.columnsIndex.length];
for (int i = 0; i < offsets.length; i++)
offsets[i] = in.readInt();
this.indexedPartSize = indexedPartSize;
this.idxInfoSerializer = idxInfoSerializer;
this.version = version;
}
/**
* Constructor called from {@link Serializer#deserializeForCache(org.apache.cassandra.io.util.DataInputPlus)}.
*/
private IndexedEntry(long dataFilePosition, DataInputPlus in, ISerializer<IndexInfo> idxInfoSerializer, Version version) throws IOException
{
super(dataFilePosition);
this.headerLength = in.readUnsignedVInt();
this.version = version;
this.deletionTime = DeletionTime.getSerializer(version).deserialize(in);
int columnsIndexCount = in.readUnsignedVInt32();
TrackedDataInputPlus trackedIn = new TrackedDataInputPlus(in);
this.columnsIndex = new IndexInfo[columnsIndexCount];
for (int i = 0; i < columnsIndexCount; i++)
this.columnsIndex[i] = idxInfoSerializer.deserialize(trackedIn);
this.offsets = null;
this.indexedPartSize = (int) trackedIn.getBytesRead();
this.idxInfoSerializer = idxInfoSerializer;
}
@Override
public boolean indexOnHeap()
{
return true;
}
@Override
public int blockCount()
{
return columnsIndex.length;
}
@Override
public DeletionTime deletionTime()
{
return deletionTime;
}
@Override
public IndexInfoRetriever openWithIndex(FileHandle indexFile)
{
indexEntrySizeHistogram.update(serializedSize(deletionTime, headerLength, columnsIndex.length, version) + indexedPartSize);
indexInfoCountHistogram.update(columnsIndex.length);
return new IndexInfoRetriever()
{
private int retrievals;
@Override
public IndexInfo columnsIndex(int index)
{
retrievals++;
return columnsIndex[index];
}
@Override
public void close()
{
indexInfoGetsHistogram.update(retrievals);
}
};
}
@Override
public long unsharedHeapSize()
{
long entrySize = 0;
for (IndexInfo idx : columnsIndex)
entrySize += idx.unsharedHeapSize();
return BASE_SIZE
+ entrySize
+ ObjectSizes.sizeOfReferenceArray(columnsIndex.length);
}
@Override
public void serialize(DataOutputPlus out, ByteBuffer indexInfo) throws IOException
{
assert indexedPartSize != Integer.MIN_VALUE;
out.writeUnsignedVInt(position);
out.writeUnsignedVInt32(serializedSize(deletionTime, headerLength, columnsIndex.length, version) + indexedPartSize);
out.writeUnsignedVInt(headerLength);
DeletionTime.getSerializer(version).serialize(deletionTime, out);
out.writeUnsignedVInt32(columnsIndex.length);
for (IndexInfo info : columnsIndex)
idxInfoSerializer.serialize(info, out);
for (int offset : offsets)
out.writeInt(offset);
}
@Override
public void serializeForCache(DataOutputPlus out) throws IOException
{
out.writeUnsignedVInt(position);
out.writeByte(CACHE_INDEXED);
out.writeUnsignedVInt(headerLength);
DeletionTime.getSerializer(version).serialize(deletionTime, out);
out.writeUnsignedVInt32(blockCount());
for (IndexInfo indexInfo : columnsIndex)
idxInfoSerializer.serialize(indexInfo, out);
}
static void skipForCache(DataInputPlus in, Version version) throws IOException
{
in.readUnsignedVInt();
DeletionTime.getSerializer(version).skip(in);
in.readUnsignedVInt();
in.readUnsignedVInt();
}
}
/**
* An entry in the row index for a row whose columns are indexed and the {@link IndexInfo} objects
* are not read into the key cache.
*/
private static final class ShallowIndexedEntry extends RowIndexEntry
{
private static final long BASE_SIZE;
static
{
BASE_SIZE = ObjectSizes.measure(new ShallowIndexedEntry(0, 0, DeletionTime.LIVE, 0, 10, 0, null, BigFormat.getInstance().getLatestVersion()));
}
private final long indexFilePosition;
private final DeletionTime deletionTime;
private final long headerLength;
private final int columnsIndexCount;
private final int indexedPartSize;
private final int offsetsOffset;
@Unmetered
private final ISerializer<IndexInfo> idxInfoSerializer;
private final int fieldsSerializedSize;
private final Version version;
/**
* See {@link #create(long, long, DeletionTime, long, int, int, List, int[], ISerializer, Version)} for a description
* of the parameters.
*/
private ShallowIndexedEntry(long dataFilePosition, long indexFilePosition,
DeletionTime deletionTime, long headerLength, int columnIndexCount,
int indexedPartSize, ISerializer<IndexInfo> idxInfoSerializer, Version version)
{
super(dataFilePosition);
assert columnIndexCount > 1;
this.indexFilePosition = indexFilePosition;
this.headerLength = headerLength;
this.deletionTime = deletionTime;
this.columnsIndexCount = columnIndexCount;
this.indexedPartSize = indexedPartSize;
this.idxInfoSerializer = idxInfoSerializer;
this.version = version;
this.fieldsSerializedSize = serializedSize(deletionTime, headerLength, columnIndexCount, this.version);
this.offsetsOffset = indexedPartSize + fieldsSerializedSize - columnsIndexCount * TypeSizes.INT_SIZE;
}
/**
* Constructor for key-cache deserialization
*/
private ShallowIndexedEntry(long dataFilePosition, DataInputPlus in, IndexInfo.Serializer idxInfoSerializer, Version version) throws IOException
{
super(dataFilePosition);
this.indexFilePosition = in.readUnsignedVInt();
this.headerLength = in.readUnsignedVInt();
this.version = version;
this.deletionTime = DeletionTime.getSerializer(version).deserialize(in);
this.columnsIndexCount = in.readUnsignedVInt32();
this.indexedPartSize = in.readUnsignedVInt32();
this.idxInfoSerializer = idxInfoSerializer;
this.fieldsSerializedSize = serializedSize(deletionTime, headerLength, columnsIndexCount, this.version);
this.offsetsOffset = indexedPartSize + fieldsSerializedSize - columnsIndexCount * TypeSizes.INT_SIZE;
}
@Override
public int blockCount()
{
return columnsIndexCount;
}
@Override
public DeletionTime deletionTime()
{
return deletionTime;
}
@Override
public IndexInfoRetriever openWithIndex(FileHandle indexFile)
{
indexEntrySizeHistogram.update(indexedPartSize + fieldsSerializedSize);
indexInfoCountHistogram.update(columnsIndexCount);
return new ShallowInfoRetriever(indexFilePosition +
VIntCoding.computeUnsignedVIntSize(position) +
VIntCoding.computeUnsignedVIntSize(indexedPartSize + fieldsSerializedSize) +
fieldsSerializedSize,
offsetsOffset - fieldsSerializedSize,
indexFile.createReader(), idxInfoSerializer);
}
@Override
public long unsharedHeapSize()
{
return BASE_SIZE;
}
@Override
public void serialize(DataOutputPlus out, ByteBuffer indexInfo) throws IOException
{
out.writeUnsignedVInt(position);
out.writeUnsignedVInt32(fieldsSerializedSize + indexInfo.limit());
out.writeUnsignedVInt(headerLength);
DeletionTime.getSerializer(version).serialize(deletionTime, out);
out.writeUnsignedVInt32(columnsIndexCount);
out.write(indexInfo);
}
@Override
public void serializeForCache(DataOutputPlus out) throws IOException
{
out.writeUnsignedVInt(position);
out.writeByte(CACHE_INDEXED_SHALLOW);
out.writeUnsignedVInt(indexFilePosition);
out.writeUnsignedVInt(headerLength);
DeletionTime.getSerializer(version).serialize(deletionTime, out);
out.writeUnsignedVInt32(columnsIndexCount);
out.writeUnsignedVInt32(indexedPartSize);
}
static void skipForCache(DataInputPlus in, Version version) throws IOException
{
in.readUnsignedVInt();
in.readUnsignedVInt();
DeletionTime.getSerializer(version).skip(in);
in.readUnsignedVInt();
in.readUnsignedVInt();
}
}
private static final class ShallowInfoRetriever extends FileIndexInfoRetriever
{
private final int offsetsOffset;
private ShallowInfoRetriever(long indexInfoFilePosition, int offsetsOffset,
FileDataInput indexReader, ISerializer<IndexInfo> idxInfoSerializer)
{
super(indexInfoFilePosition, indexReader, idxInfoSerializer);
this.offsetsOffset = offsetsOffset;
}
@Override
IndexInfo fetchIndex(int index) throws IOException
{
// seek to position in "offsets to IndexInfo" table
indexReader.seek(indexInfoFilePosition + offsetsOffset + index * TypeSizes.INT_SIZE);
// read offset of IndexInfo
int indexInfoPos = indexReader.readInt();
// seek to posision of IndexInfo
indexReader.seek(indexInfoFilePosition + indexInfoPos);
// finally, deserialize IndexInfo
return idxInfoSerializer.deserialize(indexReader);
}
}
/**
* Base class to access {@link IndexInfo} objects.
*/
public interface IndexInfoRetriever extends AutoCloseable
{
IndexInfo columnsIndex(int index) throws IOException;
void close() throws IOException;
}
/**
* Base class to access {@link IndexInfo} objects on disk that keeps already
* read {@link IndexInfo} on heap.
*/
private abstract static class FileIndexInfoRetriever implements IndexInfoRetriever
{
final long indexInfoFilePosition;
final ISerializer<IndexInfo> idxInfoSerializer;
final FileDataInput indexReader;
int retrievals;
/**
*
* @param indexInfoFilePosition offset of first serialized {@link IndexInfo} object
* @param indexReader file data input to access the index file, closed by this instance
* @param idxInfoSerializer the index serializer to deserialize {@link IndexInfo} objects
*/
FileIndexInfoRetriever(long indexInfoFilePosition, FileDataInput indexReader, ISerializer<IndexInfo> idxInfoSerializer)
{
this.indexInfoFilePosition = indexInfoFilePosition;
this.idxInfoSerializer = idxInfoSerializer;
this.indexReader = indexReader;
}
@Override
public final IndexInfo columnsIndex(int index) throws IOException
{
retrievals++;
return fetchIndex(index);
}
abstract IndexInfo fetchIndex(int index) throws IOException;
@Override
public void close() throws IOException
{
indexReader.close();
indexInfoGetsHistogram.update(retrievals);
indexInfoReadsHistogram.update(retrievals);
}
}
public static class RowIndexEntryReadSizeTooLargeException extends RejectException
{
public RowIndexEntryReadSizeTooLargeException(String message)
{
super(message);
}
}
}