From efe20e858bff5e19047a04b0a54f720e56f34489 Mon Sep 17 00:00:00 2001 From: Jon Haddad Date: Wed, 10 Jun 2026 17:14:34 -0700 Subject: [PATCH] Extract format-specific index production from the cursor writer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../io/sstable/BigCursorIndexWriter.java | 226 ++++++++++++++++++ .../io/sstable/CursorIndexWriter.java | 83 +++++++ .../io/sstable/SSTableCursorWriter.java | 179 +------------- 3 files changed, 321 insertions(+), 167 deletions(-) create mode 100644 src/java/org/apache/cassandra/io/sstable/BigCursorIndexWriter.java create mode 100644 src/java/org/apache/cassandra/io/sstable/CursorIndexWriter.java diff --git a/src/java/org/apache/cassandra/io/sstable/BigCursorIndexWriter.java b/src/java/org/apache/cassandra/io/sstable/BigCursorIndexWriter.java new file mode 100644 index 0000000000..9ebd69afc0 --- /dev/null +++ b/src/java/org/apache/cassandra/io/sstable/BigCursorIndexWriter.java @@ -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); + } +} diff --git a/src/java/org/apache/cassandra/io/sstable/CursorIndexWriter.java b/src/java/org/apache/cassandra/io/sstable/CursorIndexWriter.java new file mode 100644 index 0000000000..62cbec6d64 --- /dev/null +++ b/src/java/org/apache/cassandra/io/sstable/CursorIndexWriter.java @@ -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; +} diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java index 814b3195b4..0da0c55977 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java @@ -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) {