diff --git a/src/java/org/apache/cassandra/cql3/ResultSet.java b/src/java/org/apache/cassandra/cql3/ResultSet.java index 7d9d18dfc1..f920559cf9 100644 --- a/src/java/org/apache/cassandra/cql3/ResultSet.java +++ b/src/java/org/apache/cassandra/cql3/ResultSet.java @@ -33,6 +33,8 @@ import org.apache.cassandra.thrift.CqlRow; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.service.pager.PagingState; +import com.google.common.annotations.VisibleForTesting; + public class ResultSet { public static final Codec codec = new Codec(); @@ -286,6 +288,12 @@ public class ResultSet names.add(name); } + @VisibleForTesting + public PagingState getPagingState() + { + return pagingState; + } + public void setHasMorePages(PagingState pagingState) { this.pagingState = pagingState; diff --git a/src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java b/src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java index feb17e3c9b..1028b2a151 100644 --- a/src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java +++ b/src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java @@ -656,7 +656,7 @@ public final class JavaBasedUDFunction extends UDFunction { EcjTargetClassLoader() { - super(UDFunction.udfClassLoader); + super(UDFClassLoader.insecureClassLoader); } // This map is usually empty. diff --git a/src/java/org/apache/cassandra/cql3/functions/UDFunction.java b/src/java/org/apache/cassandra/cql3/functions/UDFunction.java index 6928a06684..f8463cddb5 100644 --- a/src/java/org/apache/cassandra/cql3/functions/UDFunction.java +++ b/src/java/org/apache/cassandra/cql3/functions/UDFunction.java @@ -636,7 +636,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct MigrationManager.announceNewFunction(this, true); } - private static class UDFClassLoader extends ClassLoader + static class UDFClassLoader extends ClassLoader { // insecureClassLoader is the C* class loader static final ClassLoader insecureClassLoader = Thread.currentThread().getContextClassLoader(); diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java index c449fdd888..d62a787bad 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java @@ -21,17 +21,22 @@ package org.apache.cassandra.distributed.impl; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.Iterator; import java.util.List; import java.util.UUID; import java.util.concurrent.Future; +import com.google.common.collect.Iterators; + +import org.apache.cassandra.config.CFMetaData; 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.Keyspace; import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.api.ICoordinator; import org.apache.cassandra.distributed.api.IInstance; @@ -41,6 +46,7 @@ import org.apache.cassandra.distributed.api.SimpleQueryResult; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.ClientWarn; import org.apache.cassandra.service.QueryState; +import org.apache.cassandra.service.pager.PagingState; import org.apache.cassandra.service.pager.QueryPager; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.tracing.Tracing; @@ -143,36 +149,53 @@ public class Coordinator implements ICoordinator prepared.validate(clientState); assert prepared instanceof SelectStatement : "Only SELECT statements can be executed with paging"; + long nanoTime = System.nanoTime(); SelectStatement selectStatement = (SelectStatement) prepared; - QueryPager pager = selectStatement.getQuery(QueryOptions.create(toCassandraCL(consistencyLevel), - boundBBValues, - false, - pageSize, - null, - null, - ProtocolVersion.CURRENT), - FBUtilities.nowInSeconds()) - .getPager(null, ProtocolVersion.CURRENT); + QueryState queryState = new QueryState(clientState); + QueryOptions initialOptions = QueryOptions.create(toCassandraCL(consistencyLevel), + boundBBValues, + false, + pageSize, + null, + null, + ProtocolVersion.CURRENT); - // Usually pager fetches a single page (see SelectStatement#execute). We need to iterate over all - // of the results lazily. - UntypedResultSet rs = UntypedResultSet.create(selectStatement, toCassandraCL(consistencyLevel), clientState, pager, pageSize); - Iterator it = new Iterator() { - Iterator iter = RowUtil.toObjects(rs); + + ResultMessage.Rows initialRows = selectStatement.execute(queryState, initialOptions, nanoTime); + Iterator iter = new Iterator() { + ResultMessage.Rows rows = selectStatement.execute(queryState, initialOptions, nanoTime); + Iterator iter = RowUtil.toIter(rows); public boolean hasNext() { - // We have to make sure iterator is not running on main thread. - return instance.sync(() -> iter.hasNext()).call(); + if (iter.hasNext()) + return true; + + if (rows.result.metadata.getPagingState() == null) + return false; + + QueryOptions nextOptions = QueryOptions.create(toCassandraCL(consistencyLevel), + boundBBValues, + true, + pageSize, + rows.result.metadata.getPagingState(), + null, + ProtocolVersion.CURRENT); + + rows = selectStatement.execute(queryState, nextOptions, nanoTime); + iter = Iterators.forArray(RowUtil.toObjects(initialRows.result.metadata.names, rows.result.rows)); + + return hasNext(); } public Object[] next() { - return instance.sync(() -> iter.next()).call(); + return iter.next(); } }; - return QueryResults.fromObjectArrayIterator(RowUtil.getColumnNames(rs.metadata()), it); + + return QueryResults.fromObjectArrayIterator(RowUtil.getColumnNames(initialRows.result.metadata.names), iter); }).call(); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java index 876ce297a9..f781a06fb2 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java @@ -19,11 +19,13 @@ package org.apache.cassandra.distributed.impl; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Collections; import java.util.Iterator; import java.util.List; import com.google.common.collect.Iterators; +import com.google.common.collect.Lists; import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row; @@ -61,44 +63,7 @@ public class RowUtil public static Object[][] toObjects(ResultMessage.Rows rows) { - Object[][] result = new Object[rows.result.rows.size()][]; - List specs = rows.result.metadata.names; - for (int i = 0; i < rows.result.rows.size(); i++) - { - List row = rows.result.rows.get(i); - result[i] = new Object[row.size()]; - for (int j = 0; j < row.size(); j++) - { - ByteBuffer bb = row.get(j); - - if (bb != null) - result[i][j] = specs.get(j).type.getSerializer().deserialize(bb); - } - } - 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; - }); + return toObjects(rows.result.metadata.names, rows.result.rows); } public static Iterator toObjects(ResultSet rs) @@ -114,5 +79,64 @@ public class RowUtil }); } + public static Object[][] toObjects(List specs, List> rows) + { + Object[][] result = new Object[rows.size()][]; + for (int i = 0; i < rows.size(); i++) + { + List row = rows.get(i); + result[i] = new Object[row.size()]; + for (int j = 0; j < row.size(); j++) + { + ByteBuffer bb = row.get(j); + if (bb != null) + result[i][j] = specs.get(j).type.getSerializer().deserialize(bb); + } + } + return result; + } + + public static Iterator toIter(UntypedResultSet rs) + { + return toIter(rs.metadata(), rs.iterator()); + } + + public static Iterator toIter(ResultMessage.Rows rows) + { + return toIterInternal(rows.result.metadata.names, rows.result.rows); + } + + public static Iterator toIter(List columnSpecs, Iterator rs) + { + Iterator> iter = Iterators.transform(rs, + (row) -> { + List bbs = new ArrayList<>(columnSpecs.size()); + for (int i = 0; i < columnSpecs.size(); i++) + { + ColumnSpecification columnSpec = columnSpecs.get(i); + bbs.add(row.getBytes(columnSpec.name.toString())); + } + return bbs; + }); + return toIterInternal(columnSpecs, Lists.newArrayList(iter)); + } + + private static Iterator toIterInternal(List columnSpecs, List> rs) + { + return Iterators.transform(rs.iterator(), + (row) -> { + Object[] objectRow = new Object[columnSpecs.size()]; + for (int i = 0; i < columnSpecs.size(); i++) + { + ColumnSpecification columnSpec = columnSpecs.get(i); + ByteBuffer bb = row.get(i); + + if (bb != null) + objectRow[i] = columnSpec.type.getSerializer().deserialize(bb); + + } + return objectRow; + }); + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/GroupByTest.java b/test/distributed/org/apache/cassandra/distributed/test/GroupByTest.java new file mode 100644 index 0000000000..1ffcb67e4f --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/GroupByTest.java @@ -0,0 +1,139 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.util.Iterator; + +import com.google.common.collect.Iterators; +import org.junit.Assert; +import org.junit.Test; + +import com.datastax.driver.core.Row; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.SimpleStatement; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.ICoordinator; + +import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; + +public class GroupByTest extends TestBaseImpl +{ + @Test + public void groupByWithDeletesAndSrpOnRows() throws Throwable + { + try (Cluster cluster = init(builder().withNodes(2).withConfig((cfg) -> cfg.set("enable_user_defined_functions", "true")).start())) + { + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck text, PRIMARY KEY (pk, ck))")); + initFunctions(cluster); + cluster.get(1).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, '1') USING TIMESTAMP 0")); + cluster.get(1).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, '2') USING TIMESTAMP 0")); + cluster.get(1).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck='0'")); + + cluster.get(2).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, '0') USING TIMESTAMP 0")); + cluster.get(2).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck='1'")); + cluster.get(2).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck='2'")); + + for (int limit : new int[]{ 0, 1, 10 }) + { + String limitClause = limit == 0 ? "" : "LIMIT " + limit; + String query = withKeyspace("SELECT concat(ck) FROM %s.tbl GROUP BY pk " + limitClause); + for (int i = 1; i <= 4; i++) + { + Iterator rows = cluster.coordinator(2).executeWithPaging(query, ConsistencyLevel.ALL, i); + assertRows(Iterators.toArray(rows, Object[].class)); + } + } + } + } + + @Test + public void testGroupByWithAggregatesAndPaging() throws Throwable + { + try (Cluster cluster = init(builder().withNodes(2).withConfig((cfg) -> cfg.set("enable_user_defined_functions", "true")).start())) + { + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v1 text, v2 text, v3 text, primary key (pk, ck))")); + initFunctions(cluster); + + cluster.coordinator(1).execute(withKeyspace("insert into %s.tbl (pk, ck, v1, v2, v3) values (1,1,'1','1','1')"), ConsistencyLevel.ALL); + cluster.coordinator(1).execute(withKeyspace("insert into %s.tbl (pk, ck, v1, v2, v3) values (1,2,'2','2','2')"), ConsistencyLevel.ALL); + cluster.coordinator(1).execute(withKeyspace("insert into %s.tbl (pk, ck, v1, v2, v3) values (1,3,'3','3','3')"), ConsistencyLevel.ALL); + + for (int i = 1; i <= 4; i++) + { + assertRows(cluster.coordinator(1).executeWithPaging(withKeyspace("select concat(v1), concat(v2), concat(v3) from %s.tbl where pk = 1 group by pk"), + ConsistencyLevel.ALL, i), + row("_ 1 2 3", "_ 1 2 3", "_ 1 2 3")); + + assertRows(cluster.coordinator(1).executeWithPaging(withKeyspace("select concat(v1), concat(v2), concat(v3) from %s.tbl where pk = 1 group by pk limit 1"), + ConsistencyLevel.ALL, i), + row("_ 1 2 3", "_ 1 2 3", "_ 1 2 3")); + + assertRows(cluster.coordinator(1).executeWithPaging(withKeyspace("select * from %s.tbl where pk = 1 group by pk"), + ConsistencyLevel.ALL, i), + row(1, 1, "1", "1", "1")); + } + } + } + + @Test + public void testGroupWithDeletesAndPaging() throws Throwable + { + try (Cluster cluster = init(builder().withNodes(2).withConfig(cfg -> cfg.with(Feature.GOSSIP, NETWORK, NATIVE_PROTOCOL)).start())) + { + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, PRIMARY KEY (pk, ck))")); + ICoordinator coordinator = cluster.coordinator(1); + coordinator.execute(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, 0)"), ConsistencyLevel.ALL); + coordinator.execute(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (1, 1)"), ConsistencyLevel.ALL); + + cluster.get(1).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck=0")); + cluster.get(2).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=1 AND ck=1")); + String query = withKeyspace("SELECT * FROM %s.tbl GROUP BY pk"); + Iterator rows = coordinator.executeWithPaging(query, ConsistencyLevel.ALL, 1); + assertRows(Iterators.toArray(rows, Object[].class)); + + try (com.datastax.driver.core.Cluster c = com.datastax.driver.core.Cluster.builder().addContactPoint("127.0.0.1").build(); + Session session = c.connect()) + { + SimpleStatement stmt = new SimpleStatement(withKeyspace("select * from %s.tbl where pk = 1 group by pk")); + stmt.setFetchSize(1); + Iterator rs = session.execute(stmt).iterator(); + Assert.assertFalse(rs.hasNext()); + } + } + } + + private static void initFunctions(Cluster cluster) + { + cluster.schemaChange(withKeyspace("CREATE FUNCTION %s.concat_strings_fn(a text, b text) " + + "RETURNS NULL ON NULL INPUT " + + "RETURNS text " + + "LANGUAGE java " + + "AS 'return a + \" \" + b;'")); + + cluster.schemaChange(withKeyspace("CREATE AGGREGATE %s.concat(text)" + + " SFUNC concat_strings_fn" + + " STYPE text" + + " INITCOND '_'")); + } +} \ No newline at end of file