You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sn...@apache.org on 2016/04/21 16:49:16 UTC
[3/3] cassandra git commit: Support large partitions on the 3.0
sstable format
Support large partitions on the 3.0 sstable format
patch by Robert Stupp; reviewed by T Jake Luciani for CASSANDRA-11206
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ef5bbedd
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ef5bbedd
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ef5bbedd
Branch: refs/heads/trunk
Commit: ef5bbedd687d75923e9a20fde9d2f78b4535241d
Parents: ae063e8
Author: Robert Stupp <sn...@snazy.de>
Authored: Thu Apr 21 16:48:26 2016 +0200
Committer: Robert Stupp <sn...@snazy.de>
Committed: Thu Apr 21 16:48:26 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
NEWS.txt | 3 +
conf/cassandra.yaml | 8 +
.../apache/cassandra/cache/AutoSavingCache.java | 4 +-
.../org/apache/cassandra/config/Config.java | 1 +
.../cassandra/config/DatabaseDescriptor.java | 11 +
.../org/apache/cassandra/db/Clustering.java | 6 +
.../cassandra/db/ClusteringComparator.java | 2 +-
.../apache/cassandra/db/ClusteringPrefix.java | 29 +
.../org/apache/cassandra/db/ColumnIndex.java | 296 +++---
.../org/apache/cassandra/db/RangeTombstone.java | 10 +
.../org/apache/cassandra/db/RowIndexEntry.java | 1006 +++++++++++++++---
.../cassandra/db/SerializationHeader.java | 20 -
.../org/apache/cassandra/db/Serializers.java | 95 +-
.../columniterator/AbstractSSTableIterator.java | 111 +-
.../db/columniterator/SSTableIterator.java | 15 +-
.../columniterator/SSTableReversedIterator.java | 19 +-
.../cassandra/db/compaction/Scrubber.java | 4 +-
.../cassandra/db/compaction/Verifier.java | 4 +-
.../UnfilteredRowIteratorWithLowerBound.java | 29 +-
.../cassandra/index/sasi/SASIIndexBuilder.java | 2 +-
.../org/apache/cassandra/io/ISerializer.java | 5 +
.../cassandra/io/sstable/IndexHelper.java | 192 ----
.../apache/cassandra/io/sstable/IndexInfo.java | 178 ++++
.../io/sstable/format/SSTableFormat.java | 6 -
.../io/sstable/format/SSTableReader.java | 10 +-
.../io/sstable/format/big/BigTableReader.java | 8 +-
.../io/sstable/format/big/BigTableScanner.java | 6 +-
.../io/sstable/format/big/BigTableWriter.java | 32 +-
.../apache/cassandra/service/CacheService.java | 12 +-
.../cassandra/cache/AutoSavingCacheTest.java | 15 +
.../apache/cassandra/cql3/KeyCacheCqlTest.java | 76 +-
.../apache/cassandra/cql3/PagingQueryTest.java | 111 ++
.../cql3/TombstonesWithIndexedSSTableTest.java | 16 +-
.../org/apache/cassandra/db/KeyCacheTest.java | 151 ++-
.../org/apache/cassandra/db/KeyspaceTest.java | 2 +-
.../apache/cassandra/db/RowIndexEntryTest.java | 768 +++++++++++--
.../cassandra/io/sstable/IndexHelperTest.java | 78 --
.../io/sstable/LargePartitionsTest.java | 219 ++++
39 files changed, 2775 insertions(+), 786 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 1b94b2d..a0c7df6 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.6
+ * Support large partitions on the 3.0 sstable format (CASSANDRA-11206)
* JSON datetime formatting needs timezone (CASSANDRA-11137)
* Add support to rebuild from specific range (CASSANDRA-10409)
* Optimize the overlapping lookup by calculating all the
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/NEWS.txt
----------------------------------------------------------------------
diff --git a/NEWS.txt b/NEWS.txt
index e073592..a177d37 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -18,6 +18,9 @@ using the provided 'sstableupgrade' tool.
New features
------------
+ - Key cache will only hold indexed entries up to the size configured by
+ column_index_cache_size_in_kb in cassandra.yaml in memory. Larger indexed entries
+ will never go into memory. See CASSANDRA-11206 for more details.
- For tables having a default_time_to_live specifying a TTL of 0 will remove the TTL
from the inserted or updated values.
- Startup is now aborted if corrupted transaction log files are found. The details
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index f9be453..582859c 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -659,6 +659,14 @@ auto_snapshot: true
# rows (as part of the key cache), so a larger granularity means
# you can cache more hot rows
column_index_size_in_kb: 64
+# Per sstable indexed key cache entries (the collation index in memory
+# mentioned above) exceeding this size will not be held on heap.
+# This means that only partition information is held on heap and the
+# index entries are read from disk.
+#
+# Note that this size refers to the size of the
+# serialized index information and not the size of the partition.
+column_index_cache_size_in_kb: 2
# Number of simultaneous compactions to allow, NOT including
# validation "compactions" for anti-entropy repair. Simultaneous
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/cache/AutoSavingCache.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cache/AutoSavingCache.java b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
index e39dcf1..2921818 100644
--- a/src/java/org/apache/cassandra/cache/AutoSavingCache.java
+++ b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
@@ -77,8 +77,10 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
* a minor version letter.
*
* Sticking with "d" is fine for 3.0 since it has never been released or used by another version
+ *
+ * "e" introduced with CASSANDRA-11206, omits IndexInfo from key-cache, stores offset into index-file
*/
- private static final String CURRENT_VERSION = "d";
+ private static final String CURRENT_VERSION = "e";
private static volatile IStreamFactory streamFactory = new IStreamFactory()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/config/Config.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index b5dea06..809966d 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -162,6 +162,7 @@ public class Config
/* if the size of columns or super-columns are more than this, indexing will kick in */
public Integer column_index_size_in_kb = 64;
+ public Integer column_index_cache_size_in_kb = 2;
public volatile int batch_size_warn_threshold_in_kb = 5;
public volatile int batch_size_fail_threshold_in_kb = 50;
public Integer unlogged_batch_across_partitions_warn_threshold = 10;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 37b9200..b98d103 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -962,6 +962,17 @@ public class DatabaseDescriptor
conf.column_index_size_in_kb = val;
}
+ public static int getColumnIndexCacheSize()
+ {
+ return conf.column_index_cache_size_in_kb * 1024;
+ }
+
+ @VisibleForTesting
+ public static void setColumnIndexCacheSize(int val)
+ {
+ conf.column_index_cache_size_in_kb = val;
+ }
+
public static int getBatchSizeWarnThreshold()
{
return conf.batch_size_warn_threshold_in_kb * 1024;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/db/Clustering.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Clustering.java b/src/java/org/apache/cassandra/db/Clustering.java
index f5ffae4..fa38ce1 100644
--- a/src/java/org/apache/cassandra/db/Clustering.java
+++ b/src/java/org/apache/cassandra/db/Clustering.java
@@ -149,6 +149,12 @@ public interface Clustering extends ClusteringPrefix
return ClusteringPrefix.serializer.valuesWithoutSizeSerializedSize(clustering, version, types);
}
+ public void skip(DataInputPlus in, int version, List<AbstractType<?>> types) throws IOException
+ {
+ if (!types.isEmpty())
+ ClusteringPrefix.serializer.skipValuesWithoutSize(in, types.size(), version, types);
+ }
+
public Clustering deserialize(DataInputPlus in, int version, List<AbstractType<?>> types) throws IOException
{
if (types.isEmpty())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/db/ClusteringComparator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ClusteringComparator.java b/src/java/org/apache/cassandra/db/ClusteringComparator.java
index f5f6ae8..3030b5a 100644
--- a/src/java/org/apache/cassandra/db/ClusteringComparator.java
+++ b/src/java/org/apache/cassandra/db/ClusteringComparator.java
@@ -29,7 +29,7 @@ import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.serializers.MarshalException;
-import static org.apache.cassandra.io.sstable.IndexHelper.IndexInfo;
+import org.apache.cassandra.io.sstable.IndexInfo;
/**
* A comparator of clustering prefixes (or more generally of {@link Clusterable}}.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/db/ClusteringPrefix.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ClusteringPrefix.java b/src/java/org/apache/cassandra/db/ClusteringPrefix.java
index 4854517..df0befc 100644
--- a/src/java/org/apache/cassandra/db/ClusteringPrefix.java
+++ b/src/java/org/apache/cassandra/db/ClusteringPrefix.java
@@ -263,6 +263,17 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
}
}
+ public void skip(DataInputPlus in, int version, List<AbstractType<?>> types) throws IOException
+ {
+ Kind kind = Kind.values()[in.readByte()];
+ // We shouldn't serialize static clusterings
+ assert kind != Kind.STATIC_CLUSTERING;
+ if (kind == Kind.CLUSTERING)
+ Clustering.serializer.skip(in, version, types);
+ else
+ RangeTombstone.Bound.serializer.skipValues(in, kind, version, types);
+ }
+
public ClusteringPrefix deserialize(DataInputPlus in, int version, List<AbstractType<?>> types) throws IOException
{
Kind kind = Kind.values()[in.readByte()];
@@ -350,6 +361,24 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
return values;
}
+ void skipValuesWithoutSize(DataInputPlus in, int size, int version, List<AbstractType<?>> types) throws IOException
+ {
+ // Callers of this method should handle the case where size = 0 (in all case we want to return a special value anyway).
+ assert size > 0;
+ int offset = 0;
+ while (offset < size)
+ {
+ long header = in.readUnsignedVInt();
+ int limit = Math.min(size, offset + 32);
+ while (offset < limit)
+ {
+ if (!isNull(header, offset) && !isEmpty(header, offset))
+ types.get(offset).skipValue(in);
+ offset++;
+ }
+ }
+ }
+
/**
* Whatever the type of a given clustering column is, its value can always be either empty or null. So we at least need to distinguish those
* 2 values, and because we want to be able to store fixed width values without appending their (fixed) size first, we need a way to encode
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/db/ColumnIndex.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnIndex.java b/src/java/org/apache/cassandra/db/ColumnIndex.java
index 1942052..4dcceff 100644
--- a/src/java/org/apache/cassandra/db/ColumnIndex.java
+++ b/src/java/org/apache/cassandra/db/ColumnIndex.java
@@ -15,184 +15,234 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.cassandra.db;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.*;
-import com.google.common.annotations.VisibleForTesting;
+import com.google.common.primitives.Ints;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.rows.*;
-import org.apache.cassandra.io.sstable.IndexHelper;
+import org.apache.cassandra.io.ISerializer;
+import org.apache.cassandra.io.sstable.IndexInfo;
import org.apache.cassandra.io.sstable.format.SSTableFlushObserver;
import org.apache.cassandra.io.sstable.format.Version;
+import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.utils.ByteBufferUtil;
+/**
+ * Column index builder used by {@link org.apache.cassandra.io.sstable.format.big.BigTableWriter}.
+ * For index entries that exceed {@link org.apache.cassandra.config.Config#column_index_cache_size_in_kb},
+ * this uses the serialization logic as in {@link RowIndexEntry}.
+ */
public class ColumnIndex
{
- public final long partitionHeaderLength;
- public final List<IndexHelper.IndexInfo> columnsIndex;
- private static final ColumnIndex EMPTY = new ColumnIndex(-1, Collections.<IndexHelper.IndexInfo>emptyList());
+ // used, if the row-index-entry reaches config column_index_cache_size_in_kb
+ private DataOutputBuffer buffer;
+ // used to track the size of the serialized size of row-index-entry (unused for buffer)
+ private int indexSamplesSerializedSize;
+ // used, until the row-index-entry reaches config column_index_cache_size_in_kb
+ public List<IndexInfo> indexSamples = new ArrayList<>();
- private ColumnIndex(long partitionHeaderLength, List<IndexHelper.IndexInfo> columnsIndex)
- {
- assert columnsIndex != null;
+ public int columnIndexCount;
+ private int[] indexOffsets;
- this.partitionHeaderLength = partitionHeaderLength;
- this.columnsIndex = columnsIndex;
- }
+ private final SerializationHeader header;
+ private final int version;
+ private final SequentialWriter writer;
+ private final long initialPosition;
+ private final ISerializer<IndexInfo> idxSerializer;
+ public long headerLength = -1;
- public static ColumnIndex writeAndBuildIndex(UnfilteredRowIterator iterator,
- SequentialWriter output,
- SerializationHeader header,
- Collection<SSTableFlushObserver> observers,
- Version version) throws IOException
- {
- assert !iterator.isEmpty() && version.storeRows();
+ private long startPosition = -1;
- Builder builder = new Builder(iterator, output, header, observers, version.correspondingMessagingVersion());
- return builder.build();
- }
+ private int written;
+ private long previousRowStart;
+
+ private ClusteringPrefix firstClustering;
+ private ClusteringPrefix lastClustering;
- @VisibleForTesting
- public static ColumnIndex nothing()
+ private DeletionTime openMarker;
+
+ private final Collection<SSTableFlushObserver> observers;
+
+ public ColumnIndex(SerializationHeader header,
+ SequentialWriter writer,
+ Version version,
+ Collection<SSTableFlushObserver> observers,
+ ISerializer<IndexInfo> indexInfoSerializer)
{
- return EMPTY;
+ this.header = header;
+ this.idxSerializer = indexInfoSerializer;
+ this.writer = writer;
+ this.version = version.correspondingMessagingVersion();
+ this.observers = observers;
+ this.initialPosition = writer.position();
}
- /**
- * Help to create an index for a column family based on size of columns,
- * and write said columns to disk.
- */
- private static class Builder
+ public void buildRowIndex(UnfilteredRowIterator iterator) throws IOException
{
- private final UnfilteredRowIterator iterator;
- private final SequentialWriter writer;
- private final SerializationHeader header;
- private final int version;
+ writePartitionHeader(iterator);
+ this.headerLength = writer.position() - initialPosition;
- private final List<IndexHelper.IndexInfo> columnsIndex = new ArrayList<>();
- private final long initialPosition;
- private long headerLength = -1;
+ while (iterator.hasNext())
+ add(iterator.next());
- private long startPosition = -1;
+ close();
+ }
- private int written;
- private long previousRowStart;
+ private void writePartitionHeader(UnfilteredRowIterator iterator) throws IOException
+ {
+ ByteBufferUtil.writeWithShortLength(iterator.partitionKey().getKey(), writer);
+ DeletionTime.serializer.serialize(iterator.partitionLevelDeletion(), writer);
+ if (header.hasStatic())
+ {
+ Row staticRow = iterator.staticRow();
- private ClusteringPrefix firstClustering;
- private ClusteringPrefix lastClustering;
+ UnfilteredSerializer.serializer.serializeStaticRow(staticRow, header, writer, version);
+ if (!observers.isEmpty())
+ observers.forEach((o) -> o.nextUnfilteredCluster(staticRow));
+ }
+ }
- private DeletionTime openMarker;
+ private long currentPosition()
+ {
+ return writer.position() - initialPosition;
+ }
- private final Collection<SSTableFlushObserver> observers;
+ public ByteBuffer buffer()
+ {
+ return buffer != null ? buffer.buffer() : null;
+ }
- public Builder(UnfilteredRowIterator iterator,
- SequentialWriter writer,
- SerializationHeader header,
- Collection<SSTableFlushObserver> observers,
- int version)
+ public int[] offsets()
+ {
+ return indexOffsets != null
+ ? Arrays.copyOf(indexOffsets, columnIndexCount)
+ : null;
+ }
+
+ private void addIndexBlock() throws IOException
+ {
+ IndexInfo cIndexInfo = new IndexInfo(firstClustering,
+ lastClustering,
+ startPosition,
+ currentPosition() - startPosition,
+ openMarker);
+
+ // indexOffsets is used for both shallow (ShallowIndexedEntry) and non-shallow IndexedEntry.
+ // For shallow ones, we need it to serialize the offsts in close().
+ // For non-shallow ones, the offsts are passed into IndexedEntry, so we don't have to
+ // calculate the offsets again.
+
+ // indexOffsets contains the offsets of the serialized IndexInfo objects.
+ // I.e. indexOffsets[0] is always 0 so we don't have to deal with a special handling
+ // for index #0 and always subtracting 1 for the index (which could be error-prone).
+ if (indexOffsets == null)
+ indexOffsets = new int[10];
+ else
{
- this.iterator = iterator;
- this.writer = writer;
- this.header = header;
- this.version = version;
- this.observers = observers == null ? Collections.emptyList() : observers;
- this.initialPosition = writer.position();
+ if (columnIndexCount >= indexOffsets.length)
+ indexOffsets = Arrays.copyOf(indexOffsets, indexOffsets.length + 10);
+ indexOffsets[columnIndexCount] =
+ buffer != null
+ ? Ints.checkedCast(buffer.position())
+ : indexSamplesSerializedSize;
}
+ columnIndexCount++;
- private void writePartitionHeader(UnfilteredRowIterator iterator) throws IOException
+ // First, we collect the IndexInfo objects until we reach Config.column_index_cache_size_in_kb in an ArrayList.
+ // When column_index_cache_size_in_kb is reached, we switch to byte-buffer mode.
+ if (buffer == null)
{
- ByteBufferUtil.writeWithShortLength(iterator.partitionKey().getKey(), writer);
- DeletionTime.serializer.serialize(iterator.partitionLevelDeletion(), writer);
- if (header.hasStatic())
+ indexSamplesSerializedSize += idxSerializer.serializedSize(cIndexInfo);
+ if (indexSamplesSerializedSize + columnIndexCount * TypeSizes.sizeof(0) > DatabaseDescriptor.getColumnIndexCacheSize())
{
- Row staticRow = iterator.staticRow();
-
- UnfilteredSerializer.serializer.serializeStaticRow(staticRow, header, writer, version);
- if (!observers.isEmpty())
- observers.forEach((o) -> o.nextUnfilteredCluster(staticRow));
+ buffer = new DataOutputBuffer(DatabaseDescriptor.getColumnIndexCacheSize() * 2);
+ for (IndexInfo indexSample : indexSamples)
+ {
+ idxSerializer.serialize(indexSample, buffer);
+ }
+ indexSamples = null;
+ }
+ else
+ {
+ indexSamples.add(cIndexInfo);
}
}
-
- public ColumnIndex build() throws IOException
+ // don't put an else here...
+ if (buffer != null)
{
- writePartitionHeader(iterator);
- this.headerLength = writer.position() - initialPosition;
-
- while (iterator.hasNext())
- add(iterator.next());
-
- return close();
+ idxSerializer.serialize(cIndexInfo, buffer);
}
- private long currentPosition()
- {
- return writer.position() - initialPosition;
- }
+ firstClustering = null;
+ }
+
+ private void add(Unfiltered unfiltered) throws IOException
+ {
+ long pos = currentPosition();
- private void addIndexBlock()
+ if (firstClustering == null)
{
- IndexHelper.IndexInfo cIndexInfo = new IndexHelper.IndexInfo(firstClustering,
- lastClustering,
- startPosition,
- currentPosition() - startPosition,
- openMarker);
- columnsIndex.add(cIndexInfo);
- firstClustering = null;
+ // Beginning of an index block. Remember the start and position
+ firstClustering = unfiltered.clustering();
+ startPosition = pos;
}
- private void add(Unfiltered unfiltered) throws IOException
- {
- long pos = currentPosition();
+ UnfilteredSerializer.serializer.serialize(unfiltered, header, writer, pos - previousRowStart, version);
- if (firstClustering == null)
- {
- // Beginning of an index block. Remember the start and position
- firstClustering = unfiltered.clustering();
- startPosition = pos;
- }
+ // notify observers about each new row
+ if (!observers.isEmpty())
+ observers.forEach((o) -> o.nextUnfilteredCluster(unfiltered));
- UnfilteredSerializer.serializer.serialize(unfiltered, header, writer, pos - previousRowStart, version);
+ lastClustering = unfiltered.clustering();
+ previousRowStart = pos;
+ ++written;
- // notify observers about each new row
- if (!observers.isEmpty())
- observers.forEach((o) -> o.nextUnfilteredCluster(unfiltered));
-
- lastClustering = unfiltered.clustering();
- previousRowStart = pos;
- ++written;
+ if (unfiltered.kind() == Unfiltered.Kind.RANGE_TOMBSTONE_MARKER)
+ {
+ RangeTombstoneMarker marker = (RangeTombstoneMarker) unfiltered;
+ openMarker = marker.isOpen(false) ? marker.openDeletionTime(false) : null;
+ }
- if (unfiltered.kind() == Unfiltered.Kind.RANGE_TOMBSTONE_MARKER)
- {
- RangeTombstoneMarker marker = (RangeTombstoneMarker)unfiltered;
- openMarker = marker.isOpen(false) ? marker.openDeletionTime(false) : null;
- }
+ // if we hit the column index size that we have to index after, go ahead and index it.
+ if (currentPosition() - startPosition >= DatabaseDescriptor.getColumnIndexSize())
+ addIndexBlock();
+ }
- // if we hit the column index size that we have to index after, go ahead and index it.
- if (currentPosition() - startPosition >= DatabaseDescriptor.getColumnIndexSize())
- addIndexBlock();
+ private void close() throws IOException
+ {
+ UnfilteredSerializer.serializer.writeEndOfPartition(writer);
- }
+ // It's possible we add no rows, just a top level deletion
+ if (written == 0)
+ return;
- private ColumnIndex close() throws IOException
- {
- UnfilteredSerializer.serializer.writeEndOfPartition(writer);
+ // the last column may have fallen on an index boundary already. if not, index it explicitly.
+ if (firstClustering != null)
+ addIndexBlock();
- // It's possible we add no rows, just a top level deletion
- if (written == 0)
- return ColumnIndex.EMPTY;
+ // If we serialize the IndexInfo objects directly in the code above into 'buffer',
+ // we have to write the offsts to these here. The offsets have already been are collected
+ // in indexOffsets[]. buffer is != null, if it exceeds Config.column_index_cache_size_in_kb.
+ // In the other case, when buffer==null, the offsets are serialized in RowIndexEntry.IndexedEntry.serialize().
+ if (buffer != null)
+ RowIndexEntry.Serializer.serializeOffsets(buffer, indexOffsets, columnIndexCount);
- // the last column may have fallen on an index boundary already. if not, index it explicitly.
- if (firstClustering != null)
- addIndexBlock();
+ // we should always have at least one computed index block, but we only write it out if there is more than that.
+ assert columnIndexCount > 0 && headerLength >= 0;
+ }
- // we should always have at least one computed index block, but we only write it out if there is more than that.
- assert columnsIndex.size() > 0 && headerLength >= 0;
- return new ColumnIndex(headerLength, columnsIndex);
- }
+ public int indexInfoSerializedSize()
+ {
+ return buffer != null
+ ? buffer.buffer().limit()
+ : indexSamplesSerializedSize + columnIndexCount * TypeSizes.sizeof(0);
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/db/RangeTombstone.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/RangeTombstone.java b/src/java/org/apache/cassandra/db/RangeTombstone.java
index 8af3b97..29feb46 100644
--- a/src/java/org/apache/cassandra/db/RangeTombstone.java
+++ b/src/java/org/apache/cassandra/db/RangeTombstone.java
@@ -202,6 +202,16 @@ public class RangeTombstone
return deserializeValues(in, kind, version, types);
}
+ public void skipValues(DataInputPlus in, Kind kind, int version,
+ List<AbstractType<?>> types) throws IOException
+ {
+ int size = in.readUnsignedShort();
+ if (size == 0)
+ return;
+
+ ClusteringPrefix.serializer.skipValuesWithoutSize(in, size, version, types);
+ }
+
public RangeTombstone.Bound deserializeValues(DataInputPlus in, Kind kind, int version,
List<AbstractType<?>> types) throws IOException
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/db/RowIndexEntry.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/RowIndexEntry.java b/src/java/org/apache/cassandra/db/RowIndexEntry.java
index 4e2f063..7fda245 100644
--- a/src/java/org/apache/cassandra/db/RowIndexEntry.java
+++ b/src/java/org/apache/cassandra/db/RowIndexEntry.java
@@ -18,50 +18,140 @@
package org.apache.cassandra.db;
import java.io.IOException;
-import java.util.ArrayList;
+import java.nio.ByteBuffer;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
-import com.google.common.primitives.Ints;
-
+import com.codahale.metrics.Histogram;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cache.IMeasurableMemory;
-import org.apache.cassandra.io.sstable.IndexHelper;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.ISerializer;
+import org.apache.cassandra.io.sstable.IndexInfo;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
-import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.io.util.FileDataInput;
+import org.apache.cassandra.io.util.SegmentedFile;
+import org.apache.cassandra.io.util.TrackedDataInputPlus;
+import org.apache.cassandra.metrics.DefaultNameFactory;
+import org.apache.cassandra.metrics.MetricNameFactory;
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) DeletionTime.localDeletionTime
+ * (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_in_kb}.
+ * </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_in_kb}.
+ * Index entries with a total serialized size of index samples up to
+ * {@code column_index_cache_size_in_kb} 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_in_kb}.</li>
+ * <li>{@link ShallowIndexedEntry} is for index entries with index samples
+ * that exceed {@link org.apache.cassandra.config.Config#column_index_cache_size_in_kb}
+ * for sstables with an offset table to the index samples.</li>
+ * <li>{@link LegacyShallowIndexedEntry} is for index entries with index samples
+ * that exceed {@link org.apache.cassandra.config.Config#column_index_cache_size_in_kb}
+ * but for legacy sstables.</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<T> implements IMeasurableMemory
{
private static final long EMPTY_SIZE = ObjectSizes.measure(new RowIndexEntry(0));
- public final long position;
+ // 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;
- public RowIndexEntry(long position)
- {
- this.position = position;
+ static final Histogram indexEntrySizeHistogram;
+ static final Histogram indexInfoCountHistogram;
+ static final Histogram indexInfoGetsHistogram;
+ 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);
}
- protected int promotedSize(IndexHelper.IndexInfo.Serializer idxSerializer)
- {
- return 0;
- }
+ public final long position;
- public static RowIndexEntry<IndexHelper.IndexInfo> create(long position, DeletionTime deletionTime, ColumnIndex index)
+ public RowIndexEntry(long position)
{
- assert index != null;
- assert deletionTime != null;
-
- // we only consider the columns summary when determining whether to create an IndexedEntry,
- // since if there are insufficient columns to be worth indexing we're going to seek to
- // the beginning of the row anyway, so we might as well read the tombstone there as well.
- if (index.columnsIndex.size() > 1)
- return new IndexedEntry(position, deletionTime, index.partitionHeaderLength, index.columnsIndex);
- else
- return new RowIndexEntry<>(position);
+ this.position = position;
}
/**
@@ -70,21 +160,17 @@ public class RowIndexEntry<T> implements IMeasurableMemory
*/
public boolean isIndexed()
{
- return !columnsIndex().isEmpty();
+ return columnsIndexCount() > 1;
}
- public DeletionTime deletionTime()
+ public boolean indexOnHeap()
{
- throw new UnsupportedOperationException();
+ return false;
}
- /**
- * @return the offset to the start of the header information for this row.
- * For some formats this may not be the start of the row.
- */
- public long headerOffset()
+ public DeletionTime deletionTime()
{
- return 0;
+ throw new UnsupportedOperationException();
}
/**
@@ -97,9 +183,9 @@ public class RowIndexEntry<T> implements IMeasurableMemory
throw new UnsupportedOperationException();
}
- public List<T> columnsIndex()
+ public int columnsIndexCount()
{
- return Collections.emptyList();
+ return 0;
}
public long unsharedHeapSize()
@@ -107,129 +193,178 @@ public class RowIndexEntry<T> implements IMeasurableMemory
return EMPTY_SIZE;
}
+ /**
+ * @param dataFilePosition position of the partition in the {@link org.apache.cassandra.io.sstable.Component.Type#DATA} file
+ * @param indexFilePosition position in the {@link org.apache.cassandra.io.sstable.Component.Type#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_in_kb}
+ * @param offsets offsets of IndexInfo offsets
+ * @param idxInfoSerializer the {@link IndexInfo} serializer
+ */
+ public static RowIndexEntry<IndexInfo> create(long dataFilePosition, long indexFilePosition,
+ DeletionTime deletionTime, long headerLength, int columnIndexCount,
+ int indexedPartSize,
+ List<IndexInfo> indexSamples, int[] offsets,
+ ISerializer<IndexInfo> idxInfoSerializer)
+ {
+ // 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_in_kb, 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);
+ // Here we have to decide whether we have serialized IndexInfo objects that exceeds
+ // Config.column_index_cache_size_in_kb (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);
+ // Last case is that there are no index samples.
+ return new RowIndexEntry<>(dataFilePosition);
+ }
+
+ public IndexInfoRetriever openWithIndex(SegmentedFile indexFile)
+ {
+ return null;
+ }
+
public interface IndexSerializer<T>
{
- void serialize(RowIndexEntry<T> rie, DataOutputPlus out) throws IOException;
- RowIndexEntry<T> deserialize(DataInputPlus in) throws IOException;
- int serializedSize(RowIndexEntry<T> rie);
+ void serialize(RowIndexEntry<T> rie, DataOutputPlus out, ByteBuffer indexInfo) throws IOException;
+ RowIndexEntry<T> deserialize(DataInputPlus in, long indexFilePosition) throws IOException;
+ void serializeForCache(RowIndexEntry<T> rie, DataOutputPlus out) throws IOException;
+ RowIndexEntry<T> deserializeForCache(DataInputPlus in) throws IOException;
+
+ long deserializePositionAndSkip(DataInputPlus in) throws IOException;
+
+ ISerializer<T> indexInfoSerializer();
}
- public static class Serializer implements IndexSerializer<IndexHelper.IndexInfo>
+ public static final class Serializer implements IndexSerializer<IndexInfo>
{
- private final IndexHelper.IndexInfo.Serializer idxSerializer;
+ private final IndexInfo.Serializer idxInfoSerializer;
private final Version version;
public Serializer(CFMetaData metadata, Version version, SerializationHeader header)
{
- this.idxSerializer = new IndexHelper.IndexInfo.Serializer(metadata, version, header);
+ this.idxInfoSerializer = metadata.serializers().indexInfoSerializer(version, header);
this.version = version;
}
- public void serialize(RowIndexEntry<IndexHelper.IndexInfo> rie, DataOutputPlus out) throws IOException
+ public IndexInfo.Serializer indexInfoSerializer()
+ {
+ return idxInfoSerializer;
+ }
+
+ public void serialize(RowIndexEntry<IndexInfo> rie, DataOutputPlus out, ByteBuffer indexInfo) throws IOException
{
assert version.storeRows() : "We read old index files but we should never write them";
- out.writeUnsignedVInt(rie.position);
- out.writeUnsignedVInt(rie.promotedSize(idxSerializer));
+ rie.serialize(out, idxInfoSerializer, indexInfo);
+ }
- if (rie.isIndexed())
- {
- out.writeUnsignedVInt(rie.headerLength());
- DeletionTime.serializer.serialize(rie.deletionTime(), out);
- out.writeUnsignedVInt(rie.columnsIndex().size());
+ public void serializeForCache(RowIndexEntry<IndexInfo> rie, DataOutputPlus out) throws IOException
+ {
+ assert version.storeRows();
- // Calculate and write the offsets to the IndexInfo objects.
+ rie.serializeForCache(out);
+ }
- int[] offsets = new int[rie.columnsIndex().size()];
+ public RowIndexEntry<IndexInfo> deserializeForCache(DataInputPlus in) throws IOException
+ {
+ assert version.storeRows();
- if (out.hasPosition())
- {
- // Out is usually a SequentialWriter, so using the file-pointer is fine to generate the offsets.
- // A DataOutputBuffer also works.
- long start = out.position();
- int i = 0;
- for (IndexHelper.IndexInfo info : rie.columnsIndex())
- {
- offsets[i] = i == 0 ? 0 : (int)(out.position() - start);
- i++;
- idxSerializer.serialize(info, out);
- }
- }
- else
- {
- // Not sure this branch will ever be needed, but if it is called, it has to calculate the
- // serialized sizes instead of simply using the file-pointer.
- int i = 0;
- int offset = 0;
- for (IndexHelper.IndexInfo info : rie.columnsIndex())
- {
- offsets[i++] = offset;
- idxSerializer.serialize(info, out);
- offset += idxSerializer.serializedSize(info);
- }
- }
+ long position = in.readUnsignedVInt();
- for (int off : offsets)
- out.writeInt(off);
+ switch (in.readByte())
+ {
+ case CACHE_NOT_INDEXED:
+ return new RowIndexEntry<>(position);
+ case CACHE_INDEXED:
+ return new IndexedEntry(position, in, idxInfoSerializer, version, true);
+ case CACHE_INDEXED_SHALLOW:
+ return new ShallowIndexedEntry(position, in, idxInfoSerializer);
+ default:
+ throw new AssertionError();
}
}
- public RowIndexEntry<IndexHelper.IndexInfo> deserialize(DataInputPlus in) throws IOException
+ public static void skipForCache(DataInputPlus in, Version version) throws IOException
{
- if (!version.storeRows())
- {
- long position = in.readLong();
-
- int size = in.readInt();
- if (size > 0)
- {
- DeletionTime deletionTime = DeletionTime.serializer.deserialize(in);
-
- int entries = in.readInt();
- List<IndexHelper.IndexInfo> columnsIndex = new ArrayList<>(entries);
+ assert version.storeRows();
- long headerLength = 0L;
- for (int i = 0; i < entries; i++)
- {
- IndexHelper.IndexInfo info = idxSerializer.deserialize(in);
- columnsIndex.add(info);
- if (i == 0)
- headerLength = info.offset;
- }
-
- return new IndexedEntry(position, deletionTime, headerLength, columnsIndex);
- }
- else
- {
- return new RowIndexEntry<>(position);
- }
+ /* long position = */in.readUnsignedVInt();
+ switch (in.readByte())
+ {
+ case CACHE_NOT_INDEXED:
+ break;
+ case CACHE_INDEXED:
+ IndexedEntry.skipForCache(in);
+ break;
+ case CACHE_INDEXED_SHALLOW:
+ ShallowIndexedEntry.skipForCache(in);
+ break;
+ default:
+ assert false;
}
+ }
+
+ public RowIndexEntry<IndexInfo> deserialize(DataInputPlus in, long indexFilePosition) throws IOException
+ {
+ if (!version.storeRows())
+ return LegacyShallowIndexedEntry.deserialize(in, indexFilePosition, idxInfoSerializer, version);
long position = in.readUnsignedVInt();
int size = (int)in.readUnsignedVInt();
- if (size > 0)
+ if (size == 0)
+ {
+ return new RowIndexEntry<>(position);
+ }
+ else
{
long headerLength = in.readUnsignedVInt();
DeletionTime deletionTime = DeletionTime.serializer.deserialize(in);
- int entries = (int)in.readUnsignedVInt();
- List<IndexHelper.IndexInfo> columnsIndex = new ArrayList<>(entries);
- for (int i = 0; i < entries; i++)
- columnsIndex.add(idxSerializer.deserialize(in));
+ int columnsIndexCount = (int) in.readUnsignedVInt();
- in.skipBytesFully(entries * TypeSizes.sizeof(0));
+ int indexedPartSize = size - serializedSize(deletionTime, headerLength, columnsIndexCount);
- return new IndexedEntry(position, deletionTime, headerLength, columnsIndex);
- }
- else
- {
- return new RowIndexEntry<>(position);
+ if (size <= DatabaseDescriptor.getColumnIndexCacheSize())
+ {
+ return new IndexedEntry(position, in, deletionTime, headerLength, columnsIndexCount,
+ idxInfoSerializer, version, indexedPartSize);
+ }
+ else
+ {
+ in.skipBytes(indexedPartSize);
+
+ return new ShallowIndexedEntry(position,
+ indexFilePosition,
+ deletionTime, headerLength, columnsIndexCount,
+ indexedPartSize, idxInfoSerializer);
+ }
}
}
- // 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 long deserializePositionAndSkip(DataInputPlus in) throws IOException
+ {
+ if (!version.storeRows())
+ return LegacyShallowIndexedEntry.deserializePositionAndSkip(in);
+
+ return ShallowIndexedEntry.deserializePositionAndSkip(in);
+ }
+
+ /**
+ * 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, Version version) throws IOException
{
return version.storeRows() ? in.readUnsignedVInt() : in.readLong();
@@ -250,51 +385,295 @@ public class RowIndexEntry<T> implements IMeasurableMemory
in.skipBytesFully(size);
}
- public int serializedSize(RowIndexEntry<IndexHelper.IndexInfo> rie)
+ public static void serializeOffsets(DataOutputBuffer out, int[] indexOffsets, int columnIndexCount) throws IOException
{
- assert version.storeRows() : "We read old index files but we should never write them";
+ for (int i = 0; i < columnIndexCount; i++)
+ out.writeInt(indexOffsets[i]);
+ }
+ }
+
+ private static int serializedSize(DeletionTime deletionTime, long headerLength, int columnIndexCount)
+ {
+ return TypeSizes.sizeofUnsignedVInt(headerLength)
+ + (int) DeletionTime.serializer.serializedSize(deletionTime)
+ + TypeSizes.sizeofUnsignedVInt(columnIndexCount);
+ }
+
+ public void serialize(DataOutputPlus out, IndexInfo.Serializer idxInfoSerializer, ByteBuffer indexInfo) throws IOException
+ {
+ out.writeUnsignedVInt(position);
+
+ out.writeUnsignedVInt(0);
+ }
+
+ public void serializeForCache(DataOutputPlus out) throws IOException
+ {
+ out.writeUnsignedVInt(position);
+
+ out.writeByte(CACHE_NOT_INDEXED);
+ }
+
+ private static final class LegacyShallowIndexedEntry extends RowIndexEntry<IndexInfo>
+ {
+ private static final long BASE_SIZE;
+ static
+ {
+ BASE_SIZE = ObjectSizes.measure(new LegacyShallowIndexedEntry(0, 0, DeletionTime.LIVE, 0, new int[0], null, 0));
+ }
+
+ private final long indexFilePosition;
+ private final int[] offsets;
+ @Unmetered
+ private final IndexInfo.Serializer idxInfoSerializer;
+ private final DeletionTime deletionTime;
+ private final long headerLength;
+ private final int serializedSize;
+
+ private LegacyShallowIndexedEntry(long dataFilePosition, long indexFilePosition,
+ DeletionTime deletionTime, long headerLength,
+ int[] offsets, IndexInfo.Serializer idxInfoSerializer,
+ int serializedSize)
+ {
+ super(dataFilePosition);
+ this.deletionTime = deletionTime;
+ this.headerLength = headerLength;
+ this.indexFilePosition = indexFilePosition;
+ this.offsets = offsets;
+ this.idxInfoSerializer = idxInfoSerializer;
+ this.serializedSize = serializedSize;
+ }
+
+ @Override
+ public DeletionTime deletionTime()
+ {
+ return deletionTime;
+ }
+
+ @Override
+ public long headerLength()
+ {
+ return headerLength;
+ }
+
+ @Override
+ public long unsharedHeapSize()
+ {
+ return BASE_SIZE + offsets.length * TypeSizes.sizeof(0);
+ }
+
+ @Override
+ public int columnsIndexCount()
+ {
+ return offsets.length;
+ }
+
+ @Override
+ public void serialize(DataOutputPlus out, IndexInfo.Serializer idxInfoSerializer, ByteBuffer indexInfo)
+ {
+ throw new UnsupportedOperationException("serializing legacy index entries is not supported");
+ }
+
+ @Override
+ public void serializeForCache(DataOutputPlus out)
+ {
+ throw new UnsupportedOperationException("serializing legacy index entries is not supported");
+ }
- int indexedSize = 0;
- if (rie.isIndexed())
+ @Override
+ public IndexInfoRetriever openWithIndex(SegmentedFile indexFile)
+ {
+ int fieldsSize = (int) DeletionTime.serializer.serializedSize(deletionTime)
+ + TypeSizes.sizeof(0); // columnIndexCount
+ indexEntrySizeHistogram.update(serializedSize);
+ indexInfoCountHistogram.update(offsets.length);
+ return new LegacyIndexInfoRetriever(indexFilePosition +
+ TypeSizes.sizeof(0L) + // position
+ TypeSizes.sizeof(0) + // indexInfoSize
+ fieldsSize,
+ offsets, indexFile.createReader(), idxInfoSerializer);
+ }
+
+ public static RowIndexEntry<IndexInfo> deserialize(DataInputPlus in, long indexFilePosition,
+ IndexInfo.Serializer idxInfoSerializer, Version version) throws IOException
+ {
+ long dataFilePosition = in.readLong();
+
+ int size = in.readInt();
+ if (size == 0)
{
- List<IndexHelper.IndexInfo> index = rie.columnsIndex();
+ return new RowIndexEntry<>(dataFilePosition);
+ }
+ else if (size <= DatabaseDescriptor.getColumnIndexCacheSize())
+ {
+ return new IndexedEntry(dataFilePosition, in, idxInfoSerializer, version, false);
+ }
+ else
+ {
+ DeletionTime deletionTime = DeletionTime.serializer.deserialize(in);
- indexedSize += TypeSizes.sizeofUnsignedVInt(rie.headerLength());
- indexedSize += DeletionTime.serializer.serializedSize(rie.deletionTime());
- indexedSize += TypeSizes.sizeofUnsignedVInt(index.size());
+ // For legacy sstables (i.e. sstables pre-"ma", pre-3.0) we have to scan all serialized IndexInfo
+ // objects to calculate the offsets array. However, it might be possible to deserialize all
+ // IndexInfo objects here - but to just skip feels more gentle to the heap/GC.
- for (IndexHelper.IndexInfo info : index)
- indexedSize += idxSerializer.serializedSize(info);
+ int entries = in.readInt();
+ int[] offsets = new int[entries];
+
+ TrackedDataInputPlus tracked = new TrackedDataInputPlus(in);
+ long start = tracked.getBytesRead();
+ long headerLength = 0L;
+ for (int i = 0; i < entries; i++)
+ {
+ offsets[i] = (int) (tracked.getBytesRead() - start);
+ if (i == 0)
+ {
+ IndexInfo info = idxInfoSerializer.deserialize(tracked);
+ headerLength = info.offset;
+ }
+ else
+ idxInfoSerializer.skip(tracked);
+ }
- indexedSize += index.size() * TypeSizes.sizeof(0);
+ return new LegacyShallowIndexedEntry(dataFilePosition, indexFilePosition, deletionTime, headerLength, offsets, idxInfoSerializer, size);
}
+ }
+
+ static long deserializePositionAndSkip(DataInputPlus in) throws IOException
+ {
+ long position = in.readLong();
+
+ int size = in.readInt();
+ if (size > 0)
+ in.skipBytesFully(size);
- return TypeSizes.sizeofUnsignedVInt(rie.position) + TypeSizes.sizeofUnsignedVInt(indexedSize) + indexedSize;
+ return position;
+ }
+ }
+
+ private static final class LegacyIndexInfoRetriever extends FileIndexInfoRetriever
+ {
+ private final int[] offsets;
+
+ private LegacyIndexInfoRetriever(long indexFilePosition, int[] offsets, FileDataInput reader, IndexInfo.Serializer idxInfoSerializer)
+ {
+ super(indexFilePosition, offsets.length, reader, idxInfoSerializer);
+ this.offsets = offsets;
+ }
+
+ IndexInfo fetchIndex(int index) throws IOException
+ {
+ retrievals++;
+
+ // seek to posision of IndexInfo
+ indexReader.seek(indexInfoFilePosition + offsets[index]);
+
+ // deserialize IndexInfo
+ return idxInfoSerializer.deserialize(indexReader);
}
}
/**
- * An entry in the row index for a row whose columns are indexed.
+ * An entry in the row index for a row whose columns are indexed - used for both legacy and current formats.
*/
- private static class IndexedEntry extends RowIndexEntry<IndexHelper.IndexInfo>
+ private static final class IndexedEntry extends RowIndexEntry<IndexInfo>
{
- private final DeletionTime deletionTime;
+ private static final long BASE_SIZE;
- // The offset in the file when the index entry end
+ static
+ {
+ BASE_SIZE = ObjectSizes.measure(new IndexedEntry(0, DeletionTime.LIVE, 0, null, null, 0, null));
+ }
+
+ private final DeletionTime deletionTime;
private final long headerLength;
- private final List<IndexHelper.IndexInfo> columnsIndex;
- private static final long BASE_SIZE =
- ObjectSizes.measure(new IndexedEntry(0, DeletionTime.LIVE, 0, Arrays.<IndexHelper.IndexInfo>asList(null, null)))
- + ObjectSizes.measure(new ArrayList<>(1));
- private IndexedEntry(long position, DeletionTime deletionTime, long headerLength, List<IndexHelper.IndexInfo> columnsIndex)
+ private final IndexInfo[] columnsIndex;
+ private final int[] offsets;
+ private final int indexedPartSize;
+ @Unmetered
+ private final ISerializer<IndexInfo> idxInfoSerializer;
+
+ private IndexedEntry(long dataFilePosition, DeletionTime deletionTime, long headerLength,
+ IndexInfo[] columnsIndex, int[] offsets,
+ int indexedPartSize, ISerializer<IndexInfo> idxInfoSerializer)
{
- super(position);
- assert deletionTime != null;
- assert columnsIndex != null && columnsIndex.size() > 1;
- this.deletionTime = deletionTime;
+ super(dataFilePosition);
+
this.headerLength = headerLength;
+ this.deletionTime = deletionTime;
+
this.columnsIndex = columnsIndex;
+ this.offsets = offsets;
+ this.indexedPartSize = indexedPartSize;
+ this.idxInfoSerializer = idxInfoSerializer;
+ }
+
+ private IndexedEntry(long dataFilePosition, DataInputPlus in,
+ DeletionTime deletionTime, long headerLength, int columnIndexCount,
+ IndexInfo.Serializer idxInfoSerializer,
+ Version version, int indexedPartSize) 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);
+
+ int[] offsets = null;
+ if (version.storeRows())
+ {
+ offsets = new int[this.columnsIndex.length];
+ for (int i = 0; i < offsets.length; i++)
+ offsets[i] = in.readInt();
+ }
+ this.offsets = offsets;
+
+ this.indexedPartSize = indexedPartSize;
+
+ this.idxInfoSerializer = idxInfoSerializer;
+ }
+
+ private IndexedEntry(long dataFilePosition, DataInputPlus in, IndexInfo.Serializer idxInfoSerializer, Version version, boolean forCache) throws IOException
+ {
+ super(dataFilePosition);
+
+ this.headerLength = in.readUnsignedVInt();
+ this.deletionTime = DeletionTime.serializer.deserialize(in);
+ int columnsIndexCount = (int) in.readUnsignedVInt();
+
+ TrackedDataInputPlus trackedIn = new TrackedDataInputPlus(in);
+
+ this.columnsIndex = new IndexInfo[columnsIndexCount];
+ for (int i = 0; i < columnsIndexCount; i++)
+ this.columnsIndex[i] = idxInfoSerializer.deserialize(trackedIn);
+
+ int[] offsets = null;
+ if (!forCache && version.storeRows())
+ {
+ offsets = new int[this.columnsIndex.length];
+ for (int i = 0; i < offsets.length; i++)
+ offsets[i] = trackedIn.readInt();
+ }
+ this.offsets = offsets;
+
+ this.indexedPartSize = (int) trackedIn.getBytesRead();
+
+ this.idxInfoSerializer = idxInfoSerializer;
+ }
+
+ @Override
+ public boolean indexOnHeap()
+ {
+ return true;
+ }
+
+ @Override
+ public int columnsIndexCount()
+ {
+ return columnsIndex.length;
}
@Override
@@ -310,36 +689,337 @@ public class RowIndexEntry<T> implements IMeasurableMemory
}
@Override
- public List<IndexHelper.IndexInfo> columnsIndex()
+ public IndexInfoRetriever openWithIndex(SegmentedFile indexFile)
+ {
+ indexEntrySizeHistogram.update(serializedSize(deletionTime, headerLength, columnsIndex.length) + indexedPartSize);
+ indexInfoCountHistogram.update(columnsIndex.length);
+ return new IndexInfoRetriever()
+ {
+ private int retrievals;
+
+ @Override
+ public IndexInfo columnsIndex(int index)
+ {
+ retrievals++;
+ return columnsIndex[index];
+ }
+
+ 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, IndexInfo.Serializer idxInfoSerializer, ByteBuffer indexInfo) throws IOException
+ {
+ assert indexedPartSize != Integer.MIN_VALUE;
+
+ out.writeUnsignedVInt(position);
+
+ out.writeUnsignedVInt(serializedSize(deletionTime, headerLength, columnsIndex.length) + indexedPartSize);
+
+ out.writeUnsignedVInt(headerLength);
+ DeletionTime.serializer.serialize(deletionTime, out);
+ out.writeUnsignedVInt(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.serializer.serialize(deletionTime, out);
+ out.writeUnsignedVInt(columnsIndexCount());
+
+ for (IndexInfo indexInfo : columnsIndex)
+ idxInfoSerializer.serialize(indexInfo, out);
+ }
+
+ static void skipForCache(DataInputPlus in) throws IOException
+ {
+ /*long headerLength =*/in.readUnsignedVInt();
+ /*DeletionTime deletionTime = */DeletionTime.serializer.skip(in);
+ /*int columnsIndexCount = (int)*/in.readUnsignedVInt();
+
+ /*int indexedPartSize = (int)*/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<IndexInfo>
+ {
+ private static final long BASE_SIZE;
+
+ static
+ {
+ BASE_SIZE = ObjectSizes.measure(new ShallowIndexedEntry(0, 0, DeletionTime.LIVE, 0, 10, 0, null));
+ }
+
+ 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;
+
+ /**
+ * See {@link #create(long, long, DeletionTime, long, int, int, List, int[], ISerializer)} for a description
+ * of the parameters.
+ */
+ private ShallowIndexedEntry(long dataFilePosition, long indexFilePosition,
+ DeletionTime deletionTime, long headerLength, int columnIndexCount,
+ int indexedPartSize, ISerializer<IndexInfo> idxInfoSerializer)
+ {
+ super(dataFilePosition);
+
+ assert columnIndexCount > 1;
+
+ this.indexFilePosition = indexFilePosition;
+ this.headerLength = headerLength;
+ this.deletionTime = deletionTime;
+ this.columnsIndexCount = columnIndexCount;
+
+ this.indexedPartSize = indexedPartSize;
+ this.idxInfoSerializer = idxInfoSerializer;
+
+ this.fieldsSerializedSize = serializedSize(deletionTime, headerLength, columnIndexCount);
+ this.offsetsOffset = indexedPartSize + fieldsSerializedSize - columnsIndexCount * TypeSizes.sizeof(0);
+ }
+
+ /**
+ * Constructor for key-cache deserialization
+ */
+ private ShallowIndexedEntry(long dataFilePosition, DataInputPlus in, IndexInfo.Serializer idxInfoSerializer) throws IOException
+ {
+ super(dataFilePosition);
+
+ this.indexFilePosition = in.readUnsignedVInt();
+
+ this.headerLength = in.readUnsignedVInt();
+ this.deletionTime = DeletionTime.serializer.deserialize(in);
+ this.columnsIndexCount = (int) in.readUnsignedVInt();
+
+ this.indexedPartSize = (int) in.readUnsignedVInt();
+
+ this.idxInfoSerializer = idxInfoSerializer;
+
+ this.fieldsSerializedSize = serializedSize(deletionTime, headerLength, columnsIndexCount);
+ this.offsetsOffset = indexedPartSize + fieldsSerializedSize - columnsIndexCount * TypeSizes.sizeof(0);
+ }
+
+ @Override
+ public int columnsIndexCount()
{
- return columnsIndex;
+ return columnsIndexCount;
}
@Override
- protected int promotedSize(IndexHelper.IndexInfo.Serializer idxSerializer)
+ public DeletionTime deletionTime()
{
- long size = TypeSizes.sizeofUnsignedVInt(headerLength)
- + DeletionTime.serializer.serializedSize(deletionTime)
- + TypeSizes.sizeofUnsignedVInt(columnsIndex.size()); // number of entries
- for (IndexHelper.IndexInfo info : columnsIndex)
- size += idxSerializer.serializedSize(info);
+ return deletionTime;
+ }
- size += columnsIndex.size() * TypeSizes.sizeof(0);
+ @Override
+ public long headerLength()
+ {
+ return headerLength;
+ }
- return Ints.checkedCast(size);
+ @Override
+ public IndexInfoRetriever openWithIndex(SegmentedFile indexFile)
+ {
+ indexEntrySizeHistogram.update(indexedPartSize + fieldsSerializedSize);
+ indexInfoCountHistogram.update(columnsIndexCount);
+ return new ShallowInfoRetriever(indexFilePosition +
+ VIntCoding.computeUnsignedVIntSize(position) +
+ VIntCoding.computeUnsignedVIntSize(indexedPartSize + fieldsSerializedSize) +
+ fieldsSerializedSize,
+ offsetsOffset - fieldsSerializedSize,
+ columnsIndexCount, indexFile.createReader(), idxInfoSerializer);
}
@Override
public long unsharedHeapSize()
{
- long entrySize = 0;
- for (IndexHelper.IndexInfo idx : columnsIndex)
- entrySize += idx.unsharedHeapSize();
+ return BASE_SIZE;
+ }
- return BASE_SIZE
- + entrySize
- + deletionTime.unsharedHeapSize()
- + ObjectSizes.sizeOfReferenceArray(columnsIndex.size());
+ @Override
+ public void serialize(DataOutputPlus out, IndexInfo.Serializer idxInfoSerializer, ByteBuffer indexInfo) throws IOException
+ {
+ out.writeUnsignedVInt(position);
+
+ out.writeUnsignedVInt(fieldsSerializedSize + indexInfo.limit());
+
+ out.writeUnsignedVInt(headerLength);
+ DeletionTime.serializer.serialize(deletionTime, out);
+ out.writeUnsignedVInt(columnsIndexCount);
+
+ out.write(indexInfo);
+ }
+
+ static long deserializePositionAndSkip(DataInputPlus in) throws IOException
+ {
+ long position = in.readUnsignedVInt();
+
+ int size = (int) in.readUnsignedVInt();
+ if (size > 0)
+ in.skipBytesFully(size);
+
+ return position;
+ }
+
+ @Override
+ public void serializeForCache(DataOutputPlus out) throws IOException
+ {
+ out.writeUnsignedVInt(position);
+ out.writeByte(CACHE_INDEXED_SHALLOW);
+
+ out.writeUnsignedVInt(indexFilePosition);
+
+ out.writeUnsignedVInt(headerLength);
+ DeletionTime.serializer.serialize(deletionTime, out);
+ out.writeUnsignedVInt(columnsIndexCount);
+
+ out.writeUnsignedVInt(indexedPartSize);
+ }
+
+ static void skipForCache(DataInputPlus in) throws IOException
+ {
+ /*long indexFilePosition =*/in.readUnsignedVInt();
+
+ /*long headerLength =*/in.readUnsignedVInt();
+ /*DeletionTime deletionTime = */DeletionTime.serializer.skip(in);
+ /*int columnsIndexCount = (int)*/in.readUnsignedVInt();
+
+ /*int indexedPartSize = (int)*/in.readUnsignedVInt();
+ }
+ }
+
+ private static final class ShallowInfoRetriever extends FileIndexInfoRetriever
+ {
+ private final int offsetsOffset;
+
+ private ShallowInfoRetriever(long indexInfoFilePosition, int offsetsOffset, int indexCount,
+ FileDataInput indexReader, ISerializer<IndexInfo> idxInfoSerializer)
+ {
+ super(indexInfoFilePosition, indexCount, indexReader, idxInfoSerializer);
+ this.offsetsOffset = offsetsOffset;
+ }
+
+ IndexInfo fetchIndex(int index) throws IOException
+ {
+ assert index >= 0 && index < indexCount;
+
+ retrievals++;
+
+ // seek to position in "offsets to IndexInfo" table
+ indexReader.seek(indexInfoFilePosition + offsetsOffset + index * TypeSizes.sizeof(0));
+
+ // 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 int indexCount;
+ final ISerializer<IndexInfo> idxInfoSerializer;
+ final FileDataInput indexReader;
+ int retrievals;
+
+ private IndexInfo[] lastIndexes;
+
+ /**
+ *
+ * @param indexInfoFilePosition offset of first serialized {@link IndexInfo} object
+ * @param indexCount number of {@link IndexInfo} objects
+ * @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, int indexCount, FileDataInput indexReader, ISerializer<IndexInfo> idxInfoSerializer)
+ {
+ this.indexInfoFilePosition = indexInfoFilePosition;
+ this.indexCount = indexCount;
+ this.idxInfoSerializer = idxInfoSerializer;
+ this.indexReader = indexReader;
+ }
+
+ public final IndexInfo columnsIndex(int index) throws IOException
+ {
+ if (lastIndexes != null
+ && lastIndexes.length > index && lastIndexes[index] != null)
+ {
+ // return a previously read/deserialized IndexInfo
+ return lastIndexes[index];
+ }
+
+ if (lastIndexes == null)
+ lastIndexes = new IndexInfo[index + 1];
+ else if (lastIndexes.length <= index)
+ lastIndexes = Arrays.copyOf(lastIndexes, index + 1);
+
+ IndexInfo indexInfo = fetchIndex(index);
+ lastIndexes[index] = indexInfo;
+
+ return indexInfo;
+ }
+
+ abstract IndexInfo fetchIndex(int index) throws IOException;
+
+ public void close() throws IOException
+ {
+ indexReader.close();
+
+ indexInfoGetsHistogram.update(retrievals);
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/db/SerializationHeader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SerializationHeader.java b/src/java/org/apache/cassandra/db/SerializationHeader.java
index 0fd1281..af2d434 100644
--- a/src/java/org/apache/cassandra/db/SerializationHeader.java
+++ b/src/java/org/apache/cassandra/db/SerializationHeader.java
@@ -29,7 +29,6 @@ import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.io.sstable.format.SSTableReader;
@@ -75,25 +74,6 @@ public class SerializationHeader
return new SerializationHeader(true, metadata, metadata.partitionColumns(), EncodingStats.NO_STATS);
}
- public static SerializationHeader forKeyCache(CFMetaData metadata)
- {
- // We don't save type information in the key cache (we could change
- // that but it's easier right now), so instead we simply use BytesType
- // for both serialization and deserialization. Note that we also only
- // serializer clustering prefixes in the key cache, so only the clusteringTypes
- // really matter.
- int size = metadata.clusteringColumns().size();
- List<AbstractType<?>> clusteringTypes = new ArrayList<>(size);
- for (int i = 0; i < size; i++)
- clusteringTypes.add(BytesType.instance);
- return new SerializationHeader(false,
- BytesType.instance,
- clusteringTypes,
- PartitionColumns.NONE,
- EncodingStats.NO_STATS,
- Collections.<ByteBuffer, AbstractType<?>>emptyMap());
- }
-
public static SerializationHeader make(CFMetaData metadata, Collection<SSTableReader> sstables)
{
// The serialization header has to be computed before the start of compaction (since it's used to write)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef5bbedd/src/java/org/apache/cassandra/db/Serializers.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Serializers.java b/src/java/org/apache/cassandra/db/Serializers.java
index cef06a3..a0503e0 100644
--- a/src/java/org/apache/cassandra/db/Serializers.java
+++ b/src/java/org/apache/cassandra/db/Serializers.java
@@ -20,10 +20,15 @@ package org.apache.cassandra.db;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.io.ISerializer;
+import org.apache.cassandra.io.sstable.IndexInfo;
+import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.sstable.format.Version;
@@ -36,36 +41,66 @@ public class Serializers
{
private final CFMetaData metadata;
+ private Map<Version, IndexInfo.Serializer> otherVersionClusteringSerializers;
+
+ private final IndexInfo.Serializer latestVersionIndexSerializer;
+
public Serializers(CFMetaData metadata)
{
this.metadata = metadata;
+ this.latestVersionIndexSerializer = new IndexInfo.Serializer(BigFormat.latestVersion,
+ indexEntryClusteringPrefixSerializer(BigFormat.latestVersion, SerializationHeader.makeWithoutStats(metadata)));
+ }
+
+ IndexInfo.Serializer indexInfoSerializer(Version version, SerializationHeader header)
+ {
+ // null header indicates streaming from pre-3.0 sstables
+ if (version.equals(BigFormat.latestVersion) && header != null)
+ return latestVersionIndexSerializer;
+
+ if (otherVersionClusteringSerializers == null)
+ otherVersionClusteringSerializers = new ConcurrentHashMap<>();
+ IndexInfo.Serializer serializer = otherVersionClusteringSerializers.get(version);
+ if (serializer == null)
+ {
+ serializer = new IndexInfo.Serializer(version,
+ indexEntryClusteringPrefixSerializer(version, header));
+ otherVersionClusteringSerializers.put(version, serializer);
+ }
+ return serializer;
}
// TODO: Once we drop support for old (pre-3.0) sstables, we can drop this method and inline the calls to
- // ClusteringPrefix.serializer in IndexHelper directly. At which point this whole class probably becomes
+ // ClusteringPrefix.serializer directly. At which point this whole class probably becomes
// unecessary (since IndexInfo.Serializer won't depend on the metadata either).
- public ISerializer<ClusteringPrefix> indexEntryClusteringPrefixSerializer(final Version version, final SerializationHeader header)
+ private ISerializer<ClusteringPrefix> indexEntryClusteringPrefixSerializer(Version version, SerializationHeader header)
{
if (!version.storeRows() || header == null) //null header indicates streaming from pre-3.0 sstables
{
return oldFormatSerializer(version);
}
- return newFormatSerializer(version, header);
+ return new NewFormatSerializer(version, header.clusteringTypes());
}
- private ISerializer<ClusteringPrefix> oldFormatSerializer(final Version version)
+ private ISerializer<ClusteringPrefix> oldFormatSerializer(Version version)
{
return new ISerializer<ClusteringPrefix>()
{
- SerializationHeader newHeader = SerializationHeader.makeWithoutStats(metadata);
+ List<AbstractType<?>> clusteringTypes = SerializationHeader.makeWithoutStats(metadata).clusteringTypes();
public void serialize(ClusteringPrefix clustering, DataOutputPlus out) throws IOException
{
//we deserialize in the old format and serialize in the new format
ClusteringPrefix.serializer.serialize(clustering, out,
version.correspondingMessagingVersion(),
- newHeader.clusteringTypes());
+ clusteringTypes);
+ }
+
+ @Override
+ public void skip(DataInputPlus in) throws IOException
+ {
+ ByteBufferUtil.skipShortLength(in);
}
public ClusteringPrefix deserialize(DataInputPlus in) throws IOException
@@ -108,31 +143,41 @@ public class Serializers
public long serializedSize(ClusteringPrefix clustering)
{
return ClusteringPrefix.serializer.serializedSize(clustering, version.correspondingMessagingVersion(),
- newHeader.clusteringTypes());
+ clusteringTypes);
}
};
}
-
- private ISerializer<ClusteringPrefix> newFormatSerializer(final Version version, final SerializationHeader header)
+ private static class NewFormatSerializer implements ISerializer<ClusteringPrefix>
{
- return new ISerializer<ClusteringPrefix>() //Reading and writing from/to the new sstable format
+ private final Version version;
+ private final List<AbstractType<?>> clusteringTypes;
+
+ NewFormatSerializer(Version version, List<AbstractType<?>> clusteringTypes)
{
- public void serialize(ClusteringPrefix clustering, DataOutputPlus out) throws IOException
- {
- ClusteringPrefix.serializer.serialize(clustering, out, version.correspondingMessagingVersion(), header.clusteringTypes());
- }
+ this.version = version;
+ this.clusteringTypes = clusteringTypes;
+ }
- public ClusteringPrefix deserialize(DataInputPlus in) throws IOException
- {
- return ClusteringPrefix.serializer.deserialize(in, version.correspondingMessagingVersion(), header.clusteringTypes());
- }
+ public void serialize(ClusteringPrefix clustering, DataOutputPlus out) throws IOException
+ {
+ ClusteringPrefix.serializer.serialize(clustering, out, version.correspondingMessagingVersion(), clusteringTypes);
+ }
- public long serializedSize(ClusteringPrefix clustering)
- {
- return ClusteringPrefix.serializer.serializedSize(clustering, version.correspondingMessagingVersion(), header.clusteringTypes());
- }
- };
- }
+ @Override
+ public void skip(DataInputPlus in) throws IOException
+ {
+ ClusteringPrefix.serializer.skip(in, version.correspondingMessagingVersion(), clusteringTypes);
+ }
+
+ public ClusteringPrefix deserialize(DataInputPlus in) throws IOException
+ {
+ return ClusteringPrefix.serializer.deserialize(in, version.correspondingMessagingVersion(), clusteringTypes);
+ }
-}
\ No newline at end of file
+ public long serializedSize(ClusteringPrefix clustering)
+ {
+ return ClusteringPrefix.serializer.serializedSize(clustering, version.correspondingMessagingVersion(), clusteringTypes);
+ }
+ }
+}