CASSANDRA-21550 Phase1 BIG splitter only

doc/modules/cassandra/pages/managing/tools/sstable/sstablesplit.adoc

@@ -24,6 +24,9 @@ sstablesplit <options> <filename>
 
 |-s, --size <size> |maximum size in MB for the output sstables (default:
 50)
+
+|--zero-copy |split a compressed BIG-format sstable without deserializing or
+recompressing rows
 |===
 
 This command should be run with Cassandra stopped. Note: the script does
@@ -94,3 +97,56 @@ sstablesplit --size 1 --no-snapshot /var/lib/cassandra/data/keyspace/eventlog-63
 
 Note: There is no output, but you can see the results in your file
 system.
+
+== Zero-copy Split
+
+Use `--zero-copy` to split a compressed BIG-format sstable by retaining its
+compression chunks and rebuilding its index and metadata. On Linux filesystems
+that support byte-range reflinks, the tool shares the data extents with the
+children. If reflinks are unavailable, it copies the compressed bytes instead;
+rows are still not deserialized or recompressed.
+
+Zero-copy splitting requires `storage_compatibility_mode: NONE` and a BIG parent
+whose exact on-disk version is supported by the splitter. It refuses uncompressed
+sstables, unknown or newer BIG versions, SSTable-attached indexes, and components
+it cannot reproduce. Split children use the `qa` format version. Binaries that do
+not understand the retained-prefix layout reject these children as an incompatible
+major version instead of attempting to read them. Do not use this option while a
+rolling upgrade is in progress. A rollback to software that supports only `p`-era
+sstables requires restoring the pre-split snapshot, when one was created; using
+`--no-snapshot` leaves no rollback copy. Older software cannot open `qa` children.
+
+The requested size remains a maximum unless one partition's compression-chunk
+span is itself larger than that size.
+
+A compression chunk crossed by a split boundary is retained by both adjacent
+children, although each child exposes only its own indexed partitions. Reflink
+filesystems share those blocks; the copy fallback duplicates the boundary
+chunk. A later normal compaction rewrites the children and reclaims the retained
+bytes.
+
+The splitter rebuilds the partition-size estimate, compression ratio, key range,
+index summary, and Bloom filter for each child. Statistics that require decoding
+the rows cannot be recovered without giving up the zero-copy algorithm, so each
+child inherits these parent-wide values: cell-count and tombstone histograms,
+total rows and columns, timestamp, TTL and deletion-time bounds, covered
+clustering bounds, partition-deletion and legacy-counter flags, commit-log
+intervals, originating host, and sstable level.
+
+Consequently, totals that are summed across sstables can be over-reported until
+the children are compacted, and single-sstable tombstone compaction can be less
+likely to select a child. Zero-copy splitting also does not purge tombstones.
+Where prompt tombstone reclamation matters, run a normal compaction after the
+split or temporarily adjust the table's tombstone-compaction settings. A normal
+compaction rewrites the rows and computes exact per-child statistics.
+
+After committing each input, the tool reports the number of children, bytes
+reflinked, bytes written, and whether any reflink was used. `reflink used=no`
+means the Data.db ranges were copied and should be budgeted as ordinary read,
+write, and temporary disk traffic.
+
+Example:
+
+....
+sstablesplit --zero-copy --size 1024 /var/lib/cassandra/data/keyspace/table/pa-1-big-Data.db
+....

src/java/org/apache/cassandra/db/compaction/CompactionManager.java

