CASSANDRA-21550 - Zero copy sstable split

CHANGES.txt

@@ -1,4 +1,5 @@
 7.0
+ * Zero-copy sstable splitting for anticompaction, and partial-sstable zero-copy streaming (CASSANDRA-XXXXX)
  * Don't increment client metrics on messaging service connection unpause (CASSANDRA-21491)
  * Add nodetool getreplicas (CASSANDRA-17665)
  * Implementation of CEP-49: Hardware-accelerated compression (CASSANDRA-20975)

NEWS.txt

@@ -86,6 +86,26 @@ New features
 
 Upgrading
 ---------
+    - The BIG sstable format has a new version, pb, and the BTI format a new version, eb. Both add a marker to
+      Statistics.db recording whether the sstable contains regions its index does not describe. New sstables are
+      written at those versions unless storage_compatibility_mode pins BIG to an earlier one (nb or oa). Older
+      sstables stay readable and are not rewritten until you run 'nodetool upgradesstables'.
+    - Zero-copy anticompaction (zero_copy_anticompaction_enabled) and partial-sstable zero-copy streaming
+      (zero_copy_partial_stream_enabled) produce, and accept, sstables that carry bytes their index does not
+      describe. Both are DISABLED by default and both are a ROLLOUT-ORDER CONTRACT that nothing negotiates over the
+      streaming protocol: enabling either one on a node while any peer that can RECEIVE a stream from it is still on
+      a build without this feature risks that peer misreading such an sstable. Reads there are unaffected, but
+      ordinary compaction enters Data.db mid-partition and fails with CorruptSSTableException, marking the sstable
+      suspect, and 'nodetool verify' and 'nodetool scrub' fail on it as well. Do not enable either flag until every
+      node that can receive a stream from the node you are enabling it on has been upgraded. If a peer has already
+      accepted such an sstable, finishing that peer's upgrade is the remedy: once it has this feature it tolerates
+      the extra bytes and compacts, verifies and scrubs the sstable normally.
+    - Zero-copy anticompaction only splits an sstable that is COMPRESSED, in the BIG or BTI format, and at version
+      pb or eb, so on a cluster upgraded to 7.0 the flag is INERT over every pre-existing sstable until 'nodetool
+      upgradesstables' has rewritten them. It also retains droppable tombstones that a rewriting anticompaction
+      would have purged, and gives its output inherited rather than recomputed cell, row and timestamp statistics.
+      Read the zero_copy_anticompaction_enabled comment in conf/cassandra.yaml for the full list of operational
+      consequences before enabling it.
 
 Deprecation
 -----------

conf/cassandra.yaml

@@ -1292,6 +1292,82 @@ sstable_preemptive_open_interval: 50MiB
 # and eventually get removed from the configuration.
 uuid_sstable_identifiers_enabled: false
 
