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

Define executeLocally() at the ReadQuery Level 

patch by Stefania Alborghetti; reviewed by Tyler Hobbs for CASSANDRA-12474
This commit is contained in:
Stefania Alborghetti 2016-08-17 17:45:34 +08:00
parent 21dda88ba0
commit f2d5cd0f7e
6 changed files with 231 additions and 6 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.10
* Define executeLocally() at the ReadQuery Level (CASSANDRA-12474)
* Extend read/write failure messages with a map of replica addresses
to error codes in the v5 native protocol (CASSANDRA-12311)
* Fix rebuild of SASI indexes with existing index files (CASSANDRA-12374)

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

@ -120,6 +120,11 @@ public class ReadExecutionController implements AutoCloseable
return index == null ? null : index.getBackingTable().orElse(null);
}
public CFMetaData metaData()
{
return baseMetadata;
}
public void close()
{
try

14
src/java/org/apache/cassandra/db/ReadQuery.java
View File

@ -50,6 +50,11 @@ public interface ReadQuery
return EmptyIterators.partition();
}
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
{
return EmptyIterators.unfilteredPartition(executionController.metaData(), false);
}
public DataLimits limits()
{
// What we return here doesn't matter much in practice. However, returning DataLimits.NONE means
@ -104,6 +109,15 @@ public interface ReadQuery
*/
public PartitionIterator executeInternal(ReadExecutionController controller);
/**
* Execute the query locally. This is similar to {@link ReadQuery#executeInternal(ReadExecutionController)}
* but it returns an unfiltered partition iterator that can be merged later on.
*
* @param controller the {@code ReadExecutionController} protecting the read.
* @return the result of the read query.
*/
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController);
/**
* Returns a pager for the query.
*

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

@ -20,10 +20,13 @@ package org.apache.cassandra.db;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.stream.Collectors;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.cassandra.cache.IRowCacheEntry;
import org.apache.cassandra.cache.RowCacheKey;
import org.apache.cassandra.cache.RowCacheSentinel;
@ -1012,12 +1015,35 @@ public class SinglePartitionReadCommand extends ReadCommand
public PartitionIterator executeInternal(ReadExecutionController controller)
{
List<PartitionIterator> partitions = new ArrayList<>(commands.size());
for (SinglePartitionReadCommand cmd : commands)
partitions.add(cmd.executeInternal(controller));
return limits.filter(UnfilteredPartitionIterators.filter(executeLocally(controller, false), nowInSec), nowInSec);
}
// Because we only have enforce the limit per command, we need to enforce it globally.
return limits.filter(PartitionIterators.concat(partitions), nowInSec);
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
{
return executeLocally(executionController, true);
}
/**
* Implementation of {@link ReadQuery#executeLocally(ReadExecutionController)}.
*
* @param executionController - the {@code ReadExecutionController} protecting the read.
* @param sort - whether to sort the inner commands by partition key, required for merging the iterator
* later on. This will be false when called by {@link ReadQuery#executeInternal(ReadExecutionController)}
* because in this case it is safe to do so as there is no merging involved and we don't want to
* change the old behavior which was to not sort by partition.
*
* @return - the iterator that can be used to retrieve the query result.
*/
private UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController, boolean sort)
{
List<Pair<DecoratedKey, UnfilteredPartitionIterator>> partitions = new ArrayList<>(commands.size());
for (SinglePartitionReadCommand cmd : commands)
partitions.add(Pair.of(cmd.partitionKey, cmd.executeLocally(executionController)));
if (sort)
Collections.sort(partitions, (p1, p2) -> p1.getLeft().compareTo(p2.getLeft()));
return UnfilteredPartitionIterators.concat(partitions.stream().map(p -> p.getRight()).collect(Collectors.toList()));
}
public QueryPager getPager(PagingState pagingState, int protocolVersion)

20
src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java
View File

