Repository: cassandra Updated Branches: refs/heads/cassandra-3.0 51e6f2446 -> 68bdf4547 refs/heads/cassandra-3.11 0e5c84ad8 -> 3e3d56ecd refs/heads/trunk 756fab87c -> 00022bfa6
Handle limit correctly on tables with strict liveness Patch by Zhao Yang; Reviewed by Paulo Motta for CASSANDRA-13883 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/68bdf454 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/68bdf454 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/68bdf454 Branch: refs/heads/cassandra-3.0 Commit: 68bdf45477417c97fa6ed3840eee39b8390fd678 Parents: 51e6f24 Author: Zhao Yang <zhaoyangsingap...@gmail.com> Authored: Tue Sep 19 18:35:24 2017 +0800 Committer: Paulo Motta <pa...@apache.org> Committed: Mon Sep 25 01:00:45 2017 -0500 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../apache/cassandra/db/ColumnFamilyStore.java | 5 +- .../org/apache/cassandra/db/ReadCommand.java | 3 +- .../db/SinglePartitionReadCommand.java | 9 ++- .../apache/cassandra/db/filter/DataLimits.java | 81 ++++++++++++++------ .../db/partitions/CachedBTreePartition.java | 3 +- .../apache/cassandra/db/rows/AbstractRow.java | 6 +- src/java/org/apache/cassandra/db/rows/Row.java | 7 +- .../cassandra/db/view/ViewUpdateGenerator.java | 6 +- .../composites/ClusteringColumnIndex.java | 5 +- .../internal/composites/PartitionKeyIndex.java | 4 +- .../apache/cassandra/service/DataResolver.java | 7 +- .../apache/cassandra/service/StorageProxy.java | 12 ++- .../service/pager/AbstractQueryPager.java | 4 +- .../service/pager/MultiPartitionPager.java | 12 ++- .../cassandra/service/pager/QueryPagers.java | 2 +- .../apache/cassandra/cql3/ViewComplexTest.java | 65 +++++++++++++++- .../org/apache/cassandra/cql3/ViewTest.java | 2 + .../apache/cassandra/db/RangeTombstoneTest.java | 34 +++++--- .../db/compaction/CompactionsPurgeTest.java | 3 +- 20 files changed, 213 insertions(+), 58 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 68da81a..9cba02b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.15 + * Handle limit correctly on tables with strict liveness (CASSANDRA-13883) * Fix missing original update in TriggerExecutor (CASSANDRA-13894) * Remove non-rpc-ready nodes from counter leader candidates (CASSANDRA-13043) * Improve short read protection performance (CASSANDRA-13794) http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/db/ColumnFamilyStore.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index e6e46b2..0d19856 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1495,7 +1495,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean // is not in the cache. We can guarantee that if either the filter is a "head filter" and the cached // partition has more live rows that queried (where live rows refers to the rows that are live now), // or if we can prove that everything the filter selects is in the cached partition based on its content. - return (filter.isHeadFilter() && limits.hasEnoughLiveData(cached, nowInSec, filter.selectsAllPartition())) + return (filter.isHeadFilter() && limits.hasEnoughLiveData(cached, + nowInSec, + filter.selectsAllPartition(), + metadata.enforceStrictLiveness())) || filter.isFullyCoveredBy(cached); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/db/ReadCommand.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index b73cdde..160b104 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -450,6 +450,7 @@ public abstract class ReadCommand implements ReadQuery private final int warningThreshold = DatabaseDescriptor.getTombstoneWarnThreshold(); private final boolean respectTombstoneThresholds = !Schema.isSystemKeyspace(ReadCommand.this.metadata().ksName); + private final boolean enforceStrictLiveness = metadata.enforceStrictLiveness(); private int liveRows = 0; private int tombstones = 0; @@ -472,7 +473,7 @@ public abstract class ReadCommand implements ReadQuery @Override public Row applyToRow(Row row) { - if (row.hasLiveData(ReadCommand.this.nowInSec())) + if (row.hasLiveData(ReadCommand.this.nowInSec(), enforceStrictLiveness)) ++liveRows; for (Cell cell : row.cells()) http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java index 00464ca..7a66eca 100644 --- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java +++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java @@ -556,6 +556,7 @@ public class SinglePartitionReadCommand extends ReadCommand try { final int rowsToCache = metadata().params.caching.rowsPerPartitionToCache(); + final boolean enforceStrictLiveness = metadata().enforceStrictLiveness(); @SuppressWarnings("resource") // we close on exception or upon closing the result of this method UnfilteredRowIterator iter = fullPartitionRead(metadata(), nowInSec(), partitionKey()).queryMemtableAndDisk(cfs, readOp); @@ -579,7 +580,7 @@ public class SinglePartitionReadCommand extends ReadCommand if (unfiltered.isRow()) { Row row = (Row) unfiltered; - if (row.hasLiveData(nowInSec())) + if (row.hasLiveData(nowInSec(), enforceStrictLiveness)) rowsCounted++; } return unfiltered; @@ -1161,10 +1162,14 @@ public class SinglePartitionReadCommand extends ReadCommand for (SinglePartitionReadCommand cmd : commands) partitions.add(cmd.executeInternal(orderGroup)); + // Note that the only difference between the command in a group must be the partition key on which + // they applied. + boolean enforceStrictLiveness = commands.get(0).metadata().enforceStrictLiveness(); // Because we only have enforce the limit per command, we need to enforce it globally. return limits.filter(PartitionIterators.concat(partitions), nowInSec, - selectsFullPartitions); + selectsFullPartitions, + enforceStrictLiveness); } public QueryPager getPager(PagingState pagingState, int protocolVersion) http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/db/filter/DataLimits.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/filter/DataLimits.java b/src/java/org/apache/cassandra/db/filter/DataLimits.java index 48ec06a..6b74293 100644 --- a/src/java/org/apache/cassandra/db/filter/DataLimits.java +++ b/src/java/org/apache/cassandra/db/filter/DataLimits.java @@ -20,6 +20,7 @@ package org.apache.cassandra.db.filter; import java.io.IOException; import java.nio.ByteBuffer; +import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.db.*; import org.apache.cassandra.db.rows.*; import org.apache.cassandra.db.partitions.*; @@ -46,7 +47,7 @@ public abstract class DataLimits public static final DataLimits NONE = new CQLLimits(NO_LIMIT) { @Override - public boolean hasEnoughLiveData(CachedPartition cached, int nowInSec, boolean countPartitionsWithOnlyStaticData) + public boolean hasEnoughLiveData(CachedPartition cached, int nowInSec, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness) { return false; } @@ -109,7 +110,10 @@ public abstract class DataLimits public abstract DataLimits forShortReadRetry(int toFetch); - public abstract boolean hasEnoughLiveData(CachedPartition cached, int nowInSec, boolean countPartitionsWithOnlyStaticData); + public abstract boolean hasEnoughLiveData(CachedPartition cached, + int nowInSec, + boolean countPartitionsWithOnlyStaticData, + boolean enforceStrictLiveness); /** * Returns a new {@code Counter} for this limits. @@ -120,9 +124,14 @@ public abstract class DataLimits * {@code RowIterator} (since it only returns live rows), false otherwise. * @param countPartitionsWithOnlyStaticData if {@code true} the partitions with only static data should be counted * as 1 valid row. + * @param enforceStrictLiveness whether the row should be purged if there is no PK liveness info, + * normally retrieved from {@link CFMetaData#enforceStrictLiveness()} * @return a new {@code Counter} for this limits. */ - public abstract Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData); + public abstract Counter newCounter(int nowInSec, + boolean assumeLiveData, + boolean countPartitionsWithOnlyStaticData, + boolean enforceStrictLiveness); /** * The max number of results this limits enforces. @@ -140,19 +149,27 @@ public abstract class DataLimits int nowInSec, boolean countPartitionsWithOnlyStaticData) { - return this.newCounter(nowInSec, false, countPartitionsWithOnlyStaticData).applyTo(iter); + return this.newCounter(nowInSec, + false, + countPartitionsWithOnlyStaticData, + iter.metadata().enforceStrictLiveness()) + .applyTo(iter); } public UnfilteredRowIterator filter(UnfilteredRowIterator iter, int nowInSec, boolean countPartitionsWithOnlyStaticData) { - return this.newCounter(nowInSec, false, countPartitionsWithOnlyStaticData).applyTo(iter); + return this.newCounter(nowInSec, + false, + countPartitionsWithOnlyStaticData, + iter.metadata().enforceStrictLiveness()) + .applyTo(iter); } - public PartitionIterator filter(PartitionIterator iter, int nowInSec, boolean countPartitionsWithOnlyStaticData) + public PartitionIterator filter(PartitionIterator iter, int nowInSec, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness) { - return this.newCounter(nowInSec, true, countPartitionsWithOnlyStaticData).applyTo(iter); + return this.newCounter(nowInSec, true, countPartitionsWithOnlyStaticData, enforceStrictLiveness).applyTo(iter); } /** @@ -301,7 +318,7 @@ public abstract class DataLimits return new CQLLimits(toFetch, NO_LIMIT, isDistinct); } - public boolean hasEnoughLiveData(CachedPartition cached, int nowInSec, boolean countPartitionsWithOnlyStaticData) + public boolean hasEnoughLiveData(CachedPartition cached, int nowInSec, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness) { // We want the number of row that are currently live. Getting that precise number forces // us to iterate the cached partition in general, but we can avoid that if: @@ -316,7 +333,7 @@ public abstract class DataLimits // Otherwise, we need to re-count - DataLimits.Counter counter = newCounter(nowInSec, false, countPartitionsWithOnlyStaticData); + DataLimits.Counter counter = newCounter(nowInSec, false, countPartitionsWithOnlyStaticData, enforceStrictLiveness); try (UnfilteredRowIterator cacheIter = cached.unfilteredIterator(ColumnFilter.selection(cached.columns()), Slices.ALL, false); UnfilteredRowIterator iter = counter.applyTo(cacheIter)) { @@ -327,9 +344,12 @@ public abstract class DataLimits } } - public Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData) + public Counter newCounter(int nowInSec, + boolean assumeLiveData, + boolean countPartitionsWithOnlyStaticData, + boolean enforceStrictLiveness) { - return new CQLCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData); + return new CQLCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData, enforceStrictLiveness); } public int count() @@ -360,25 +380,30 @@ public abstract class DataLimits protected int rowInCurrentPartition; protected boolean hasLiveStaticRow; + private final boolean enforceStrictLiveness; - public CQLCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData) + public CQLCounter(int nowInSec, + boolean assumeLiveData, + boolean countPartitionsWithOnlyStaticData, + boolean enforceStrictLiveness) { this.nowInSec = nowInSec; this.assumeLiveData = assumeLiveData; this.countPartitionsWithOnlyStaticData = countPartitionsWithOnlyStaticData; + this.enforceStrictLiveness = enforceStrictLiveness; } @Override public void applyToPartition(DecoratedKey partitionKey, Row staticRow) { rowInCurrentPartition = 0; - hasLiveStaticRow = !staticRow.isEmpty() && (assumeLiveData || staticRow.hasLiveData(nowInSec)); + hasLiveStaticRow = !staticRow.isEmpty() && (assumeLiveData || staticRow.hasLiveData(nowInSec, enforceStrictLiveness)); } @Override public Row applyToRow(Row row) { - if (assumeLiveData || row.hasLiveData(nowInSec)) + if (assumeLiveData || row.hasLiveData(nowInSec, enforceStrictLiveness)) incrementRowCount(); return row; } @@ -473,16 +498,19 @@ public abstract class DataLimits } @Override - public Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData) + public Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness) { - return new PagingAwareCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData); + return new PagingAwareCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData, enforceStrictLiveness); } private class PagingAwareCounter extends CQLCounter { - private PagingAwareCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData) + private PagingAwareCounter(int nowInSec, + boolean assumeLiveData, + boolean countPartitionsWithOnlyStaticData, + boolean enforceStrictLiveness) { - super(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData); + super(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData, enforceStrictLiveness); } @Override @@ -556,7 +584,7 @@ public abstract class DataLimits return new ThriftLimits(1, toFetch); } - public boolean hasEnoughLiveData(CachedPartition cached, int nowInSec, boolean countPartitionsWithOnlyStaticData) + public boolean hasEnoughLiveData(CachedPartition cached, int nowInSec, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness) { // We want the number of cells that are currently live. Getting that precise number forces // us to iterate the cached partition in general, but we can avoid that if: @@ -570,7 +598,7 @@ public abstract class DataLimits return false; // Otherwise, we need to re-count - DataLimits.Counter counter = newCounter(nowInSec, false, countPartitionsWithOnlyStaticData); + DataLimits.Counter counter = newCounter(nowInSec, false, countPartitionsWithOnlyStaticData, enforceStrictLiveness); try (UnfilteredRowIterator cacheIter = cached.unfilteredIterator(ColumnFilter.selection(cached.columns()), Slices.ALL, false); UnfilteredRowIterator iter = counter.applyTo(cacheIter)) { @@ -581,7 +609,7 @@ public abstract class DataLimits } } - public Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData) + public Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness) { return new ThriftCounter(nowInSec, assumeLiveData); } @@ -711,23 +739,26 @@ public abstract class DataLimits } @Override - public Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData) + public Counter newCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness) { - return new SuperColumnCountingCounter(nowInSec, assumeLiveData); + return new SuperColumnCountingCounter(nowInSec, assumeLiveData, enforceStrictLiveness); } protected class SuperColumnCountingCounter extends ThriftCounter { - public SuperColumnCountingCounter(int nowInSec, boolean assumeLiveData) + private final boolean enforceStrictLiveness; + + public SuperColumnCountingCounter(int nowInSec, boolean assumeLiveData, boolean enforceStrictLiveness) { super(nowInSec, assumeLiveData); + this.enforceStrictLiveness = enforceStrictLiveness; } @Override public Row applyToRow(Row row) { // In the internal format, a row == a super column, so that's what we want to count. - if (assumeLiveData || row.hasLiveData(nowInSec)) + if (assumeLiveData || row.hasLiveData(nowInSec, enforceStrictLiveness)) { ++cellsCounted; if (++cellsInCurrentPartition >= cellPerPartitionLimit) http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java b/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java index afe1cc3..9c6ab59 100644 --- a/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java +++ b/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java @@ -91,10 +91,11 @@ public class CachedBTreePartition extends ImmutableBTreePartition implements Cac int rowsWithNonExpiringCells = 0; int nonTombstoneCellCount = 0; int nonExpiringLiveCells = 0; + boolean enforceStrictLiveness = iterator.metadata().enforceStrictLiveness(); for (Row row : BTree.<Row>iterable(holder.tree)) { - if (row.hasLiveData(nowInSec)) + if (row.hasLiveData(nowInSec, enforceStrictLiveness)) ++cachedLiveRows; int nonExpiringLiveCellsThisRow = 0; http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/db/rows/AbstractRow.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/rows/AbstractRow.java b/src/java/org/apache/cassandra/db/rows/AbstractRow.java index 59addeb..67ed219 100644 --- a/src/java/org/apache/cassandra/db/rows/AbstractRow.java +++ b/src/java/org/apache/cassandra/db/rows/AbstractRow.java @@ -42,11 +42,13 @@ public abstract class AbstractRow extends AbstractCollection<ColumnData> impleme return Unfiltered.Kind.ROW; } - public boolean hasLiveData(int nowInSec) + @Override + public boolean hasLiveData(int nowInSec, boolean enforceStrictLiveness) { if (primaryKeyLivenessInfo().isLive(nowInSec)) return true; - + else if (enforceStrictLiveness) + return false; return Iterables.any(cells(), cell -> cell.isLive(nowInSec)); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/db/rows/Row.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index 3bcc220..3c97e09 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -103,8 +103,13 @@ public interface Row extends Unfiltered, Collection<ColumnData> /** * Whether the row has some live information (i.e. it's not just deletion informations). + * + * @param nowInSec the current time to decide what is deleted and what isn't + * @param enforceStrictLiveness whether the row should be purged if there is no PK liveness info, + * normally retrieved from {@link CFMetaData#enforceStrictLiveness()} + * @return true if there is some live information */ - public boolean hasLiveData(int nowInSec); + public boolean hasLiveData(int nowInSec, boolean enforceStrictLiveness); /** * Returns a cell for a simple column. http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/db/view/ViewUpdateGenerator.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/view/ViewUpdateGenerator.java b/src/java/org/apache/cassandra/db/view/ViewUpdateGenerator.java index 0c8e078..341c511 100644 --- a/src/java/org/apache/cassandra/db/view/ViewUpdateGenerator.java +++ b/src/java/org/apache/cassandra/db/view/ViewUpdateGenerator.java @@ -51,6 +51,7 @@ public class ViewUpdateGenerator private final ByteBuffer[] basePartitionKey; private final CFMetaData viewMetadata; + private final boolean baseEnforceStrictLiveness; private final Map<DecoratedKey, PartitionUpdate> updates = new HashMap<>(); @@ -87,6 +88,7 @@ public class ViewUpdateGenerator this.nowInSec = nowInSec; this.baseMetadata = view.getDefinition().baseTableMetadata(); + this.baseEnforceStrictLiveness = baseMetadata.enforceStrictLiveness(); this.baseDecoratedKey = basePartitionKey; this.basePartitionKey = extractKeyComponents(basePartitionKey, baseMetadata.getKeyValidator()); @@ -186,8 +188,8 @@ public class ViewUpdateGenerator // The view entry is necessarily the same pre and post update. // Note that we allow existingBaseRow to be null and treat it as empty (see MultiViewUpdateBuilder.generateViewsMutations). - boolean existingHasLiveData = existingBaseRow != null && existingBaseRow.hasLiveData(nowInSec); - boolean mergedHasLiveData = mergedBaseRow.hasLiveData(nowInSec); + boolean existingHasLiveData = existingBaseRow != null && existingBaseRow.hasLiveData(nowInSec, baseEnforceStrictLiveness); + boolean mergedHasLiveData = mergedBaseRow.hasLiveData(nowInSec, baseEnforceStrictLiveness); return existingHasLiveData ? (mergedHasLiveData ? UpdateAction.UPDATE_EXISTING : UpdateAction.DELETE_OLD) : (mergedHasLiveData ? UpdateAction.NEW_ENTRY : UpdateAction.NONE); http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/index/internal/composites/ClusteringColumnIndex.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/index/internal/composites/ClusteringColumnIndex.java b/src/java/org/apache/cassandra/index/internal/composites/ClusteringColumnIndex.java index b932602..cace6de 100644 --- a/src/java/org/apache/cassandra/index/internal/composites/ClusteringColumnIndex.java +++ b/src/java/org/apache/cassandra/index/internal/composites/ClusteringColumnIndex.java @@ -45,9 +45,12 @@ import org.apache.cassandra.schema.IndexMetadata; */ public class ClusteringColumnIndex extends CassandraIndex { + private final boolean enforceStrictLiveness; + public ClusteringColumnIndex(ColumnFamilyStore baseCfs, IndexMetadata indexDef) { super(baseCfs, indexDef); + this.enforceStrictLiveness = baseCfs.metadata.enforceStrictLiveness(); } @@ -95,6 +98,6 @@ public class ClusteringColumnIndex extends CassandraIndex public boolean isStale(Row data, ByteBuffer indexValue, int nowInSec) { - return !data.hasLiveData(nowInSec); + return !data.hasLiveData(nowInSec, enforceStrictLiveness); } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/index/internal/composites/PartitionKeyIndex.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/index/internal/composites/PartitionKeyIndex.java b/src/java/org/apache/cassandra/index/internal/composites/PartitionKeyIndex.java index 2c0b5aa..d854102 100644 --- a/src/java/org/apache/cassandra/index/internal/composites/PartitionKeyIndex.java +++ b/src/java/org/apache/cassandra/index/internal/composites/PartitionKeyIndex.java @@ -47,9 +47,11 @@ import org.apache.cassandra.schema.IndexMetadata; */ public class PartitionKeyIndex extends CassandraIndex { + private final boolean enforceStrictLiveness; public PartitionKeyIndex(ColumnFamilyStore baseCfs, IndexMetadata indexDef) { super(baseCfs, indexDef); + this.enforceStrictLiveness = baseCfs.metadata.enforceStrictLiveness(); } public ByteBuffer getIndexedValue(ByteBuffer partitionKey, @@ -90,6 +92,6 @@ public class PartitionKeyIndex extends CassandraIndex public boolean isStale(Row data, ByteBuffer indexValue, int nowInSec) { - return !data.hasLiveData(nowInSec); + return !data.hasLiveData(nowInSec, enforceStrictLiveness); } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/service/DataResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/DataResolver.java b/src/java/org/apache/cassandra/service/DataResolver.java index f1eedd1..61bffe5 100644 --- a/src/java/org/apache/cassandra/service/DataResolver.java +++ b/src/java/org/apache/cassandra/service/DataResolver.java @@ -44,9 +44,12 @@ public class DataResolver extends ResponseResolver @VisibleForTesting final List<AsyncOneResponse> repairResults = Collections.synchronizedList(new ArrayList<>()); + private final boolean enforceStrictLiveness; + DataResolver(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistency, int maxResponseCount) { super(keyspace, command, consistency, maxResponseCount); + this.enforceStrictLiveness = command.metadata().enforceStrictLiveness(); } public PartitionIterator getData() @@ -98,7 +101,7 @@ public class DataResolver extends ResponseResolver */ DataLimits.Counter mergedResultCounter = - command.limits().newCounter(command.nowInSec(), true, command.selectsFullPartition()); + command.limits().newCounter(command.nowInSec(), true, command.selectsFullPartition(), enforceStrictLiveness); UnfilteredPartitionIterator merged = mergeWithShortReadProtection(iters, sources, mergedResultCounter); FilteredPartitions filtered = @@ -125,7 +128,7 @@ public class DataResolver extends ResponseResolver for (int i = 0; i < results.size(); i++) { DataLimits.Counter singleResultCounter = - command.limits().newCounter(command.nowInSec(), false, command.selectsFullPartition()).onlyCount(); + command.limits().newCounter(command.nowInSec(), false, command.selectsFullPartition(), enforceStrictLiveness).onlyCount(); ShortReadResponseProtection protection = new ShortReadResponseProtection(sources[i], singleResultCounter, mergedResultCounter); http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/service/StorageProxy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 6bf275d..5af2ad0 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -1598,10 +1598,13 @@ public class StorageProxy implements StorageProxyMBean try { PartitionIterator result = fetchRows(group.commands, consistencyLevel); + // Note that the only difference between the command in a group must be the partition key on which + // they applied. + boolean enforceStrictLiveness = group.commands.get(0).metadata().enforceStrictLiveness(); // If we have more than one command, then despite each read command honoring the limit, the total result // might not honor it and so we should enforce it if (group.commands.size() > 1) - result = group.limits().filter(result, group.nowInSec(), group.selectsFullPartition()); + result = group.limits().filter(result, group.nowInSec(), group.selectsFullPartition(), enforceStrictLiveness); return result; } catch (UnavailableException e) @@ -1995,6 +1998,7 @@ public class StorageProxy implements StorageProxyMBean private final PartitionRangeReadCommand command; private final Keyspace keyspace; private final ConsistencyLevel consistency; + private final boolean enforceStrictLiveness; private final long startTime; private DataLimits.Counter counter; @@ -2015,6 +2019,7 @@ public class StorageProxy implements StorageProxyMBean this.totalRangeCount = ranges.rangeCount(); this.consistency = consistency; this.keyspace = keyspace; + this.enforceStrictLiveness = command.metadata().enforceStrictLiveness(); } public RowIterator computeNext() @@ -2119,7 +2124,7 @@ public class StorageProxy implements StorageProxyMBean Tracing.trace("Submitted {} concurrent range requests", concurrentQueries.size()); // We want to count the results for the sake of updating the concurrency factor (see updateConcurrencyFactor) but we don't want to // enforce any particular limit at this point (this could break code than rely on postReconciliationProcessing), hence the DataLimits.NONE. - counter = DataLimits.NONE.newCounter(command.nowInSec(), true, command.selectsFullPartition()); + counter = DataLimits.NONE.newCounter(command.nowInSec(), true, command.selectsFullPartition(), enforceStrictLiveness); return counter.applyTo(PartitionIterators.concat(concurrentQueries)); } @@ -2163,7 +2168,8 @@ public class StorageProxy implements StorageProxyMBean return command.limits().filter(command.postReconciliationProcessing(new RangeCommandIterator(ranges, command, concurrencyFactor, keyspace, consistencyLevel)), command.nowInSec(), - command.selectsFullPartition()); + command.selectsFullPartition(), + command.metadata().enforceStrictLiveness()); } public Map<String, List<String>> getSchemaVersions() http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java b/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java index ffd1b82..f44aa24 100644 --- a/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java +++ b/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java @@ -32,6 +32,7 @@ abstract class AbstractQueryPager implements QueryPager protected final ReadCommand command; protected final DataLimits limits; protected final int protocolVersion; + private final boolean enforceStrictLiveness; private int remaining; @@ -48,6 +49,7 @@ abstract class AbstractQueryPager implements QueryPager this.command = command; this.protocolVersion = protocolVersion; this.limits = command.limits(); + this.enforceStrictLiveness = command.metadata().enforceStrictLiveness(); this.remaining = limits.count(); this.remainingInPartition = limits.perPartitionCount(); @@ -126,7 +128,7 @@ abstract class AbstractQueryPager implements QueryPager private Pager(DataLimits pageLimits, int nowInSec) { - this.counter = pageLimits.newCounter(nowInSec, true, command.selectsFullPartition()); + this.counter = pageLimits.newCounter(nowInSec, true, command.selectsFullPartition(), enforceStrictLiveness); this.pageLimits = pageLimits; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java b/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java index 11bbc0e..344c64d 100644 --- a/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java +++ b/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java @@ -126,7 +126,11 @@ public class MultiPartitionPager implements QueryPager { int toQuery = Math.min(remaining, pageSize); PagersIterator iter = new PagersIterator(toQuery, consistency, clientState, null); - DataLimits.Counter counter = limit.forPaging(toQuery).newCounter(nowInSec, true, selectsFullPartitions); + /** + * It's safe to set it as false since all PartitionIterators have been filtered by each SPRC. + */ + boolean enforceStrictLiveness = false; + DataLimits.Counter counter = limit.forPaging(toQuery).newCounter(nowInSec, true, selectsFullPartitions, enforceStrictLiveness); iter.setCounter(counter); return counter.applyTo(iter); } @@ -136,7 +140,11 @@ public class MultiPartitionPager implements QueryPager { int toQuery = Math.min(remaining, pageSize); PagersIterator iter = new PagersIterator(toQuery, null, null, orderGroup); - DataLimits.Counter counter = limit.forPaging(toQuery).newCounter(nowInSec, true, selectsFullPartitions); + /** + * It's safe to set it as false since all PartitionIterators have been filtered by each SPRC. + */ + boolean enforceStrictLiveness = false; + DataLimits.Counter counter = limit.forPaging(toQuery).newCounter(nowInSec, true, selectsFullPartitions, enforceStrictLiveness); iter.setCounter(counter); return counter.applyTo(iter); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/src/java/org/apache/cassandra/service/pager/QueryPagers.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/pager/QueryPagers.java b/src/java/org/apache/cassandra/service/pager/QueryPagers.java index c26bf3f..6bc1f80 100644 --- a/src/java/org/apache/cassandra/service/pager/QueryPagers.java +++ b/src/java/org/apache/cassandra/service/pager/QueryPagers.java @@ -55,7 +55,7 @@ public class QueryPagers { try (PartitionIterator iter = pager.fetchPage(pageSize, consistencyLevel, state)) { - DataLimits.Counter counter = limits.newCounter(nowInSec, true, command.selectsFullPartition()); + DataLimits.Counter counter = limits.newCounter(nowInSec, true, command.selectsFullPartition(), metadata.enforceStrictLiveness()); PartitionIterators.consume(counter.applyTo(iter)); count += counter.counted(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java b/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java index 9e32620..d0f80f8 100644 --- a/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java +++ b/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java @@ -796,6 +796,7 @@ public class ViewComplexTest extends CQLTester if (flush) FBUtilities.waitOnFutures(ks.flush()); assertRowsIgnoringOrder(execute("SELECT v1, p, v2, WRITETIME(v2) from mv"), row(2, 3, 3, 6L)); + assertRowsIgnoringOrder(execute("SELECT v1, p, v2, WRITETIME(v2) from mv limit 1"), row(2, 3, 3, 6L)); // change v1's to 1 and remove existing view row with ts8 updateView("UPdate %s using timestamp 8 set v1 = 1 where p = 3;"); if (flush) @@ -804,6 +805,65 @@ public class ViewComplexTest extends CQLTester } @Test + public void testExpiredLivenessLimitWithFlush() throws Throwable + { + // CASSANDRA-13883 + testExpiredLivenessLimit(true); + } + + @Test + public void testExpiredLivenessLimitWithoutFlush() throws Throwable + { + // CASSANDRA-13883 + testExpiredLivenessLimit(false); + } + + private void testExpiredLivenessLimit(boolean flush) throws Throwable + { + createTable("CREATE TABLE %s (k int PRIMARY KEY, a int, b int);"); + + execute("USE " + keyspace()); + executeNet(protocolVersion, "USE " + keyspace()); + Keyspace ks = Keyspace.open(keyspace()); + + createView("mv1", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE k IS NOT NULL AND a IS NOT NULL PRIMARY KEY (k, a);"); + createView("mv2", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE k IS NOT NULL AND a IS NOT NULL PRIMARY KEY (a, k);"); + ks.getColumnFamilyStore("mv1").disableAutoCompaction(); + ks.getColumnFamilyStore("mv2").disableAutoCompaction(); + + for (int i = 1; i <= 100; i++) + updateView("INSERT INTO %s(k, a, b) VALUES (?, ?, ?);", i, i, i); + for (int i = 1; i <= 100; i++) + { + if (i % 50 == 0) + continue; + // create expired liveness + updateView("DELETE a FROM %s WHERE k = ?;", i); + } + if (flush) + { + ks.getColumnFamilyStore("mv1").forceBlockingFlush(); + ks.getColumnFamilyStore("mv2").forceBlockingFlush(); + } + + for (String view : Arrays.asList("mv1", "mv2")) + { + // paging + assertEquals(1, executeNetWithPaging(String.format("SELECT k,a,b FROM %s limit 1", view), 1).all().size()); + assertEquals(2, executeNetWithPaging(String.format("SELECT k,a,b FROM %s limit 2", view), 1).all().size()); + assertEquals(2, executeNetWithPaging(String.format("SELECT k,a,b FROM %s", view), 1).all().size()); + assertRowsNet(executeNetWithPaging(String.format("SELECT k,a,b FROM %s ", view), 1), + row(50, 50, 50), + row(100, 100, 100)); + // limit + assertEquals(1, execute(String.format("SELECT k,a,b FROM %s limit 1", view)).size()); + assertRowsIgnoringOrder(execute(String.format("SELECT k,a,b FROM %s limit 2", view)), + row(50, 50, 50), + row(100, 100, 100)); + } + } + + @Test public void testUpdateWithColumnTimestampBiggerThanPkWithFlush() throws Throwable { // CASSANDRA-11500 @@ -848,6 +908,7 @@ public class ViewComplexTest extends CQLTester FBUtilities.waitOnFutures(ks.flush()); ks.getColumnFamilyStore("mv").forceMajorCompaction(); assertRowsIgnoringOrder(execute("SELECT k,a,b from mv"), row(1, 2, 2)); + assertRowsIgnoringOrder(execute("SELECT k,a,b from mv limit 1"), row(1, 2, 2)); updateView("UPDATE %s USING TIMESTAMP 11 SET a = 1 WHERE k = 1;"); if (flush) FBUtilities.waitOnFutures(ks.flush()); @@ -1048,12 +1109,12 @@ public class ViewComplexTest extends CQLTester .sorted(Comparator.comparingInt(s -> s.descriptor.generation)) .map(s -> s.getFilename()) .collect(Collectors.toList()); - System.out.println("SSTables " + sstables); String dataFiles = String.join(",", Arrays.asList(sstables.get(1), sstables.get(2))); CompactionManager.instance.forceUserDefinedCompaction(dataFiles); } // cell-tombstone in sstable 4 is not compacted away, because the shadowable tombstone is shadowed by new row. assertRowsIgnoringOrder(execute("SELECT v1, p, v2, WRITETIME(v2) from mv"), row(1, 3, null, null)); + assertRowsIgnoringOrder(execute("SELECT v1, p, v2, WRITETIME(v2) from mv limit 1"), row(1, 3, null, null)); } @Test @@ -1172,6 +1233,7 @@ public class ViewComplexTest extends CQLTester FBUtilities.waitOnFutures(ks.flush()); // deleted column in MV remained dead assertRowsIgnoringOrder(execute("SELECT * from mv"), row(1, 3, null)); + assertRowsIgnoringOrder(execute("SELECT * from mv limit 1"), row(1, 3, null)); // insert values TS=2, it should be considered dead due to previous tombstone executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET v2 = ? WHERE p = ?", 4, 3); @@ -1183,6 +1245,7 @@ public class ViewComplexTest extends CQLTester ks.getColumnFamilyStore("mv").forceMajorCompaction(); assertRows(execute("SELECT v1, p, v2, WRITETIME(v2) from mv"), row(1, 3, 4, 3L)); + assertRows(execute("SELECT v1, p, v2, WRITETIME(v2) from mv limit 1"), row(1, 3, 4, 3L)); } @Test http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/test/unit/org/apache/cassandra/cql3/ViewTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/cql3/ViewTest.java b/test/unit/org/apache/cassandra/cql3/ViewTest.java index 84b2773..1107a64 100644 --- a/test/unit/org/apache/cassandra/cql3/ViewTest.java +++ b/test/unit/org/apache/cassandra/cql3/ViewTest.java @@ -363,7 +363,9 @@ public class ViewTest extends CQLTester updateView("UPDATE %s USING TIMESTAMP 2 SET val = ? WHERE k = ?", 1, 0); updateView("UPDATE %s USING TIMESTAMP 4 SET c = ? WHERE k = ?", 2, 0); updateView("UPDATE %s USING TIMESTAMP 3 SET val = ? WHERE k = ?", 2, 0); + assertRows(execute("SELECT c, k, val FROM mv_rctstest"), row(2, 0, 2)); + assertRows(execute("SELECT c, k, val FROM mv_rctstest limit 1"), row(2, 0, 2)); } @Test http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java b/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java index d0cc890..967a85c 100644 --- a/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java +++ b/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java @@ -76,6 +76,7 @@ public class RangeTombstoneTest { Keyspace keyspace = Keyspace.open(KSNAME); ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CFNAME); + boolean enforceStrictLiveness = cfs.metadata.enforceStrictLiveness(); // Inserting data String key = "k1"; @@ -112,17 +113,21 @@ public class RangeTombstoneTest int nowInSec = FBUtilities.nowInSeconds(); for (int i : live) - assertTrue("Row " + i + " should be live", partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec)); + assertTrue("Row " + i + " should be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); for (int i : dead) - assertFalse("Row " + i + " shouldn't be live", partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec)); + assertFalse("Row " + i + " shouldn't be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); // Queries by slices partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(7).toIncl(30).build()); for (int i : new int[]{ 7, 8, 9, 11, 13, 15, 17, 28, 29, 30 }) - assertTrue("Row " + i + " should be live", partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec)); + assertTrue("Row " + i + " should be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); for (int i : new int[]{ 10, 12, 14, 16, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27 }) - assertFalse("Row " + i + " shouldn't be live", partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec)); + assertFalse("Row " + i + " shouldn't be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); } @Test @@ -385,6 +390,7 @@ public class RangeTombstoneTest CompactionManager.instance.disableAutoCompaction(); Keyspace keyspace = Keyspace.open(KSNAME); ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CFNAME); + boolean enforceStrictLiveness = cfs.metadata.enforceStrictLiveness(); // Inserting data String key = "k2"; @@ -408,22 +414,30 @@ public class RangeTombstoneTest int nowInSec = FBUtilities.nowInSeconds(); for (int i = 0; i < 5; i++) - assertTrue("Row " + i + " should be live", partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec)); + assertTrue("Row " + i + " should be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); for (int i = 16; i < 20; i++) - assertTrue("Row " + i + " should be live", partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec)); + assertTrue("Row " + i + " should be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); for (int i = 5; i <= 15; i++) - assertFalse("Row " + i + " shouldn't be live", partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec)); + assertFalse("Row " + i + " shouldn't be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); // Compact everything and re-test CompactionManager.instance.performMaximal(cfs, false); partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build()); for (int i = 0; i < 5; i++) - assertTrue("Row " + i + " should be live", partition.getRow(new Clustering(bb(i))).hasLiveData(FBUtilities.nowInSeconds())); + assertTrue("Row " + i + " should be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(FBUtilities.nowInSeconds(), + enforceStrictLiveness)); for (int i = 16; i < 20; i++) - assertTrue("Row " + i + " should be live", partition.getRow(new Clustering(bb(i))).hasLiveData(FBUtilities.nowInSeconds())); + assertTrue("Row " + i + " should be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(FBUtilities.nowInSeconds(), + enforceStrictLiveness)); for (int i = 5; i <= 15; i++) - assertFalse("Row " + i + " shouldn't be live", partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec)); + assertFalse("Row " + i + " shouldn't be live", + partition.getRow(new Clustering(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); } @Test http://git-wip-us.apache.org/repos/asf/cassandra/blob/68bdf454/test/unit/org/apache/cassandra/db/compaction/CompactionsPurgeTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionsPurgeTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionsPurgeTest.java index 436b916..f02f4c2 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionsPurgeTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionsPurgeTest.java @@ -320,6 +320,7 @@ public class CompactionsPurgeTest Keyspace keyspace = Keyspace.open(KEYSPACE2); String cfName = "Standard1"; ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName); + final boolean enforceStrictLiveness = cfs.metadata.enforceStrictLiveness(); String key3 = "key3"; // inserts @@ -352,7 +353,7 @@ public class CompactionsPurgeTest ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key3).build()); assertEquals(2, partition.rowCount()); for (Row row : partition) - assertFalse(row.hasLiveData(FBUtilities.nowInSeconds())); + assertFalse(row.hasLiveData(FBUtilities.nowInSeconds(), enforceStrictLiveness)); } @Test --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org