+# Anticompact by copying compression chunks verbatim instead of rewriting rows, when an sstable's full / transient /
+# unrepaired partitions form contiguous token runs; interleaved ranges (what vnodes produce) fall back to the rewrite.
+# The copy is still throttled by compaction_throughput, is an ANTICOMPACTION in compactionstats, and stops on nodetool
+# stop ANTICOMPACTION / --id, TRUNCATE and DROP.
+#
+# Only a COMPRESSED sstable in the BIG or BTI format is eligible. An uncompressed table makes this flag a SILENT
+# NO-OP: an uncompressed split is a different algorithm - the cut is exact rather than chunk-aligned, and CRC.db,
+# whose 64KiB grid is addressed from origin 0, has to be regenerated wholesale rather than sliced - so the split
+# declines rather than emit a child with a misaligned CRC.db. A compression dictionary (CEP-49) declines for a
+# related reason: a chunk is only decompressible against the dictionary its parent recorded. Eligibility also needs
+# the parent's Statistics.db to carry the unindexed-regions marker, which arrived with sstable version pb (BIG) and
+# eb (BTI), so on an upgraded cluster this stays INERT over every pre-existing sstable until `nodetool
+# upgradesstables` has rewritten it.
+#
+# Off by default, because for the sstables it does handle it changes what anticompaction does:
+#
+# - It NO LONGER PURGES TOMBSTONES for the sstables it handles: droppable tombstones and shadowed data are retained,
+#   so disk usage can rise until the children are compacted normally. Retention only - nothing is resurrected.
+# - Children from the middle or end of the parent begin with a dead prefix: the bytes before their first indexed
+#   partition, the head of the copied boundary compression chunk. Every read path on this build tolerates it, and such
+#   a child can still be zero-copy streamed whole, so this is a rollout-order contract that nothing negotiates over
+#   the streaming protocol. On a peer WITHOUT the splitter's scrubber and verifier seeks, compaction enters Data.db
+#   mid-partition and fails with CorruptSSTableException, marking it suspect; `nodetool verify` and `nodetool scrub`
+#   fail too. Leave off until every node that can receive a stream from this one has those seeks.
+# - Child statistics are deliberately imprecise: recomputing them means deserializing every row, the cost this path
+#   avoids. Partition count and partition size histogram are exact; cell count, row count and the tombstone-drop
+#   histogram are inherited whole from the parent. Per-table aggregates (tablestats mean row count, estimated column
+#   count, droppable tombstone ratio) over-report by roughly the child count, and single-sstable tombstone compaction
+#   fires less readily than tombstone_threshold suggests, so the tombstones retained above are purged later; lower
+#   tombstone_threshold or set unchecked_tombstone_compaction on tables that mind. All errors are conservative; none
+#   can lose or resurrect data.
+# - maxTimestamp and maxLocalDeletionTime are inherited whole as well, and those two have effects of their own. A
+#   child every one of whose partitions has expired is no longer dropped as a unit by the fully-expired-sstable
+#   check, so it waits for an ordinary compaction to remove it - more retention, on top of the tombstones above.
+#   And under TWCS every child is bucketed into the PARENT's time window however narrow its own data is: a split
+#   never scatters children across windows, but neither does it let a child join the window its own timestamps
+#   belong to.
+# - A table with a STORAGE-ATTACHED INDEX is never handled by this path and always takes the rewrite, which writes
+#   index components inline. A split rebuilds only the components it knows about, so a child would be live and readable
+#   with no index components at all - and SAI leaves such an sstable out of its view without reporting it invalid, so
+#   every row in it would match no index predicate, silently. The refusal is logged at INFO once per anticompaction,
+#   and BytesZeroCopyAnticompaction therefore stays 0 for indexed tables. Backed up by a second refusal on any
+#   component the splitter cannot itself produce, so a future component type declines rather than being dropped.
+# zero_copy_anticompaction_enabled: false
+
+# Have the zero-copy splitter SHARE each child's Data.db extents with the parent instead of copying them, via the
+# Linux FICLONERANGE ioctl ("reflink"): the split writes no data blocks and needs no additional disk space, so a 1 TiB
+# anticompaction stops needing 1 TiB of headroom. Needs a filesystem that can share extents - xfs made with `-m
+# reflink=1` (the mkfs default since xfsprogs 5.1), or btrfs - and is discovered by trying. A filesystem that simply
+# cannot share extents answers EOPNOTSUPP, ENOTTY, ENOSYS or EXDEV; that verdict is remembered per data directory,
+# logged once at INFO, and every later split in that directory copies as before, as on ext4. EINVAL and EPERM are
+# deliberately NOT remembered, because neither says anything about the filesystem - EINVAL is the kernel's answer to
+# any bad argument and EPERM is per file - so they are logged at WARN on EVERY occurrence and retried on every split.
+# Either way the range is copied and the child is identical, which is why this sub-knob of an already opted-in
+# operation defaults to true; turn it off only for these side effects:
+#
+# - Up to 64 KiB of alignment padding at the head of each child's Data.db - the ioctl needs block-aligned offsets and
+#   a compression chunk boundary is aligned to nothing. Children under 1 MiB are copied regardless.
+# - `du` counts shared extents once per file but `df` only once in total, so per-table disk usage over-reports until
+#   the parent is unlinked and its extents become the children's alone. Page cache is per inode too, so bytes read
+#   through both the parent and a child are cached twice for as long as both are live - during a split, briefly.
+# zero_copy_split_reflink_enabled: true
+
+# Write Digest.crc32 for the children of a zero-copy split. It costs one full sequential read of every child, which
+# once the extents above are shared is the ENTIRE remaining cost of a split; without it a split reads only Index.db, a
+# couple of percent of the data. Nothing requires the component: sstable opening asserts only Data, Index and
+# Statistics, entire-sstable streaming skips absent components and never validates the digest, and snapshot and backup
+# tooling copies whichever component files exist. A compressed sstable is self-checking anyway: every compression
+# chunk carries an inline CRC32 that this path preserves and every read verifies. Its only unique coverage is bytes no
+# read touches - the child's dead prefix and its alignment padding. What its absence costs is verification SPEED,
+# not strength: `nodetool verify` and `nodetool import --verify-sstables` treat a missing digest as a reason to run
+# a full extended verification, row by row instead of one whole-file CRC (`nodetool verify -q` and an import
+# without --verify-sstables never look at it). Because verifying and importing are routine, and a child's file set
+# can travel through backup and restore without a digest ever being regenerated, this defaults to true.
+# zero_copy_split_digest_enabled: true
+
 # When enabled, permits Cassandra to zero-copy stream entire eligible
 # SSTables between nodes, including every component.
 # This speeds up the network transfer significantly subject to
