From d7258ac8f31eedc9c28f8c5e381893d192209d48 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Sun, 8 Dec 2024 16:10:01 +0300 Subject: [PATCH] Add table metric PurgeableTombstoneScannedHistogram and a tracing event for scanned purgeable tombstones patch by Dmitry Konstantinov; reviewed by Chris Lohfink, Stefan Miklosovic for CASSANDRA-20132 --- CHANGES.txt | 1 + conf/cassandra.yaml | 14 + conf/cassandra_latest.yaml | 14 + .../pages/managing/operating/metrics.adoc | 3 + .../org/apache/cassandra/config/Config.java | 30 ++ .../cassandra/config/DatabaseDescriptor.java | 11 + .../org/apache/cassandra/db/ReadCommand.java | 139 ++++++ .../db/virtual/TableMetricTables.java | 1 + .../cassandra/metrics/KeyspaceMetrics.java | 3 + .../cassandra/metrics/TableMetrics.java | 3 + .../config/DatabaseDescriptorRefTest.java | 1 + .../apache/cassandra/db/ReadCommandTest.java | 465 ++++++++++++++++-- 12 files changed, 641 insertions(+), 44 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 4a749676fb..28ab975f79 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.1 + * Add table metric PurgeableTombstoneScannedHistogram and a tracing event for scanned purgeable tombstones (CASSANDRA-20132) * Make sure we can parse the expanded CQL before writing it to the log or sending it to replicas (CASSANDRA-20218) * Add format_bytes and format_time functions (CASSANDRA-19546) * Fix error when trying to assign a tuple to target type not being a tuple (CASSANDRA-20237) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 55fc50f6c6..58f8945d65 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -1964,6 +1964,20 @@ transparent_data_encryption_options: tombstone_warn_threshold: 1000 tombstone_failure_threshold: 100000 +# Controls the granularity of purgeable tombstones reported to PurgeableTombstoneScannedHistogram table metric +# Possible values: +# 'disabled' - do not collect the metric at all +# 'row' - track only partition/range/row level tombstone, +# a good compromise between overheads and usability. +# For CPU-bound workload you may get less than 1% of overhead for throughput. +# For IO-bound workload the overhead is negligible. +# 'cell' - track partition/range/row/cell level tombstones. +# This is the most granular option, +# but it has some performance overheads due to iteration over cells. +# For CPU-bound workload you may get about 5% of overhead for throughput. +# For IO-bound workload the overhead is almost negligible. +# tombstone_read_purgeable_metric_granularity: disabled + # Filtering and secondary index queries at read consistency levels above ONE/LOCAL_ONE use a # mechanism called replica filtering protection to ensure that results from stale replicas do # not violate consistency. (See CASSANDRA-8272 and CASSANDRA-15907 for more details.) This diff --git a/conf/cassandra_latest.yaml b/conf/cassandra_latest.yaml index 387565fe18..ce1d97c476 100644 --- a/conf/cassandra_latest.yaml +++ b/conf/cassandra_latest.yaml @@ -1838,6 +1838,20 @@ transparent_data_encryption_options: tombstone_warn_threshold: 1000 tombstone_failure_threshold: 100000 +# Controls the granularity of purgeable tombstones reported to PurgeableTombstoneScannedHistogram table metric +# Possible values: +# 'disabled' - do not collect the metric at all +# 'row' - track only partition/range/row level tombstone, +# a good compromise between overheads and usability. +# For CPU-bound workload you may get less than 1% of overhead for throughput. +# For IO-bound workload the overhead is negligible. +# 'cell' - track partition/range/row/cell level tombstones. +# This is the most granular option, +# but it has some performance overheads due to iteration over cells. +# For CPU-bound workload you may get about 5% of overhead for throughput. +# For IO-bound workload the overhead is almost negligible. +# tombstone_read_purgeable_metric_granularity: disabled + # Filtering and secondary index queries at read consistency levels above ONE/LOCAL_ONE use a # mechanism called replica filtering protection to ensure that results from stale replicas do # not violate consistency. (See CASSANDRA-8272 and CASSANDRA-15907 for more details.) This diff --git a/doc/modules/cassandra/pages/managing/operating/metrics.adoc b/doc/modules/cassandra/pages/managing/operating/metrics.adoc index 78dbd16586..4f3d66652c 100644 --- a/doc/modules/cassandra/pages/managing/operating/metrics.adoc +++ b/doc/modules/cassandra/pages/managing/operating/metrics.adoc @@ -181,6 +181,9 @@ by compression meta data. |TombstoneScannedHistogram |Histogram |Histogram of tombstones scanned in queries on this table. +|PurgeableTombstoneScannedHistogram |Histogram |Histogram of purgeable tombstones scanned +in queries on this table. Use tombstone_read_purgeable_metric_granularity property in cassandra.yaml to enable it. + |LiveScannedHistogram |Histogram |Histogram of live cells scanned in queries on this table. diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 932423d248..77f7d9a341 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -544,6 +544,8 @@ public class Config public volatile int tombstone_warn_threshold = 1000; public volatile int tombstone_failure_threshold = 100000; + public TombstonesMetricGranularity tombstone_read_purgeable_metric_granularity = TombstonesMetricGranularity.disabled; + public final ReplicaFilteringProtectionOptions replica_filtering_protection = new ReplicaFilteringProtectionOptions(); @Replaces(oldName = "index_summary_capacity_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_LONG, deprecated = true) @@ -1327,6 +1329,34 @@ public class Config } } + /** + * Allow to control the granularity of metrics related to tombstones. + * It is a trade-off between granularity of a metric vs performance overheads. + * See CASSANDRA-20132 for more details. + */ + public enum TombstonesMetricGranularity + { + /** + * Do not collect the metric at all. + */ + disabled, + /** + * Track only partition/range/row level tombstone, + * a good compromise between overheads and usability. + * For CPU-bound workload you may get less than 1% of overhead for throughput. + * For IO-bound workload the overhead is negligible. + */ + row, + /** + * Track partition/range/row/cell level tombstones. + * This is the most granular option, + * but it has some performance overheads due to iteration over cells. + * For CPU-bound workload you may get about 5% of overhead for throughput. + * For IO-bound workload the overhead is almost negligible. + */ + cell + } + private static final Set SENSITIVE_KEYS = new HashSet() {{ add("client_encryption_options"); add("server_encryption_options"); diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 8a09f4fd76..b784571d01 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -5523,4 +5523,15 @@ public class DatabaseDescriptor { return conf.password_validator_reconfiguration_enabled; } + + public static Config.TombstonesMetricGranularity getPurgeableTobmstonesMetricGranularity() + { + return conf.tombstone_read_purgeable_metric_granularity; + } + + @VisibleForTesting + public static void setPurgeableTobmstonesMetricGranularity(Config.TombstonesMetricGranularity granularity) + { + conf.tombstone_read_purgeable_metric_granularity = granularity; + } } diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 3bb50cbcfa..e4ea5f12d7 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -466,6 +466,7 @@ public abstract class ReadCommand extends AbstractReadQuery iterator = withQuerySizeTracking(iterator); iterator = maybeSlowDownForTesting(iterator); iterator = withQueryCancellation(iterator); + iterator = maybeRecordPurgeableTombstones(iterator, cfs); iterator = RTBoundValidator.validate(withoutPurgeableTombstones(iterator, cfs, executionController), Stage.PURGED, false); iterator = withMetricsRecording(iterator, cfs.metric, startTimeNanos); @@ -894,6 +895,144 @@ public abstract class ReadCommand extends AbstractReadQuery return Transformation.apply(iterator, new WithoutPurgeableTombstones()); } + + /** + * Wraps the provided iterator so that metrics on count of purgeable tombstones are tracked and traced. + * It tracks only tombstones with localDeletionTime < now - gc_grace_period. + * Other (non-purgeable) tombstones will be tracked by regular Cassandra logic later. + */ + private UnfilteredPartitionIterator maybeRecordPurgeableTombstones(UnfilteredPartitionIterator iter, + ColumnFamilyStore cfs) + { + class PurgeableTombstonesMetricRecording extends Transformation + { + private int purgeableTombstones = 0; + + @Override + public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator iter) + { + if (!iter.partitionLevelDeletion().isLive()) + purgeableTombstones++; + return Transformation.apply(iter, this); + } + + @Override + public Row applyToStatic(Row row) + { + return applyToRow(row); + } + + @Override + public Row applyToRow(Row row) + { + final long nowInSec = nowInSec(); + boolean hasTombstones = false; + + if (isPurgeableCellTombstonesTrackingEnabled()) + { + for (Cell cell : row.cells()) + { + if (!cell.isLive(nowInSec) && isPurgeable(cell.localDeletionTime(), nowInSec)) + { + purgeableTombstones++; + hasTombstones = true; // allows to avoid counting an extra tombstone if the whole row expired + } + } + } + + // we replicate the logic is used for non-purged tombstones metric here + if (!row.primaryKeyLivenessInfo().isLive(nowInSec) + && row.hasDeletion(nowInSec) + && isPurgeable(row.deletion().time(), nowInSec) + && !hasTombstones) + { + // We're counting primary key deletions only here. + purgeableTombstones++; + } + + return row; + } + + @Override + public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker) + { + final long nowInSec = nowInSec(); + + // for boundary markers - increment metric only if both - close and open - markers are purgeable + if (marker.isBoundary()) + { + countIfBothPurgeable(marker.closeDeletionTime(false), + marker.openDeletionTime(false), + nowInSec); + } + // for bound markers - just increment if it is purgeable + else if (marker instanceof RangeTombstoneBoundMarker) + { + countIfPurgeable(((RangeTombstoneBoundMarker) marker).deletionTime(), nowInSec); + } + + return marker; + } + + @Override + public void onClose() + { + cfs.metric.purgeableTombstoneScannedHistogram.update(purgeableTombstones); + if (purgeableTombstones > 0) + Tracing.trace("Read {} purgeable tombstone cells", purgeableTombstones); + } + + /** + * Increments if both - close and open - deletion times less than (now - gc_grace_period) + */ + private void countIfBothPurgeable(DeletionTime closeDeletionTime, + DeletionTime openDeletionTime, + long nowInSec) + { + if (isPurgeable(closeDeletionTime, nowInSec) && isPurgeable(openDeletionTime, nowInSec)) + purgeableTombstones++; + } + + /** + * Increments if deletion time less than (now - gc_grace_period) + */ + private void countIfPurgeable(DeletionTime deletionTime, + long nowInSec) + { + if (isPurgeable(deletionTime, nowInSec)) + purgeableTombstones++; + } + + /** + * Checks that deletion time < now - gc_grace_period + */ + private boolean isPurgeable(DeletionTime deletionTime, + long nowInSec) + { + return isPurgeable(deletionTime.localDeletionTime(), nowInSec); + } + + /** + * Checks that deletion time < now - gc_grace_period + */ + private boolean isPurgeable(long localDeletionTime, + long nowInSec) + { + return localDeletionTime < cfs.gcBefore(nowInSec); + } + + private boolean isPurgeableCellTombstonesTrackingEnabled() + { + return DatabaseDescriptor.getPurgeableTobmstonesMetricGranularity() == Config.TombstonesMetricGranularity.cell; + } + } + + if (DatabaseDescriptor.getPurgeableTobmstonesMetricGranularity() != Config.TombstonesMetricGranularity.disabled) + return Transformation.apply(iter, new PurgeableTombstonesMetricRecording()); + else + return iter; + } + /** * Return the queried token(s) for logging */ diff --git a/src/java/org/apache/cassandra/db/virtual/TableMetricTables.java b/src/java/org/apache/cassandra/db/virtual/TableMetricTables.java index 5528c92011..e1defe51e6 100644 --- a/src/java/org/apache/cassandra/db/virtual/TableMetricTables.java +++ b/src/java/org/apache/cassandra/db/virtual/TableMetricTables.java @@ -75,6 +75,7 @@ public class TableMetricTables new LatencyTableMetric(name, "local_write_latency", t -> t.writeLatency.latency), new LatencyTableMetric(name, "coordinator_write_latency", t -> t.coordinatorWriteLatency), new HistogramTableMetric(name, "tombstones_per_read", t -> t.tombstoneScannedHistogram.cf), + new HistogramTableMetric(name, "purgeable_tombstones_per_read", t -> t.purgeableTombstoneScannedHistogram.cf), new HistogramTableMetric(name, "rows_per_read", t -> t.liveScannedHistogram.cf), new StorageTableMetric(name, "disk_usage", (TableMetrics t) -> t.totalDiskSpaceUsed), new StorageTableMetric(name, "max_partition_size", (TableMetrics t) -> t.maxPartitionSize), diff --git a/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java b/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java index 237fd03e2d..a1916bebd0 100644 --- a/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java +++ b/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java @@ -85,6 +85,8 @@ public class KeyspaceMetrics public final Histogram sstablesPerRangeReadHistogram; /** Tombstones scanned in queries on this Keyspace */ public final Histogram tombstoneScannedHistogram; + /** Purgeable tombstones scanned in queries on this Keyspace */ + public final Histogram purgeableTombstoneScannedHistogram; /** Live cells scanned in queries on this Keyspace */ public final Histogram liveScannedHistogram; /** Column update time delta on this Keyspace */ @@ -234,6 +236,7 @@ public class KeyspaceMetrics sstablesPerReadHistogram = createKeyspaceHistogram("SSTablesPerReadHistogram", true); sstablesPerRangeReadHistogram = createKeyspaceHistogram("SSTablesPerRangeReadHistogram", true); tombstoneScannedHistogram = createKeyspaceHistogram("TombstoneScannedHistogram", false); + purgeableTombstoneScannedHistogram = createKeyspaceHistogram("PurgeableTombstoneScannedHistogram", false); liveScannedHistogram = createKeyspaceHistogram("LiveScannedHistogram", false); colUpdateTimeDeltaHistogram = createKeyspaceHistogram("ColUpdateTimeDeltaHistogram", false); viewLockAcquireTime = createKeyspaceTimer("ViewLockAcquireTime"); diff --git a/src/java/org/apache/cassandra/metrics/TableMetrics.java b/src/java/org/apache/cassandra/metrics/TableMetrics.java index ab1ab6eb73..fabb0814e4 100644 --- a/src/java/org/apache/cassandra/metrics/TableMetrics.java +++ b/src/java/org/apache/cassandra/metrics/TableMetrics.java @@ -156,6 +156,8 @@ public class TableMetrics public final Gauge compressionMetadataOffHeapMemoryUsed; /** Tombstones scanned in queries on this CF */ public final TableHistogram tombstoneScannedHistogram; + /** Purgeable tombstones scanned in queries on this CF */ + public final TableHistogram purgeableTombstoneScannedHistogram; /** Live rows scanned in queries on this CF */ public final TableHistogram liveScannedHistogram; /** Column update time delta on this CF */ @@ -768,6 +770,7 @@ public class TableMetrics additionalWriteLatencyNanos = createTableGauge("AdditionalWriteLatencyNanos", () -> MICROSECONDS.toNanos(cfs.additionalWriteLatencyMicros)); tombstoneScannedHistogram = createTableHistogram("TombstoneScannedHistogram", cfs.keyspace.metric.tombstoneScannedHistogram, false); + purgeableTombstoneScannedHistogram = createTableHistogram("PurgeableTombstoneScannedHistogram", cfs.keyspace.metric.purgeableTombstoneScannedHistogram, true); liveScannedHistogram = createTableHistogram("LiveScannedHistogram", cfs.keyspace.metric.liveScannedHistogram, false); colUpdateTimeDeltaHistogram = createTableHistogram("ColUpdateTimeDeltaHistogram", cfs.keyspace.metric.colUpdateTimeDeltaHistogram, false); coordinatorReadLatency = createTableTimer("CoordinatorReadLatency"); diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java index 89af53b175..34656eac55 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java @@ -103,6 +103,7 @@ public class DatabaseDescriptorRefTest "org.apache.cassandra.config.ConfigurationLoader", "org.apache.cassandra.config.Config$CorruptedTombstoneStrategy", "org.apache.cassandra.config.Config$BatchlogEndpointStrategy", + "org.apache.cassandra.config.Config$TombstonesMetricGranularity", "org.apache.cassandra.config.DatabaseDescriptor$ByteUnit", "org.apache.cassandra.config.DataRateSpec", "org.apache.cassandra.config.DataRateSpec$DataRateUnit", diff --git a/test/unit/org/apache/cassandra/db/ReadCommandTest.java b/test/unit/org/apache/cassandra/db/ReadCommandTest.java index cc4ceee695..51a49c4365 100644 --- a/test/unit/org/apache/cassandra/db/ReadCommandTest.java +++ b/test/unit/org/apache/cassandra/db/ReadCommandTest.java @@ -35,6 +35,7 @@ import org.junit.Test; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; +import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter; import org.apache.cassandra.db.filter.ColumnFilter; @@ -93,6 +94,9 @@ import static org.junit.Assert.fail; public class ReadCommandTest { + private static final String CREATE = "1"; + private static final String DELETE = "-1"; + private static final String KEYSPACE = "ReadCommandTest"; private static final String CF1 = "Standard1"; private static final String CF2 = "Standard2"; @@ -103,6 +107,12 @@ public class ReadCommandTest private static final String CF7 = "Counter7"; private static final String CF8 = "Standard8"; private static final String CF9 = "Standard9"; + private static final String CF10 = "Standard10"; + private static final String CF11 = "Standard11"; + private static final String CF12 = "Standard12"; + private static final String CF13 = "Standard13"; + private static final String CF14 = "Standard14"; + private static final InetAddressAndPort REPAIR_COORDINATOR; static { @@ -194,6 +204,61 @@ public class ReadCommandTest .addClusteringColumn("col", ReversedType.getInstance(Int32Type.instance)) .addRegularColumn("a", AsciiType.instance); + TableMetadata.Builder metadata10 = + TableMetadata.builder(KEYSPACE, CF10) + .addPartitionKeyColumn("key", BytesType.instance) + .addClusteringColumn("col", AsciiType.instance) + .addRegularColumn("a", AsciiType.instance) + .addRegularColumn("b", AsciiType.instance) + .addRegularColumn("c", AsciiType.instance) + .addRegularColumn("d", AsciiType.instance) + .addRegularColumn("e", AsciiType.instance) + .addRegularColumn("f", AsciiType.instance); + + TableMetadata.Builder metadata11 = + TableMetadata.builder(KEYSPACE, CF11) + .addPartitionKeyColumn("key", BytesType.instance) + .addClusteringColumn("col", AsciiType.instance) + .addRegularColumn("a", AsciiType.instance) + .addRegularColumn("b", AsciiType.instance) + .addRegularColumn("c", AsciiType.instance) + .addRegularColumn("d", AsciiType.instance) + .addRegularColumn("e", AsciiType.instance) + .addRegularColumn("f", AsciiType.instance); + + TableMetadata.Builder metadata12 = + TableMetadata.builder(KEYSPACE, CF12) + .addPartitionKeyColumn("key", BytesType.instance) + .addClusteringColumn("col", AsciiType.instance) + .addRegularColumn("a", AsciiType.instance) + .addRegularColumn("b", AsciiType.instance) + .addRegularColumn("c", AsciiType.instance) + .addRegularColumn("d", AsciiType.instance) + .addRegularColumn("e", AsciiType.instance) + .addRegularColumn("f", AsciiType.instance); + + TableMetadata.Builder metadata13 = + TableMetadata.builder(KEYSPACE, CF13) + .addPartitionKeyColumn("key", BytesType.instance) + .addClusteringColumn("col", AsciiType.instance) + .addRegularColumn("a", AsciiType.instance) + .addRegularColumn("b", AsciiType.instance) + .addRegularColumn("c", AsciiType.instance) + .addRegularColumn("d", AsciiType.instance) + .addRegularColumn("e", AsciiType.instance) + .addRegularColumn("f", AsciiType.instance); + + TableMetadata.Builder metadata14 = + TableMetadata.builder(KEYSPACE, CF14) + .addPartitionKeyColumn("key", BytesType.instance) + .addClusteringColumn("col", AsciiType.instance) + .addRegularColumn("a", AsciiType.instance) + .addRegularColumn("b", AsciiType.instance) + .addRegularColumn("c", AsciiType.instance) + .addRegularColumn("d", AsciiType.instance) + .addRegularColumn("e", AsciiType.instance) + .addRegularColumn("f", AsciiType.instance); + SchemaLoader.prepareServer(); SchemaLoader.createKeyspace(KEYSPACE, KeyspaceParams.simple(1), @@ -205,7 +270,12 @@ public class ReadCommandTest metadata6, metadata7, metadata8, - metadata9); + metadata9, + metadata10, + metadata11, + metadata12, + metadata13, + metadata14); LocalSessionAccessor.startup(); } @@ -332,23 +402,23 @@ public class ReadCommandTest String[][][] groups = new String[][][] { new String[][] { - new String[] { "1", "key1", "aa", "a" }, // "1" indicates to create the data, "-1" to delete the row - new String[] { "1", "key2", "bb", "b" }, - new String[] { "1", "key3", "cc", "c" } + new String[] { CREATE, "key1", "aa", "a" }, + new String[] { CREATE, "key2", "bb", "b" }, + new String[] { CREATE, "key3", "cc", "c" } }, new String[][] { - new String[] { "1", "key3", "dd", "d" }, - new String[] { "1", "key2", "ee", "e" }, - new String[] { "1", "key1", "ff", "f" } + new String[] { CREATE, "key3", "dd", "d" }, + new String[] { CREATE, "key2", "ee", "e" }, + new String[] { CREATE, "key1", "ff", "f" } }, new String[][] { - new String[] { "1", "key6", "aa", "a" }, - new String[] { "1", "key5", "bb", "b" }, - new String[] { "1", "key4", "cc", "c" } + new String[] { CREATE, "key6", "aa", "a" }, + new String[] { CREATE, "key5", "bb", "b" }, + new String[] { CREATE, "key4", "cc", "c" } }, new String[][] { - new String[] { "-1", "key6", "aa", "a" }, - new String[] { "-1", "key2", "bb", "b" } + new String[] { DELETE, "key6", "aa", "a" }, + new String[] { DELETE, "key2", "bb", "b" } } }; @@ -371,7 +441,7 @@ public class ReadCommandTest for (String[] data : group) { - if (data[0].equals("1")) + if (data[0].equals(CREATE)) { new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes(data[1])) .clustering(data[2]) @@ -493,33 +563,32 @@ public class ReadCommandTest String[][][] groups = new String[][][] { new String[][] { - new String[] { "1", "key1", "aa", "a" }, // "1" indicates to create the data, "-1" to delete the - // row - new String[] { "1", "key2", "bb", "b" }, - new String[] { "1", "key3", "cc", "c" } + new String[] { CREATE, "key1", "aa", "a" }, + new String[] { CREATE, "key2", "bb", "b" }, + new String[] { CREATE, "key3", "cc", "c" } }, new String[][] { - new String[] { "1", "key3", "dd", "d" }, - new String[] { "1", "key2", "ee", "e" }, - new String[] { "1", "key1", "ff", "f" } + new String[] { CREATE, "key3", "dd", "d" }, + new String[] { CREATE, "key2", "ee", "e" }, + new String[] { CREATE, "key1", "ff", "f" } }, new String[][] { - new String[] { "1", "key6", "aa", "a" }, - new String[] { "1", "key5", "bb", "b" }, - new String[] { "1", "key4", "cc", "c" } + new String[] { CREATE, "key6", "aa", "a" }, + new String[] { CREATE, "key5", "bb", "b" }, + new String[] { CREATE, "key4", "cc", "c" } }, new String[][] { - new String[] { "1", "key2", "aa", "a" }, - new String[] { "1", "key2", "cc", "c" }, - new String[] { "1", "key2", "dd", "d" } + new String[] { CREATE, "key2", "aa", "a" }, + new String[] { CREATE, "key2", "cc", "c" }, + new String[] { CREATE, "key2", "dd", "d" } }, new String[][] { - new String[] { "-1", "key6", "aa", "a" }, - new String[] { "-1", "key2", "bb", "b" }, - new String[] { "-1", "key2", "ee", "e" }, - new String[] { "-1", "key2", "aa", "a" }, - new String[] { "-1", "key2", "cc", "c" }, - new String[] { "-1", "key2", "dd", "d" } + new String[] { DELETE, "key6", "aa", "a" }, + new String[] { DELETE, "key2", "bb", "b" }, + new String[] { DELETE, "key2", "ee", "e" }, + new String[] { DELETE, "key2", "aa", "a" }, + new String[] { DELETE, "key2", "cc", "c" }, + new String[] { DELETE, "key2", "dd", "d" } } }; @@ -539,7 +608,7 @@ public class ReadCommandTest for (String[] data : group) { - if (data[0].equals("1")) + if (data[0].equals(CREATE)) { new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes(data[1])) .clustering(data[2]) @@ -582,20 +651,19 @@ public class ReadCommandTest String[][][] groups = new String[][][] { new String[][] { - new String[] { "1", "key1", "aa", "a" }, // "1" indicates to create the data, "-1" to delete the - // row - new String[] { "1", "key2", "bb", "b" }, - new String[] { "1", "key3", "cc", "c" } + new String[] { CREATE, "key1", "aa", "a" }, + new String[] { CREATE, "key2", "bb", "b" }, + new String[] { CREATE, "key3", "cc", "c" } }, new String[][] { - new String[] { "1", "key3", "dd", "d" }, - new String[] { "1", "key2", "ee", "e" }, - new String[] { "1", "key1", "ff", "f" } + new String[] { CREATE, "key3", "dd", "d" }, + new String[] { CREATE, "key2", "ee", "e" }, + new String[] { CREATE, "key1", "ff", "f" } }, new String[][] { - new String[] { "1", "key6", "aa", "a" }, - new String[] { "1", "key5", "bb", "b" }, - new String[] { "1", "key4", "cc", "c" } + new String[] { CREATE, "key6", "aa", "a" }, + new String[] { CREATE, "key5", "bb", "b" }, + new String[] { CREATE, "key4", "cc", "c" } } }; @@ -615,7 +683,7 @@ public class ReadCommandTest for (String[] data : group) { - if (data[0].equals("1")) + if (data[0].equals(CREATE)) { new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes(data[1])) .clustering(data[2]) @@ -651,6 +719,315 @@ public class ReadCommandTest assertEquals(1, cfs.metric.tombstoneScannedHistogram.cf.getSnapshot().getMax()); } + @Test + public void testCountPurgeableRowTombstones() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF10); + TestWriteOperation[] operations = new TestWriteOperation[] + { + TestWriteOperation.insert("key1", "aa", "a"), + TestWriteOperation.insert("key1", "ff", "f"), + + TestWriteOperation.insert("key2", "aa", "e"), + TestWriteOperation.deleteRow("key2", "aa", PURGEABLE_DELETION), + TestWriteOperation.deleteRow("key2", "bb", NEW_DELETION), + TestWriteOperation.deleteRow("key2", "cc", PURGEABLE_DELETION), + TestWriteOperation.deleteRow("key2", "dd", PURGEABLE_DELETION), + TestWriteOperation.deleteRow("key2", "ee", NEW_DELETION), + }; + + runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.row); + + assertEquals(2, cfs.metric.purgeableTombstoneScannedHistogram.cf.getCount()); + assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin()); + assertEquals(3, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax()); + } + @Test + public void testCountPurgeablePartitionTombstones() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF11); + TestWriteOperation[] operations = new TestWriteOperation[] + { + TestWriteOperation.insert("key1", "aa", "a"), + TestWriteOperation.insert("key1", "ff", "f"), + + TestWriteOperation.insert("key2", "aa", "a"), + TestWriteOperation.insert("key2", "cc", "c"), + TestWriteOperation.insert("key2", "dd", "d"), + + TestWriteOperation.deletePartition("key2", PURGEABLE_DELETION), + TestWriteOperation.deletePartition("key3", NEW_DELETION) + }; + runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.row); + + assertEquals(3, cfs.metric.purgeableTombstoneScannedHistogram.cf.getCount()); + assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin()); + assertEquals(1, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax()); + } + + @Test + public void testCountPurgeableCellTombstones() throws Exception + { + DatabaseDescriptor.setPurgeableTobmstonesMetricGranularity(Config.TombstonesMetricGranularity.cell); + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF12); + TestWriteOperation[] operations = new TestWriteOperation[] + { + TestWriteOperation.insert("key1", "aa", "a"), + TestWriteOperation.insert("key1", "ff", "f"), + + TestWriteOperation.insert("key2", "aa", "a"), + TestWriteOperation.deleteCell("key2", "aa", "b", PURGEABLE_DELETION), + TestWriteOperation.deleteCell("key2", "aa", "f", NEW_DELETION), + TestWriteOperation.insert("key2", "cc", "c"), + TestWriteOperation.insert("key2", "dd", "d") + }; + runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.cell); + + assertEquals(2, cfs.metric.purgeableTombstoneScannedHistogram.cf.getCount()); + assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin()); + assertEquals(1, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax()); + + } + + /** + * Test purgeable tombstones count for range tombstones with non-overlapping ranges, + * i.e. only Bound (not Boundary) Markers will be created and counted + */ + @Test + public void testCountPurgeableRangeTombstones_nonOverlappingRanges() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF13); + TestWriteOperation[] operations = new TestWriteOperation[] + { + TestWriteOperation.insert("key1", "aa", "a"), + TestWriteOperation.insert("key1", "ff", "f"), + + TestWriteOperation.insert("key2", "aa", "a"), + TestWriteOperation.insert("key2", "cc", "c"), + TestWriteOperation.insert("key2", "dd", "d"), + + TestWriteOperation.deleteRange("key2", "aa", "bb", NEW_DELETION), + TestWriteOperation.deleteRange("key2", "dd", "ee", PURGEABLE_DELETION), + TestWriteOperation.deleteRange("key2", "ff", "ff", PURGEABLE_DELETION) + }; + runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.row); + + assertEquals(2, cfs.metric.purgeableTombstoneScannedHistogram.cf.getCount()); + assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin()); + assertEquals(4, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax()); + } + + /** + * Test purgeable tombstones count for range tombstones with overlapping ranges + */ + @Test + public void testCountPurgeableRangeTombstones_overlappingRanges() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF14); + + TestWriteOperation[] operations = new TestWriteOperation[] + { + TestWriteOperation.insert("key1", "aa", "a"), + TestWriteOperation.insert("key1", "ff", "f"), + + TestWriteOperation.insert("key2", "aa", "a"), + TestWriteOperation.insert("key2", "bb", "b"), + TestWriteOperation.insert("key2", "cc", "c"), + TestWriteOperation.insert("key2", "dd", "d"), + TestWriteOperation.insert("key2", "ee", "e"), + + // this range tombstone is non-purgeable and overlaps with the next one, + // so it will create one non-purgeable bound marker + // and one non-purgeable boundary marker so TWO NON-PURGEABLE tombstones + TestWriteOperation.deleteRange("key2", "aa", "bb", NEW_DELETION), + + // this range tombstone is purgeable and overlaps with previous and next ones, + // so it will create one non-purgeable bound marker + // and one non-purgeable boundary marker so TWO non-purgeable tombstones will be counted + TestWriteOperation.deleteRange("key2", "bb", "ee", PURGEABLE_DELETION), + + // this range tombstone is purgeable and overlaps with previous one, + // it has a different deletion time to not combine into a single range, + // so it will create one non-purgeable boundary marker (same as previous one) + // and one purgeable bound marker, so it will increment purgeable tombstones counter on one, + // we expect TWO purgeable tombstones in total + TestWriteOperation.deleteRange("key2", "ee", "ff", PURGEABLE_DELETION - 1) + }; + + runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.row); + + assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin()); + assertEquals(2, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax()); + } + + + private static void runTestWriteOperationsAndReadResults(ColumnFamilyStore cfs, TestWriteOperation[] operations, Config.TombstonesMetricGranularity granularity) throws IOException + { + Config.TombstonesMetricGranularity original = DatabaseDescriptor.getPurgeableTobmstonesMetricGranularity(); + DatabaseDescriptor.setPurgeableTobmstonesMetricGranularity(granularity); + try + { + Set usedPartitionKeys = runWriteOperations(cfs, operations); + runPartitionReadCommands(cfs, usedPartitionKeys); + } + finally + { + DatabaseDescriptor.setPurgeableTobmstonesMetricGranularity(original); + cfs.truncateBlocking(); + } + } + + private static void runPartitionReadCommands(ColumnFamilyStore cfs, Set partitionKeys) throws IOException + { + List commands = new ArrayList<>(partitionKeys.size()); + long nowInSeconds = FBUtilities.nowInSeconds(); // all reads within a group must have the same nowInSec + for (String partitionKey : partitionKeys) + { + commands.add(getWholePartitionReadCommand(cfs, partitionKey, nowInSeconds)); + } + executeReadCommands(commands); + } + + private static Set runWriteOperations(ColumnFamilyStore cfs, TestWriteOperation[] operations) + { + Set usedPartitionKeys = new HashSet<>(); + for (TestWriteOperation operation : operations) + { + if (operation.type == OperationType.CREATE) + { + new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes(operation.partitionKey)) + .clustering(operation.clusteringKey) + .add(operation.columnName, ByteBufferUtil.bytes(operation.columnValue)) + .build() + .apply(); + } + else if (operation.type == OperationType.DELETE_PARTITION) + { + new Mutation(PartitionUpdate.simpleBuilder(cfs.metadata(), ByteBufferUtil.bytes(operation.partitionKey)) + .nowInSec(operation.deletionTime) + .delete() + .build()).apply(); + } + else if (operation.type == OperationType.DELETE_RANGE) + { + new RowUpdateBuilder(cfs.metadata(), operation.deletionTime, 0L, ByteBufferUtil.bytes(operation.partitionKey)) + .addRangeTombstone(operation.clusteringRangeStart, operation.clusteringRangeEnd).build().apply(); + } + else if (operation.type == OperationType.DELETE_ROW) + { + RowUpdateBuilder.deleteRowAt(cfs.metadata(), 0, operation.deletionTime, + ByteBufferUtil.bytes(operation.partitionKey), operation.clusteringKey + ).apply(); + } + else if (operation.type == OperationType.DELETE_CELL) + { + new RowUpdateBuilder(cfs.metadata(), operation.deletionTime, 0L, ByteBufferUtil.bytes(operation.partitionKey)) + .clustering(operation.clusteringKey) + .delete(operation.columnName) + .build().apply(); + } + + usedPartitionKeys.add(operation.partitionKey); + } + return usedPartitionKeys; + } + + private static final long NEW_DELETION = FBUtilities.nowInSeconds(); + private static final long PURGEABLE_DELETION = 42; + + private enum OperationType + { + CREATE, + DELETE_PARTITION, + DELETE_RANGE, + DELETE_ROW, + DELETE_CELL + } + + private static class TestWriteOperation + { + OperationType type; + String partitionKey; + String clusteringKey; + + String clusteringRangeStart, clusteringRangeEnd; + String columnName; + String columnValue = "bla"; + + long deletionTime; + + public TestWriteOperation(OperationType type, String partitionKey, + String clusteringKey, String clusteringRangeStart, String clusteringRangeEnd, + String columnName, long deletionTime) + { + this.type = type; + this.partitionKey = partitionKey; + this.clusteringKey = clusteringKey; + this.clusteringRangeStart = clusteringRangeStart; + this.clusteringRangeEnd = clusteringRangeEnd; + this.columnName = columnName; + this.deletionTime = deletionTime; + } + + public static TestWriteOperation insert(String partitionKey, String clusteringKey, + String columnName) + { + return new TestWriteOperation(OperationType.CREATE, partitionKey, clusteringKey, null, null, columnName, 0); + } + + public static TestWriteOperation deletePartition(String partitionKey, long deletionTime) + { + return new TestWriteOperation(OperationType.DELETE_PARTITION, partitionKey, + null, null, null, null, deletionTime); + } + + public static TestWriteOperation deleteRange(String partitionKey, String clusteringRangeStart, String clusteringRangeEnd, long deletionTime) + { + return new TestWriteOperation(OperationType.DELETE_RANGE, partitionKey, + null, clusteringRangeStart, clusteringRangeEnd, null, deletionTime); + } + + public static TestWriteOperation deleteRow(String partitionKey, String clusteringKey, long deletionTime) + { + return new TestWriteOperation(OperationType.DELETE_ROW, partitionKey, clusteringKey, + null, null, null, deletionTime); + } + + public static TestWriteOperation deleteCell(String partitionKey, String clusteringKey, String columnName, long deletionTime) + { + return new TestWriteOperation(OperationType.DELETE_CELL, partitionKey, clusteringKey, + null, null, columnName, deletionTime); + } + + + } + + private static void executeReadCommands(List commands) throws IOException + { + ReadQuery query = SinglePartitionReadCommand.Group.create(commands, DataLimits.NONE); + + try (ReadExecutionController executionController = query.executionController(); + UnfilteredPartitionIterator iter = query.executeLocally(executionController); + DataOutputBuffer buffer = new DataOutputBuffer()) + { + UnfilteredPartitionIterators.serializerForIntraNode().serialize(iter, + query.columnFilter(), + buffer, + MessagingService.current_version); + } + } + + private static SinglePartitionReadCommand getWholePartitionReadCommand(ColumnFamilyStore cfs, String partitionKey, long nowInSeconds) + { + ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata(), false).build(); + RowFilter rowFilter = RowFilter.create(true); + Slice slice = Slice.make(BufferClusteringBound.BOTTOM, BufferClusteringBound.TOP); + ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.with(cfs.metadata().comparator, slice), false); + return SinglePartitionReadCommand.create(cfs.metadata(), nowInSeconds, + columnFilter, rowFilter, + DataLimits.NONE, Util.dk(partitionKey), sliceFilter); + } + @Test public void testSinglePartitionSliceRepairedDataTracking() throws Exception {