diff --git a/CHANGES.txt b/CHANGES.txt index f46b23ef46..1c73c7d334 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/db/ReadExecutionController.java b/src/java/org/apache/cassandra/db/ReadExecutionController.java index 7ddc8dff6c..56bb0d30ee 100644 --- a/src/java/org/apache/cassandra/db/ReadExecutionController.java +++ b/src/java/org/apache/cassandra/db/ReadExecutionController.java @@ -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 diff --git a/src/java/org/apache/cassandra/db/ReadQuery.java b/src/java/org/apache/cassandra/db/ReadQuery.java index d74834c2ff..c6b2b3e9e8 100644 --- a/src/java/org/apache/cassandra/db/ReadQuery.java +++ b/src/java/org/apache/cassandra/db/ReadQuery.java @@ -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. * diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java index cea8c0ce18..ad73a17627 100644 --- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java +++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java @@ -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 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> 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) diff --git a/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java b/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java index b7f6793732..852d95ec70 100644 --- a/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java +++ b/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java @@ -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 iterators) + { + if (iterators.size() == 1) + return iterators.get(0); + + class Extend implements MorePartitions + { + 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 iterators, int nowInSec, MergeListener listener) { // TODO: we could have a somewhat faster version if we were to merge the UnfilteredRowIterators directly as RowIterators diff --git a/test/unit/org/apache/cassandra/db/ReadCommandTest.java b/test/unit/org/apache/cassandra/db/ReadCommandTest.java index fddd347de4..2aef2a774d 100644 --- a/test/unit/org/apache/cassandra/db/ReadCommandTest.java +++ b/test/unit/org/apache/cassandra/db/ReadCommandTest.java @@ -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 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 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 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 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); + } + } }