mirror of https://github.com/apache/cassandra
Extract format-specific index production from the cursor writer
Pure refactor: SSTableCursorWriter's index machinery — promoted index blocks serialized incrementally, the tail-counted promotion decision, Index.db entries, bloom filter and summary appends — moves verbatim into BigCursorIndexWriter behind a new CursorIndexWriter seam (startPartition / rowWritten / staticRowWritten / endPartition / block clock). The seam is event-shaped because unfiltered descriptors are transient: implementations needing block-boundary clusterings must capture them at row time, which is also how the BIG code already worked (eager first-clustering serialization, reusable last-clustering copy). One deliberate substitution: the tail block passes DeletionTime.LIVE where the old code read the writer's open marker — equivalent because range tombstones close before partition end. No behavior change: this branch's full differential suite passes byte-identically before and after. Kept purely as enabling infrastructure — later fixes (2GiB index-offset widening, disabled bloom filter tolerance) are written against BigCursorIndexWriter's class structure rather than the original monolithic writer.
This commit is contained in:
Jon Haddad
parent
afc5c61598
commit
efe20e858b
|
|
@ -0,0 +1,226 @@
|
|||
/*
|
||||
* 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;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import com.google.common.primitives.Ints;
|
||||
import org.agrona.collections.IntArrayList;
|
||||
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.db.ClusteringPrefix;
|
||||
import org.apache.cassandra.db.DeletionTime;
|
||||
import org.apache.cassandra.io.FSWriteError;
|
||||
import org.apache.cassandra.io.sstable.format.big.BigFormatPartitionWriter;
|
||||
import org.apache.cassandra.io.sstable.format.big.BigTableWriter;
|
||||
import org.apache.cassandra.io.sstable.format.big.RowIndexEntry;
|
||||
import org.apache.cassandra.io.util.DataOutputBuffer;
|
||||
import org.apache.cassandra.io.util.SequentialWriter;
|
||||
import org.apache.cassandra.utils.ByteArrayUtil;
|
||||
import org.apache.cassandra.utils.BloomFilter;
|
||||
|
||||
/**
|
||||
* BIG-format index production for the cursor writer: promoted index blocks serialized
|
||||
* incrementally (IndexInfo wire format) and Index.db entries with the tail-counted
|
||||
* promotion decision, byte-identical to the iterator path
|
||||
* (BigFormatPartitionWriter + RowIndexEntry.create + BigTableWriter.IndexWriter.append).
|
||||
*/
|
||||
public class BigCursorIndexWriter extends CursorIndexWriter
|
||||
{
|
||||
private final BigTableWriter.IndexWriter indexWriter;
|
||||
private final DeletionTime.Serializer deletionTimeSerializer;
|
||||
/**
|
||||
* See: {@link BloomFilter#reusableIndexes}
|
||||
*/
|
||||
private final long[] reusableIndexes = new long[21];
|
||||
|
||||
private final DataOutputBuffer rowIndexEntries = new DataOutputBuffer();
|
||||
private final IntArrayList rowIndexEntriesOffsets = new IntArrayList();
|
||||
private final ClusteringDescriptor rowIndexEntryLastClustering;
|
||||
private boolean hasDistinctLastClustering = false;
|
||||
private int rowIndexEntryOffset;
|
||||
private final int indexBlockThreshold;
|
||||
|
||||
public BigCursorIndexWriter(BigTableWriter.IndexWriter indexWriter,
|
||||
DeletionTime.Serializer deletionTimeSerializer,
|
||||
ClusteringDescriptor rowIndexEntryLastClustering)
|
||||
{
|
||||
this.indexWriter = indexWriter;
|
||||
this.deletionTimeSerializer = deletionTimeSerializer;
|
||||
this.rowIndexEntryLastClustering = rowIndexEntryLastClustering;
|
||||
this.indexBlockThreshold = DatabaseDescriptor.getColumnIndexSize(BigFormatPartitionWriter.DEFAULT_GRANULARITY);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void reset()
|
||||
{
|
||||
rowIndexEntries.clear();
|
||||
rowIndexEntriesOffsets.clear();
|
||||
rowIndexEntryOffset = 0;
|
||||
hasDistinctLastClustering = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void rowWritten(UnfilteredDescriptor unfilteredDescriptor, long unfilteredStartPosition,
|
||||
long unfilteredEndPosition, DeletionTime openMarker) throws IOException
|
||||
{
|
||||
// write the first clustering into rowIndexEntries buffer (we will need it unless we never write the first entry)
|
||||
if (currentOffsetInPartition(unfilteredStartPosition) == indexBlockStartOffset || (rowIndexEntryOffset == rowIndexEntries.position()))
|
||||
{
|
||||
writeClusteringToRowIndexEntries(unfilteredDescriptor);
|
||||
}
|
||||
else
|
||||
{
|
||||
rowIndexEntryLastClustering.copy(unfilteredDescriptor);
|
||||
hasDistinctLastClustering = true;
|
||||
}
|
||||
|
||||
/** {@link BigFormatPartitionWriter#addUnfiltered(org.apache.cassandra.db.rows.Unfiltered)} */
|
||||
// if we hit the index block size that we have to index after, go ahead and index it.
|
||||
long indexBlockSize = currentOffsetInPartition(unfilteredEndPosition) - indexBlockStartOffset;
|
||||
if (indexBlockSize >= this.indexBlockThreshold)
|
||||
addIndexBlock(unfilteredEndPosition, indexBlockSize, openMarker);
|
||||
}
|
||||
|
||||
/**
|
||||
* See:
|
||||
* {@link BigFormatPartitionWriter#addIndexBlock()}
|
||||
* - {@link org.apache.cassandra.io.sstable.IndexInfo.Serializer#serialize(org.apache.cassandra.io.sstable.IndexInfo, org.apache.cassandra.io.util.DataOutputPlus)}
|
||||
*/
|
||||
private void addIndexBlock(long endOfRowPosition, long indexBlockSize, DeletionTime openMarker) throws IOException
|
||||
{
|
||||
if (rowIndexEntriesOffsets.isEmpty() && rowIndexEntryOffset != 0) {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
|
||||
// serialize the index info
|
||||
/** {@link org.apache.cassandra.io.sstable.IndexInfo.Serializer#serialize(org.apache.cassandra.io.sstable.IndexInfo, org.apache.cassandra.io.util.DataOutputPlus)}*/
|
||||
rowIndexEntriesOffsets.addInt(rowIndexEntryOffset);
|
||||
|
||||
// first clustering is already in, write last entry
|
||||
if (!hasDistinctLastClustering)
|
||||
{
|
||||
// first entry is the last entry, copy it
|
||||
byte[] entriesData = rowIndexEntries.getData();
|
||||
long endOfFirstEntry = rowIndexEntries.position();
|
||||
rowIndexEntries.write(entriesData, rowIndexEntryOffset, (int) (endOfFirstEntry - rowIndexEntryOffset));
|
||||
}
|
||||
else
|
||||
{
|
||||
writeClusteringToRowIndexEntries(rowIndexEntryLastClustering);
|
||||
rowIndexEntryLastClustering.resetClustering();
|
||||
}
|
||||
hasDistinctLastClustering = false;
|
||||
|
||||
rowIndexEntries.writeUnsignedVInt((long)indexBlockStartOffset);
|
||||
rowIndexEntries.writeVInt(indexBlockSize - IndexInfo.Serializer.WIDTH_BASE);
|
||||
|
||||
boolean isDeleteTimePresent = !openMarker.isLive();
|
||||
rowIndexEntries.writeBoolean(isDeleteTimePresent);
|
||||
if (isDeleteTimePresent)
|
||||
deletionTimeSerializer.serialize(openMarker, rowIndexEntries);
|
||||
// next block starts
|
||||
rowIndexEntryOffset = Ints.checkedCast(rowIndexEntries.position());
|
||||
notePosition(endOfRowPosition);
|
||||
}
|
||||
|
||||
private void writeClusteringToRowIndexEntries(ClusteringDescriptor clustering) throws IOException
|
||||
{
|
||||
ClusteringPrefix.Kind kind = clustering.clusteringKind();
|
||||
rowIndexEntries.writeByte(kind.ordinal());
|
||||
if (kind != ClusteringPrefix.Kind.CLUSTERING)
|
||||
rowIndexEntries.writeShort(clustering.clusteringColumnsBound());
|
||||
rowIndexEntries.write(clustering.clusteringBytes(), 0, clustering.clusteringLength());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void endPartition(byte[] key, int keyLength, int headerLength,
|
||||
DeletionTime partitionDeletionTime, long partitionEnd) throws IOException
|
||||
{
|
||||
/**
|
||||
* {@link BigTableWriter#createRowIndexEntry(org.apache.cassandra.db.DecoratedKey, DeletionTime, long)}
|
||||
* {@link BigTableWriter.IndexWriter#append(org.apache.cassandra.db.DecoratedKey, RowIndexEntry, long, java.nio.ByteBuffer)}
|
||||
*
|
||||
*/
|
||||
SequentialWriter indexFileWriter = indexWriter.writer;
|
||||
((BloomFilter)indexWriter.bf).add(key, 0, keyLength, reusableIndexes);
|
||||
long indexStart = indexFileWriter.position();
|
||||
try
|
||||
{
|
||||
ByteArrayUtil.writeWithShortLength(key, 0, keyLength, indexFileWriter);
|
||||
|
||||
indexFileWriter.writeUnsignedVInt(partitionStart);
|
||||
|
||||
// The trailing block must be counted BEFORE deciding whether to promote the index:
|
||||
// the iterator (BigFormatPartitionWriter.finish + RowIndexEntry.create) promotes when
|
||||
// the total block count INCLUDING the tail is > 1. A tail size of exactly 1 means only
|
||||
// the end-of-partition marker remains since the last cut (the iterator's
|
||||
// firstClustering == null case) and no tail block exists.
|
||||
// The tail width itself includes the end-of-partition marker byte, matching the
|
||||
// iterator, which indexes the final block AFTER SortedTablePartitionWriter.finish()
|
||||
// has written the marker.
|
||||
long tailBlockSize = (partitionEnd - partitionStart) - indexBlockStartOffset;
|
||||
boolean hasTailBlock = tailBlockSize > 1;
|
||||
int totalBlocks = rowIndexEntriesOffsets.size() + (hasTailBlock ? 1 : 0);
|
||||
|
||||
/** See: {@link org.apache.cassandra.io.sstable.format.big.RowIndexEntry#create} */
|
||||
if (totalBlocks <= 1)
|
||||
{
|
||||
/**
|
||||
* {@link RowIndexEntry#serialize(org.apache.cassandra.io.util.DataOutputPlus, java.nio.ByteBuffer)}
|
||||
*/
|
||||
indexFileWriter.writeUnsignedVInt32(0); // size
|
||||
}
|
||||
else {
|
||||
// add last block
|
||||
if (hasTailBlock) {
|
||||
addIndexBlock(partitionEnd, tailBlockSize, DeletionTime.LIVE);
|
||||
}
|
||||
// if we have intermeddiate index info elements we also need to serialize the partitionDeletionTime
|
||||
/** {@link RowIndexEntry.IndexedEntry#serialize(org.apache.cassandra.io.util.DataOutputPlus, java.nio.ByteBuffer) */
|
||||
// size up to the offsets?
|
||||
int endOfEntries = rowIndexEntries.getLength();
|
||||
// Write the headerLength, partitionDeletionTime and rowIndexEntriesOffsets.size() after the entries,
|
||||
// just to calculate size.
|
||||
rowIndexEntries.writeUnsignedVInt((long)headerLength);
|
||||
deletionTimeSerializer.serialize(partitionDeletionTime, rowIndexEntries);
|
||||
|
||||
rowIndexEntries.writeUnsignedVInt32(rowIndexEntriesOffsets.size()); // number of entries
|
||||
|
||||
// bytes until offsets
|
||||
int entriesAndOffsetsSize = rowIndexEntries.getLength() + rowIndexEntriesOffsets.size() * 4;
|
||||
assert entriesAndOffsetsSize > 0;
|
||||
indexFileWriter.writeUnsignedVInt32(entriesAndOffsetsSize); // size != 0
|
||||
// copy the header elements
|
||||
indexFileWriter.write(rowIndexEntries.getData(), endOfEntries, rowIndexEntries.getLength() - endOfEntries);
|
||||
indexFileWriter.write(rowIndexEntries.getData(), 0, endOfEntries);
|
||||
for (int i = 0; i < rowIndexEntriesOffsets.size(); i++)
|
||||
{
|
||||
int offset = rowIndexEntriesOffsets.get(i);
|
||||
indexFileWriter.writeInt(offset);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (IOException e)
|
||||
{
|
||||
throw new FSWriteError(e, indexFileWriter.getPath());
|
||||
}
|
||||
indexWriter.summary.maybeAddEntry(key, 0, keyLength, indexStart);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
/*
|
||||
* 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;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import org.apache.cassandra.db.DeletionTime;
|
||||
|
||||
/**
|
||||
* Format-specific partition/row index production for {@link SSTableCursorWriter}.
|
||||
*
|
||||
* The seam is EVENT-shaped because {@link UnfilteredDescriptor}s are transient (reused per
|
||||
* row): implementations that need block-boundary clusterings must capture them at
|
||||
* {@link #rowWritten} time, not at block boundaries. The cursor writer owns all data-file
|
||||
* serialization and the open range-tombstone marker; implementations own everything about
|
||||
* their index format (BIG: promoted index blocks + Index.db entries + bloom filter/summary;
|
||||
* other formats own their own index structures accordingly).
|
||||
*/
|
||||
public abstract class CursorIndexWriter
|
||||
{
|
||||
protected long partitionStart;
|
||||
// Offset within the current partition where the current index block begins.
|
||||
protected int indexBlockStartOffset;
|
||||
|
||||
public final void startPartition(long partitionStartPosition, long positionAfterHeader)
|
||||
{
|
||||
this.partitionStart = partitionStartPosition;
|
||||
reset();
|
||||
notePosition(positionAfterHeader);
|
||||
}
|
||||
|
||||
/** Static rows reset the block clock without participating in row index blocks. */
|
||||
public final void staticRowWritten(long position)
|
||||
{
|
||||
notePosition(position);
|
||||
}
|
||||
|
||||
public final int indexBlockStartOffset()
|
||||
{
|
||||
return indexBlockStartOffset;
|
||||
}
|
||||
|
||||
protected final void notePosition(long endOfRowPosition)
|
||||
{
|
||||
indexBlockStartOffset = (int) (endOfRowPosition - partitionStart);
|
||||
}
|
||||
|
||||
protected final long currentOffsetInPartition(long position)
|
||||
{
|
||||
return position - partitionStart;
|
||||
}
|
||||
|
||||
/** Clear per-partition state. Called by {@link #startPartition}. */
|
||||
protected abstract void reset();
|
||||
|
||||
/**
|
||||
* A non-static row or range tombstone marker was written to [rowStart, rowEnd).
|
||||
* The descriptor's clustering describes it; openMarker is the range deletion open at
|
||||
* the END of this unfiltered (LIVE if none).
|
||||
*/
|
||||
public abstract void rowWritten(UnfilteredDescriptor descriptor, long rowStart, long rowEnd,
|
||||
DeletionTime openMarker) throws IOException;
|
||||
|
||||
/** The partition (including its end-of-partition marker) ends at partitionEnd. */
|
||||
public abstract void endPartition(byte[] key, int keyLength, int headerLength,
|
||||
DeletionTime partitionDeletionTime, long partitionEnd) throws IOException;
|
||||
}
|
||||
|
|
@ -76,10 +76,6 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
private final DeletionTime.Serializer deletionTimeSerializer;
|
||||
private final MetadataCollector metadataCollector;
|
||||
private final SerializationHeader serializationHeader;
|
||||
/**
|
||||
* See: {@link BloomFilter#reusableIndexes}
|
||||
*/
|
||||
private final long[] reusableIndexes = new long[21];
|
||||
private final boolean hasStaticColumns;
|
||||
|
||||
private long partitionStart;
|
||||
|
|
@ -101,14 +97,9 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
private ColumnMetadata[] columns; // points to static/regular
|
||||
private int columnsWrittenCount = 0;
|
||||
private int nextCellIndex = 0;
|
||||
// Index info
|
||||
private final DataOutputBuffer rowIndexEntries = new DataOutputBuffer();
|
||||
private final IntArrayList rowIndexEntriesOffsets = new IntArrayList();
|
||||
private final ClusteringDescriptor rowIndexEntryLastClustering;
|
||||
private boolean hasDistinctLastClustering = false;
|
||||
private int indexBlockStartOffset;
|
||||
private int rowIndexEntryOffset;
|
||||
private final int indexBlockThreshold;
|
||||
// Format-specific index production (BIG: promoted blocks + Index.db + bloom filter/summary;
|
||||
// other formats own their own index structures accordingly)
|
||||
private final CursorIndexWriter cursorIndexWriter;
|
||||
|
||||
private SSTableCursorWriter(
|
||||
Descriptor desc,
|
||||
|
|
@ -127,8 +118,9 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
hasStaticColumns = serializationHeader.hasStatic();
|
||||
staticColumns = hasStaticColumns ? serializationHeader.columns(true).toArray(EMPTY_COL_META) : EMPTY_COL_META;
|
||||
regularColumns = serializationHeader.columns(false).toArray(EMPTY_COL_META);
|
||||
this.indexBlockThreshold = DatabaseDescriptor.getColumnIndexSize(BigFormatPartitionWriter.DEFAULT_GRANULARITY);
|
||||
rowIndexEntryLastClustering = new ClusteringDescriptor(serializationHeader.clusteringTypes().toArray(AbstractType[]::new));
|
||||
this.cursorIndexWriter = new BigCursorIndexWriter((BigTableWriter.IndexWriter) indexWriter,
|
||||
this.deletionTimeSerializer,
|
||||
new ClusteringDescriptor(serializationHeader.clusteringTypes().toArray(AbstractType[]::new)));
|
||||
}
|
||||
|
||||
public SSTableCursorWriter(SortedTableWriter<?,?> ssTableWriter)
|
||||
|
|
@ -164,17 +156,13 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
|
||||
public int writePartitionStart(byte[] partitionKey, int partitionKeyLength, DeletionTime partitionDeletionTime) throws IOException
|
||||
{
|
||||
rowIndexEntries.clear();
|
||||
rowIndexEntriesOffsets.clear();
|
||||
rowIndexEntryOffset = 0;
|
||||
openMarker.resetLive();
|
||||
hasDistinctLastClustering = false;
|
||||
|
||||
partitionStart = dataWriter.position();
|
||||
previousRowStartOffset = 0;
|
||||
writePartitionHeader(partitionKey, partitionKeyLength, partitionDeletionTime);
|
||||
updateIndexBlockStartOffset(dataWriter.position());
|
||||
return indexBlockStartOffset;
|
||||
cursorIndexWriter.startPartition(partitionStart, dataWriter.position());
|
||||
return cursorIndexWriter.indexBlockStartOffset();
|
||||
}
|
||||
|
||||
public void writePartitionEnd(byte[] partitionKey, int partitionKeyLength, DeletionTime partitionDeletionTime, int headerLength) throws IOException
|
||||
|
|
@ -194,82 +182,9 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
// this is implemented differently for BIG/BTI
|
||||
createRowIndexEntry(key, partitionLevelDeletion, partitionEnd - 1);
|
||||
*/
|
||||
appendBIGIndex(partitionKey, partitionKeyLength, partitionStart, headerLength, partitionDeletionTime, partitionEnd);
|
||||
cursorIndexWriter.endPartition(partitionKey, partitionKeyLength, headerLength, partitionDeletionTime, partitionEnd);
|
||||
}
|
||||
|
||||
private void appendBIGIndex(byte[] key, int keyLength, long partitionStart, int headerLength, DeletionTime partitionDeletionTime, long partitionEnd) throws IOException
|
||||
{
|
||||
/**
|
||||
* {@link BigTableWriter#createRowIndexEntry(DecoratedKey, DeletionTime, long)}
|
||||
* {@link BigTableWriter.IndexWriter#append(DecoratedKey, RowIndexEntry, long, ByteBuffer)}
|
||||
*
|
||||
*/
|
||||
BigTableWriter.IndexWriter indexWriter = (BigTableWriter.IndexWriter) this.indexWriter;
|
||||
SequentialWriter indexFileWriter = indexWriter.writer;
|
||||
((BloomFilter)indexWriter.bf).add(key, 0, keyLength, reusableIndexes);
|
||||
long indexStart = indexFileWriter.position();
|
||||
try
|
||||
{
|
||||
ByteArrayUtil.writeWithShortLength(key, 0, keyLength, indexFileWriter);
|
||||
|
||||
indexFileWriter.writeUnsignedVInt(partitionStart);
|
||||
|
||||
// The trailing block must be counted BEFORE deciding whether to promote the index:
|
||||
// the iterator (BigFormatPartitionWriter.finish + RowIndexEntry.create) promotes when
|
||||
// the total block count INCLUDING the tail is > 1. A tail size of exactly 1 means only
|
||||
// the end-of-partition marker remains since the last cut (the iterator's
|
||||
// firstClustering == null case) and no tail block exists.
|
||||
// The tail width itself includes the end-of-partition marker byte, matching the
|
||||
// iterator, which indexes the final block AFTER SortedTablePartitionWriter.finish()
|
||||
// has written the marker.
|
||||
long tailBlockSize = (partitionEnd - partitionStart) - indexBlockStartOffset;
|
||||
boolean hasTailBlock = tailBlockSize > 1;
|
||||
int totalBlocks = rowIndexEntriesOffsets.size() + (hasTailBlock ? 1 : 0);
|
||||
|
||||
/** See: {@link org.apache.cassandra.io.sstable.format.big.RowIndexEntry#create} */
|
||||
if (totalBlocks <= 1)
|
||||
{
|
||||
/**
|
||||
* {@link RowIndexEntry#serialize(DataOutputPlus, ByteBuffer)}
|
||||
*/
|
||||
indexFileWriter.writeUnsignedVInt32(0); // size
|
||||
}
|
||||
else {
|
||||
// add last block
|
||||
if (hasTailBlock) {
|
||||
addIndexBlock(partitionEnd, tailBlockSize);
|
||||
}
|
||||
// if we have intermeddiate index info elements we also need to serialize the partitionDeletionTime
|
||||
/** {@link RowIndexEntry.IndexedEntry#serialize(DataOutputPlus, ByteBuffer) */
|
||||
// size up to the offsets?
|
||||
int endOfEntries = rowIndexEntries.getLength();
|
||||
// Write the headerLength, partitionDeletionTime and rowIndexEntriesOffsets.size() after the entries,
|
||||
// just to calculate size.
|
||||
rowIndexEntries.writeUnsignedVInt((long)headerLength);
|
||||
deletionTimeSerializer.serialize(partitionDeletionTime, rowIndexEntries);
|
||||
|
||||
rowIndexEntries.writeUnsignedVInt32(rowIndexEntriesOffsets.size()); // number of entries
|
||||
|
||||
// bytes until offsets
|
||||
int entriesAndOffsetsSize = rowIndexEntries.getLength() + rowIndexEntriesOffsets.size() * 4;
|
||||
assert entriesAndOffsetsSize > 0;
|
||||
indexFileWriter.writeUnsignedVInt32(entriesAndOffsetsSize); // size != 0
|
||||
// copy the header elements
|
||||
indexFileWriter.write(rowIndexEntries.getData(), endOfEntries, rowIndexEntries.getLength() - endOfEntries);
|
||||
indexFileWriter.write(rowIndexEntries.getData(), 0, endOfEntries);
|
||||
for (int i = 0; i < rowIndexEntriesOffsets.size(); i++)
|
||||
{
|
||||
int offset = rowIndexEntriesOffsets.get(i);
|
||||
indexFileWriter.writeInt(offset);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (IOException e)
|
||||
{
|
||||
throw new FSWriteError(e, indexFileWriter.getPath());
|
||||
}
|
||||
indexWriter.summary.maybeAddEntry(key, 0, keyLength, indexStart);
|
||||
}
|
||||
|
||||
final long guardrailsPartitionSizeWarning = Guardrails.partitionSize.warnValue(null);
|
||||
final long guardrailsPartitionTombstonesWarning = Guardrails.partitionTombstones.warnValue(null);
|
||||
|
|
@ -325,7 +240,7 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
missingColumns.clear();
|
||||
writeRowEnd(null, false);
|
||||
|
||||
updateIndexBlockStartOffset(dataWriter.position());
|
||||
cursorIndexWriter.staticRowWritten(dataWriter.position());
|
||||
return true;
|
||||
}
|
||||
|
||||
|
|
@ -530,7 +445,7 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
|
||||
if (isStatic)
|
||||
{
|
||||
updateIndexBlockStartOffset(dataWriter.position());
|
||||
cursorIndexWriter.staticRowWritten(dataWriter.position());
|
||||
}
|
||||
else
|
||||
{
|
||||
|
|
@ -598,22 +513,7 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
boolean updateClusteringMetadata) throws IOException
|
||||
{
|
||||
if (updateClusteringMetadata) updateClusteringMetadata(unfilteredDescriptor);
|
||||
// write the first clustering into rowIndexEntries buffer (we will need it unless we never write the first entry)
|
||||
if (currentOffsetInPartition(unfilteredStartPosition) == indexBlockStartOffset || (rowIndexEntryOffset == rowIndexEntries.position()))
|
||||
{
|
||||
writeClusteringToRowIndexEntries(unfilteredDescriptor);
|
||||
}
|
||||
else
|
||||
{
|
||||
rowIndexEntryLastClustering.copy(unfilteredDescriptor);
|
||||
hasDistinctLastClustering = true;
|
||||
}
|
||||
|
||||
/** {@link BigFormatPartitionWriter#addUnfiltered(Unfiltered)} */
|
||||
// if we hit the index block size that we have to index after, go ahead and index it.
|
||||
long indexBlockSize = currentOffsetInPartition(unfilteredEndPosition) - indexBlockStartOffset;
|
||||
if (indexBlockSize >= this.indexBlockThreshold)
|
||||
addIndexBlock(unfilteredEndPosition, indexBlockSize);
|
||||
cursorIndexWriter.rowWritten(unfilteredDescriptor, unfilteredStartPosition, unfilteredEndPosition, openMarker);
|
||||
}
|
||||
|
||||
public void updateClusteringMetadata(UnfilteredDescriptor unfilteredDescriptor)
|
||||
|
|
@ -621,61 +521,6 @@ public class SSTableCursorWriter implements AutoCloseable
|
|||
metadataCollector.updateClusteringValues(unfilteredDescriptor);
|
||||
}
|
||||
|
||||
/**
|
||||
* See:
|
||||
* {@link BigFormatPartitionWriter#addIndexBlock()}
|
||||
* - {@link org.apache.cassandra.io.sstable.IndexInfo.Serializer#serialize(org.apache.cassandra.io.sstable.IndexInfo, org.apache.cassandra.io.util.DataOutputPlus)}
|
||||
*/
|
||||
private void addIndexBlock(long endOfRowPosition, long indexBlockSize) throws IOException
|
||||
{
|
||||
if (rowIndexEntriesOffsets.isEmpty() && rowIndexEntryOffset != 0) {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
|
||||
// serialize the index info
|
||||
/** {@link org.apache.cassandra.io.sstable.IndexInfo.Serializer#serialize(org.apache.cassandra.io.sstable.IndexInfo, org.apache.cassandra.io.util.DataOutputPlus)}*/
|
||||
rowIndexEntriesOffsets.addInt(rowIndexEntryOffset);
|
||||
|
||||
// first clustering is already in, write last entry
|
||||
if (!hasDistinctLastClustering)
|
||||
{
|
||||
// first entry is the last entry, copy it
|
||||
byte[] entriesData = rowIndexEntries.getData();
|
||||
long endOfFirstEntry = rowIndexEntries.position();
|
||||
rowIndexEntries.write(entriesData, rowIndexEntryOffset, (int) (endOfFirstEntry - rowIndexEntryOffset));
|
||||
}
|
||||
else
|
||||
{
|
||||
writeClusteringToRowIndexEntries(rowIndexEntryLastClustering);
|
||||
rowIndexEntryLastClustering.resetClustering();
|
||||
}
|
||||
hasDistinctLastClustering = false;
|
||||
|
||||
rowIndexEntries.writeUnsignedVInt((long)indexBlockStartOffset);
|
||||
rowIndexEntries.writeVInt(indexBlockSize - IndexInfo.Serializer.WIDTH_BASE);
|
||||
|
||||
boolean isDeleteTimePresent = !openMarker.isLive();
|
||||
rowIndexEntries.writeBoolean(isDeleteTimePresent);
|
||||
if (isDeleteTimePresent)
|
||||
deletionTimeSerializer.serialize(openMarker, rowIndexEntries);
|
||||
// next block starts
|
||||
rowIndexEntryOffset = Ints.checkedCast(rowIndexEntries.position());
|
||||
updateIndexBlockStartOffset(endOfRowPosition);
|
||||
}
|
||||
|
||||
private void updateIndexBlockStartOffset(long endOfRowPosition)
|
||||
{
|
||||
indexBlockStartOffset = (int) (endOfRowPosition - partitionStart);
|
||||
}
|
||||
|
||||
private void writeClusteringToRowIndexEntries(ClusteringDescriptor clustering) throws IOException
|
||||
{
|
||||
ClusteringPrefix.Kind kind = clustering.clusteringKind();
|
||||
rowIndexEntries.writeByte(kind.ordinal());
|
||||
if (kind != ClusteringPrefix.Kind.CLUSTERING)
|
||||
rowIndexEntries.writeShort(clustering.clusteringColumnsBound());
|
||||
rowIndexEntries.write(clustering.clusteringBytes(), 0, clustering.clusteringLength());
|
||||
}
|
||||
|
||||
private long currentOffsetInPartition(long position)
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in New Issue