@@ -1304,6 +1380,45 @@ uuid_sstable_identifiers_enabled: false
 # disabled if internode encryption is enabled.
 # stream_entire_sstables: true
 
+# Stream a PARTIAL sstable through the entire-sstable (zero-copy) path instead of the row-by-row one, by sending a
+# verbatim run of the parent's compression chunks and synthesizing the other components. The saving is receiver-side:
+# no decompressing, deserializing, re-serializing or recompressing every row, no rebuilding the index, filter and
+# summary it is handed instead. Requested ranges are widened to the sstable's grid (see the next setting). The BIG and
+# BTI formats are both handled, compressed or not; a format that is neither, legacy counter shards, a pre-4.0 bloom
+# filter, a compression dictionary and anything the arithmetic cannot express fall back. stream_entire_sstables gates
+# this too: it is that protocol and rate limiter (entire_sstable_stream_throughput_outbound) in use. Like the split,
+# a slice needs a parent at sstable version pb (BIG) or eb (BTI) or later, so on an upgraded cluster this stays INERT
+# over pre-existing sstables until `nodetool upgradesstables` has rewritten them.
+#
+# Off by default: another rollout-order contract, with more to tolerate than a split child's dead prefix - a received
+# sstable can also carry unindexed bytes BETWEEN partitions. A build WITHOUT this patch fails on both, as described
+# for zero_copy_anticompaction_enabled, and interior gaps can additionally make it hand back partitions the index does
+# not list. Leave off until every node that can RECEIVE a stream has those changes.
+# The received sstable has a split child's statistics imprecision, plus inherited min/max timestamp and clustering
+# bounds, and purges no tombstones in transit; entire-sstable streaming already copies statistics verbatim, so this is
+# a difference of degree. Digest.crc32 is not sent - the bytes reach the socket by sendfile, so the sender cannot
+# checksum them without a second full read - so the RECEIVER computes it while writing the component: the sstable that
+# lands has one and `nodetool verify` on it is a whole-file CRC.
+# An sstable that would stream components a slice cannot synthesize - every storage-attached index component - falls
+# back to the row-by-row path, which builds those components on the receiver through the ordinary flush observers. A
+# WHOLE sstable is unaffected: entire-sstable streaming sends index components verbatim, which is correct there because
+# the row ids do not change. Each refusal is counted by StreamingMetrics.SlicedZeroCopyStreamsRefused.
+# zero_copy_partial_stream_enabled: false
+
+# The most dead space a partial zero-copy stream may carry before it falls back to the row-by-row path, as a fraction
+# of EVERYTHING IT WOULD TRANSFER rather than of the sliced sstable alone. A stream cannot start part-way into a grid
+# cell (a compression chunk, or a CRC.db chunk when the sstable is uncompressed), so the cells covering the requested
+# ranges also carry bytes nobody asked for: the head of the first cell, partitions between sections less than a cell
+# apart, and the tail of the last cell (a compressed sstable cannot cut its final chunk). None of it is indexed, so no
+# read can reach it - it is transferred and stored for nothing until the sstable is compacted. Only the first two of
+# those three lie within the slice's uncompressed length; the tail lies past it and is still sent, so the ratio
+# compared against this setting is (interior dead bytes + tail) / (uncompressed length + tail). Without the tail on
+# both sides a slice whose only waste is a partial final chunk would report 0% dead and be accepted even at 0.0.
+# Dead space is bounded by roughly a cell per section boundary: immaterial for a wide range, dominant for a narrow one
+# (a 4 KiB range inside one 16 KiB chunk is 75% dead). At the 0.25 default a transfer may be up to a third larger than
+# the data it is for; 0.0 permits only ranges that begin AND end on cell boundaries, 1.0 disables the check.
+# zero_copy_partial_stream_max_dead_space_ratio: 0.25
+
 # Throttles entire SSTable outbound streaming file transfers on
 # this node to the given total throughput in Mbps.
 # Setting this value to 0 it disables throttling.

doc/modules/cassandra/pages/architecture/streaming.adoc

@@ -69,13 +69,27 @@ stream_entire_sstables: true
 ....
 
 Zero copy streaming is enabled by default.
+
 === SSTables Eligible for Zero Copy Streaming
 
