natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
Paulo Motta 2017-09-25 01:01:31 -05:00
parent 0e5c84ad8f 68bdf45477
commit 3e3d56ecd4
22 changed files with 226 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.11.1
* Handle limit correctly on tables with strict liveness (CASSANDRA-13883)
* AbstractTokenTreeBuilder#serializedSize returns wrong value when there is a single leaf and overflow collisions (CASSANDRA-13869)
* Add a compaction option to TWCS to ignore sstables overlapping checks (CASSANDRA-13418)
* BTree.Builder memory leak (CASSANDRA-13754)

5
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -1633,7 +1633,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);
}

3
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -463,6 +463,7 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery
private final int warningThreshold = DatabaseDescriptor.getTombstoneWarnThreshold();
private final boolean respectTombstoneThresholds = !SchemaConstants.isSystemKeyspace(ReadCommand.this.metadata().ksName);
private final boolean enforceStrictLiveness = metadata.enforceStrictLiveness();
private int liveRows = 0;
private int tombstones = 0;
@ -485,7 +486,7 @@ public abstract class ReadCommand extends MonitorableImpl 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())

9
src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
View File

@ -574,6 +574,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, executionController);
@ -597,7 +598,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;
@ -1200,9 +1201,13 @@ public class SinglePartitionReadCommand extends ReadCommand
public PartitionIterator executeInternal(ReadExecutionController controller)
{
// 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();
return limits.filter(UnfilteredPartitionIterators.filter(executeLocally(controller, false), nowInSec),
nowInSec,
selectsFullPartitions);
selectsFullPartitions,
enforceStrictLiveness);
}
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)

111
src/java/org/apache/cassandra/db/filter/DataLimits.java
View File

