package org.apache.cassandra.service.pager;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.transport.ProtocolVersion;
public class PartitionRangeQueryPager extends AbstractQueryPager
{
private volatile DecoratedKey lastReturnedKey;
private volatile PagingState.RowMark lastReturnedRow;
public PartitionRangeQueryPager(PartitionRangeReadCommand command, PagingState state, ProtocolVersion protocolVersion)
{
super(command, protocolVersion);
if (state != null)
{
lastReturnedKey = command.metadata().decorateKey(state.partitionKey);
lastReturnedRow = state.rowMark;
restoreState(lastReturnedKey, state.remaining, state.remainingInPartition);
}
}
public PartitionRangeQueryPager(ReadCommand command,
ProtocolVersion protocolVersion,
DecoratedKey lastReturnedKey,
PagingState.RowMark lastReturnedRow,
int remaining,
int remainingInPartition)
{
super(command, protocolVersion);
this.lastReturnedKey = lastReturnedKey;
this.lastReturnedRow = lastReturnedRow;
restoreState(lastReturnedKey, remaining, remainingInPartition);
}
public PartitionRangeQueryPager withUpdatedLimit(DataLimits newLimits)
{
return new PartitionRangeQueryPager(command.withUpdatedLimit(newLimits),
protocolVersion,
lastReturnedKey,
lastReturnedRow,
maxRemaining(),
remainingInPartition());
}
public PagingState state()
{
return lastReturnedKey == null
? null
: new PagingState(lastReturnedKey.getKey(), lastReturnedRow, maxRemaining(), remainingInPartition());
}
protected ReadCommand nextPageReadCommand(int pageSize)
throws RequestExecutionException
{
DataLimits limits;
DataRange fullRange = ((PartitionRangeReadCommand)command).dataRange();
DataRange pageRange;
if (lastReturnedKey == null)
{
pageRange = fullRange;
limits = command.limits().forPaging(pageSize);
}
else
{
boolean includeLastKey = remainingInPartition() > 0 && lastReturnedRow != null;
AbstractBounds<PartitionPosition> bounds = makeKeyBounds(lastReturnedKey, includeLastKey);
if (includeLastKey)
{
pageRange = fullRange.forPaging(bounds, command.metadata().comparator, lastReturnedRow.clustering(command.metadata()), false);
limits = command.limits().forPaging(pageSize, lastReturnedKey.getKey(), remainingInPartition());
}
else
{
pageRange = fullRange.forSubRange(bounds);
limits = command.limits().forPaging(pageSize);
}
}
return ((PartitionRangeReadCommand) command).withUpdatedLimitsAndDataRange(limits, pageRange);
}
protected void recordLast(DecoratedKey key, Row last)
{
if (last != null)
{
lastReturnedKey = key;
if (last.clustering() != Clustering.STATIC_CLUSTERING)
lastReturnedRow = PagingState.RowMark.create(command.metadata(), last, protocolVersion);
}
}
protected boolean isPreviouslyReturnedPartition(DecoratedKey key)
{
return key.equals(lastReturnedKey);
}
private AbstractBounds<PartitionPosition> makeKeyBounds(PartitionPosition lastReturnedKey, boolean includeLastKey)
{
AbstractBounds<PartitionPosition> bounds = ((PartitionRangeReadCommand)command).dataRange().keyRange();
if (bounds instanceof Range || bounds instanceof Bounds)
{
return includeLastKey
? new Bounds<PartitionPosition>(lastReturnedKey, bounds.right)
: new Range<PartitionPosition>(lastReturnedKey, bounds.right);
}
else
{
return includeLastKey
? new IncludingExcludingBounds<PartitionPosition>(lastReturnedKey, bounds.right)
: new ExcludingBounds<PartitionPosition>(lastReturnedKey, bounds.right);
}
}
}