@@ -711,8 +711,8 @@ public AllSSTableOpStatus performSSTableRewrite(final ColumnFamilyStore cfs,
                                                     int jobs) throws InterruptedException, ExecutionException
     {
         return performSSTableRewrite(cfs, (sstable) -> {
-            // Skip if descriptor version matches current version
-            if (skipIfCurrentVersion && sstable.descriptor.version.equals(sstable.descriptor.getFormat().getLatestVersion()))
+            // Skip if the SSTable already uses a version considered current by its format
+            if (skipIfCurrentVersion && Upgrader.isCurrentVersion(sstable.descriptor))
                 return false;
 
             // Skip if SSTable creation time is past given timestamp

src/java/org/apache/cassandra/db/compaction/StatefulCursor.java

@@ -68,6 +68,7 @@ class StatefulCursor extends SSTableCursorReader
     public StatefulCursor(SSTableReader reader, DiskAccessMode diskAccessMode)
     {
         super(reader, diskAccessMode);
+        bytesReadPositionSnapshot = position();
         currPartition = new PartitionDescriptor(reader.getPartitioner().createReusableKey(0));
         prevPartition = new PartitionDescriptor(reader.getPartitioner().createReusableKey(0));
         unfiltered = new UnfilteredDescriptor(reader.header.clusteringTypes().toArray(AbstractType[]::new));

src/java/org/apache/cassandra/db/compaction/Upgrader.java

@@ -29,6 +29,7 @@
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.SSTableRewriter;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.format.SSTableWriter;
 import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
@@ -69,6 +70,22 @@ public Upgrader(ColumnFamilyStore cfs, LifecycleTransaction txn, OutputHandler o
         this.estimatedRows = (long) Math.ceil((double) estimatedTotalKeys / estimatedSSTables);
     }
 
+    /**
+     * Whether an SSTable uses a version considered current by its format.
+     */
+    public static boolean isCurrentVersion(Descriptor descriptor)
+    {
+        return descriptor.version.isLatestVersion();
+    }
+
+    /**
+     * Whether an SSTable uses a current version of the selected output format.
+     */
+    public static boolean isCurrentVersion(Descriptor descriptor, SSTableFormat<?, ?> selectedFormat)
+    {
+        return descriptor.getFormat().name().equals(selectedFormat.name()) && isCurrentVersion(descriptor);
+    }
+
     private SSTableWriter createCompactionWriter(StatsMetadata metadata)
     {
         MetadataCollector sstableMetadataCollector = new MetadataCollector(cfs.getComparator());
@@ -130,4 +147,3 @@ public LongPredicate getPurgeEvaluator(DecoratedKey key)
         }
     }
 }
-

src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java

@@ -34,6 +34,8 @@
 import org.apache.cassandra.streaming.StreamOperation;
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.streaming.StreamingDataOutputPlus;
+import org.apache.cassandra.utils.CassandraVersion;
+import org.apache.cassandra.utils.StorageCompatibilityMode;
 import org.apache.cassandra.utils.TimeUUID;
 import org.apache.cassandra.utils.concurrent.Ref;
 
@@ -42,6 +44,8 @@
  */
 public class CassandraOutgoingFile implements OutgoingStream
 {
+    private static final int SPLIT_PREFIX_CAPABLE_MAJOR = 7;
+
     private final Ref<SSTableReader> ref;
     private final long estimatedKeys;
     private final List<SSTableReader.PartitionPositionBounds> sections;
@@ -50,10 +54,18 @@ public class CassandraOutgoingFile implements OutgoingStream
     private final StreamOperation operation;
     private final CassandraStreamHeader header;
     private final List<Range<Token>> ranges;
+    private final boolean peerSupportsSplitPrefix;
 
     public CassandraOutgoingFile(StreamOperation operation, Ref<SSTableReader> ref,
                                  List<SSTableReader.PartitionPositionBounds> sections, List<Range<Token>> normalizedRanges,
                                  long estimatedKeys)
+    {
+        this(operation, ref, sections, normalizedRanges, estimatedKeys, null);
+    }
+
+    public CassandraOutgoingFile(StreamOperation operation, Ref<SSTableReader> ref,
+                                 List<SSTableReader.PartitionPositionBounds> sections, List<Range<Token>> normalizedRanges,
+                                 long estimatedKeys, CassandraVersion peerVersion)
     {
         Preconditions.checkNotNull(ref.get());
         Range.assertNormalized(normalizedRanges);
@@ -62,6 +74,7 @@ public CassandraOutgoingFile(StreamOperation operation, Ref<SSTableReader> ref,
         this.estimatedKeys = estimatedKeys;
         this.sections = sections;
         this.ranges = normalizedRanges;
+        this.peerSupportsSplitPrefix = peerVersion != null && peerVersion.major >= SPLIT_PREFIX_CAPABLE_MAJOR;
 
         SSTableReader sstable = ref.get();
 
@@ -188,11 +201,14 @@ public void write(StreamSession session, StreamingDataOutputPlus out, int versio
     @VisibleForTesting
     public boolean computeShouldStreamEntireSSTables()
     {
-        // don't stream if full sstable transfers are disabled, legacy counter shards are present,
-        // or sstable uses old bloom filter format (pre-4.0) which is incompatible with zero-copy streaming
+        // A pre-7.0 reader does not understand the split-prefix position or the qa descriptor that protects it. Fall
+        // back to partition streaming, which rewrites rows in the peer's format, unless the peer advertises support.
         if (!DatabaseDescriptor.streamEntireSSTables() ||
             ref.get().getSSTableMetadata().hasLegacyCounterShards ||
-            ref.get().descriptor.version.hasOldBfFormat())
+            ref.get().descriptor.version.hasOldBfFormat() ||
+            (ref.get().descriptor.version.hasSplitPrefixMarker()
+             && (DatabaseDescriptor.getStorageCompatibilityMode() != StorageCompatibilityMode.NONE
+                 || !peerSupportsSplitPrefix)))
             return false;
 
         return contained(sections, ref.get());
@@ -204,9 +220,19 @@ public boolean contained(List<SSTableReader.PartitionPositionBounds> sections, S
         if (sections == null || sections.isEmpty())
             return false;
 
-        // if transfer sections contain entire sstable
+        // MOVED_START hides an already-compacted prefix without removing it from the physical components. Sending
+        // those components whole would restore data this reader no longer owns on the receiving node.
+        if (sstable.openReason == SSTableReader.OpenReason.MOVED_START)
+            return false;
+
         long transferLength = sections.stream().mapToLong(p -> p.upperPosition - p.lowerPosition).sum();
-        return transferLength == sstable.uncompressedLength();
+        if (!sstable.hasSplitPrefix())
+            return transferLength == sstable.uncompressedLength();
+
+        // A split child may retain a dead prefix from its first compression chunk. Compare with the full live span
+        // rather than the physical data length so that such a child remains eligible for entire-SSTable streaming.
+        SSTableReader.PartitionPositionBounds fullRange = sstable.getPositionsForFullRange();
+        return fullRange != null && transferLength == fullRange.upperPosition - fullRange.lowerPosition;
     }
 
     @Override

src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java

@@ -41,6 +41,7 @@
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
+import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.RangesAtEndpoint;
 import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.service.ActiveRepairService;
@@ -51,6 +52,11 @@
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.streaming.TableStreamManager;
 import org.apache.cassandra.streaming.messages.StreamMessageHeader;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.Directory;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.utils.CassandraVersion;
 import org.apache.cassandra.utils.TimeUUID;
 import org.apache.cassandra.utils.concurrent.Ref;
 import org.apache.cassandra.utils.concurrent.Refs;
@@ -141,6 +147,7 @@ else if (pendingRepair == ActiveRepairService.NO_PENDING_REPAIR)
 
             List<Range<Token>> normalizedFullRanges = Range.normalize(replicas.onlyFull().ranges());
             List<Range<Token>> normalizedAllRanges = Range.normalize(replicas.ranges());
+            CassandraVersion peerVersion = peerVersion(session.peer);
             //Create outgoing file streams for ranges possibly skipping repaired ranges in sstables
             List<OutgoingStream> streams = new ArrayList<>(refs.size());
             for (SSTableReader sstable : refs)
@@ -155,7 +162,7 @@ else if (pendingRepair == ActiveRepairService.NO_PENDING_REPAIR)
                     continue;
                 }
                 streams.add(new CassandraOutgoingFile(session.getStreamOperation(), ref, sections, ranges,
-                                                      sstable.estimatedKeysForRanges(ranges)));
+                                                      sstable.estimatedKeysForRanges(ranges), peerVersion));
             }
 
             return streams;
@@ -166,4 +173,16 @@ else if (pendingRepair == ActiveRepairService.NO_PENDING_REPAIR)
             throw t;
         }
     }