-Zero copy streaming is used if all partitions within the SSTable need to
-be transmitted. This is common when using `LeveledCompactionStrategy` or
-when partitioning SSTables by token range has been enabled. All
-partition keys in the SSTables are iterated over to determine the
-eligibility for Zero Copy streaming.
+An SSTable is streamed whole through the zero copy path if all partitions
+within it need to be transmitted. This is common when using
+`LeveledCompactionStrategy` or when partitioning SSTables by token range
+has been enabled. All partition keys in the SSTables are iterated over to
+determine the eligibility for Zero Copy streaming.
+
+An SSTable only part of which needs to be transmitted can also use the
+zero copy path, as a *slice*: a verbatim run of the SSTable's compression
+chunks covering the requested ranges, with the other components
+synthesized for it on the sender. This must be enabled explicitly with
+`zero_copy_partial_stream_enabled`, because a slice can carry bytes its
+index does not describe, which a node that does not yet have this feature
+will misread. Read that setting's comment in `cassandra.yaml`, and the
+`Upgrading` notes in `NEWS.txt`, before enabling it. Requested ranges are
+widened to the SSTable's compression grid, so ranges too narrow or too
+scattered to be worth that widening fall back to partition-by-partition
+streaming; `zero_copy_partial_stream_max_dead_space_ratio` sets how much
+widening is acceptable.
 
 === Benefits of Zero Copy Streaming
 
@@ -120,7 +134,9 @@ inter_dc_stream_throughput_outbound: 24MiB/s
 
 Zero Copy Streaming streams entire SSTables. SSTables are made up of
 multiple components in separate files. SSTable components streamed are
-listed in Table 1.
+listed in Table 1. A partial SSTable slice streams the same components,
+synthesized for the sliced range, with the one exception noted against
+`Digest.crc32`.
 
 Table 1. SSTable Components
 
@@ -141,7 +157,11 @@ length, chunk offsets etc.
 
 |Statistics.db |Statistical metadata about the content of the SSTable.
 
-|Digest.crc32 |Holds CRC32 checksum of the data file size_bytes.
+|Digest.crc32 |Holds CRC32 checksum of the data file size_bytes. Not
+streamed for a partial SSTable slice: the data reaches the socket by
+`sendfile` and is never in process on the sender, which therefore cannot
+checksum it without a second full read, so the receiver computes this
+component as it writes the data file.
 
 |CRC.db |Holds the CRC32 for chunks in an uncompressed file.
 

doc/modules/cassandra/pages/managing/operating/compaction/overview.adoc

@@ -68,7 +68,11 @@ This type of compaction is typically triggered on neighbouring nodes after a nod
 Secondary index rebuild::
 A compaction is triggered if the secondary indexes are rebuilt on a node.
 Anticompaction::
-After repair, the ranges that were actually repaired are split out of the SSTables that existed when repair started. This type of compaction rewrites SSTables to accomplish this task.
+After repair, the ranges that were actually repaired are split out of the SSTables that existed when repair started.
+This type of compaction normally rewrites SSTables to accomplish this task.
+If `zero_copy_anticompaction_enabled` is set and an SSTable's repaired, transient and unrepaired partitions happen to form contiguous token runs, that SSTable is instead split by copying its compression chunks verbatim, which is far cheaper but does not purge tombstones and leaves the children with statistics inherited from the parent rather than recomputed.
+Read the `zero_copy_anticompaction_enabled` comment in `cassandra.yaml` before enabling it.
+Because a verbatim copy retains droppable tombstones and inherits the parent's tombstone histogram and timestamps, the delayed purge is felt through <<single-sstable-tombstone-compaction,single SSTable tombstone compaction>> and through xref:cassandra:managing/operating/compaction/tombstones.adoc#fully-expired-sstables[fully expired SSTables] no longer being dropped whole; `tombstone_threshold` and `unchecked_tombstone_compaction` in <<compaction-options,Common options>> are the levers for tables that are sensitive to it.
 Sub range compaction::
 It is possible to only compact a given sub range - this action is useful if you know a token that has been misbehaving - either gathering many updates or many deletes.
 The command `nodetool compact -st x -et y` will pick all SSTables containing the range between x and y and issue a compaction for those SSTables. 
@@ -142,6 +146,7 @@ so if you have a tiny disk and a big disk backing two data directories,
 the big one will be limited the by the small one. One work around to
 this is to create more data directories backed by the big disk.
 
+[[single-sstable-tombstone-compaction]]
 == Single SSTable tombstone compaction
 
 When an SSTable is written a histogram with the tombstone expiry times