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:
Alex Petrov 2019-02-25 16:30:36 +01:00
parent 7b462ec467 127cf8ffbf
commit 093fe7dfe9
6 changed files with 335 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
src/java/org/apache/cassandra/cql3/UntypedResultSet.java
View File

@ -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<UntypedResultSet.Row>
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<UntypedResultSet.Row>
}
}
/**
* 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<ColumnSpecification> 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<Row> iterator()
{
return new AbstractIterator<Row>()
{
private Iterator<List<ByteBuffer>> 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<ColumnSpecification> metadata()
{
return metadata;
}
}
private static class FromResultList extends UntypedResultSet
{
private final List<Map<String, ByteBuffer>> cqlRows;

4
test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
View File

@ -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<Object[]> executeWithPaging(String query, Enum<?> consistencyLevel, int pageSize, Object... boundValues);
}

66
test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
View File

@ -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<ByteBuffer> 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<Object[]> 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<ByteBuffer> 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<Object[]>() {
Iterator<Object[]> 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));
}
}

26
test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java
View File

@ -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<Object[]> toObjects(UntypedResultSet rs)
{
return toObjects(rs.metadata(), rs.iterator());
}
public static Iterator<Object[]> toObjects(List<ColumnSpecification> columnSpecs, Iterator<UntypedResultSet.Row> 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;
});
}
}

122
test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java
View File

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

40
test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java
View File

@ -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<Object[]> actual, Iterator<Object[]> 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<Object[]> 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<Object[]> iter)
{
List<Object[]> 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;