diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java index d896071ad7..c551d42463 100644 --- a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java +++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java @@ -22,6 +22,9 @@ import java.net.InetAddress; import java.nio.ByteBuffer; import java.util.*; +import com.google.common.annotations.VisibleForTesting; + +import org.apache.cassandra.service.ClientState; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.utils.AbstractIterator; @@ -53,6 +56,20 @@ public abstract class UntypedResultSet implements Iterable return new FromPager(select, pager, pageSize); } + /** + * This method is intended for testing purposes, since it executes query on cluster + * and not on the local node only. + */ + @VisibleForTesting + public static UntypedResultSet create(SelectStatement select, + ConsistencyLevel cl, + ClientState clientState, + QueryPager pager, + int pageSize) + { + return new FromDistributedPager(select, cl, clientState, pager, pageSize); + } + public boolean isEmpty() { return size() == 0; @@ -106,6 +123,71 @@ public abstract class UntypedResultSet implements Iterable } } + /** + * Pager that calls `execute` rather than `executeInternal` + */ + private static class FromDistributedPager extends UntypedResultSet + { + private final SelectStatement select; + private final ConsistencyLevel cl; + private final ClientState clientState; + private final QueryPager pager; + private final int pageSize; + private final List metadata; + + private FromDistributedPager(SelectStatement select, + ConsistencyLevel cl, + ClientState clientState, + QueryPager pager, int pageSize) + { + this.select = select; + this.cl = cl; + this.clientState = clientState; + this.pager = pager; + this.pageSize = pageSize; + this.metadata = select.getResultMetadata().requestNames(); + } + + public int size() + { + throw new UnsupportedOperationException(); + } + + public Row one() + { + throw new UnsupportedOperationException(); + } + + public Iterator iterator() + { + return new AbstractIterator() + { + private Iterator> currentPage; + + protected Row computeNext() + { + int nowInSec = FBUtilities.nowInSeconds(); + while (currentPage == null || !currentPage.hasNext()) + { + if (pager.isExhausted()) + return endOfData(); + + try (PartitionIterator iter = pager.fetchPage(pageSize, cl, clientState, System.nanoTime())) + { + currentPage = select.process(iter, nowInSec).rows.iterator(); + } + } + return new Row(metadata, currentPage.next()); + } + }; + } + + public List metadata() + { + return metadata; + } + } + private static class FromResultList extends UntypedResultSet { private final List> cqlRows; diff --git a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java index c7bdb36d30..59afd8bde6 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java @@ -18,10 +18,14 @@ package org.apache.cassandra.distributed.api; +import java.util.Iterator; + // The cross-version API requires that a Coordinator can be constructed without any constructor arguments public interface ICoordinator { // a bit hacky, but ConsistencyLevel draws in too many dependent classes, so we cannot have a cross-version // method signature that accepts ConsistencyLevel directly. So we just accept an Enum and cast. Object[][] execute(String query, Enum consistencyLevel, Object... boundValues); + + Iterator executeWithPaging(String query, Enum consistencyLevel, int pageSize, Object... boundValues); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java index 038648c832..ec5e58f8b4 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java @@ -18,20 +18,26 @@ package org.apache.cassandra.distributed.impl; +import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Iterator; import java.util.List; import org.apache.cassandra.cql3.CQLStatement; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.cql3.statements.SelectStatement; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.distributed.api.ICoordinator; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.QueryState; +import org.apache.cassandra.service.pager.QueryPager; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.transport.messages.ResultMessage; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; public class Coordinator implements ICoordinator { @@ -46,7 +52,8 @@ public class Coordinator implements ICoordinator { return instance.sync(() -> { ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name()); - CQLStatement prepared = QueryProcessor.getStatement(query, ClientState.forInternalCalls()).statement; + ClientState clientState = makeFakeClientState(); + CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement; List boundBBValues = new ArrayList<>(); for (Object boundValue : boundValues) { @@ -57,7 +64,7 @@ public class Coordinator implements ICoordinator QueryOptions.create(consistencyLevel, boundBBValues, false, - 10, + Integer.MAX_VALUE, null, null, ProtocolVersion.CURRENT), @@ -73,4 +80,59 @@ public class Coordinator implements ICoordinator } }).call(); } + + @Override + public Iterator executeWithPaging(String query, Enum consistencyLevelOrigin, int pageSize, Object... boundValues) + { + if (pageSize <= 0) + throw new IllegalArgumentException("Page size should be strictly positive but was " + pageSize); + + return instance.sync(() -> { + ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name()); + ClientState clientState = makeFakeClientState(); + CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement; + List boundBBValues = new ArrayList<>(); + for (Object boundValue : boundValues) + { + boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue)); + } + + prepared.validate(QueryState.forInternalCalls().getClientState()); + assert prepared instanceof SelectStatement : "Only SELECT statements can be executed with paging"; + + SelectStatement selectStatement = (SelectStatement) prepared; + + QueryPager pager = selectStatement.getQuery(QueryOptions.create(consistencyLevel, + boundBBValues, + false, + pageSize, + null, + null, + ProtocolVersion.CURRENT), + FBUtilities.nowInSeconds()) + .getPager(null, ProtocolVersion.CURRENT); + + // Usually pager fetches a single page (see SelectStatement#execute). We need to iterate over all + // of the results lazily. + return new Iterator() { + Iterator iter = RowUtil.toObjects(UntypedResultSet.create(selectStatement, consistencyLevel, clientState, pager, pageSize)); + + public boolean hasNext() + { + // We have to make sure iterator is not running on main thread. + return instance.sync(() -> iter.hasNext()).call(); + } + + public Object[] next() + { + return instance.sync(() -> iter.next()).call(); + } + }; + }).call(); + } + + private static final ClientState makeFakeClientState() + { + return ClientState.forExternalCalls(new InetSocketAddress(FBUtilities.getLocalAddress(), 9042)); + } } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java index c3b129bac8..e83b85e6eb 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java @@ -19,9 +19,13 @@ package org.apache.cassandra.distributed.impl; import java.nio.ByteBuffer; +import java.util.Iterator; import java.util.List; +import com.google.common.collect.Iterators; + import org.apache.cassandra.cql3.ColumnSpecification; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.transport.messages.ResultMessage; public class RowUtil @@ -44,4 +48,26 @@ public class RowUtil } return result; } + + public static Iterator toObjects(UntypedResultSet rs) + { + return toObjects(rs.metadata(), rs.iterator()); + } + + public static Iterator toObjects(List columnSpecs, Iterator rs) + { + return Iterators.transform(rs, + (row) -> { + Object[] objectRow = new Object[columnSpecs.size()]; + for (int i = 0; i < columnSpecs.size(); i++) + { + ColumnSpecification columnSpec = columnSpecs.get(i); + ByteBuffer bb = row.getBytes(columnSpec.name.toString()); + + if (bb != null) + objectRow[i] = columnSpec.type.getSerializer().deserialize(bb); + } + return objectRow; + }); + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java index 75131b3792..3578b2c750 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java @@ -27,7 +27,7 @@ import org.apache.cassandra.distributed.Cluster; public class DistributedReadWritePathTest extends DistributedTestBase { @Test - public void coordinatorRead() throws Throwable + public void coordinatorReadTest() throws Throwable { try (Cluster cluster = init(Cluster.create(3))) { @@ -47,7 +47,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase } @Test - public void coordinatorWrite() throws Throwable + public void coordinatorWriteTest() throws Throwable { try (Cluster cluster = init(Cluster.create(3))) { @@ -149,4 +149,122 @@ public class DistributedReadWritePathTest extends DistributedTestBase Assert.assertTrue(thrown.getCause().getCause().getCause().getMessage().contains("Unknown column v2 during deserialization")); } } + @Test + public void simplePagedReadsTest() throws Throwable + { + try (Cluster cluster = init(Cluster.create(3))) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + + int size = 100; + Object[][] results = new Object[size][]; + for (int i = 0; i < size; i++) + { + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, ?, ?)", + ConsistencyLevel.QUORUM, + i, i); + results[i] = new Object[] { 1, i, i}; + } + + // First, make sure that non-paged reads are able to fetch the results + assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl", ConsistencyLevel.QUORUM), + results); + + // Make sure paged read returns same results with different page sizes + for (int pageSize : new int[] { 1, 2, 3, 5, 10, 20, 50}) + { + assertRows(cluster.coordinator(1).executeWithPaging("SELECT * FROM " + KEYSPACE + ".tbl", + ConsistencyLevel.QUORUM, + pageSize), + results); + } + } + } + + @Test + public void pagingWithRepairTest() throws Throwable + { + try (Cluster cluster = init(Cluster.create(3))) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + + int size = 100; + Object[][] results = new Object[size][]; + for (int i = 0; i < size; i++) + { + // Make sure that data lands on different nodes and not coordinator + cluster.get(i % 2 == 0 ? 2 : 3).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, ?, ?)", + i, i); + + results[i] = new Object[] { 1, i, i}; + } + + // Make sure paged read returns same results with different page sizes + for (int pageSize : new int[] { 1, 2, 3, 5, 10, 20, 50}) + { + assertRows(cluster.coordinator(1).executeWithPaging("SELECT * FROM " + KEYSPACE + ".tbl", + ConsistencyLevel.ALL, + pageSize), + results); + } + + assertRows(cluster.get(1).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl"), + results); + } + } + + @Test + public void pagingTests() throws Throwable + { + try (Cluster cluster = init(Cluster.create(3)); + Cluster singleNode = init(Cluster.create(1))) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + singleNode.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + + for (int i = 0; i < 10; i++) + { + for (int j = 0; j < 10; j++) + { + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, ?, ?)", + ConsistencyLevel.QUORUM, + i, j, i + i); + singleNode.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, ?, ?)", + ConsistencyLevel.QUORUM, + i, j, i + i); + } + } + + int[] pageSizes = new int[] { 1, 2, 3, 5, 10, 20, 50}; + String[] statements = new String [] {"SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck > 5", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck >= 5", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck > 5 AND ck <= 10", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck > 5 LIMIT 3", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck >= 5 LIMIT 2", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck > 5 AND ck <= 10 LIMIT 2", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck > 5 ORDER BY ck DESC", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck >= 5 ORDER BY ck DESC", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck > 5 AND ck <= 10 ORDER BY ck DESC", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck > 5 ORDER BY ck DESC LIMIT 3", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck >= 5 ORDER BY ck DESC LIMIT 2", + "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1 AND ck > 5 AND ck <= 10 ORDER BY ck DESC LIMIT 2", + "SELECT DISTINCT pk FROM " + KEYSPACE + ".tbl LIMIT 3", + "SELECT DISTINCT pk FROM " + KEYSPACE + ".tbl WHERE pk IN (3,5,8,10)", + "SELECT DISTINCT pk FROM " + KEYSPACE + ".tbl WHERE pk IN (3,5,8,10) LIMIT 2" + }; + for (String statement : statements) + { + for (int pageSize : pageSizes) + { + assertRows(cluster.coordinator(1) + .executeWithPaging(statement, + ConsistencyLevel.QUORUM, pageSize), + singleNode.coordinator(1) + .executeWithPaging(statement, + ConsistencyLevel.QUORUM, Integer.MAX_VALUE)); + } + } + + } + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java index e2c5303302..18ca17f04d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java @@ -18,7 +18,10 @@ package org.apache.cassandra.distributed.test; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Iterator; +import java.util.List; import com.google.common.collect.Iterators; import org.junit.After; @@ -64,9 +67,35 @@ public class DistributedTestBase } } - public static String rowsNotEqualErrorMessage(Object[][] actual, Object[][] expected) + public static void assertRow(Object[] actual, Object... expected) + { + Assert.assertTrue(rowNotEqualErrorMessage(actual, expected), + Arrays.equals(actual, expected)); + } + + public static void assertRows(Iterator actual, Iterator expected) + { + while (actual.hasNext() && expected.hasNext()) + assertRow(actual.next(), expected.next()); + + Assert.assertEquals("Resultsets have different sizes", actual.hasNext(), expected.hasNext()); + } + + public static void assertRows(Iterator actual, Object[]... expected) + { + assertRows(actual, Iterators.forArray(expected)); + } + + public static String rowNotEqualErrorMessage(Object[] actual, Object[] expected) { return String.format("Expected: %s\nActual:%s\n", + Arrays.toString(expected), + Arrays.toString(actual)); + } + + public static String rowsNotEqualErrorMessage(Object[][] actual, Object[][] expected) + { + return String.format("Expected: %s\nActual: %s\n", rowsToString(expected), rowsToString(actual)); } @@ -88,6 +117,15 @@ public class DistributedTestBase return builder.toString(); } + public static Object[][] toObjectArray(Iterator iter) + { + List res = new ArrayList<>(); + while (iter.hasNext()) + res.add(iter.next()); + + return res.toArray(new Object[res.size()][]); + } + public static Object[] row(Object... expected) { return expected;