+
+    private static CassandraVersion peerVersion(InetAddressAndPort peer)
+    {
+        ClusterMetadata metadata = ClusterMetadata.currentNullable();
+        if (metadata == null)
+            return null;
+
+        Directory directory = metadata.directory;
+        NodeId peerId = directory.peerId(peer);
+        NodeVersion version = peerId == null ? null : directory.version(peerId);
+        return version == null ? null : version.cassandraVersion;
+    }
 }

src/java/org/apache/cassandra/index/SecondaryIndexManager.java

@@ -578,6 +578,24 @@ public boolean validateSSTableAttachedIndexes(Collection<SSTableReader> sstables
         return complete;
     }
 
+    /**
+     * Returns whether this table has an index that declares itself SSTable-attached through
+     * {@link Index#isSSTableAttached()}.
+     * <p>
+     * This is intentionally a table-level check rather than an inspection of an SSTable's components. It remains
+     * true while an index build is in progress, before the index components exist on every SSTable.
+     */
+    public boolean hasSSTableAttachedIndexes()
+    {
+        for (Index.Group group : indexGroups.values())
+        {
+            if (group.getIndexes().stream().anyMatch(Index::isSSTableAttached))
+                return true;
+        }
+
+        return false;
+    }
+
     /**
      * Incrementally builds indexes for the specified SSTables in a blocking fashion.
      * <p>

src/java/org/apache/cassandra/io/compress/CompressionMetadata.java

@@ -218,6 +218,16 @@ public ICompressor compressor()
         }
     }
 
+    /**
+     * Returns the dictionary used by the chunks described by this metadata, if any.
+     * Copied chunks must retain the same dictionary to remain readable.
+     */
+    @Nullable
+    public CompressionDictionary compressionDictionary()
+    {
+        return compressionDictionary;
+    }
+
     static ICompressor resolveCompressor(ICompressor compressor, CompressionDictionary dictionary)
     {
         if (dictionary == null)

src/java/org/apache/cassandra/io/sstable/SSTableCursorReader.java

@@ -323,6 +323,7 @@ int readCellHeader() throws IOException
     private final SSTableReader ssTableReader;
     private final RandomAccessReader dataReader;
     private final DeletionTime.Serializer deletionTimeSerializer;
+    private final long firstPartitionPosition;
 
     private final CellCursor staticRowCellCursor = new CellCursor();
     private final CellCursor rowCellCursor = new CellCursor();
@@ -357,7 +358,16 @@ public static SSTableCursorReader fromDescriptor(Descriptor desc) throws IOExcep
     {
         TableMetadata metadata = Util.metadataFromSSTable(desc);
         SSTableReader reader = SSTableReader.openNoValidation(null, desc, TableMetadataRef.forOfflineTools(metadata));
-        return new SSTableCursorReader(reader, metadata, reader.ref(), null);
+        Ref<SSTableReader> ref = reader.selfRef();
+        try
+        {
+            return new SSTableCursorReader(reader, metadata, ref, null);
+        }
+        catch (RuntimeException | Error e)
+        {
+            ref.close();
+            throw e;
+        }
     }
 
     public SSTableCursorReader(SSTableReader reader)
@@ -387,7 +397,21 @@ private SSTableCursorReader(SSTableReader reader, TableMetadata metadata, Ref<SS
         serializationHeader = reader.header;
         sstableHasDroppedColumns = anyDroppedColumn(deserializationHelper, serializationHeader);
 
+        SSTableReader.PartitionPositionBounds fullRange = reader.getPositionsForFullRange();
+        // A null range means MOVED_START has consumed the whole logical reader. Pin its only legal seek to physical
+        // EOF so a caller cannot move a DONE cursor back into bytes this reader no longer owns.
+        firstPartitionPosition = fullRange == null ? reader.uncompressedLength() : fullRange.lowerPosition;
         dataReader = reader.openDataReaderForScan(diskAccessMode);
+        try
+        {
+            if (fullRange == null || firstPartitionPosition > 0 || reader.uncompressedLength() == 0)
+                seekPartitionInRange(firstPartitionPosition);
+        }
+        catch (RuntimeException | Error e)
+        {
+            dataReader.close();
+            throw e;
+        }
         // the HEADER decides whether this sstable can contain static rows: after
         // ALTER TABLE ... DROP of the last static column, current metadata has no static
         // columns but older sstables legitimately still carry static rows
@@ -423,6 +447,24 @@ private void resetOnPartitionStart()
 
     public int seekPartition(long position)
     {
+        long endPosition = uncompressedLength();
+        if (position < firstPartitionPosition || position > endPosition)
+            throw new IllegalArgumentException("Cannot seek outside cursor range [" + firstPartitionPosition +
+                                               ", " + endPosition + "]: " + position);
+
+        return seekPartitionInRange(position);
+    }
+
+    private int seekPartitionInRange(long position)
+    {
+        if (position == uncompressedLength())
+        {
+            dataReader.seek(position);
+            state = DONE;
+            resetOnPartitionStart();
+            return state;
+        }
+
         state = SEEK;
         if (position == 0)
         {
@@ -441,7 +483,11 @@ public int seekPartition(long position)
                 return corruptSSTable(e);
             }
             // end of partition
-            if (!UnfilteredSerializer.isEndOfPartition(basicUnfilteredFlags)) {
+            if (!UnfilteredSerializer.isEndOfPartition(basicUnfilteredFlags))
+            {
+                if (position == firstPartitionPosition)
+                    return corruptSSTable("Authenticated first partition at " + position +
+                                          " is not preceded by an end-of-partition marker");
                 throw new IllegalArgumentException("Seeking to a partition at: " + position + " did not result in a valid state");
             }
             state = dataReader.isEOF() ? DONE : PARTITION_START;