@ -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.aggregation.GroupMaker;
import org.apache.cassandra.db.aggregation.GroupingState;
@ -49,7 +50,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;
}
@ -150,7 +151,10 @@ public abstract class DataLimits
throw new UnsupportedOperationException();
}
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.
@ -161,9 +165,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.
@ -187,19 +196,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);
}
/**
@ -212,14 +229,16 @@ public abstract class DataLimits
{
protected final int nowInSec;
protected final boolean assumeLiveData;
private final boolean enforceStrictLiveness;
// false means we do not propagate our stop signals onto the iterator, we only count
private boolean enforceLimits = true;
protected Counter(int nowInSec, boolean assumeLiveData)
protected Counter(int nowInSec, boolean assumeLiveData, boolean enforceStrictLiveness)
{
this.nowInSec = nowInSec;
this.assumeLiveData = assumeLiveData;
this.enforceStrictLiveness = enforceStrictLiveness;
}
public Counter onlyCount()
@ -278,7 +297,7 @@ public abstract class DataLimits
protected boolean isLive(Row row)
{
return assumeLiveData || row.hasLiveData(nowInSec);
return assumeLiveData || row.hasLiveData(nowInSec, enforceStrictLiveness);
}
@Override
@ -383,7 +402,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:
@ -398,7 +417,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))
{
@ -409,9 +428,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()
@ -445,9 +467,12 @@ public abstract class DataLimits
protected boolean hasLiveStaticRow;
public CQLCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData)
public CQLCounter(int nowInSec,
boolean assumeLiveData,
boolean countPartitionsWithOnlyStaticData,
boolean enforceStrictLiveness)
{
super(nowInSec, assumeLiveData);
super(nowInSec, assumeLiveData, enforceStrictLiveness);
this.countPartitionsWithOnlyStaticData = countPartitionsWithOnlyStaticData;
}
@ -572,16 +597,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
@ -717,9 +745,12 @@ 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 GroupByAwareCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData);
return new GroupByAwareCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData, enforceStrictLiveness);
}
@Override
@ -814,9 +845,12 @@ public abstract class DataLimits
protected boolean hasReturnedRowsFromCurrentPartition;
private GroupByAwareCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData)
private GroupByAwareCounter(int nowInSec,
boolean assumeLiveData,
boolean countPartitionsWithOnlyStaticData,
boolean enforceStrictLiveness)
{
super(nowInSec, assumeLiveData);
super(nowInSec, assumeLiveData, enforceStrictLiveness);
this.groupMaker = groupBySpec.newGroupMaker(state);
this.countPartitionsWithOnlyStaticData = countPartitionsWithOnlyStaticData;
@ -1062,10 +1096,10 @@ 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)
{
assert state == GroupingState.EMPTY_STATE || lastReturnedKey.equals(state.partitionKey());
return new PagingGroupByAwareCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData);
return new PagingGroupByAwareCounter(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData, enforceStrictLiveness);
}
@Override
@ -1076,9 +1110,9 @@ public abstract class DataLimits
private class PagingGroupByAwareCounter extends GroupByAwareCounter
{
private PagingGroupByAwareCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData)
private PagingGroupByAwareCounter(int nowInSec, boolean assumeLiveData, boolean countPartitionsWithOnlyStaticData, boolean enforceStrictLiveness)
{
super(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData);
super(nowInSec, assumeLiveData, countPartitionsWithOnlyStaticData, enforceStrictLiveness);
}
@Override
@ -1151,7 +1185,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:
@ -1165,7 +1199,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))
{
@ -1176,9 +1210,9 @@ 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);
return new ThriftCounter(nowInSec, assumeLiveData, enforceStrictLiveness);
}
public int count()
@ -1209,9 +1243,9 @@ public abstract class DataLimits
protected int cellsCounted;
protected int cellsInCurrentPartition;
public ThriftCounter(int nowInSec, boolean assumeLiveData)
public ThriftCounter(int nowInSec, boolean assumeLiveData, boolean enforceStrictLiveness)
{
super(nowInSec, assumeLiveData);
super(nowInSec, assumeLiveData, enforceStrictLiveness);
}
@Override
@ -1317,16 +1351,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 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);
super(nowInSec, assumeLiveData, enforceStrictLiveness);
this.enforceStrictLiveness = enforceStrictLiveness;
}
@Override

3
src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java
View File

@ -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;

6
src/java/org/apache/cassandra/db/rows/AbstractRow.java
View File

@ -46,11 +46,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));
}

7
src/java/org/apache/cassandra/db/rows/Row.java
View File

@ -106,8 +106,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.

6
src/java/org/apache/cassandra/db/view/ViewUpdateGenerator.java
View File

@ -50,6 +50,7 @@ public class ViewUpdateGenerator
private final ByteBuffer[] basePartitionKey;
private final CFMetaData viewMetadata;
private final boolean baseEnforceStrictLiveness;
private final Map<DecoratedKey, PartitionUpdate> updates = new HashMap<>();
@ -86,6 +87,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());
@ -185,8 +187,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);

5
src/java/org/apache/cassandra/index/internal/composites/ClusteringColumnIndex.java
View File

@ -47,9 +47,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();
}
@ -97,6 +100,6 @@ public class ClusteringColumnIndex extends CassandraIndex
public boolean isStale(Row data, ByteBuffer indexValue, int nowInSec)
{
return !data.hasLiveData(nowInSec);
return !data.hasLiveData(nowInSec, enforceStrictLiveness);
}
}

4
src/java/org/apache/cassandra/index/internal/composites/PartitionKeyIndex.java
View File

@ -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);
}
}

6
src/java/org/apache/cassandra/service/DataResolver.java
View File

@ -44,11 +44,13 @@ public class DataResolver extends ResponseResolver
@VisibleForTesting
final List<AsyncOneResponse> repairResults = Collections.synchronizedList(new ArrayList<>());
private final long queryStartNanoTime;
private final boolean enforceStrictLiveness;
DataResolver(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistency, int maxResponseCount, long queryStartNanoTime)
{
super(keyspace, command, consistency, maxResponseCount);
this.queryStartNanoTime = queryStartNanoTime;
this.enforceStrictLiveness = command.metadata().enforceStrictLiveness();
}
public PartitionIterator getData()
@ -100,7 +102,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 =
@ -127,7 +129,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, queryStartNanoTime);

12
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -1679,10 +1679,13 @@ public class StorageProxy implements StorageProxyMBean
try
{
PartitionIterator result = fetchRows(group.commands, consistencyLevel, queryStartNanoTime);
// 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)
@ -2103,6 +2106,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 final long queryStartNanoTime;
@ -2125,6 +2129,7 @@ public class StorageProxy implements StorageProxyMBean
this.consistency = consistency;
this.keyspace = keyspace;
this.queryStartNanoTime = queryStartNanoTime;
this.enforceStrictLiveness = command.metadata().enforceStrictLiveness();
}
public RowIterator computeNext()
@ -2238,7 +2243,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));
}
@ -2282,7 +2287,8 @@ public class StorageProxy implements StorageProxyMBean
return command.limits().filter(command.postReconciliationProcessing(new RangeCommandIterator(ranges, command, concurrencyFactor, keyspace, consistencyLevel, queryStartNanoTime)),
command.nowInSec(),
command.selectsFullPartition());
command.selectsFullPartition(),
command.metadata().enforceStrictLiveness());
}
public Map<String, List<String>> getSchemaVersions()

4
src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java
View File

@ -31,6 +31,7 @@ abstract class AbstractQueryPager implements QueryPager
protected final ReadCommand command;
protected final DataLimits limits;
protected final ProtocolVersion protocolVersion;
private final boolean enforceStrictLiveness;
private int remaining;
@ -47,6 +48,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;
}

2
src/java/org/apache/cassandra/service/pager/QueryPagers.java
View File

@ -56,7 +56,7 @@ public class QueryPagers
{
try (PartitionIterator iter = pager.fetchPage(pageSize, consistencyLevel, state, queryStartNanoTime))
{
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();
}

3
test/long/org/apache/cassandra/cql3/CachingBench.java
View File

@ -340,8 +340,9 @@ public class CachingBench extends CQLTester
int countRows(ColumnFamilyStore cfs)
{
boolean enforceStrictLiveness = cfs.metadata.enforceStrictLiveness();
int nowInSec = FBUtilities.nowInSeconds();
return count(cfs, x -> x.isRow() && ((Row) x).hasLiveData(nowInSec));
return count(cfs, x -> x.isRow() && ((Row) x).hasLiveData(nowInSec, enforceStrictLiveness));
}
private int count(ColumnFamilyStore cfs, Predicate<Unfiltered> predicate)

3
test/long/org/apache/cassandra/cql3/GcCompactionBench.java
View File

@ -339,8 +339,9 @@ public class GcCompactionBench extends CQLTester
int countRows(ColumnFamilyStore cfs)
{
boolean enforceStrictLiveness = cfs.metadata.enforceStrictLiveness();
int nowInSec = FBUtilities.nowInSeconds();
return count(cfs, x -> x.isRow() && ((Row) x).hasLiveData(nowInSec));
return count(cfs, x -> x.isRow() && ((Row) x).hasLiveData(nowInSec, enforceStrictLiveness));
}
private int count(ColumnFamilyStore cfs, Predicate<Unfiltered> predicate)

3
test/unit/org/apache/cassandra/cql3/GcCompactionTest.java
View File

@ -339,8 +339,9 @@ public class GcCompactionTest extends CQLTester
int countRows(SSTableReader reader)
{
boolean enforceStrictLiveness = reader.metadata.enforceStrictLiveness();
int nowInSec = FBUtilities.nowInSeconds();
return count(reader, x -> x.isRow() && ((Row) x).hasLiveData(nowInSec) ? 1 : 0, x -> 0);
return count(reader, x -> x.isRow() && ((Row) x).hasLiveData(nowInSec, enforceStrictLiveness) ? 1 : 0, x -> 0);
}
int countCells(SSTableReader reader)

65
test/unit/org/apache/cassandra/cql3/ViewComplexTest.java
View File

@ -797,6 +797,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
assertRowsIgnoringOrder(execute("SELECT v1, p, v2, WRITETIME(v2) from mv"), row(1, 3, 3, 6L));
}
@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
{
@ -849,6 +909,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());
@ -1049,12 +1110,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
@ -1173,6 +1234,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);
@ -1184,6 +1246,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

2
test/unit/org/apache/cassandra/cql3/ViewTest.java
View File

@ -377,7 +377,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

35
test/unit/org/apache/cassandra/db/RangeTombstoneTest.java
View File

@ -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(Clustering.make(bb(i))).hasLiveData(nowInSec));
assertTrue("Row " + i + " should be live",
partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness));
for (int i : dead)
assertFalse("Row " + i + " shouldn't be live", partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec));
assertFalse("Row " + i + " shouldn't be live",
partition.getRow(Clustering.make(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(Clustering.make(bb(i))).hasLiveData(nowInSec));
assertTrue("Row " + i + " should be live",
partition.getRow(Clustering.make(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(Clustering.make(bb(i))).hasLiveData(nowInSec));
assertFalse("Row " + i + " shouldn't be live",
partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness));
}
@Test
@ -385,7 +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 +413,30 @@ public class RangeTombstoneTest
int nowInSec = FBUtilities.nowInSeconds();
for (int i = 0; i < 5; i++)
assertTrue("Row " + i + " should be live", partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec));
assertTrue("Row " + i + " should be live",
partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness));
for (int i = 16; i < 20; i++)
assertTrue("Row " + i + " should be live", partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec));
assertTrue("Row " + i + " should be live",
partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness));
for (int i = 5; i <= 15; i++)
assertFalse("Row " + i + " shouldn't be live", partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec));
assertFalse("Row " + i + " shouldn't be live",
partition.getRow(Clustering.make(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(Clustering.make(bb(i))).hasLiveData(FBUtilities.nowInSeconds()));
assertTrue("Row " + i + " should be live",
partition.getRow(Clustering.make(bb(i))).hasLiveData(FBUtilities.nowInSeconds(),
enforceStrictLiveness));
for (int i = 16; i < 20; i++)
assertTrue("Row " + i + " should be live", partition.getRow(Clustering.make(bb(i))).hasLiveData(FBUtilities.nowInSeconds()));
assertTrue("Row " + i + " should be live",
partition.getRow(Clustering.make(bb(i))).hasLiveData(FBUtilities.nowInSeconds(),
enforceStrictLiveness));
for (int i = 5; i <= 15; i++)
assertFalse("Row " + i + " shouldn't be live", partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec));
assertFalse("Row " + i + " shouldn't be live",
partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness));
}
@Test

3
test/unit/org/apache/cassandra/db/compaction/CompactionsPurgeTest.java
View File

@ -323,6 +323,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
@ -357,7 +358,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