@ -27,6 +27,7 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.FilteredPartitions;
import org.apache.cassandra.db.transform.MorePartitions;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -77,6 +78,25 @@ public abstract class UnfilteredPartitionIterators
return Transformation.apply(toReturn, new Close());
}
public static UnfilteredPartitionIterator concat(final List<UnfilteredPartitionIterator> iterators)
{
if (iterators.size() == 1)
return iterators.get(0);
class Extend implements MorePartitions<UnfilteredPartitionIterator>
{
int i = 1;
public UnfilteredPartitionIterator moreContents()
{
if (i >= iterators.size())
return null;
return iterators.get(i++);
}
}
return MorePartitions.extend(iterators.get(0), new Extend());
}
public static PartitionIterator mergeAndFilter(List<UnfilteredPartitionIterator> iterators, int nowInSec, MergeListener listener)
{
// TODO: we could have a somewhat faster version if we were to merge the UnfilteredRowIterators directly as RowIterators

161
test/unit/org/apache/cassandra/db/ReadCommandTest.java
View File

@ -18,6 +18,8 @@
package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.junit.BeforeClass;
@ -27,12 +29,28 @@ import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.partitions.FilteredPartition;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.rows.SerializationHelper;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import static org.junit.Assert.assertEquals;
@ -41,6 +59,7 @@ public class ReadCommandTest
private static final String KEYSPACE = "ReadCommandTest";
private static final String CF1 = "Standard1";
private static final String CF2 = "Standard2";
private static final String CF3 = "Standard3";
@BeforeClass
public static void defineSchema() throws ConfigurationException
@ -55,11 +74,22 @@ public class ReadCommandTest
.addRegularColumn("a", AsciiType.instance)
.addRegularColumn("b", AsciiType.instance).build();
CFMetaData metadata3 = CFMetaData.Builder.create(KEYSPACE, CF3)
.addPartitionKey("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).build();
SchemaLoader.prepareServer();
SchemaLoader.createKeyspace(KEYSPACE,
KeyspaceParams.simple(1),
metadata1,
metadata2);
metadata2,
metadata3);
}
@Test
@ -149,4 +179,133 @@ public class ReadCommandTest
readCommand.abort();
assertEquals(0, Util.getAll(readCommand).size());
}
@Test
public void testSinglePartitionGroupMerge() throws Exception
{
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF3);
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[][] {
new String[] { "1", "key3", "dd", "d" },
new String[] { "1", "key2", "ee", "e" },
new String[] { "1", "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[][] {
new String[] { "-1", "key6", "aa", "a" },
new String[] { "-1", "key2", "bb", "b" }
}
};
// Given the data above, when the keys are sorted and the deletions removed, we should
// get these clustering rows in this order
String[] expectedRows = new String[] { "aa", "ff", "ee", "cc", "dd", "cc", "bb"};
List<ByteBuffer> buffers = new ArrayList<>(groups.length);
int nowInSeconds = FBUtilities.nowInSeconds();
ColumnFilter columnFilter = ColumnFilter.allColumnsBuilder(cfs.metadata).build();
RowFilter rowFilter = RowFilter.create();
Slice slice = Slice.make(ClusteringBound.BOTTOM, ClusteringBound.TOP);
ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.with(cfs.metadata.comparator, slice), false);
for (String[][] group : groups)
{
cfs.truncateBlocking();
List<SinglePartitionReadCommand> commands = new ArrayList<>(group.length);
for (String[] data : group)
{
if (data[0].equals("1"))
{
new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes(data[1]))
.clustering(data[2])
.add(data[3], ByteBufferUtil.bytes("blah"))
.build()
.apply();
}
else
{
RowUpdateBuilder.deleteRow(cfs.metadata, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(data[1]), data[2]).apply();
}
commands.add(SinglePartitionReadCommand.create(cfs.metadata, nowInSeconds, columnFilter, rowFilter, DataLimits.NONE, Util.dk(data[1]), sliceFilter));
}
cfs.forceBlockingFlush();
ReadQuery query = new SinglePartitionReadCommand.Group(commands, DataLimits.NONE);
try (ReadExecutionController executionController = query.executionController();
UnfilteredPartitionIterator iter = query.executeLocally(executionController);
DataOutputBuffer buffer = new DataOutputBuffer())
{
UnfilteredPartitionIterators.serializerForIntraNode().serialize(iter,
columnFilter,
buffer,
MessagingService.current_version);
buffers.add(buffer.buffer());
}
}
// deserialize, merge and check the results are all there
List<UnfilteredPartitionIterator> iterators = new ArrayList<>();
for (ByteBuffer buffer : buffers)
{
try (DataInputBuffer in = new DataInputBuffer(buffer, true))
{
iterators.add(UnfilteredPartitionIterators.serializerForIntraNode().deserialize(in,
MessagingService.current_version,
cfs.metadata,
columnFilter,
SerializationHelper.Flag.LOCAL));
}
}
try(PartitionIterator partitionIterator = UnfilteredPartitionIterators.mergeAndFilter(iterators,
nowInSeconds,
new UnfilteredPartitionIterators.MergeListener()
{
public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List<UnfilteredRowIterator> versions)
{
return null;
}
public void close()
{
}
}))
{
int i = 0;
int numPartitions = 0;
while (partitionIterator.hasNext())
{
numPartitions++;
try(RowIterator rowIterator = partitionIterator.next())
{
while (rowIterator.hasNext())
{
Row row = rowIterator.next();
assertEquals("col=" + expectedRows[i++], row.clustering().toString(cfs.metadata));
//System.out.print(row.toString(cfs.metadata, true));
}
}
}
assertEquals(5, numPartitions);
assertEquals(expectedRows.length, i);
}
}
}