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 2021-02-15 12:17:01 +01:00
parent 9c5b626f3e e538df5e50
commit b4beebd55a
6 changed files with 252 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/java/org/apache/cassandra/cql3/ResultSet.java
View File

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

2
src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java
View File

@ -656,7 +656,7 @@ public final class JavaBasedUDFunction extends UDFunction
{
EcjTargetClassLoader()
{
super(UDFunction.udfClassLoader);
super(UDFClassLoader.insecureClassLoader);
}
// This map is usually empty.

2
src/java/org/apache/cassandra/cql3/functions/UDFunction.java
View File

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

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

@ -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<Object[]> it = new Iterator<Object[]>() {
Iterator<Object[]> iter = RowUtil.toObjects(rs);
ResultMessage.Rows initialRows = selectStatement.execute(queryState, initialOptions, nanoTime);
Iterator<Object[]> iter = new Iterator<Object[]>() {
ResultMessage.Rows rows = selectStatement.execute(queryState, initialOptions, nanoTime);
Iterator<Object[]> 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();
}

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

@ -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<ColumnSpecification> specs = rows.result.metadata.names;
for (int i = 0; i < rows.result.rows.size(); i++)
{
List<ByteBuffer> 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<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;
});
return toObjects(rows.result.metadata.names, rows.result.rows);
}
public static Iterator<Object[]> toObjects(ResultSet rs)
@ -114,5 +79,64 @@ public class RowUtil
});
}
public static Object[][] toObjects(List<ColumnSpecification> specs, List<List<ByteBuffer>> rows)
{
Object[][] result = new Object[rows.size()][];
for (int i = 0; i < rows.size(); i++)
{
List<ByteBuffer> 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<Object[]> toIter(UntypedResultSet rs)
{
return toIter(rs.metadata(), rs.iterator());
}
public static Iterator<Object[]> toIter(ResultMessage.Rows rows)
{
return toIterInternal(rows.result.metadata.names, rows.result.rows);
}
public static Iterator<Object[]> toIter(List<ColumnSpecification> columnSpecs, Iterator<UntypedResultSet.Row> rs)
{
Iterator<List<ByteBuffer>> iter = Iterators.transform(rs,
(row) -> {
List<ByteBuffer> 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<Object[]> toIterInternal(List<ColumnSpecification> columnSpecs, List<List<ByteBuffer>> 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;
});
}
}

139
test/distributed/org/apache/cassandra/distributed/test/GroupByTest.java Normal file
View File

@ -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<Object[]> 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<Object[]> 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<Row> 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 '_'"));
}
}